- {message.status === 'error' &&
- !isStreaming(message) &&
- message.content.trim() ? (
-
- )}
-
- {/* Status (e.g. "Generating code...") Apollo may stream
- after the text answer, while we wait for code. Same
- visual as the pre-text loading indicator. */}
- {isStreaming(message) && streamingStatus && (
-
-
-
+ {displayMessages.map(message => {
+ const segments = timelineSegments(message);
+ const showMessageAddButtons =
+ !isStreaming(message) && showAddButtons && !message.code;
+
+ return (
+
+
+ {message.role === 'assistant' ? (
+
+
+ {message.status === 'error' &&
+ !isStreaming(message) &&
+ message.content.trim() ? (
+
+
+
+
+ {message.content}
+
+
-
- {streamingStatus}
-
-
- )}
+ ) : segments ? (
+
+ ) : (
+
+ )}
- {!isStreaming(message) && message.code && (
-
+ {/* Transient thinking status — the registry clears it
+ when text or a persistent status segment arrives. */}
+ {isStreaming(message) && streamingStatus && (
-
{
- setExpandedYaml(prev => {
- const next = new Set(prev);
- if (next.has(message.id)) {
- next.delete(message.id);
+
+
+
+ {streamingStatus}
+
+
+ )}
+
+ {!isStreaming(message) && message.code && (
+
+
+ {
+ setExpandedYaml(prev => {
+ const next = new Set(prev);
+ if (next.has(message.id)) {
+ next.delete(message.id);
+ } else {
+ next.add(message.id);
+ }
+ return next;
+ });
+ }}
+ className="flex items-center gap-2 hover:opacity-75 transition-opacity"
+ >
+
+
+
+ {message.job_id
+ ? 'Generated Job Code'
+ : 'Generated Workflow'}
+
+
+ {
+ if (message.job_id) {
+ onApplyJobCode?.(message.code!, message.id);
} else {
- next.add(message.id);
+ onApplyWorkflow?.(message.code!, message.id);
}
- return next;
- });
- }}
- className="flex items-center gap-2 hover:opacity-75 transition-opacity"
- >
- {
+ if (message.from_global) {
+ // Per-step diff from the full workflow YAML
+ onPreviewGlobalStep?.(
+ message.code!,
+ message.id
+ );
+ } else {
+ onPreviewJobCode?.(message.code!, message.id);
+ }
+ }}
+ isApplying={!!applyingMessageId}
+ isPreviewActive={previewingMessageId === message.id}
+ isWriteDisabled={isWriteDisabled}
+ />
+
+ {expandedYaml.has(message.id) && (
+
-
- {message.job_id
- ? 'Generated Job Code'
- : 'Generated Workflow'}
+ {message.code}
+
+ )}
+
+ )}
+
+ {!isStreaming(message) &&
+ message.status === 'error' &&
+ !message.content.trim() && (
+
+
+ Failed to send message. Please try again.
-
- {
- if (message.job_id) {
- onApplyJobCode?.(message.code!, message.id);
- } else {
- onApplyWorkflow?.(message.code!, message.id);
- }
- }}
- onPreview={() => {
- if (message.from_global) {
- // Per-step diff from the full workflow YAML
- onPreviewGlobalStep?.(message.code!, message.id);
- } else {
- onPreviewJobCode?.(message.code!, message.id);
+ {onRetryMessage && (
+ onRetryMessage(message.id)}
+ className={cn(
+ 'inline-flex items-center gap-1.5 px-3 py-1.5',
+ 'text-xs font-medium rounded-md',
+ 'bg-red-100 text-red-700 hover:bg-red-200',
+ 'transition-colors duration-150',
+ 'focus:outline-none focus:ring-2 focus:ring-red-500 focus:ring-offset-1'
+ )}
+ >
+
+ )}
+
+ )}
+
+ {!isStreaming(message) &&
+ message.status === 'processing' && (
+
+ )}
+
+
+ {formatTimestamp(message.inserted_at)}
+ •
+ {
+ void (async () => {
+ const success = await doCopy(message.content);
+ if (success) {
+ setCopiedMessageId(message.id);
+ setTimeout(() => setCopiedMessageId(null), 2000);
}
- }}
- isApplying={!!applyingMessageId}
- isPreviewActive={previewingMessageId === message.id}
- isWriteDisabled={isWriteDisabled}
+ })();
+ }}
+ className={cn(
+ 'flex items-center gap-1 transition-colors duration-200',
+ copiedMessageId === message.id
+ ? 'text-green-600'
+ : 'text-gray-400 hover:text-gray-600'
+ )}
+ title={
+ copiedMessageId === message.id
+ ? 'Copied!'
+ : 'Copy message'
+ }
+ >
+
+ {copiedMessageId === message.id ? 'Copied' : 'Copy'}
+
+
+
+
+
+ ) : (
+
+
+
+
+ {message.content}
- {expandedYaml.has(message.id) && (
-
- {message.code}
-
- )}
- )}
- {!isStreaming(message) &&
- message.status === 'error' &&
- !message.content.trim() && (
+ {message.status === 'error' && (
-
- Failed to send message. Please try again.
+
+ Failed to send
{onRetryMessage && (
onRetryMessage(message.id)}
className={cn(
- 'inline-flex items-center gap-1.5 px-3 py-1.5',
+ 'inline-flex items-center gap-1 px-2 py-1',
'text-xs font-medium rounded-md',
'bg-red-100 text-red-700 hover:bg-red-200',
'transition-colors duration-150',
'focus:outline-none focus:ring-2 focus:ring-red-500 focus:ring-offset-1'
)}
>
-
)}
)}
- {!isStreaming(message) && message.status === 'processing' && (
-
- )}
-
-
- {formatTimestamp(message.inserted_at)}
- •
- {
- void (async () => {
- const success = await doCopy(message.content);
- if (success) {
- setCopiedMessageId(message.id);
- setTimeout(() => setCopiedMessageId(null), 2000);
- }
- })();
- }}
- className={cn(
- 'flex items-center gap-1 transition-colors duration-200',
- copiedMessageId === message.id
- ? 'text-green-600'
- : 'text-gray-400 hover:text-gray-600'
+
+ {formatUserName(message.user) ? (
+ <>
+ Sent by {formatUserName(message.user)} •{' '}
+ {formatTimestamp(message.inserted_at)}
+
+ ) : (
+ formatTimestamp(message.inserted_at)
)}
- title={
- copiedMessageId === message.id
- ? 'Copied!'
- : 'Copy message'
- }
- >
-
- {copiedMessageId === message.id ? 'Copied' : 'Copy'}
-
-
+
-
- ) : (
-
-
-
-
- {message.content}
-
-
-
- {message.status === 'error' && (
-
-
- Failed to send
-
- {onRetryMessage && (
- onRetryMessage(message.id)}
- className={cn(
- 'inline-flex items-center gap-1 px-2 py-1',
- 'text-xs font-medium rounded-md',
- 'bg-red-100 text-red-700 hover:bg-red-200',
- 'transition-colors duration-150',
- 'focus:outline-none focus:ring-2 focus:ring-red-500 focus:ring-offset-1'
- )}
- >
-
- )}
-
- )}
-
-
- {formatUserName(message.user) ? (
- <>
- Sent by {formatUserName(message.user)} •{' '}
- {formatTimestamp(message.inserted_at)}
-
- ) : (
- formatTimestamp(message.inserted_at)
- )}
-
-
-
- )}
+ )}
+
-
- ))}
+ );
+ })}
{isLoading && !streamingContent && (
{
);
};
+/**
+ * Get the woven text/status streaming timeline. Populated for every stream
+ * (text chunks are mirrored in); status segments only occur on global
+ * assistant streams today.
+ */
+export const useAIStreamingSegments = () => {
+ const store = useAIStore();
+ return useSyncExternalStore(
+ store.subscribe,
+ store.withSelector(state => state.streamingSegments)
+ );
+};
+
/**
* Get streaming changes (code edits or workflow YAML sent before text streams)
*/
diff --git a/assets/js/collaborative-editor/hooks/useAIWorkflowApplications.ts b/assets/js/collaborative-editor/hooks/useAIWorkflowApplications.ts
index a2a5c77dfc..01244a6c90 100644
--- a/assets/js/collaborative-editor/hooks/useAIWorkflowApplications.ts
+++ b/assets/js/collaborative-editor/hooks/useAIWorkflowApplications.ts
@@ -20,6 +20,7 @@ import type {
StreamingApplyState,
WorkflowTemplateContext,
} from '../types/ai-assistant';
+import { STREAMING_MESSAGE_ID } from '../types/ai-assistant';
import type { AIModeResult } from './useAIMode';
import { NOT_CONNECTED_ALERT, STILL_CONNECTING_ALERT } from './useWorkflow';
@@ -109,6 +110,7 @@ export function useAIWorkflowApplications({
currentSession,
currentUserId,
aiMode,
+ isGlobalSession,
isNewWorkflow,
isSessionConnected,
isSessionConnecting,
@@ -133,6 +135,12 @@ export function useAIWorkflowApplications({
} | null;
currentUserId: string | undefined;
aiMode: AIModeResult | null;
+ /**
+ * Whether the active session is a global assistant session. Mid-stream
+ * applies (STREAMING_MESSAGE_ID) have no session message to look up, so this
+ * flag decides whether they get the global page-independent treatment.
+ */
+ isGlobalSession: boolean;
isNewWorkflow: boolean;
/**
* Workflow-session socket connectivity (distinct from `connectionState`,
@@ -260,11 +268,16 @@ export function useAIWorkflowApplications({
): Promise<'applied' | 'gated' | 'failed'> => {
if (!aiMode) return 'failed';
// Global messages carry a full workflow YAML and may be applied even
- // while a job is open (job_code mode). Non-global workflow chat keeps
- // the workflow_template-only guard so its Apply stays a no-op when a
- // job is open.
- const isGlobal = !!currentSession?.messages.find(m => m.id === messageId)
- ?.from_global;
+ // while a job is open (job_code mode). Mid-stream applies
+ // (STREAMING_MESSAGE_ID) have no session message to look up, so the caller
+ // tells us via isGlobalSession whether the active session is global.
+ // A non-global workflow chat stream (e.g. the user navigated to a job
+ // mid-stream) keeps the workflow_template-only guard so its Apply
+ // stays a no-op when a job is open.
+ const isGlobal =
+ (messageId === STREAMING_MESSAGE_ID && isGlobalSession) ||
+ !!currentSession?.messages.find(m => m.id === messageId)?.from_global;
+
if (aiMode.page !== 'workflow_template' && !isGlobal) {
console.error(
'[AI Assistant] Cannot apply workflow - not in workflow mode',
@@ -300,7 +313,7 @@ export function useAIWorkflowApplications({
// Any non-streaming apply supersedes a pending streaming apply — the
// canvas will no longer hold the streamed YAML after this import.
- if (messageId !== '__streaming__') {
+ if (messageId !== STREAMING_MESSAGE_ID) {
streamingApplyActions.clear();
}
@@ -327,7 +340,7 @@ export function useAIWorkflowApplications({
await importWorkflow(workflowStateWithCreds);
applySucceeded = true;
- if (messageId === '__streaming__') {
+ if (messageId === STREAMING_MESSAGE_ID) {
// Record the applied YAML so the auto-apply effect can skip the
// duplicate import when the final new_message carries the same YAML.
// Set only after a successful import, so failed applies never
@@ -377,6 +390,7 @@ export function useAIWorkflowApplications({
[
aiMode,
currentSession,
+ isGlobalSession,
importWorkflow,
startApplyingWorkflow,
doneApplyingWorkflow,
@@ -414,7 +428,7 @@ export function useAIWorkflowApplications({
* re-apply loop); the user can re-apply via the still-enabled manual button
* and save failures recover via the shared Retry toast.
*
- * The '__streaming__' pseudo-message is deliberately NOT routed through here
+ * The STREAMING_MESSAGE_ID pseudo-message is deliberately NOT routed through here
* (it calls handleApplyWorkflow directly) so it never lands in
* appliedMessageIdsRef and can be superseded by the final new_message.
*/
@@ -472,7 +486,7 @@ export function useAIWorkflowApplications({
// If we're previewing from streaming and the real message arrives,
// just update the message ID without re-rendering the diff
- if (previewingMessageId === '__streaming__') {
+ if (previewingMessageId === STREAMING_MESSAGE_ID) {
setPreviewingMessageId(messageId);
return;
}
@@ -522,7 +536,7 @@ export function useAIWorkflowApplications({
// Same dedup guards as handlePreviewJobCode
if (previewingMessageId === messageId) return;
- if (previewingMessageId === '__streaming__') {
+ if (previewingMessageId === STREAMING_MESSAGE_ID) {
setPreviewingMessageId(messageId);
return;
}
@@ -678,7 +692,16 @@ export function useAIWorkflowApplications({
if (!currentSession) return;
const messages = currentSession.messages;
- if (page !== 'workflow_template' || !messages.length) return;
+ // Global sessions apply page-independently (their streaming applies can
+ // finish while a job is open), so they must reconcile here too —
+ // otherwise a streamingApply record would stay stuck and a failed save
+ // would never retry until the user navigated back to the canvas.
+ if (
+ (page !== 'workflow_template' && !isGlobalSession) ||
+ !messages.length
+ ) {
+ return;
+ }
if (connectionState !== 'connected') return;
// Don't auto-apply when readonly (except for new workflow creation)
if (!canApplyChanges) return;
@@ -754,6 +777,7 @@ export function useAIWorkflowApplications({
}, [
currentSession,
page,
+ isGlobalSession,
sessionId,
connectionState,
launchApply,
diff --git a/assets/js/collaborative-editor/lib/AIChannelRegistry.ts b/assets/js/collaborative-editor/lib/AIChannelRegistry.ts
index 3c99e44850..e7c96c990a 100644
--- a/assets/js/collaborative-editor/lib/AIChannelRegistry.ts
+++ b/assets/js/collaborative-editor/lib/AIChannelRegistry.ts
@@ -114,6 +114,7 @@ interface ChannelEntry {
messageStatusChanged: ChannelCallback;
streamingChunk: ChannelCallback;
streamingStatus: ChannelCallback;
+ streamingSegment: ChannelCallback;
streamingChanges: ChannelCallback;
streamingError: ChannelCallback;
};
@@ -137,6 +138,11 @@ export class AIChannelRegistry {
private streamingBuffer = '';
private streamingDrainPos = 0;
private streamingDrainTimer: ReturnType | null = null;
+ // Status markers pinned to buffer positions. A status arriving over the
+ // wire is emitted into the store's streamingSegments timeline only once
+ // every character buffered before it has drained, preserving wire order
+ // (same guarantee drainThenRun provides for new_message).
+ private pendingStatusMarkers: Array<{ pos: number; text: string }> = [];
// Delay in ms between each letter. 15ms ≈ 65 chars/sec.
private static readonly LETTER_INTERVAL_MS = 15;
// Callback to run after the buffer finishes draining (e.g., finalize message)
@@ -163,10 +169,41 @@ export class AIChannelRegistry {
this.startDraining();
}
+ /**
+ * Enqueue a status marker at the current end of the streaming buffer so it
+ * enters the store's streamingSegments timeline only after the text that
+ * preceded it on the wire has drained.
+ */
+ private bufferStreamingStatusSegment(text: string): void {
+ this.pendingStatusMarkers.push({
+ pos: this.streamingBuffer.length,
+ text,
+ });
+ this.startDraining();
+ }
+
+ /**
+ * Emit any status markers whose buffer position has been reached by the
+ * char drain (i.e. all text before them has already been appended).
+ */
+ private flushDueStatusMarkers(): void {
+ let next = this.pendingStatusMarkers.at(0);
+ while (next && next.pos <= this.streamingDrainPos) {
+ this.pendingStatusMarkers.shift();
+ this.store._appendStreamingSegment({
+ type: 'status',
+ content: next.text,
+ });
+ next = this.pendingStatusMarkers.at(0);
+ }
+ }
+
private startDraining(): void {
if (this.streamingDrainTimer !== null) return;
this.streamingDrainTimer = setInterval(() => {
+ this.flushDueStatusMarkers();
+
if (this.streamingDrainPos >= this.streamingBuffer.length) {
// Buffer fully drained — if a callback is waiting, run it now
if (this.streamingDrainCallback) {
@@ -191,6 +228,7 @@ export class AIChannelRegistry {
}
this.streamingBuffer = '';
this.streamingDrainPos = 0;
+ this.pendingStatusMarkers = [];
}
/**
@@ -199,7 +237,9 @@ export class AIChannelRegistry {
*/
private drainThenRun(callback: () => void): void {
if (this.streamingDrainPos >= this.streamingBuffer.length) {
- // Nothing left to drain
+ // Nothing left to drain — emit any statuses already due before the
+ // stream finalizes, so the timeline matches wire order to the end.
+ this.flushDueStatusMarkers();
this.stopDraining();
callback();
} else {
@@ -608,6 +648,7 @@ export class AIChannelRegistry {
);
entry.channel.off('streaming_chunk', entry.handlers.streamingChunk);
entry.channel.off('streaming_status', entry.handlers.streamingStatus);
+ entry.channel.off('streaming_segment', entry.handlers.streamingSegment);
entry.channel.off('streaming_changes', entry.handlers.streamingChanges);
entry.channel.off('streaming_error', entry.handlers.streamingError);
entry.channel.leave();
@@ -674,11 +715,33 @@ export class AIChannelRegistry {
this.bufferStreamingChunk(typedPayload.content);
};
+ // Transient "thinking" updates: scalar only. They replace each other and
+ // are cleared by any subsequent event (text chunk or status segment).
+ // They never enter the persistent segments timeline.
const streamingStatusHandler: ChannelCallback = (payload: unknown) => {
const typedPayload = payload as { text: string };
this.store.setStreamingStatus(typedPayload.text);
};
+ // Persistent completed-action statuses (same shape as a
+ // response_segments entry): supersede any active thinking status at
+ // network arrival, and enter the woven timeline through the char drain
+ // so they land after the text that preceded them on the wire.
+ const streamingSegmentHandler: ChannelCallback = (payload: unknown) => {
+ const typedPayload = payload as {
+ segment?: { type?: string; content?: string };
+ };
+ // Only status segments exist on the wire today; anything else is a
+ // contract change and is ignored until the client learns about it.
+ if (typedPayload.segment?.type !== 'status') return;
+ if (typeof typedPayload.segment.content !== 'string') return;
+
+ if (this.store.getSnapshot().streamingStatus) {
+ this.store.setStreamingStatus(null);
+ }
+ this.bufferStreamingStatusSegment(typedPayload.segment.content);
+ };
+
const streamingChangesHandler: ChannelCallback = (payload: unknown) => {
const typedPayload = payload as {
changes: Record;
@@ -698,6 +761,7 @@ export class AIChannelRegistry {
channel.on('message_status_changed', messageStatusChangedHandler);
channel.on('streaming_chunk', streamingChunkHandler);
channel.on('streaming_status', streamingStatusHandler);
+ channel.on('streaming_segment', streamingSegmentHandler);
channel.on('streaming_changes', streamingChangesHandler);
channel.on('streaming_error', streamingErrorHandler);
@@ -709,6 +773,7 @@ export class AIChannelRegistry {
messageStatusChanged: messageStatusChangedHandler,
streamingChunk: streamingChunkHandler,
streamingStatus: streamingStatusHandler,
+ streamingSegment: streamingSegmentHandler,
streamingChanges: streamingChangesHandler,
streamingError: streamingErrorHandler,
};
@@ -820,6 +885,7 @@ export class AIChannelRegistry {
);
entry.channel.off('streaming_chunk', entry.handlers.streamingChunk);
entry.channel.off('streaming_status', entry.handlers.streamingStatus);
+ entry.channel.off('streaming_segment', entry.handlers.streamingSegment);
entry.channel.off('streaming_changes', entry.handlers.streamingChanges);
entry.channel.leave();
diff --git a/assets/js/collaborative-editor/stores/createAIAssistantStore.ts b/assets/js/collaborative-editor/stores/createAIAssistantStore.ts
index b642e16e4b..709df1ab77 100644
--- a/assets/js/collaborative-editor/stores/createAIAssistantStore.ts
+++ b/assets/js/collaborative-editor/stores/createAIAssistantStore.ts
@@ -61,6 +61,7 @@ import type {
JobCodeContext,
Message,
MessageStatus,
+ ResponseSegment,
Session,
SessionListResponse,
SessionSummary,
@@ -89,6 +90,7 @@ export const createAIAssistantStore = (): AIAssistantStore => {
streamingContent: null,
streamingStatus: null,
streamingChanges: null,
+ streamingSegments: [],
streamingApply: null,
sessionList: [],
sessionListLoading: false,
@@ -162,6 +164,7 @@ export const createAIAssistantStore = (): AIAssistantStore => {
draft.streamingContent = null;
draft.streamingStatus = null;
draft.streamingChanges = null;
+ draft.streamingSegments = [];
});
notify('disconnect');
@@ -209,6 +212,7 @@ export const createAIAssistantStore = (): AIAssistantStore => {
draft.streamingContent = null;
draft.streamingStatus = null;
draft.streamingChanges = null;
+ draft.streamingSegments = [];
draft.streamingApply = null;
});
@@ -418,6 +422,7 @@ export const createAIAssistantStore = (): AIAssistantStore => {
draft.streamingContent = null;
draft.streamingStatus = null;
draft.streamingChanges = null;
+ draft.streamingSegments = [];
} else if (message.status === 'processing') {
draft.isLoading = true;
}
@@ -447,6 +452,7 @@ export const createAIAssistantStore = (): AIAssistantStore => {
if (status === 'error') {
draft.streamingContent = null;
draft.streamingStatus = null;
+ draft.streamingSegments = [];
}
if (status === 'processing') {
draft.isLoading = true;
@@ -568,10 +574,31 @@ export const createAIAssistantStore = (): AIAssistantStore => {
const _appendStreamingChunk = (content: string) => {
state = produce(state, draft => {
draft.streamingContent = (draft.streamingContent || '') + content;
+
+ // streamingContent stays the flat source of truth; the timeline is a
+ // parallel view of the same text, split by status segments.
+ const lastSegment = draft.streamingSegments.at(-1);
+ if (lastSegment && lastSegment.type === 'text') {
+ lastSegment.content += content;
+ } else {
+ draft.streamingSegments.push({ type: 'text', content });
+ }
});
notify('_appendStreamingChunk');
};
+ /**
+ * Append a status segment to the streaming timeline. Only the channel
+ * registry's char drain may call this — that is what keeps wire order.
+ * @internal
+ */
+ const _appendStreamingSegment = (segment: ResponseSegment) => {
+ state = produce(state, draft => {
+ draft.streamingSegments.push(segment);
+ });
+ notify('_appendStreamingSegment');
+ };
+
const setStreamingStatus = (text: string | null) => {
state = produce(state, draft => {
draft.streamingStatus = text;
@@ -593,6 +620,7 @@ export const createAIAssistantStore = (): AIAssistantStore => {
draft.streamingContent = null;
draft.streamingStatus = null;
draft.streamingChanges = null;
+ draft.streamingSegments = [];
});
notify('_clearStreaming');
};
@@ -720,6 +748,7 @@ export const createAIAssistantStore = (): AIAssistantStore => {
_initializeContext,
_setProcessingState,
_appendStreamingChunk,
+ _appendStreamingSegment,
setStreamingStatus,
_setStreamingChanges,
_clearStreaming,
diff --git a/assets/js/collaborative-editor/types/ai-assistant.ts b/assets/js/collaborative-editor/types/ai-assistant.ts
index 47bf189170..1a1da00294 100644
--- a/assets/js/collaborative-editor/types/ai-assistant.ts
+++ b/assets/js/collaborative-editor/types/ai-assistant.ts
@@ -41,6 +41,23 @@ export interface MessageUser {
last_name: string | null;
}
+/**
+ * Message id of the in-flight streaming placeholder. Applies triggered
+ * mid-stream carry this id instead of a persisted message id.
+ */
+export const STREAMING_MESSAGE_ID = '__streaming__' as const;
+
+/**
+ * A single entry in an assistant message's display timeline: either a chunk
+ * of answer text or a status update ("Adding step...") woven between texts.
+ * Mirrors the backend `response_segments` contract
+ * (`{"type": "text" | "status", "content": string}`).
+ */
+export interface ResponseSegment {
+ type: 'text' | 'status';
+ content: string;
+}
+
/**
* Message represents a single chat message in the AI assistant
*/
@@ -59,6 +76,11 @@ export interface Message {
* messages carry a full workflow YAML in `code` and never a `job_id`.
*/
from_global?: boolean;
+ /**
+ * Interleaved text/status timeline for global assistant replies.
+ * `null`/absent for legacy and non-global messages (render flat `content`).
+ */
+ response_segments?: ResponseSegment[] | null;
}
/**
@@ -167,6 +189,12 @@ export interface AIAssistantState {
streamingContent: string | null;
streamingStatus: string | null;
streamingChanges: Record | null;
+ /**
+ * Woven text/status timeline built up while a reply streams in.
+ * Append-only during a stream; fed exclusively by the char drain so wire
+ * order is preserved. Reset alongside the other streaming fields.
+ */
+ streamingSegments: ResponseSegment[];
streamingApply: StreamingApplyState | null;
sessionList: SessionSummary[];
@@ -224,6 +252,7 @@ export interface AIAssistantStore {
) => void;
_setProcessingState: (isProcessing: boolean) => void;
_appendStreamingChunk: (content: string) => void;
+ _appendStreamingSegment: (segment: ResponseSegment) => void;
setStreamingStatus: (text: string | null) => void;
_setStreamingChanges: (changes: Record) => void;
_clearStreaming: () => void;
diff --git a/assets/test/collaborative-editor/components/MessageList.test.tsx b/assets/test/collaborative-editor/components/MessageList.test.tsx
index 1aba7f478a..a99fa96bab 100644
--- a/assets/test/collaborative-editor/components/MessageList.test.tsx
+++ b/assets/test/collaborative-editor/components/MessageList.test.tsx
@@ -950,6 +950,182 @@ describe('MessageList', () => {
});
});
+ describe('Response Segments Timeline', () => {
+ it('renders text blocks and settled status rows in segment order', () => {
+ const messages = [
+ createMockAIMessage({
+ role: 'assistant',
+ content: 'Final answer',
+ response_segments: [
+ { type: 'text', content: 'Adding a step first.' },
+ { type: 'status', content: 'Adding step send-to-gmail...' },
+ { type: 'text', content: 'Final answer' },
+ { type: 'status', content: 'Validating workflow...' },
+ ],
+ }),
+ ];
+
+ render(,
+ appliedMessageIdsRef = { current: new Set() }
+ ) =>
+ renderHook(
+ ({ currentSession, streamingApply }: Props) =>
+ useAIWorkflowApplications({
+ sessionId: 'session-1',
+ page: 'job_code',
+ currentSession,
+ currentUserId: 'user-123',
+ aiMode: createMockAIMode('job_code'),
+ isGlobalSession: true,
+ workflowActions: saveWorkflow
+ ? { ...mockWorkflowActions, saveWorkflow }
+ : mockWorkflowActions,
+ monacoRef: createMockMonacoRef(),
+ jobs: [],
+ canApplyChanges: true,
+ connectionState: 'connected' as ConnectionState,
+ setPreviewingMessageId: mockSetPreviewingMessageId,
+ previewingMessageId: null,
+ setApplyingMessageId: mockSetApplyingMessageId,
+ isNewWorkflow: true,
+ isSessionConnected: true,
+ isSessionConnecting: false,
+ appliedMessageIdsRef,
+ streamingApply,
+ streamingApplyActions: mockStreamingApplyActions,
+ }),
+ {
+ initialProps: {
+ currentSession: { messages: [userMessage] },
+ streamingApply: null,
+ } as Props,
+ }
+ );
+
+ it('clears a matching streaming apply record while on the job page', async () => {
+ const appliedMessageIdsRef = { current: new Set() };
+ const { rerender } = renderOnJobPage(undefined, appliedMessageIdsRef);
+
+ rerender({
+ currentSession: {
+ messages: [userMessage, assistantMessage] as (typeof userMessage)[],
+ },
+ streamingApply: { yaml: 'name: Test', saveFailed: false },
+ });
+
+ await waitFor(() => {
+ expect(appliedMessageIdsRef.current.has('msg-1')).toBe(true);
+ expect(mockStreamingApplyActions.clear).toHaveBeenCalled();
+ });
+ expect(mockImportWorkflow).not.toHaveBeenCalled();
+ });
+
+ it('retries an owed save while on the job page', async () => {
+ const successfulSaveWorkflow = vi.fn(() => Promise.resolve(true));
+ const { rerender } = renderOnJobPage(successfulSaveWorkflow);
+
+ rerender({
+ currentSession: {
+ messages: [userMessage, assistantMessage] as (typeof userMessage)[],
+ },
+ streamingApply: { yaml: 'name: Test', saveFailed: true },
+ });
+
+ await waitFor(() => {
+ expect(successfulSaveWorkflow).toHaveBeenCalled();
+ expect(mockStreamingApplyActions.clear).toHaveBeenCalled();
+ });
+ expect(mockImportWorkflow).not.toHaveBeenCalled();
+ });
+});
diff --git a/assets/test/collaborative-editor/lib/AIChannelRegistry.test.ts b/assets/test/collaborative-editor/lib/AIChannelRegistry.test.ts
index d88ab0f983..6aa117c700 100644
--- a/assets/test/collaborative-editor/lib/AIChannelRegistry.test.ts
+++ b/assets/test/collaborative-editor/lib/AIChannelRegistry.test.ts
@@ -71,4 +71,106 @@ describe('AIChannelRegistry streaming', () => {
expect(store.getSnapshot().streamingStatus).toBe('Writing code...');
expect(store.getSnapshot().streamingContent).toBe('Answer');
});
+
+ it('preserves wire order of text and status segments in the streaming timeline', () => {
+ channel._test.emit('streaming_chunk', { content: 'First' });
+ channel._test.emit('streaming_segment', {
+ segment: { type: 'status', content: 'Added step' },
+ });
+ channel._test.emit('streaming_chunk', { content: 'Second' });
+
+ // The segment must not enter the timeline before the text preceding it
+ // on the wire has drained ("First" = 5 chars at 15ms each).
+ vi.advanceTimersByTime(4 * 15);
+ expect(store.getSnapshot().streamingSegments).toEqual([
+ { type: 'text', content: 'Firs' },
+ ]);
+
+ // Drain everything.
+ vi.advanceTimersByTime(1000);
+ expect(store.getSnapshot().streamingSegments).toEqual([
+ { type: 'text', content: 'First' },
+ { type: 'status', content: 'Added step' },
+ { type: 'text', content: 'Second' },
+ ]);
+ expect(store.getSnapshot().streamingContent).toBe('FirstSecond');
+ });
+
+ it('keeps thinking statuses out of the timeline and supersedes them with status segments', () => {
+ // Thinking events only touch the scalar, never the timeline.
+ channel._test.emit('streaming_status', { text: 'Reviewing workflow...' });
+ expect(store.getSnapshot().streamingStatus).toBe('Reviewing workflow...');
+ expect(store.getSnapshot().streamingSegments).toEqual([]);
+
+ // A persistent status segment clears the thinking scalar at arrival...
+ channel._test.emit('streaming_segment', {
+ segment: { type: 'status', content: 'Reviewed workflow' },
+ });
+ expect(store.getSnapshot().streamingStatus).toBeNull();
+
+ // ...and lands in the timeline via the drain.
+ vi.advanceTimersByTime(200);
+ expect(store.getSnapshot().streamingSegments).toEqual([
+ { type: 'status', content: 'Reviewed workflow' },
+ ]);
+ });
+
+ it('flushes a trailing status into the timeline before the final message lands', () => {
+ // Track timeline appends so we can assert the status entered the
+ // timeline before finalization cleared the streaming state.
+ const appended: unknown[] = [];
+ const originalAppend = store._appendStreamingSegment.bind(store);
+ store._appendStreamingSegment = segment => {
+ appended.push(segment);
+ originalAppend(segment);
+ };
+
+ channel._test.emit('streaming_chunk', { content: 'Hi' });
+ channel._test.emit('streaming_segment', {
+ segment: { type: 'status', content: 'Validated workflow' },
+ });
+ channel._test.emit('new_message', {
+ message: {
+ id: 'final-1',
+ role: 'assistant',
+ content: 'Hi',
+ status: 'success',
+ },
+ });
+
+ // Drain everything; drainThenRun must flush the due status marker
+ // before running the finalize callback.
+ vi.advanceTimersByTime(1000);
+
+ expect(appended).toContainEqual({
+ type: 'status',
+ content: 'Validated workflow',
+ });
+ expect(
+ store.getSnapshot().messages.find(m => m.id === 'final-1')
+ ).toBeTruthy();
+ });
+
+ it('does not leak pending status markers from an errored stream into the next one', () => {
+ // A status is pinned behind text that will never finish draining.
+ channel._test.emit('streaming_chunk', { content: 'Long answer text' });
+ channel._test.emit('streaming_segment', {
+ segment: { type: 'status', content: 'Stale status from dead stream' },
+ });
+
+ // The stream errors before the drain reaches the status marker.
+ vi.advanceTimersByTime(2 * 15);
+ channel._test.emit('streaming_error', { error: 'boom' });
+ expect(store.getSnapshot().streamingSegments).toEqual([]);
+
+ // A fresh stream starts and fully drains.
+ channel._test.emit('streaming_chunk', { content: 'New' });
+ vi.advanceTimersByTime(500);
+
+ // Only the new stream's text is in the timeline — the dead stream's
+ // status marker must not resurface.
+ expect(store.getSnapshot().streamingSegments).toEqual([
+ { type: 'text', content: 'New' },
+ ]);
+ });
});
diff --git a/assets/test/collaborative-editor/stores/createAIAssistantStore.test.ts b/assets/test/collaborative-editor/stores/createAIAssistantStore.test.ts
index 038dcf2278..e3829e0be8 100644
--- a/assets/test/collaborative-editor/stores/createAIAssistantStore.test.ts
+++ b/assets/test/collaborative-editor/stores/createAIAssistantStore.test.ts
@@ -364,6 +364,92 @@ describe('createAIAssistantStore', () => {
});
});
+ describe('Streaming Segments', () => {
+ it('stacks status segments and opens a new text segment after a status', () => {
+ store._appendStreamingChunk('First ');
+ store._appendStreamingChunk('answer');
+ store._appendStreamingSegment({
+ type: 'status',
+ content: 'Edited workflow structure',
+ });
+ store._appendStreamingSegment({
+ type: 'status',
+ content: 'Added step send-to-gmail',
+ });
+ store._appendStreamingChunk('Done');
+
+ // Every status segment is a completed action from Apollo's dedicated
+ // status event — they all persist, in wire order, no collapsing.
+ expect(store.getSnapshot().streamingSegments).toEqual([
+ { type: 'text', content: 'First answer' },
+ { type: 'status', content: 'Edited workflow structure' },
+ { type: 'status', content: 'Added step send-to-gmail' },
+ { type: 'text', content: 'Done' },
+ ]);
+ });
+
+ it('keeps a leading status segment when text starts', () => {
+ store._appendStreamingSegment({
+ type: 'status',
+ content: 'Edited workflow structure',
+ });
+ store._appendStreamingChunk('Answer');
+
+ expect(store.getSnapshot().streamingSegments).toEqual([
+ { type: 'status', content: 'Edited workflow structure' },
+ { type: 'text', content: 'Answer' },
+ ]);
+ });
+
+ it('survives scalar-status clearing (setStreamingStatus and streaming changes)', () => {
+ store._appendStreamingChunk('Answer');
+ store._appendStreamingSegment({ type: 'status', content: 'Working...' });
+
+ store.setStreamingStatus(null);
+ store._setStreamingChanges({ code: 'fn(s => s)' });
+
+ expect(store.getSnapshot().streamingSegments).toEqual([
+ { type: 'text', content: 'Answer' },
+ { type: 'status', content: 'Working...' },
+ ]);
+ });
+
+ it('resets when the final assistant message lands', () => {
+ store._appendStreamingChunk('Answer');
+ store._appendStreamingSegment({ type: 'status', content: 'Working...' });
+
+ store._addMessage(
+ createMockAIMessage({ role: 'assistant', status: 'success' })
+ );
+
+ expect(store.getSnapshot().streamingSegments).toEqual([]);
+ });
+
+ it('resets on message error', () => {
+ const message = createMockAIMessage({
+ role: 'assistant',
+ status: 'processing',
+ });
+ store._addMessage(message);
+ store._appendStreamingChunk('Answer');
+ store._appendStreamingSegment({ type: 'status', content: 'Working...' });
+
+ store._updateMessageStatus(message.id, 'error');
+
+ expect(store.getSnapshot().streamingSegments).toEqual([]);
+ });
+
+ it('resets on clearSession and disconnect', () => {
+ store._appendStreamingChunk('Answer');
+ store.clearSession();
+ expect(store.getSnapshot().streamingSegments).toEqual([]);
+
+ store._appendStreamingChunk('Answer again');
+ store.disconnect();
+ expect(store.getSnapshot().streamingSegments).toEqual([]);
+ });
+ });
+
describe('Streaming Apply', () => {
it('records, flags, and clears the streaming apply lifecycle', () => {
store._setStreamingApply('name: Test');
diff --git a/lib/lightning/ai_assistant/ai_assistant.ex b/lib/lightning/ai_assistant/ai_assistant.ex
index 6bb7ac7eaa..7d09c56c04 100644
--- a/lib/lightning/ai_assistant/ai_assistant.ex
+++ b/lib/lightning/ai_assistant/ai_assistant.ex
@@ -1206,6 +1206,31 @@ defmodule Lightning.AiAssistant do
acc
end
+ # Bridge event: status — a persistent, completed-action status, the same
+ # shape as a persisted `response_segments` entry, so the client renders
+ # live and reloaded status segments identically. Transient "thinking"
+ # updates arrive separately as Anthropic thinking events (see
+ # handle_stream_event/2).
+ defp handle_sse_event(session_id, %{event: "status", data: data}, acc) do
+ case Jason.decode(data) do
+ {:ok, %{"type" => "status", "content" => content} = segment}
+ when is_binary(content) ->
+ # Take only the contract fields so stray keys from Apollo never
+ # reach the client.
+ broadcast_streaming_segment(
+ session_id,
+ Map.take(segment, ["type", "content"])
+ )
+
+ _ ->
+ Logger.warning(
+ "Dropping malformed status event for session #{session_id}"
+ )
+ end
+
+ acc
+ end
+
# Bridge event: log — skip Python stdout
defp handle_sse_event(_session_id, %{event: "log"}, acc), do: acc
@@ -1274,6 +1299,14 @@ defmodule Lightning.AiAssistant do
)
end
+ defp broadcast_streaming_segment(session_id, segment) do
+ Lightning.broadcast(
+ "ai_session:#{session_id}",
+ {:ai_assistant, :streaming_segment,
+ %{segment: segment, session_id: session_id}}
+ )
+ end
+
defp broadcast_streaming_error(session_id, error) do
Lightning.broadcast(
"ai_session:#{session_id}",
@@ -1297,7 +1330,9 @@ defmodule Lightning.AiAssistant do
case error_response do
{:ok, %Tesla.Env{status: status, body: body}}
when status not in @success_status_range ->
- error_message = body["message"]
+ error_message =
+ error_message_from_body(body) ||
+ "AI server returned an error (HTTP #{status})."
Logger.error(
"AI query failed for session #{session.id}: #{error_message}"
@@ -1322,6 +1357,13 @@ defmodule Lightning.AiAssistant do
end
end
+ # Streaming requests carry a lazy Stream (a fun or %Stream{} struct) as the
+ # body, so error responses can't be indexed like decoded JSON maps.
+ defp error_message_from_body(body) when is_map(body) and not is_struct(body),
+ do: body["message"]
+
+ defp error_message_from_body(_body), do: nil
+
defp build_job_message(body) do
message = body["history"] |> Enum.reverse() |> hd()
message_attrs = Map.take(message, ["role", "content"])
@@ -1353,16 +1395,59 @@ defmodule Lightning.AiAssistant do
defp build_global_message(body) do
code = extract_global_workflow_yaml(body["attachments"])
- message_attrs = %{
- role: :assistant,
- content: body["response"],
- meta: %{"from_global" => true}
- }
+ message_attrs =
+ %{
+ role: :assistant,
+ content: body["response"],
+ meta: %{"from_global" => true}
+ }
+ |> put_response_segments(
+ normalize_response_segments(body["response_segments"])
+ )
opts = [usage: body["usage"] || %{}, meta: body["meta"], code: code]
{message_attrs, opts}
end
+ # Flat replies omit the key entirely: casting nil into embeds_many is an
+ # error, and an absent key leaves the column NULL for legacy parity.
+ defp put_response_segments(attrs, nil), do: attrs
+
+ defp put_response_segments(attrs, segments),
+ do: Map.put(attrs, :response_segments, segments)
+
+ # `response_segments` is the display timeline of the streamed reply (text and status
+ # segments in stream order); `response` stays the flat answer that history
+ # is rebuilt from. Apollo is an external boundary, so invalid or oversized
+ # segments are dropped (and counted in the logs) rather than failing the
+ # save: absent or all-invalid segments mean a flat legacy message. The
+ # segment contract itself lives in `ChatMessage.Segment`.
+ defp normalize_response_segments(segments) when is_list(segments) do
+ {valid, dropped} =
+ Enum.split_with(segments, fn segment ->
+ is_map(segment) and
+ ChatMessage.Segment.changeset(%ChatMessage.Segment{}, segment).valid?
+ end)
+
+ max_segments = ChatMessage.max_response_segments()
+ {kept, truncated} = Enum.split(valid, max_segments)
+
+ if dropped != [] or truncated != [] do
+ Logger.warning(
+ "Discarding response segments from Apollo payload: " <>
+ "#{length(dropped)} invalid, #{length(truncated)} over the " <>
+ "#{max_segments}-segment cap"
+ )
+ end
+
+ case kept do
+ [] -> nil
+ kept -> Enum.map(kept, &Map.take(&1, ["type", "content"]))
+ end
+ end
+
+ defp normalize_response_segments(_segments), do: nil
+
# Global chat always returns a full workflow YAML (job bodies embedded).
# The frontend handles per-step diffing and full-workflow apply.
defp extract_global_workflow_yaml(attachments) when is_list(attachments) do
diff --git a/lib/lightning/ai_assistant/chat_message.ex b/lib/lightning/ai_assistant/chat_message.ex
index 2c3753b684..2552c7ab8d 100644
--- a/lib/lightning/ai_assistant/chat_message.ex
+++ b/lib/lightning/ai_assistant/chat_message.ex
@@ -10,6 +10,9 @@ defmodule Lightning.AiAssistant.ChatMessage do
* `content` - The text content of the message (required, 1-10,000 characters)
* `code` - Optional code associated with the message (e.g., generated workflows)
+ * `response_segments` - Optional display timeline of text and status segments
+ for assistant messages (global chat); `[]` for flat messages (the column
+ is NULL, which `embeds_many` loads as an empty list)
* `role` - Who sent the message: `:user` or `:assistant`
* `status` - Processing status: `:pending`, `:success`, `:error`, or `:cancelled`
* `is_deleted` - Soft deletion flag (defaults to false)
@@ -25,6 +28,45 @@ defmodule Lightning.AiAssistant.ChatMessage do
alias Lightning.Workflows.Job
+ defmodule Segment do
+ @moduledoc """
+ One entry in an assistant reply's display timeline: a chunk of answer
+ text, or a completed-action status ("Added step send-to-gmail...") woven
+ between texts in stream order. Display-only — the flat `content` field
+ stays the canonical answer.
+ """
+
+ use Ecto.Schema
+ import Ecto.Changeset
+
+ @max_content_length 10_000
+
+ @type t() :: %__MODULE__{type: :text | :status, content: String.t()}
+
+ @derive {Jason.Encoder, only: [:type, :content]}
+ @primary_key false
+ embedded_schema do
+ field :type, Ecto.Enum, values: [:text, :status]
+ field :content, :string
+ end
+
+ @doc false
+ def changeset(segment, attrs) do
+ segment
+ |> cast(attrs, [:type, :content])
+ |> validate_required([:type, :content])
+ |> validate_length(:content, max: @max_content_length)
+ end
+
+ @doc "Maximum length of a single segment's content (matches `content`'s cap)."
+ def max_content_length, do: @max_content_length
+ end
+
+ # A reply's segment count is naturally bounded by the model's output size;
+ # this cap only guards against a runaway or buggy Apollo response bloating
+ # rows that get re-serialized on every channel join.
+ @max_response_segments 200
+
@type role() :: :user | :assistant
@type status() :: :pending | :processing | :success | :error | :cancelled
@@ -32,6 +74,7 @@ defmodule Lightning.AiAssistant.ChatMessage do
id: Ecto.UUID.t(),
content: String.t() | nil,
code: String.t() | nil,
+ response_segments: [Segment.t()],
role: role(),
status: status(),
job_id: Ecto.UUID.t() | nil,
@@ -51,6 +94,8 @@ defmodule Lightning.AiAssistant.ChatMessage do
field :code, :string
field :role, Ecto.Enum, values: [:user, :assistant]
+ embeds_many :response_segments, Segment, on_replace: :delete
+
field :status, Ecto.Enum,
values: [:pending, :processing, :success, :error, :cancelled]
@@ -79,6 +124,9 @@ defmodule Lightning.AiAssistant.ChatMessage do
* `content` and `role` are required
* `content` must be between 1 and 10,000 characters
+ * `response_segments`, when present, must be `#{@max_response_segments}` or
+ fewer segments, each with a `type` of `text` or `status` and a `content`
+ string of at most `#{Segment.max_content_length()}` characters
* User messages (role: `:user`) require an associated user
* Status defaults based on role: `:pending` for users, `:success` for assistant
* If status is explicitly provided, it takes precedence over role-based defaults
@@ -97,6 +145,8 @@ defmodule Lightning.AiAssistant.ChatMessage do
:processing_started_at,
:processing_completed_at
])
+ |> cast_embed(:response_segments)
+ |> validate_length(:response_segments, max: @max_response_segments)
|> validate_required([:content, :role])
|> validate_length(:content, min: 1, max: 10_000)
|> maybe_put_user_assoc(attrs[:user] || attrs["user"])
@@ -105,6 +155,9 @@ defmodule Lightning.AiAssistant.ChatMessage do
|> set_default_status_by_role()
end
+ @doc "Maximum number of segments accepted on a message."
+ def max_response_segments, do: @max_response_segments
+
@doc """
Creates a changeset for updating message status.
diff --git a/lib/lightning/ai_assistant/message_processor.ex b/lib/lightning/ai_assistant/message_processor.ex
index 6747022db1..3e7fbcfb20 100644
--- a/lib/lightning/ai_assistant/message_processor.ex
+++ b/lib/lightning/ai_assistant/message_processor.ex
@@ -129,7 +129,9 @@ defmodule Lightning.AiAssistant.MessageProcessor do
@spec dispatch_message_processing(
AiAssistant.ChatSession.t(),
ChatMessage.t()
- ) :: {:ok, AiAssistant.ChatSession.t()} | {:error, String.t()}
+ ) ::
+ {:ok, AiAssistant.ChatSession.t()}
+ | {:error, String.t() | Ecto.Changeset.t()}
defp dispatch_message_processing(session, message) do
if global_chat?(session) do
process_global_message(session, message)
@@ -148,7 +150,8 @@ defmodule Lightning.AiAssistant.MessageProcessor do
end
@spec process_global_message(AiAssistant.ChatSession.t(), ChatMessage.t()) ::
- {:ok, AiAssistant.ChatSession.t()} | {:error, String.t()}
+ {:ok, AiAssistant.ChatSession.t()}
+ | {:error, String.t() | Ecto.Changeset.t()}
defp process_global_message(session, message) do
workflow_yaml = message.code
page = get_in(session.meta, ["message_options", "page"])
@@ -179,7 +182,8 @@ defmodule Lightning.AiAssistant.MessageProcessor do
@spec handle_processing_result(
ChatMessage.t(),
- {:ok, AiAssistant.ChatSession.t()} | {:error, String.t()}
+ {:ok, AiAssistant.ChatSession.t()}
+ | {:error, String.t() | Ecto.Changeset.t()}
) :: {:ok, AiAssistant.ChatSession.t()} | {:error, String.t()}
defp handle_processing_result(message, result) do
case result do
@@ -189,6 +193,17 @@ defmodule Lightning.AiAssistant.MessageProcessor do
{:ok, updated_session}
+ {:error, %Ecto.Changeset{} = changeset} ->
+ {:ok, _updated_session, _updated_message} =
+ update_message_status(message, :error)
+
+ Logger.error(
+ "[MessageProcessor] Failed to save assistant response for message " <>
+ "#{message.id}: invalid changeset: #{inspect(changeset.errors)}"
+ )
+
+ {:error, "Failed to save assistant response"}
+
{:error, error_message} ->
{:ok, _updated_session, _updated_message} =
update_message_status(message, :error)
@@ -199,7 +214,8 @@ defmodule Lightning.AiAssistant.MessageProcessor do
@doc false
@spec process_job_message(AiAssistant.ChatSession.t(), ChatMessage.t()) ::
- {:ok, AiAssistant.ChatSession.t()} | {:error, String.t()}
+ {:ok, AiAssistant.ChatSession.t()}
+ | {:error, String.t() | Ecto.Changeset.t()}
defp process_job_message(session, message) do
enriched_session = AiAssistant.enrich_session_with_job_context(session)
@@ -256,7 +272,8 @@ defmodule Lightning.AiAssistant.MessageProcessor do
@doc false
@spec process_workflow_message(AiAssistant.ChatSession.t(), ChatMessage.t()) ::
- {:ok, AiAssistant.ChatSession.t()} | {:error, String.t()}
+ {:ok, AiAssistant.ChatSession.t()}
+ | {:error, String.t() | Ecto.Changeset.t()}
defp process_workflow_message(session, message) do
code = message.code || workflow_code_from_session(session)
diff --git a/lib/lightning_web/channels/ai_assistant_channel.ex b/lib/lightning_web/channels/ai_assistant_channel.ex
index e098ad37e9..d6e667a2cc 100644
--- a/lib/lightning_web/channels/ai_assistant_channel.ex
+++ b/lib/lightning_web/channels/ai_assistant_channel.ex
@@ -351,6 +351,18 @@ defmodule LightningWeb.AiAssistantChannel do
{:noreply, socket}
end
+ # Streaming: a persistent completed-action status segment (same shape as a
+ # response_segments entry)
+ @impl true
+ def handle_info(
+ {:ai_assistant, :streaming_segment,
+ %{segment: segment, session_id: session_id}},
+ %{assigns: %{session_id: session_id}} = socket
+ ) do
+ broadcast(socket, "streaming_segment", %{segment: segment})
+ {:noreply, socket}
+ end
+
# Streaming: error during stream
@impl true
def handle_info(
@@ -853,6 +865,7 @@ defmodule LightningWeb.AiAssistantChannel do
id: message.id,
content: message.content,
code: message.code,
+ response_segments: message.response_segments,
role: to_string(message.role),
status: to_string(message.status),
inserted_at: message.inserted_at,
diff --git a/priv/repo/migrations/20260802120000_add_response_segments_to_ai_chat_messages.exs b/priv/repo/migrations/20260802120000_add_response_segments_to_ai_chat_messages.exs
new file mode 100644
index 0000000000..a4944e3863
--- /dev/null
+++ b/priv/repo/migrations/20260802120000_add_response_segments_to_ai_chat_messages.exs
@@ -0,0 +1,11 @@
+defmodule Lightning.Repo.Migrations.AddResponseSegmentsToAiChatMessages do
+ use Ecto.Migration
+
+ def change do
+ alter table(:ai_chat_messages) do
+ # jsonb holding an array of {type, content} segment objects
+ # (embeds_many on the schema); nil for flat legacy messages
+ add :response_segments, :map, null: true
+ end
+ end
+end
diff --git a/test/lightning/ai_assistant/chat_message_test.exs b/test/lightning/ai_assistant/chat_message_test.exs
index a705cc0709..a6326fd777 100644
--- a/test/lightning/ai_assistant/chat_message_test.exs
+++ b/test/lightning/ai_assistant/chat_message_test.exs
@@ -152,6 +152,97 @@ defmodule Lightning.AiAssistant.ChatMessageTest do
assert changeset.valid?
end
+ test "accepts a valid segments timeline and casts to Segment embeds" do
+ segments = [
+ %{"type" => "text", "content" => "Adding a step..."},
+ %{"type" => "status", "content" => "Validating workflow..."},
+ %{"type" => "text", "content" => "Done!"}
+ ]
+
+ changeset =
+ ChatMessage.changeset(%ChatMessage{}, %{
+ content: "Adding a step...\n\nDone!",
+ role: :assistant,
+ response_segments: segments
+ })
+
+ assert changeset.valid?
+
+ assert [
+ %ChatMessage.Segment{type: :text, content: "Adding a step..."},
+ %ChatMessage.Segment{
+ type: :status,
+ content: "Validating workflow..."
+ },
+ %ChatMessage.Segment{type: :text, content: "Done!"}
+ ] = Ecto.Changeset.apply_changes(changeset).response_segments
+ end
+
+ test "leaves segments empty when the key is absent (flat message)" do
+ changeset =
+ ChatMessage.changeset(%ChatMessage{}, %{
+ content: "Flat response",
+ role: :assistant
+ })
+
+ assert changeset.valid?
+
+ assert Ecto.Changeset.apply_changes(changeset).response_segments == []
+ end
+
+ test "rejects segments with unknown types or non-binary content" do
+ invalid_segments = [
+ [%{"type" => "thinking", "content" => "hmm"}],
+ [%{"type" => "text", "content" => 123}],
+ [%{"content" => "no type"}],
+ [%{"type" => "text", "content" => "ok"}, %{"type" => "status"}]
+ ]
+
+ for segments <- invalid_segments do
+ changeset =
+ ChatMessage.changeset(%ChatMessage{}, %{
+ content: "Test message",
+ role: :assistant,
+ response_segments: segments
+ })
+
+ refute changeset.valid?, "expected #{inspect(segments)} to be invalid"
+ end
+ end
+
+ test "rejects segments over the count cap and content over the length cap" do
+ too_many =
+ for i <- 1..(ChatMessage.max_response_segments() + 1) do
+ %{"type" => "text", "content" => "segment #{i}"}
+ end
+
+ changeset =
+ ChatMessage.changeset(%ChatMessage{}, %{
+ content: "Test message",
+ role: :assistant,
+ response_segments: too_many
+ })
+
+ refute changeset.valid?
+
+ oversized = [
+ %{
+ "type" => "text",
+ "content" =>
+ String.duplicate("x", ChatMessage.Segment.max_content_length() + 1)
+ }
+ ]
+
+ changeset =
+ ChatMessage.changeset(%ChatMessage{}, %{
+ content: "Test message",
+ role: :assistant,
+ response_segments: oversized
+ })
+
+ refute changeset.valid?
+ end
+
test "sets pending status by default for user messages" do
user = insert(:user)
diff --git a/test/lightning/ai_assistant/message_processor_test.exs b/test/lightning/ai_assistant/message_processor_test.exs
index e208f16412..546aad72ec 100644
--- a/test/lightning/ai_assistant/message_processor_test.exs
+++ b/test/lightning/ai_assistant/message_processor_test.exs
@@ -7,6 +7,7 @@ defmodule Lightning.AiAssistant.MessageProcessorTest do
import Lightning.Factories
alias Lightning.AiAssistant
+ alias Lightning.AiAssistant.ChatMessage
alias Lightning.AiAssistant.MessageProcessor
# Note: Integration tests for I/O data scrubbing are tested at lower levels:
@@ -329,6 +330,90 @@ defmodule Lightning.AiAssistant.MessageProcessorTest do
assistant_msg = Enum.find(reloaded.messages, &(&1.role == :assistant))
assert assistant_msg != nil
assert assistant_msg.content == "Global response"
+ # Flat-string responses have no timeline
+ assert assistant_msg.response_segments == []
+ end
+
+ test "persists the segments timeline alongside the flat response",
+ %{user: user, project: project} do
+ workflow = insert(:workflow, project: project)
+
+ global_meta = %{
+ "message_options" => %{
+ "use_global_assistant" => true,
+ "page" => "/projects/p1/workflows/w1"
+ }
+ }
+
+ session =
+ insert(:chat_session,
+ user: user,
+ session_type: "workflow_template",
+ project: project,
+ workflow: workflow,
+ job_id: nil,
+ meta: global_meta
+ )
+
+ {:ok, updated_session} =
+ AiAssistant.save_message(
+ session,
+ %{
+ role: :user,
+ content: "add a step to my workflow",
+ user: user,
+ code: "workflow:\n name: test"
+ },
+ meta: global_meta
+ )
+
+ user_message = Enum.find(updated_session.messages, &(&1.role == :user))
+
+ Mox.stub(
+ Lightning.Tesla.Mock,
+ :call,
+ Lightning.AiAssistantHelpers.streaming_or_sync_response(%{
+ "response" => "Here's the final result.",
+ "response_segments" => [
+ %{"type" => "text", "content" => "I'm going to add a step."},
+ %{"type" => "status", "content" => "Adding step send-to-gmail..."},
+ %{"type" => "text", "content" => "Here's the final result."},
+ %{"type" => "unknown", "content" => "dropped during normalization"}
+ ],
+ "attachments" => [
+ %{"type" => "workflow_yaml", "content" => "workflow:\n name: new"}
+ ],
+ "usage" => %{}
+ })
+ )
+
+ assert :ok =
+ perform_job(MessageProcessor, %{"message_id" => user_message.id})
+
+ reloaded = AiAssistant.get_session!(session.id)
+ assistant_msg = Enum.find(reloaded.messages, &(&1.role == :assistant))
+
+ # Content is the flat response verbatim; segments keep the full timeline
+ # (minus unrecognised entries); code comes from attachments.
+ assert %{
+ content: "Here's the final result.",
+ response_segments: [
+ %ChatMessage.Segment{
+ type: :text,
+ content: "I'm going to add a step."
+ },
+ %ChatMessage.Segment{
+ type: :status,
+ content: "Adding step send-to-gmail..."
+ },
+ %ChatMessage.Segment{
+ type: :text,
+ content: "Here's the final result."
+ }
+ ],
+ code: "workflow:\n name: new",
+ meta: %{"from_global" => true}
+ } = assistant_msg
end
test "dispatches to workflow chat when use_global_assistant is not set", %{
@@ -385,6 +470,292 @@ defmodule Lightning.AiAssistant.MessageProcessorTest do
assert assistant_msg != nil
assert assistant_msg.content == "Workflow response"
end
+
+ test "clamps segments over the cap instead of failing the save",
+ %{user: user, project: project} do
+ workflow = insert(:workflow, project: project)
+
+ global_meta = %{
+ "message_options" => %{
+ "use_global_assistant" => true,
+ "page" => "/projects/p1/workflows/w1"
+ }
+ }
+
+ session =
+ insert(:chat_session,
+ user: user,
+ session_type: "workflow_template",
+ project: project,
+ workflow: workflow,
+ job_id: nil,
+ meta: global_meta
+ )
+
+ {:ok, updated_session} =
+ AiAssistant.save_message(
+ session,
+ %{role: :user, content: "add a step", user: user},
+ meta: global_meta
+ )
+
+ user_message = Enum.find(updated_session.messages, &(&1.role == :user))
+
+ max = ChatMessage.max_response_segments()
+
+ segments =
+ for i <- 1..(max + 1) do
+ %{"type" => "text", "content" => "segment #{i}"}
+ end
+
+ Mox.stub(
+ Lightning.Tesla.Mock,
+ :call,
+ Lightning.AiAssistantHelpers.streaming_or_sync_response(%{
+ "response" => "Done!",
+ "response_segments" => segments,
+ "usage" => %{}
+ })
+ )
+
+ log =
+ ExUnit.CaptureLog.capture_log(fn ->
+ assert :ok =
+ perform_job(MessageProcessor, %{
+ "message_id" => user_message.id
+ })
+ end)
+
+ reloaded = AiAssistant.get_session!(session.id)
+ assistant_msg = Enum.find(reloaded.messages, &(&1.role == :assistant))
+
+ assert length(assistant_msg.response_segments) == max
+ assert log =~ "over the #{max}-segment cap"
+ end
+
+ test "persists a flat message when every segment is invalid",
+ %{user: user, project: project} do
+ workflow = insert(:workflow, project: project)
+
+ global_meta = %{
+ "message_options" => %{
+ "use_global_assistant" => true,
+ "page" => "/projects/p1/workflows/w1"
+ }
+ }
+
+ session =
+ insert(:chat_session,
+ user: user,
+ session_type: "workflow_template",
+ project: project,
+ workflow: workflow,
+ job_id: nil,
+ meta: global_meta
+ )
+
+ {:ok, updated_session} =
+ AiAssistant.save_message(
+ session,
+ %{role: :user, content: "add a step", user: user},
+ meta: global_meta
+ )
+
+ user_message = Enum.find(updated_session.messages, &(&1.role == :user))
+
+ Mox.stub(
+ Lightning.Tesla.Mock,
+ :call,
+ Lightning.AiAssistantHelpers.streaming_or_sync_response(%{
+ "response" => "Done!",
+ "response_segments" => [
+ %{"type" => "thinking", "content" => "unknown type"},
+ %{"content" => "no type"}
+ ],
+ "usage" => %{}
+ })
+ )
+
+ log =
+ ExUnit.CaptureLog.capture_log(fn ->
+ assert :ok =
+ perform_job(MessageProcessor, %{
+ "message_id" => user_message.id
+ })
+ end)
+
+ reloaded = AiAssistant.get_session!(session.id)
+ assistant_msg = Enum.find(reloaded.messages, &(&1.role == :assistant))
+
+ # All-invalid segments degrade to a flat message (the key is omitted,
+ # so the column stays NULL and loads as []), never a save failure.
+ assert assistant_msg.content == "Done!"
+ assert assistant_msg.response_segments == []
+ assert log =~ "2 invalid"
+ end
+
+ test "broadcasts valid status events as streaming segments and drops malformed ones",
+ %{user: user, project: project} do
+ workflow = insert(:workflow, project: project)
+
+ global_meta = %{
+ "message_options" => %{
+ "use_global_assistant" => true,
+ "page" => "/projects/p1/workflows/w1"
+ }
+ }
+
+ session =
+ insert(:chat_session,
+ user: user,
+ session_type: "workflow_template",
+ project: project,
+ workflow: workflow,
+ job_id: nil,
+ meta: global_meta
+ )
+
+ {:ok, updated_session} =
+ AiAssistant.save_message(
+ session,
+ %{role: :user, content: "add a step", user: user},
+ meta: global_meta
+ )
+
+ user_message = Enum.find(updated_session.messages, &(&1.role == :user))
+
+ segment = %{
+ "type" => "status",
+ "content" => "Adding step send-to-gmail..."
+ }
+
+ sse_body =
+ "event: status\ndata: #{Jason.encode!(segment)}\n\n" <>
+ "event: status\ndata: {not json\n\n" <>
+ "event: status\ndata: #{Jason.encode!(%{"type" => "status", "content" => 123})}\n\n" <>
+ "event: complete\ndata: #{Jason.encode!(%{"response" => "Done!", "usage" => %{}})}\n\n"
+
+ Mox.stub(Lightning.Tesla.Mock, :call, fn %{url: url}, _opts ->
+ assert url =~ "/stream"
+
+ {:ok,
+ %Tesla.Env{
+ status: 200,
+ headers: [{"content-type", "text/event-stream"}],
+ body: sse_body
+ }}
+ end)
+
+ Lightning.subscribe("ai_session:#{session.id}")
+ session_id = session.id
+
+ log =
+ ExUnit.CaptureLog.capture_log(fn ->
+ assert :ok =
+ perform_job(MessageProcessor, %{
+ "message_id" => user_message.id
+ })
+ end)
+
+ # The well-formed status event reaches the session topic verbatim...
+ assert_receive {:ai_assistant, :streaming_segment,
+ %{segment: ^segment, session_id: ^session_id}}
+
+ # ...while the two malformed ones are dropped with a warning, never
+ # broadcast.
+ refute_receive {:ai_assistant, :streaming_segment, _}, 100
+ assert log =~ "Dropping malformed status event"
+ end
+
+ test "marks the message as error when a streaming request returns non-2xx",
+ %{user: user, project: project} do
+ workflow = insert(:workflow, project: project)
+
+ session =
+ insert(:chat_session,
+ user: user,
+ session_type: "workflow_template",
+ project: project,
+ workflow: workflow,
+ job_id: nil,
+ meta: %{}
+ )
+
+ {:ok, updated_session} =
+ AiAssistant.save_message(
+ session,
+ %{role: :user, content: "generate a workflow", user: user},
+ []
+ )
+
+ user_message = Enum.find(updated_session.messages, &(&1.role == :user))
+
+ # Streaming requests carry a lazy Stream body even on error responses,
+ # so the error handler must not index it like a decoded JSON map.
+ Mox.expect(Lightning.Tesla.Mock, :call, fn %{url: url}, _opts ->
+ assert url =~ "/services/workflow_chat/stream"
+
+ {:ok,
+ %Tesla.Env{
+ status: 500,
+ body: Stream.map(["upstream exploded"], & &1)
+ }}
+ end)
+
+ assert :ok =
+ perform_job(MessageProcessor, %{
+ "message_id" => user_message.id
+ })
+
+ reloaded = AiAssistant.get_session!(session.id)
+ assert Enum.find(reloaded.messages, &(&1.role == :user)).status == :error
+ refute Enum.find(reloaded.messages, &(&1.role == :assistant))
+ end
+
+ test "marks the message as error when the assistant reply fails to save",
+ %{user: user, project: project} do
+ workflow = insert(:workflow, project: project)
+
+ session =
+ insert(:chat_session,
+ user: user,
+ session_type: "workflow_template",
+ project: project,
+ workflow: workflow,
+ job_id: nil,
+ meta: %{}
+ )
+
+ {:ok, updated_session} =
+ AiAssistant.save_message(
+ session,
+ %{role: :user, content: "generate a workflow", user: user},
+ []
+ )
+
+ user_message = Enum.find(updated_session.messages, &(&1.role == :user))
+
+ # A response over the 10k content cap fails the ChatMessage changeset,
+ # exercising the {:error, %Ecto.Changeset{}} branch in
+ # handle_processing_result/2.
+ Mox.stub(
+ Lightning.Tesla.Mock,
+ :call,
+ Lightning.AiAssistantHelpers.streaming_or_sync_response(%{
+ "response" => String.duplicate("x", 10_001),
+ "usage" => %{}
+ })
+ )
+
+ assert :ok =
+ perform_job(MessageProcessor, %{
+ "message_id" => user_message.id
+ })
+
+ reloaded = AiAssistant.get_session!(session.id)
+ assert Enum.find(reloaded.messages, &(&1.role == :user)).status == :error
+ refute Enum.find(reloaded.messages, &(&1.role == :assistant))
+ end
end
describe "fetch_and_scrub_io_data/1 via process_job_message/2" do
diff --git a/test/lightning_web/channels/ai_assistant_channel_test.exs b/test/lightning_web/channels/ai_assistant_channel_test.exs
index 246c132fe5..6e787bbfc9 100644
--- a/test/lightning_web/channels/ai_assistant_channel_test.exs
+++ b/test/lightning_web/channels/ai_assistant_channel_test.exs
@@ -253,6 +253,65 @@ defmodule LightningWeb.AiAssistantChannelTest do
%{from_global: false}
] = messages
end
+
+ test "serializes segments timeline when present", %{
+ socket: socket,
+ job: job,
+ user: user
+ } do
+ segments = [
+ %Lightning.AiAssistant.ChatMessage.Segment{
+ type: :text,
+ content: "Adding a step..."
+ },
+ %Lightning.AiAssistant.ChatMessage.Segment{
+ type: :status,
+ content: "Validating workflow..."
+ },
+ %Lightning.AiAssistant.ChatMessage.Segment{
+ type: :text,
+ content: "Done!"
+ }
+ ]
+
+ session =
+ insert(:chat_session,
+ job: job,
+ user: user,
+ session_type: "job_code",
+ messages: [
+ %{
+ role: :assistant,
+ content: "Adding a step...\n\nDone!",
+ status: :success,
+ meta: %{"from_global" => true},
+ response_segments: segments
+ },
+ %{
+ role: :assistant,
+ content: "Flat response",
+ status: :success,
+ inserted_at: DateTime.utc_now() |> DateTime.add(1)
+ }
+ ]
+ )
+
+ assert {:ok, %{messages: messages}, _socket} =
+ subscribe_and_join(
+ socket,
+ AiAssistantChannel,
+ "ai_assistant:job_code:#{session.id}",
+ %{}
+ )
+
+ assert [
+ %{
+ response_segments: ^segments,
+ content: "Adding a step...\n\nDone!"
+ },
+ %{response_segments: [], content: "Flat response"}
+ ] = messages
+ end
end
describe "workflow_template sessions" do
@@ -3604,6 +3663,21 @@ defmodule LightningWeb.AiAssistantChannelTest do
assert_broadcast "streaming_changes", %{changes: ^changes}
end
+ test "forwards streaming_segment to channel", %{
+ socket: socket,
+ session_id: session_id
+ } do
+ segment = %{"type" => "status", "content" => "Edited workflow structure"}
+
+ send(
+ socket.channel_pid,
+ {:ai_assistant, :streaming_segment,
+ %{segment: segment, session_id: session_id}}
+ )
+
+ assert_broadcast "streaming_segment", %{segment: ^segment}
+ end
+
test "forwards streaming_error to channel", %{
socket: socket,
session_id: session_id
diff --git a/test/support/ai_assistant_helpers.ex b/test/support/ai_assistant_helpers.ex
index 202e1a76eb..fa2183221b 100644
--- a/test/support/ai_assistant_helpers.ex
+++ b/test/support/ai_assistant_helpers.ex
@@ -29,10 +29,23 @@ defmodule Lightning.AiAssistantHelpers do
endpoints. For streaming URLs (containing "/stream"), returns an SSE
`event: complete` response. For other URLs, returns a regular JSON body.
+ The body is encoded verbatim, so it can include the global chat planner
+ fields (`"response_segments"` timeline alongside the flat `"response"`).
+
## Examples
stub_ai_response(%{"history" => [%{"role" => "assistant", "content" => "Hi"}]})
+ streaming_or_sync_response(%{
+ "response" => "Done!",
+ "response_segments" => [
+ %{"type" => "text", "content" => "Adding a step..."},
+ %{"type" => "status", "content" => "Validating workflow..."},
+ %{"type" => "text", "content" => "Done!"}
+ ],
+ "attachments" => [%{"type" => "workflow_yaml", "content" => "..."}]
+ })
+
"""
def streaming_or_sync_response(body) do
fn