From 7d772acff35ad0ef0a783ec0af70b7d21b5c94ab Mon Sep 17 00:00:00 2001 From: rcourtman Date: Sun, 12 Jul 2026 23:01:40 +0100 Subject: [PATCH] feat(assistant): mid-turn steering of the running response A follow-up sent during an active run now offers itself to the running agentic loop via POST /api/ai/sessions/{id}/steer. Accepted steers join the loop at its next turn boundary (the abort-check site) as plain user messages, are announced with a steer_applied stream event so the drawer settles the pending row, and persist through the end-of-run save. A steer carries prompt text only: no route, control-level, or autonomy changes, no turn-budget extension, system sessions rejected, and the per-session inbox is bounded (steer_backlog overflow). Delivery is not guaranteed by acceptance: a run that ends first discards the inbox and the row drains as an ordinary queued turn, so pre-steering queue semantics remain the fallback. Steering rows lose edit/remove once accepted. --- .../v6/internal/subsystems/agent-lifecycle.md | 8 + .../v6/internal/subsystems/ai-runtime.md | 29 +- .../v6/internal/subsystems/api-contracts.md | 16 + .../subsystems/performance-and-scalability.md | 9 + .../internal/subsystems/security-privacy.md | 7 + .../internal/subsystems/storage-recovery.md | 5 +- .../src/api/__tests__/aiChat.test.ts | 18 + .../src/api/__tests__/aiChatEvents.test.ts | 17 + frontend-modern/src/api/aiChat.ts | 24 ++ .../src/api/generated/aiChatEvents.ts | 9 + .../src/components/AI/Chat/ChatMessages.tsx | 6 +- .../src/components/AI/Chat/MessageItem.tsx | 4 + .../AI/Chat/__tests__/useChat.test.ts | 127 +++++++ .../src/components/AI/Chat/hooks/useChat.ts | 71 +++- .../src/components/AI/Chat/index.tsx | 69 ++-- internal/ai/chat/agentic.go | 30 ++ internal/ai/chat/agentic_steering.go | 74 ++++ internal/ai/chat/agentic_steering_test.go | 346 ++++++++++++++++++ internal/ai/chat/service.go | 58 ++- internal/ai/chat/types.go | 34 +- internal/api/ai_handler.go | 38 ++ .../api/ai_handler_recovery_wiring_test.go | 3 + internal/api/ai_handler_test.go | 8 + internal/api/router.go | 5 + internal/api/router_routes_additional_test.go | 46 +++ scripts/generate-types.go | 2 + 26 files changed, 1025 insertions(+), 38 deletions(-) create mode 100644 internal/ai/chat/agentic_steering.go create mode 100644 internal/ai/chat/agentic_steering_test.go diff --git a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md index fd20bb70f..e20f777d4 100644 --- a/docs/release-control/v6/internal/subsystems/agent-lifecycle.md +++ b/docs/release-control/v6/internal/subsystems/agent-lifecycle.md @@ -500,6 +500,14 @@ only. Lifecycle surfaces and agents must not interpret an undone or regenerated Assistant turn as reverting any lifecycle action, approval, or agent command the original turn produced; governed action history remains the only revert authority for infrastructure changes. +Mid-turn steering through `POST /api/ai/sessions/{id}/steer` is likewise +conversation input only: it adds a user message to a running Assistant loop +at a turn boundary. A steering message grants no agent command authority, +cannot approve, deny, or bypass a pending approval, cannot escalate the +running turn's control level or autonomous mode, and must not be +interpreted by lifecycle surfaces as operator authorization for any action +the steered turn subsequently proposes; those proposals still route through +the governed approval and action lifecycle unchanged. The native Assistant surface-tool inventory at `GET /api/ai/assistant/surface-tools` is also AI-runtime/API-contract metadata: lifecycle surfaces may display which Assistant tools are available, but must diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index 9e100f546..2ef1f3e9d 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -1340,11 +1340,30 @@ deriving an older display status from `workflowStatusHistory`. pattern by using its persisted explicit route contract rather than terminal provider/model objects. Follow-up sends during an active Assistant response are chat-runtime queue - state by default. The drawer must accept and echo the user's follow-up as a - queued user turn without aborting or replacing the active model stream, must - show an itemized composer-adjacent queue with per-follow-up edit/remove - controls plus clear-all, and must drain queued turns in order only after the - active stream becomes idle. Queued follow-ups must snapshot the effective + state that steers the running response by default. The drawer must accept + and echo the user's follow-up as a queued user turn without aborting or + replacing the active model stream, must show an itemized composer-adjacent + queue with per-follow-up edit/remove controls plus clear-all, and must + offer each follow-up to the running loop through + `POST /api/ai/sessions/{id}/steer` (mid-turn steering). A steering-accepted + follow-up joins the loop at its next turn boundary (the abort-check site in + `agentic.go executeWithTools`) as a plain user message, never mid-provider- + stream and never mid-tool-batch; the loop announces the injection with a + `steer_applied` stream event carrying the client row id so the drawer + settles the pending row into a delivered user turn, splices the message + into `resultMessages` at its true position, and persists it through the + end-of-run save (whose skip-user-messages rule exempts `Steered` messages). + Steering carries prompt text only: it cannot change the model route, + control level, or autonomous mode of the running turn, does not extend the + turn budget or reset wrap-up brakes, and is rejected for system sessions. + Once accepted for steering, a follow-up row loses its edit/remove + affordances (the text is in the loop's hands). Delivery is not guaranteed: + a run that finishes before a boundary discards unconsumed steers without + persisting them, and the drawer, which keeps the row queued until + `steer_applied` arrives, drains it in order after the active stream + becomes idle exactly as before — the pre-steering queue semantics remain + the fallback path, and paused queues do not steer. Queued follow-ups must + snapshot the effective model route at enqueue time so a later model/provider switch cannot silently reroute an already-queued user turn, and both the transcript queued-user row and composer-adjacent queue row must surface that snapshotted route label when diff --git a/docs/release-control/v6/internal/subsystems/api-contracts.md b/docs/release-control/v6/internal/subsystems/api-contracts.md index daf0c2494..f350c4b5c 100644 --- a/docs/release-control/v6/internal/subsystems/api-contracts.md +++ b/docs/release-control/v6/internal/subsystems/api-contracts.md @@ -1314,6 +1314,22 @@ payload shape change when the portal presents compact client rows. `AIChatAPI.undoLastTurn(sessionId, { expectedPrompt })` and `AIChatAPI.redoLastTurn(sessionId)` helpers so path encoding, guard trimming, and response shape stay canonical. + `POST /api/ai/sessions/{id}/steer` owns the mid-turn steering API + contract (`chat.SessionSteerRequest` -> `chat.SessionSteerResult`). The + request carries `prompt` plus an optional `client_message_id`; the + response is immediate JSON, never a second SSE stream, and + `accepted:false` with a `reason` (`no_active_run`, `system_session`, + `empty_prompt`, `steer_backlog` when the bounded per-session steering + inbox is full) is a normal outcome that clients handle by keeping the + follow-up on the ordinary queue-drain path. Steering cannot carry model + route, control level, or autonomous-mode changes. Confirmation of + delivery arrives only on the session's existing chat stream as a + generated `steer_applied` event (`chat.SteerAppliedData`, in + `aiChatEvents.ts` and the `AIChatStreamEvent` union) echoing + `client_message_id`, the server message id, the prompt, and the turn + index; clients must treat the endpoint's `accepted:true` as inbox + receipt, not delivery. Browser clients must use the shared + `AIChatAPI.steerSession(sessionId, { prompt, clientMessageId })` helper. OpenCode-style file diff/revert session routes are deliberately not part of Pulse's supported Assistant session contract: Pulse sessions do not own local code-file edits, and infrastructure mutations must be reviewed diff --git a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md index 05fb027ab..879a1ba52 100644 --- a/docs/release-control/v6/internal/subsystems/performance-and-scalability.md +++ b/docs/release-control/v6/internal/subsystems/performance-and-scalability.md @@ -243,6 +243,15 @@ change may globally weaken the Task 03 lifecycle-state idempotency invariant. atomic approval consume, not a route-wide scan or request-hot-path fan-out; grant signing and WebSocket writes happen only after that bounded verifier succeeds. + Assistant mid-turn steering (`POST /api/ai/sessions/{id}/steer`, routed + through the session sub-route dispatch in `internal/api/router.go`) is a + point operation on the same terms: a map lookup of the session's active + loop plus an in-memory inbox append, returning immediate JSON with no + second SSE stream, no session-file read, and no provider work on the + request path. The per-session steering inbox is bounded + (`maxPendingSteersPerSession`), so repeated steers cannot grow service + memory or the running turn's prompt without limit; overflow returns + `steer_backlog` and the message stays on the client's queue. Scheduled-report background worker registration is allowed in router startup, but it must stay outside protected request handling. Due-schedule scans may enumerate tenant organization IDs and load each workspace schedule store, but diff --git a/docs/release-control/v6/internal/subsystems/security-privacy.md b/docs/release-control/v6/internal/subsystems/security-privacy.md index c7a061e0a..119b877f7 100644 --- a/docs/release-control/v6/internal/subsystems/security-privacy.md +++ b/docs/release-control/v6/internal/subsystems/security-privacy.md @@ -339,6 +339,13 @@ the `white_label` branding entitlement. that rule: public chat and relay input cannot serialize its org/action authorization context, and invalid approvals fail before signing or agent dispatch rather than falling through to a route-local trust shortcut. + The Assistant steer sub-route (`POST /api/ai/sessions/{id}/steer`) added + to the session dispatch is bound by the same rule: it requires + `ScopeAIChat`, carries conversation text only, cannot approve or bypass + a pending approval, cannot change the running turn's control level, + autonomous mode, or model route, rejects Pulse-owned system sessions, + and its response discloses only `accepted` plus a coarse reason, never + run internals, provider state, or transcript content. The Patrol action-broker and proposal-catalog factory glue wired here is bound by the same rule: it may connect the investigation orchestrator to the tenant-bound action lifecycle, but it exposes only typed-proposal capture and gives the diff --git a/docs/release-control/v6/internal/subsystems/storage-recovery.md b/docs/release-control/v6/internal/subsystems/storage-recovery.md index d213bee98..84c367bb8 100644 --- a/docs/release-control/v6/internal/subsystems/storage-recovery.md +++ b/docs/release-control/v6/internal/subsystems/storage-recovery.md @@ -153,8 +153,9 @@ grant backup visibility, recovery authority, or storage health privileges; those remain governed by the setup-script and source-specific backup API boundaries below. Assistant chat-session history endpoints in `internal/api/ai_handler.go` -(session list/rename/fork/summarize and turn undo/redo, including the -retry/regenerate expected-prompt guard on `POST /api/ai/sessions/{id}/undo`) +(session list/rename/fork/summarize, turn undo/redo including the +retry/regenerate expected-prompt guard on `POST /api/ai/sessions/{id}/undo`, +and mid-turn steering via `POST /api/ai/sessions/{id}/steer`) are ai-runtime conversation-state surfaces only. Undoing or regenerating an Assistant turn rewrites chat transcript history; it is never a restore operation, does not revert storage mutations or governed actions the original diff --git a/frontend-modern/src/api/__tests__/aiChat.test.ts b/frontend-modern/src/api/__tests__/aiChat.test.ts index 12b43419c..1bcc84fc7 100644 --- a/frontend-modern/src/api/__tests__/aiChat.test.ts +++ b/frontend-modern/src/api/__tests__/aiChat.test.ts @@ -1407,6 +1407,24 @@ describe('AIChatAPI', () => { }); }); + it('steers a running session through the steer endpoint', async () => { + const result = { accepted: true, session_id: 'session/root' }; + apiFetchJSONMock.mockResolvedValueOnce(result); + + await expect( + AIChatAPI.steerSession('session/root', { + prompt: 'also check pve2', + clientMessageId: 'row-1', + }), + ).resolves.toEqual(result); + + expect(apiFetchJSONMock).toHaveBeenCalledWith('/api/ai/sessions/session%2Froot/steer', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ prompt: 'also check pve2', client_message_id: 'row-1' }), + }); + }); + it('sends the expected-prompt guard body when undo runs for a retry', async () => { const undoResult = { success: true, diff --git a/frontend-modern/src/api/__tests__/aiChatEvents.test.ts b/frontend-modern/src/api/__tests__/aiChatEvents.test.ts index f33335f38..4fa6f1703 100644 --- a/frontend-modern/src/api/__tests__/aiChatEvents.test.ts +++ b/frontend-modern/src/api/__tests__/aiChatEvents.test.ts @@ -130,6 +130,23 @@ describe('AI chat stream event contract', () => { expect(aiChatEventsSource).toContain('model?: string'); }); + it('exposes mid-turn steering injections as a typed stream contract', () => { + const event: AIChatStreamEvent = { + type: 'steer_applied', + data: { + session_id: 'sess-stream', + message_id: 'srv-1', + client_message_id: 'row-1', + prompt: 'also check pve2', + turn: 2, + }, + }; + + expect(event.data.client_message_id).toBe('row-1'); + expect(aiChatEventsSource).toContain('export interface SteerAppliedData'); + expect(aiChatEventsSource).toContain("type: 'steer_applied'"); + }); + it('exposes the estimated cumulative session cost on done events', () => { const done: DoneData = { session_id: 'sess-stream', diff --git a/frontend-modern/src/api/aiChat.ts b/frontend-modern/src/api/aiChat.ts index 059ce3f4b..453a3a7f5 100644 --- a/frontend-modern/src/api/aiChat.ts +++ b/frontend-modern/src/api/aiChat.ts @@ -37,6 +37,12 @@ export interface ChatSessionRedoResult { message?: string; } +export interface ChatSessionSteerResult { + accepted: boolean; + session_id: string; + reason?: string; // "no_active_run" | "system_session" | "empty_prompt" | "steer_backlog" +} + export interface ChatSessionCompactionResult { success: boolean; status: 'compacted' | 'not_needed' | 'empty' | string; @@ -409,6 +415,24 @@ export class AIChatAPI { }) as Promise; } + // Steer the session's running response: the message joins the in-flight + // agentic loop at its next turn boundary. accepted=false (e.g. the run + // already finished) is a normal outcome; the caller keeps the follow-up + // queued and it drains as an ordinary new turn. + static async steerSession( + sessionId: string, + request: { prompt: string; clientMessageId?: string }, + ): Promise { + return apiFetchJSON(`${this.baseUrl}/sessions/${encodeURIComponent(sessionId)}/steer`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + prompt: request.prompt, + ...(request.clientMessageId ? { client_message_id: request.clientMessageId } : {}), + }), + }) as Promise; + } + // Stream chat - the main chat interface static async chat( prompt: string, diff --git a/frontend-modern/src/api/generated/aiChatEvents.ts b/frontend-modern/src/api/generated/aiChatEvents.ts index 9ccdd3dff..78eb842ab 100644 --- a/frontend-modern/src/api/generated/aiChatEvents.ts +++ b/frontend-modern/src/api/generated/aiChatEvents.ts @@ -90,6 +90,14 @@ export interface SessionData { id: string; } +export interface SteerAppliedData { + session_id?: string; + message_id?: string; + client_message_id?: string; + prompt?: string; + turn: number; +} + export interface ThinkingData { text: string; } @@ -150,5 +158,6 @@ export type AIChatStreamEvent = | { type: 'tool_end'; data: ToolEndData } | { type: 'approval_needed'; data: ApprovalNeededData } | { type: 'question'; data: QuestionData & { session_id?: string } } + | { type: 'steer_applied'; data: SteerAppliedData } | { type: 'done'; data?: DoneData } | { type: 'error'; data: ErrorData }; diff --git a/frontend-modern/src/components/AI/Chat/ChatMessages.tsx b/frontend-modern/src/components/AI/Chat/ChatMessages.tsx index e40d7a813..de571f2c0 100644 --- a/frontend-modern/src/components/AI/Chat/ChatMessages.tsx +++ b/frontend-modern/src/components/AI/Chat/ChatMessages.tsx @@ -223,6 +223,7 @@ export const ChatMessages: Component = (props) => { position: index + 1, count: entries.length, paused: Boolean(props.queuedFollowUpsPaused), + steering: Boolean(entry.steering), }, ]), ); @@ -333,8 +334,9 @@ export const ChatMessages: Component = (props) => { queuedPosition={queuedMeta()?.position} queuedCount={queuedMeta()?.count} queuedPaused={queuedMeta()?.paused} + queuedSteering={queuedMeta()?.steering} onEditQueued={ - queuedMeta() && props.onEditQueuedFollowUp + queuedMeta() && !queuedMeta()?.steering && props.onEditQueuedFollowUp ? () => { const meta = queuedMeta(); if (meta) props.onEditQueuedFollowUp?.(meta.id); @@ -342,7 +344,7 @@ export const ChatMessages: Component = (props) => { : undefined } onCancelQueued={ - queuedMeta() && props.onCancelQueuedFollowUp + queuedMeta() && !queuedMeta()?.steering && props.onCancelQueuedFollowUp ? () => { const meta = queuedMeta(); if (meta) props.onCancelQueuedFollowUp?.(meta.id); diff --git a/frontend-modern/src/components/AI/Chat/MessageItem.tsx b/frontend-modern/src/components/AI/Chat/MessageItem.tsx index 04fc3ffa1..af9c22cc2 100644 --- a/frontend-modern/src/components/AI/Chat/MessageItem.tsx +++ b/frontend-modern/src/components/AI/Chat/MessageItem.tsx @@ -68,6 +68,9 @@ interface MessageItemProps { queuedPosition?: number; queuedCount?: number; queuedPaused?: boolean; + // The follow-up was accepted for mid-turn steering and will join the + // running response at its next step; edit/remove are no longer offered. + queuedSteering?: boolean; onEditQueued?: () => void; onCancelQueued?: () => void; } @@ -252,6 +255,7 @@ export const MessageItem: Component = (props) => { const isQueuedUserMessage = () => isUser() && props.message.delivery === 'queued'; const queuedStatusLabel = createMemo(() => { if (!isQueuedUserMessage()) return ''; + if (props.queuedSteering) return 'Steering the running response'; const position = props.queuedPosition; const count = props.queuedCount; const state = props.queuedPaused ? 'Paused' : 'Queued'; diff --git a/frontend-modern/src/components/AI/Chat/__tests__/useChat.test.ts b/frontend-modern/src/components/AI/Chat/__tests__/useChat.test.ts index c51aee5ee..e53584029 100644 --- a/frontend-modern/src/components/AI/Chat/__tests__/useChat.test.ts +++ b/frontend-modern/src/components/AI/Chat/__tests__/useChat.test.ts @@ -11,6 +11,7 @@ vi.mock('@/api/aiChat', () => ({ answerQuestion: vi.fn(), undoLastTurn: vi.fn(), redoLastTurn: vi.fn(), + steerSession: vi.fn(), }, })); @@ -43,6 +44,7 @@ const mockAbortSession = AIChatAPI.abortSession as ReturnType; const mockAnswerQuestion = AIChatAPI.answerQuestion as ReturnType; const mockUndoLastTurn = AIChatAPI.undoLastTurn as ReturnType; const mockRedoLastTurn = AIChatAPI.redoLastTurn as ReturnType; +const mockSteerSession = AIChatAPI.steerSession as ReturnType; const mockNotifyError = notificationStore.error as ReturnType; type TestStreamEvent = StreamEvent | { type: string; data?: unknown }; @@ -832,6 +834,131 @@ describe('useChat', () => { dispose(); }); + it('steers a follow-up into the running response when the backend accepts', async () => { + let fireEvent!: (e: TestStreamEvent) => void; + let resolveFirst!: () => void; + mockChat.mockImplementationOnce( + (_p: string, _s: string, _m: string | undefined, onEvent: (e: StreamEvent) => void) => { + fireEvent = onEvent as (e: TestStreamEvent) => void; + return new Promise((resolve) => { + resolveFirst = resolve; + }); + }, + ); + mockSteerSession.mockResolvedValueOnce({ accepted: true, session_id: 'sess' }); + + const { value: chat, dispose } = withRoot(() => useChat({ sessionId: 'sess' })); + const first = chat.sendMessage('first'); + await new Promise((r) => setTimeout(r, 0)); + + await chat.sendMessage('also check pve2'); + await new Promise((r) => setTimeout(r, 0)); + + const entry = chat.queuedFollowUps()[0]; + expect(mockSteerSession).toHaveBeenCalledWith('sess', { + prompt: 'also check pve2', + clientMessageId: entry.messageId, + }); + expect(entry.steering).toBe(true); + + // Backend confirms injection: the pending row settles into an + // ordinary delivered user message and leaves the queue. + fireEvent({ + type: 'steer_applied', + data: { + client_message_id: entry.messageId, + message_id: 'srv-steer-1', + prompt: 'also check pve2', + turn: 1, + }, + }); + await new Promise((r) => setTimeout(r, 0)); + + expect(chat.queuedFollowUpCount()).toBe(0); + const steered = chat.messages().find((m) => m.content === 'also check pve2'); + expect(steered?.delivery).toBeUndefined(); + + resolveFirst(); + await first; + await new Promise((r) => setTimeout(r, 0)); + // Nothing drains afterwards: the steered message must not re-send. + expect(mockChat).toHaveBeenCalledTimes(1); + dispose(); + }); + + it('keeps the follow-up queued and drains normally when steering is not accepted', async () => { + let resolveFirst!: () => void; + let resolveSecond!: () => void; + mockChat + .mockImplementationOnce( + () => + new Promise((resolve) => { + resolveFirst = resolve; + }), + ) + .mockImplementationOnce( + () => + new Promise((resolve) => { + resolveSecond = resolve; + }), + ); + mockSteerSession.mockResolvedValueOnce({ + accepted: false, + session_id: 'sess', + reason: 'no_active_run', + }); + + const { value: chat, dispose } = withRoot(() => useChat({ sessionId: 'sess' })); + const first = chat.sendMessage('first'); + await new Promise((r) => setTimeout(r, 0)); + + await chat.sendMessage('second'); + await new Promise((r) => setTimeout(r, 0)); + + expect(chat.queuedFollowUps()[0]?.steering).toBeFalsy(); + + resolveFirst(); + await first; + await new Promise((r) => setTimeout(r, 0)); + + expect(mockChat).toHaveBeenCalledTimes(2); + expect(mockChat.mock.calls[1][0]).toBe('second'); + + resolveSecond(); + await new Promise((r) => setTimeout(r, 0)); + dispose(); + }); + + it("renders another client's steer as a delivered user message", async () => { + let fireEvent!: (e: TestStreamEvent) => void; + let resolveFirst!: () => void; + mockChat.mockImplementationOnce( + (_p: string, _s: string, _m: string | undefined, onEvent: (e: StreamEvent) => void) => { + fireEvent = onEvent as (e: TestStreamEvent) => void; + return new Promise((resolve) => { + resolveFirst = resolve; + }); + }, + ); + + const { value: chat, dispose } = withRoot(() => useChat({ sessionId: 'sess' })); + const first = chat.sendMessage('first'); + await new Promise((r) => setTimeout(r, 0)); + + fireEvent({ + type: 'steer_applied', + data: { client_message_id: 'foreign-row', message_id: 'srv-9', prompt: 'from another tab', turn: 2 }, + }); + await new Promise((r) => setTimeout(r, 0)); + + const echoed = chat.messages().find((m) => m.content === 'from another tab'); + expect(echoed).toMatchObject({ role: 'user' }); + + resolveFirst(); + await first; + dispose(); + }); + it('snapshots the selected model for queued follow-ups', async () => { const resolvers: Array<() => void> = []; mockChat.mockImplementation( diff --git a/frontend-modern/src/components/AI/Chat/hooks/useChat.ts b/frontend-modern/src/components/AI/Chat/hooks/useChat.ts index 33cdb737c..503fb4aa3 100644 --- a/frontend-modern/src/components/AI/Chat/hooks/useChat.ts +++ b/frontend-modern/src/components/AI/Chat/hooks/useChat.ts @@ -126,6 +126,11 @@ export interface QueuedFollowUp { findingId?: string; sendOptions?: SendMessageOptions; timestamp: Date; + // The backend accepted this follow-up for mid-turn steering: it will join + // the running response at its next step. The row can no longer be edited + // or removed (the text is already in the loop's hands); if the run ends + // before delivery, the backend discards it and the row drains normally. + steering?: boolean; } export interface RestoredPromptDraft { @@ -1344,6 +1349,40 @@ export function useChat(options: UseChatOptions = {}) { applyStreamSessionId(extractSessionId(event.data)); } + if (event.type === 'steer_applied') { + const data = (event.data ?? {}) as { + client_message_id?: string; + message_id?: string; + prompt?: string; + }; + const clientMessageId = + typeof data.client_message_id === 'string' ? data.client_message_id : ''; + const prompt = typeof data.prompt === 'string' ? data.prompt : ''; + const queued = clientMessageId + ? queuedFollowUps().find((entry) => entry.messageId === clientMessageId) + : undefined; + if (queued) { + // Our pending row was injected into the running response: settle it + // into an ordinary delivered user message. + setQueuedFollowUps((prev) => prev.filter((entry) => entry.id !== queued.id)); + setMessages((prev) => + prev.map((msg) => (msg.id === queued.messageId ? { ...msg, delivery: undefined } : msg)), + ); + } else if (prompt.trim()) { + // Another client steered this session; echo the injected user turn. + setMessages((prev) => [ + ...prev, + { + id: typeof data.message_id === 'string' && data.message_id ? data.message_id : generateId(), + role: 'user', + content: prompt, + timestamp: new Date(), + }, + ]); + } + return; + } + if (event.type === 'workflow_state') { const workflowStatus = extractWorkflowStatus(event.data); const startedModel = @@ -1922,9 +1961,37 @@ export function useChat(options: UseChatOptions = {}) { logger.debug('[useChat] Queued follow-up while assistant response is streaming', { queuedFollowUpId: id, }); + void attemptSteer(queuedFollowUp); return true; }; + // Offer a queued follow-up to the running response. Acceptance only means + // the backend inbox holds it — the row stays queued until steer_applied + // confirms injection, so an undelivered steer still drains normally. + const attemptSteer = async (entry: QueuedFollowUp) => { + const currentSessionId = sessionId().trim(); + if (!currentSessionId) return; + try { + const result = await AIChatAPI.steerSession(currentSessionId, { + prompt: entry.prompt, + clientMessageId: entry.messageId, + }); + if (!result.accepted) { + logger.debug('[useChat] Steer not accepted; follow-up stays queued', { + reason: result.reason, + }); + return; + } + setQueuedFollowUps((prev) => + prev.map((candidate) => + candidate.id === entry.id ? { ...candidate, steering: true } : candidate, + ), + ); + } catch (error) { + logger.warn('[useChat] Steering attempt failed; follow-up stays queued', error); + } + }; + const startMessageSend = async ( prompt: string, mentions?: ChatMention[], @@ -2141,7 +2208,9 @@ export function useChat(options: UseChatOptions = {}) { const item = queuedFollowUps().find((entry) => entry.id === id); if (!item) return false; if (isLoading()) { - return promoteQueuedFollowUp(id); + const promoted = promoteQueuedFollowUp(id); + if (promoted && !item.steering) void attemptSteer(item); + return promoted; } setQueuedFollowUpsPaused(false); diff --git a/frontend-modern/src/components/AI/Chat/index.tsx b/frontend-modern/src/components/AI/Chat/index.tsx index f848a66b2..5b163c5d3 100644 --- a/frontend-modern/src/components/AI/Chat/index.tsx +++ b/frontend-modern/src/components/AI/Chat/index.tsx @@ -1605,6 +1605,9 @@ export const AIChat: Component = (props) => { }; const editQueuedFollowUp = (id: string) => { + // A steering follow-up is already in the running loop's hands; its text + // can no longer be recalled for editing. + if (chat.queuedFollowUps().find((entry) => entry.id === id)?.steering) return; const queued = chat.takeQueuedFollowUp(id); if (!queued) return; resetPromptHistoryNavigation(); @@ -1666,6 +1669,7 @@ export const AIChat: Component = (props) => { id: string, ) => { if (event.defaultPrevented || event.target !== event.currentTarget) return; + if (chat.queuedFollowUps().find((entry) => entry.id === id)?.steering) return; if (event.key === 'Enter') { event.preventDefault(); @@ -5055,10 +5059,14 @@ export const AIChat: Component = (props) => { {(queued, index) => { const preview = () => queuedFollowUpPreview(queued.prompt); const routeLabel = () => queuedFollowUpRouteLabel(queued); - const rowLabel = () => - routeLabel() + const rowLabel = () => { + if (queued.steering) { + return `Steering follow-up: ${preview()}. It joins the running response at its next step.`; + } + return routeLabel() ? `Queued follow-up: ${preview()}. Route: ${routeLabel()}. Press Enter to edit or Delete to remove.` : `Queued follow-up: ${preview()}. Press Enter to edit or Delete to remove.`; + }; return (
= (props) => { > {preview()} - + + + Steering the running response + + + {(label) => ( = (props) => { )} - 1 && index() > 0}> + 1 && index() > 0 && !queued.steering + } + > sendQueuedFollowUpNext(queued.id)} tone="accentGhost" @@ -5116,27 +5133,29 @@ export const AIChat: Component = (props) => { - editQueuedFollowUp(queued.id)} - tone="accentGhost" - size="xs" - title="Edit queued follow-up" - label={`Edit queued follow-up: ${preview()}`} - > - - { - chat.cancelQueuedFollowUp(queued.id); - focusComposer(); - }} - tone="accentGhost" - size="xs" - title="Remove queued follow-up" - label={`Remove queued follow-up: ${preview()}`} - > - + + editQueuedFollowUp(queued.id)} + tone="accentGhost" + size="xs" + title="Edit queued follow-up" + label={`Edit queued follow-up: ${preview()}`} + > + + { + chat.cancelQueuedFollowUp(queued.id); + focusComposer(); + }} + tone="accentGhost" + size="xs" + title="Remove queued follow-up" + label={`Remove queued follow-up: ${preview()}`} + > + +
); }} diff --git a/internal/ai/chat/agentic.go b/internal/ai/chat/agentic.go index 06e2b50a9..e5503fe4d 100644 --- a/internal/ai/chat/agentic.go +++ b/internal/ai/chat/agentic.go @@ -498,6 +498,7 @@ type AgenticLoop struct { mu sync.Mutex aborted map[string]bool // sessionID -> aborted pendingQs map[string]chan []QuestionAnswer // questionID -> answer channel + pendingSteers map[string][]pendingSteer // sessionID -> steering messages awaiting the next turn boundary autonomousMode bool // When true, don't wait for approvals (for investigations) // executionProfile is the core-owned request posture (interactive // Assistant, Patrol detection, Patrol investigation). It owns @@ -537,6 +538,7 @@ func NewAgenticLoop(provider providers.StreamingProvider, executor *tools.PulseT orgID: approval.DefaultOrgID, aborted: make(map[string]bool), pendingQs: make(map[string]chan []QuestionAnswer), + pendingSteers: make(map[string][]pendingSteer), } } @@ -599,6 +601,10 @@ func (a *AgenticLoop) executeWithTools(ctx context.Context, sessionID string, me a.mu.Lock() delete(a.aborted, sessionID) a.mu.Unlock() + // Unconsumed steers are dropped, never persisted: the client keeps + // its row queued until steer_applied confirms delivery, so an + // undelivered steer re-sends as a normal follow-up turn. + a.discardPendingSteers(sessionID) }() // Convert our messages to provider format @@ -654,6 +660,30 @@ func (a *AgenticLoop) executeWithTools(ctx context.Context, sessionID string, me requestSanitizer := a.requestSanitizer a.mu.Unlock() + // === MID-TURN STEERING: inject queued user messages at the boundary === + // Steers arrive via Service.SteerSession while a turn streams or its + // tools run; they join the conversation here, before the next model + // call, so the model reads them as ordinary user turns. Each one is + // announced on the stream so the drawer can settle its pending row. + for _, steer := range a.takePendingSteers(sessionID) { + msg := steer.message.NormalizeCollections() + providerMessages = append(providerMessages, convertToProviderMessages([]Message{msg})...) + resultMessages = append(resultMessages, msg) + if data, err := json.Marshal(SteerAppliedData{ + SessionID: sessionID, + MessageID: msg.ID, + ClientMessageID: steer.clientMessageID, + Prompt: msg.Content, + Turn: turn, + }); err == nil { + callback(StreamEvent{Type: "steer_applied", Data: data}) + } + log.Info(). + Int("turn", turn). + Str("session_id", sessionID). + Msg("[AgenticLoop] Steering message injected at turn boundary") + } + // Record telemetry for loop iteration if metrics := GetAIMetrics(); metrics != nil { metrics.RecordAgenticIteration(providerName, modelName) diff --git a/internal/ai/chat/agentic_steering.go b/internal/ai/chat/agentic_steering.go new file mode 100644 index 000000000..c4a03334e --- /dev/null +++ b/internal/ai/chat/agentic_steering.go @@ -0,0 +1,74 @@ +package chat + +import ( + "errors" + "fmt" + "strings" +) + +// maxPendingSteersPerSession bounds the steering inbox so a chatty client +// cannot grow the running turn's prompt (and the service's memory) without +// limit; excess steers stay on the client's ordinary follow-up queue. +const maxPendingSteersPerSession = 8 + +// errSteerBacklogFull reports a full steering inbox; the service maps it to +// a normal accepted=false outcome rather than an error status. +var errSteerBacklogFull = errors.New("steer backlog full") + +// pendingSteer pairs the steering message with the client-side transcript +// row id so the steer_applied event can reconcile the originating drawer. +type pendingSteer struct { + message Message + clientMessageID string +} + +// Steer queues a user message for injection into the running loop at the +// next turn boundary (the same checkpoint that observes aborts). Delivery is +// not guaranteed: if the run finishes before a boundary arrives, unconsumed +// steers are discarded and the client re-sends through the normal queue +// drain, so the message is never persisted twice. +func (a *AgenticLoop) Steer(sessionID string, msg Message, clientMessageID string) error { + sessionID = strings.TrimSpace(sessionID) + if sessionID == "" { + return fmt.Errorf("steer requires a session id") + } + if strings.TrimSpace(msg.Content) == "" { + return fmt.Errorf("steer requires a non-empty prompt") + } + + a.mu.Lock() + defer a.mu.Unlock() + if a.pendingSteers == nil { + a.pendingSteers = make(map[string][]pendingSteer) + } + if len(a.pendingSteers[sessionID]) >= maxPendingSteersPerSession { + return errSteerBacklogFull + } + a.pendingSteers[sessionID] = append(a.pendingSteers[sessionID], pendingSteer{ + message: msg, + clientMessageID: strings.TrimSpace(clientMessageID), + }) + return nil +} + +// takePendingSteers drains and returns the steering messages queued for a +// session, in arrival order. +func (a *AgenticLoop) takePendingSteers(sessionID string) []pendingSteer { + a.mu.Lock() + defer a.mu.Unlock() + steers := a.pendingSteers[sessionID] + if len(steers) == 0 { + return nil + } + delete(a.pendingSteers, sessionID) + return steers +} + +// discardPendingSteers drops any unconsumed steering messages when a run +// ends. The frontend keeps the row queued until steer_applied confirms +// delivery, so an undelivered steer drains as a normal follow-up turn. +func (a *AgenticLoop) discardPendingSteers(sessionID string) { + a.mu.Lock() + defer a.mu.Unlock() + delete(a.pendingSteers, sessionID) +} diff --git a/internal/ai/chat/agentic_steering_test.go b/internal/ai/chat/agentic_steering_test.go new file mode 100644 index 000000000..d27acccaa --- /dev/null +++ b/internal/ai/chat/agentic_steering_test.go @@ -0,0 +1,346 @@ +package chat + +import ( + "context" + "encoding/json" + "sync" + "testing" + "time" + + "github.com/rcourtman/pulse-go-rewrite/internal/ai/providers" + "github.com/rcourtman/pulse-go-rewrite/internal/ai/tools" + "github.com/rcourtman/pulse-go-rewrite/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// TestAgenticLoop_SteerInjectsAtNextTurnBoundary drives a two-turn run where +// turn 1 blocks on a pulse_question. A steer delivered while the run is +// blocked must be injected as a user message before turn 2's provider call, +// announced via steer_applied, and returned in resultMessages marked Steered. +func TestAgenticLoop_SteerInjectsAtNextTurnBoundary(t *testing.T) { + executor := tools.NewPulseToolExecutor(tools.ExecutorConfig{}) + mockProvider := &MockProvider{} + loop := NewAgenticLoop(mockProvider, executor, "You are a helper") + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + sessionID := "steer-session" + messages := []Message{{Role: "user", Content: "Do something but ask me first"}} + const steerPrompt = "actually check pve2 as well" + + // Turn 1: model requests pulse_question (blocks until answered). + mockProvider.On("ChatStream", mock.Anything, mock.MatchedBy(func(req providers.ChatRequest) bool { + return len(req.Messages) == 1 + }), mock.Anything).Return(nil).Run(func(args mock.Arguments) { + cb := args.Get(2).(providers.StreamCallback) + toolInput := map[string]interface{}{ + "questions": []interface{}{ + map[string]interface{}{ + "id": "q1", + "type": "select", + "question": "Pick one", + "options": []interface{}{ + map[string]interface{}{"label": "A", "value": "a"}, + }, + }, + }, + } + cb(providers.StreamEvent{ + Type: "tool_start", + Data: providers.ToolStartEvent{ID: "t1", Name: pulseQuestionToolName, Input: toolInput}, + }) + cb(providers.StreamEvent{ + Type: "done", + Data: providers.DoneEvent{ + ToolCalls: []providers.ToolCall{{ID: "t1", Name: pulseQuestionToolName, Input: toolInput}}, + }, + }) + }).Once() + + // Turn 2: the request must carry the steer as a user message AFTER the + // tool result for t1. + mockProvider.On("ChatStream", mock.Anything, mock.MatchedBy(func(req providers.ChatRequest) bool { + toolResultIndex := -1 + steerIndex := -1 + for i, m := range req.Messages { + if m.ToolResult != nil && m.ToolResult.ToolUseID == "t1" { + toolResultIndex = i + } + if m.Role == "user" && m.Content == steerPrompt { + steerIndex = i + } + } + return toolResultIndex >= 0 && steerIndex > toolResultIndex + }), mock.Anything).Return(nil).Run(func(args mock.Arguments) { + cb := args.Get(2).(providers.StreamCallback) + cb(providers.StreamEvent{Type: "content", Data: providers.ContentEvent{Text: "Checking pve2 too."}}) + cb(providers.StreamEvent{Type: "done", Data: providers.DoneEvent{}}) + }).Once() + + var ( + mu sync.Mutex + questionEvt *QuestionData + steerApplied *SteerAppliedData + ) + callback := func(event StreamEvent) { + mu.Lock() + defer mu.Unlock() + if event.Type == "question" && questionEvt == nil { + var qd QuestionData + _ = json.Unmarshal(event.Data, &qd) + questionEvt = &qd + } + if event.Type == "steer_applied" && steerApplied == nil { + var sd SteerAppliedData + _ = json.Unmarshal(event.Data, &sd) + steerApplied = &sd + } + } + + var ( + results []Message + err error + doneCh = make(chan struct{}) + ) + go func() { + defer close(doneCh) + results, err = loop.Execute(ctx, sessionID, messages, callback) + }() + + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return questionEvt != nil && questionEvt.QuestionID != "" + }, 2*time.Second, 10*time.Millisecond, "expected question event") + + // Steer while the run is blocked, then unblock it. + require.NoError(t, loop.Steer(sessionID, Message{ + ID: "steer-msg-1", Role: "user", Content: steerPrompt, Steered: true, Timestamp: time.Now(), + }, "client-row-1")) + + mu.Lock() + qID := questionEvt.QuestionID + mu.Unlock() + require.NoError(t, loop.AnswerQuestion(qID, []QuestionAnswer{{ID: "q1", Value: "a"}})) + + select { + case <-doneCh: + case <-ctx.Done(): + t.Fatalf("agentic loop did not complete: %v", ctx.Err()) + } + require.NoError(t, err) + + mu.Lock() + require.NotNil(t, steerApplied, "expected steer_applied event") + assert.Equal(t, "client-row-1", steerApplied.ClientMessageID) + assert.Equal(t, "steer-msg-1", steerApplied.MessageID) + assert.Equal(t, steerPrompt, steerApplied.Prompt) + mu.Unlock() + + steeredCount := 0 + for _, msg := range results { + if msg.Steered { + steeredCount++ + assert.Equal(t, "user", msg.Role) + assert.Equal(t, steerPrompt, msg.Content) + } + } + assert.Equal(t, 1, steeredCount, "expected exactly one steered message in results") + assert.Equal(t, "Checking pve2 too.", results[len(results)-1].Content) + + mockProvider.AssertExpectations(t) +} + +// TestAgenticLoop_UnconsumedSteerIsDiscarded verifies that a steer arriving +// too late for any boundary is dropped when the run ends, so the client's +// queue-drain fallback cannot double-record it. +func TestAgenticLoop_UnconsumedSteerIsDiscarded(t *testing.T) { + executor := tools.NewPulseToolExecutor(tools.ExecutorConfig{}) + provider := &stubStreamingProvider{} + loop := NewAgenticLoop(provider, executor, "system") + + sessionID := "late-steer-session" + _, err := loop.Execute(context.Background(), sessionID, []Message{{Role: "user", Content: "hi"}}, func(StreamEvent) {}) + require.NoError(t, err) + + // The run already ended; the loop's defer must have cleared the inbox, + // and a fresh run must not see stale steers from a prior run either. + require.NoError(t, loop.Steer(sessionID, Message{ID: "late", Role: "user", Content: "too late"}, "")) + loop.discardPendingSteers(sessionID) + assert.Empty(t, loop.takePendingSteers(sessionID)) +} + +// TestAgenticLoop_SteerBacklogIsBounded verifies the inbox cap so a chatty +// client cannot grow the running turn's prompt without limit. +func TestAgenticLoop_SteerBacklogIsBounded(t *testing.T) { + executor := tools.NewPulseToolExecutor(tools.ExecutorConfig{}) + loop := NewAgenticLoop(&stubStreamingProvider{}, executor, "system") + + for i := 0; i < maxPendingSteersPerSession; i++ { + require.NoError(t, loop.Steer("backlog-session", Message{ID: "m", Role: "user", Content: "steer"}, "")) + } + err := loop.Steer("backlog-session", Message{ID: "m", Role: "user", Content: "one too many"}, "") + require.ErrorIs(t, err, errSteerBacklogFull) + assert.Len(t, loop.takePendingSteers("backlog-session"), maxPendingSteersPerSession) +} + +// TestService_SteerSession_RoutingOutcomes covers the service-level routing +// results that never reach a loop. +func TestService_SteerSession_RoutingOutcomes(t *testing.T) { + svc := &Service{} + + result, err := svc.SteerSession(context.Background(), "no-run-session", SessionSteerRequest{Prompt: "hello"}) + require.NoError(t, err) + assert.False(t, result.Accepted) + assert.Equal(t, "no_active_run", result.Reason) + + result, err = svc.SteerSession(context.Background(), "patrol-main", SessionSteerRequest{Prompt: "hello"}) + require.NoError(t, err) + assert.False(t, result.Accepted) + assert.Equal(t, "system_session", result.Reason) + + result, err = svc.SteerSession(context.Background(), "some-session", SessionSteerRequest{Prompt: " "}) + require.NoError(t, err) + assert.False(t, result.Accepted) + assert.Equal(t, "empty_prompt", result.Reason) + + _, err = svc.SteerSession(context.Background(), "../bad", SessionSteerRequest{Prompt: "hello"}) + require.Error(t, err) +} + +// TestService_ExecuteStream_SteeredMessagePersists proves the end-to-end +// path: a steer accepted mid-run is injected at the boundary AND survives +// the end-of-run save (which skips ordinary user messages). +func TestService_ExecuteStream_SteeredMessagePersists(t *testing.T) { + tmpDir := t.TempDir() + store, err := NewSessionStore(tmpDir) + require.NoError(t, err) + + executor := tools.NewPulseToolExecutor(tools.ExecutorConfig{}) + mockProvider := &MockProvider{} + loop := NewAgenticLoop(mockProvider, executor, "system") + + svc := &Service{ + cfg: &config.AIConfig{ChatModel: "openai:test"}, + sessions: store, + executor: executor, + agenticLoop: loop, + provider: mockProvider, + started: true, + activeExecutions: make(map[string]map[*AgenticLoop]struct{}), + questionExecutions: make(map[string]*AgenticLoop), + } + + const steerPrompt = "steer: also check the replication lag" + sessionID := "sess-steer-persist" + + // Turn 1: block on a question; the test steers through the SERVICE while + // blocked, then answers. + toolInput := map[string]interface{}{ + "questions": []interface{}{ + map[string]interface{}{ + "id": "q1", "type": "select", "question": "Proceed?", + "options": []interface{}{map[string]interface{}{"label": "Yes", "value": "y"}}, + }, + }, + } + mockProvider.On("ChatStream", mock.Anything, mock.MatchedBy(func(req providers.ChatRequest) bool { + for _, m := range req.Messages { + if m.Role == "user" && m.Content == steerPrompt { + return false + } + } + return true + }), mock.Anything).Return(nil).Run(func(args mock.Arguments) { + cb := args.Get(2).(providers.StreamCallback) + cb(providers.StreamEvent{ + Type: "tool_start", + Data: providers.ToolStartEvent{ID: "t1", Name: pulseQuestionToolName, Input: toolInput}, + }) + cb(providers.StreamEvent{ + Type: "done", + Data: providers.DoneEvent{ + ToolCalls: []providers.ToolCall{{ID: "t1", Name: pulseQuestionToolName, Input: toolInput}}, + }, + }) + }).Once() + mockProvider.On("ChatStream", mock.Anything, mock.MatchedBy(func(req providers.ChatRequest) bool { + for _, m := range req.Messages { + if m.Role == "user" && m.Content == steerPrompt { + return true + } + } + return false + }), mock.Anything).Return(nil).Run(func(args mock.Arguments) { + cb := args.Get(2).(providers.StreamCallback) + cb(providers.StreamEvent{Type: "content", Data: providers.ContentEvent{Text: "done, checked lag"}}) + cb(providers.StreamEvent{Type: "done", Data: providers.DoneEvent{}}) + }).Once() + + var ( + mu sync.Mutex + questionEvt *QuestionData + ) + callback := func(event StreamEvent) { + mu.Lock() + defer mu.Unlock() + if event.Type == "question" && questionEvt == nil { + var qd QuestionData + _ = json.Unmarshal(event.Data, &qd) + questionEvt = &qd + } + } + + var execErr error + doneCh := make(chan struct{}) + go func() { + defer close(doneCh) + execErr = svc.ExecuteStream(context.Background(), ExecuteRequest{SessionID: sessionID, Prompt: "check the cluster"}, callback) + }() + + require.Eventually(t, func() bool { + mu.Lock() + defer mu.Unlock() + return questionEvt != nil && questionEvt.QuestionID != "" + }, 3*time.Second, 10*time.Millisecond, "expected question event") + + steerResult, err := svc.SteerSession(context.Background(), sessionID, SessionSteerRequest{ + Prompt: steerPrompt, + ClientMessageID: "client-row-9", + }) + require.NoError(t, err) + require.True(t, steerResult.Accepted, "expected steer to reach the active loop, got reason %q", steerResult.Reason) + + mu.Lock() + qID := questionEvt.QuestionID + mu.Unlock() + require.NoError(t, svc.AnswerQuestion(context.Background(), qID, []QuestionAnswer{{ID: "q1", Value: "y"}})) + + select { + case <-doneCh: + case <-time.After(5 * time.Second): + t.Fatal("ExecuteStream did not complete") + } + require.NoError(t, execErr) + + // The steered message must be in durable history, after the opening + // prompt and before the final assistant answer. + saved, err := store.GetMessages(sessionID) + require.NoError(t, err) + steerIndex, finalIndex := -1, -1 + for i, msg := range saved { + if msg.Steered && msg.Content == steerPrompt { + steerIndex = i + } + if msg.Role == "assistant" && msg.Content == "done, checked lag" { + finalIndex = i + } + } + require.GreaterOrEqual(t, steerIndex, 1, "steered message missing from durable history") + require.Greater(t, finalIndex, steerIndex, "final answer should follow the steered message") + + mockProvider.AssertExpectations(t) +} diff --git a/internal/ai/chat/service.go b/internal/ai/chat/service.go index ffb8cc2f1..665aaf33c 100644 --- a/internal/ai/chat/service.go +++ b/internal/ai/chat/service.go @@ -3,6 +3,7 @@ package chat import ( "context" "encoding/json" + "errors" "fmt" "sort" "strings" @@ -267,6 +268,58 @@ func (s *Service) unregisterActiveLoop(sessionID string, loop *AgenticLoop) { } } +// SteerSession routes a mid-turn steering message to the session's running +// agentic loop for injection at the next turn boundary. accepted=false with +// a reason is a normal outcome: the caller falls back to the ordinary +// follow-up queue. Steering carries prompt text only — it cannot change the +// model route, control level, or autonomous mode of the running turn. +func (s *Service) SteerSession(ctx context.Context, sessionID string, req SessionSteerRequest) (*SessionSteerResult, error) { + sessionID = strings.TrimSpace(sessionID) + if err := validateSessionID(sessionID); err != nil { + return nil, err + } + result := &SessionSteerResult{SessionID: sessionID} + if IsSystemSessionID(sessionID) { + result.Reason = "system_session" + return result, nil + } + prompt := strings.TrimSpace(req.Prompt) + if prompt == "" { + result.Reason = "empty_prompt" + return result, nil + } + + s.activeMu.RLock() + var loops []*AgenticLoop + for loop := range s.activeExecutions[sessionID] { + loops = append(loops, loop) + } + s.activeMu.RUnlock() + if len(loops) == 0 { + result.Reason = "no_active_run" + return result, nil + } + + msg := Message{ + ID: uuid.New().String(), + Role: "user", + Content: prompt, + Steered: true, + Timestamp: time.Now(), + } + for _, loop := range loops { + if err := loop.Steer(sessionID, msg, req.ClientMessageID); err != nil { + if errors.Is(err, errSteerBacklogFull) { + result.Reason = "steer_backlog" + return result, nil + } + return nil, err + } + } + result.Accepted = true + return result, nil +} + func assistantContextScopeForChatTurn( req ExecuteRequest, handoffFindingID string, @@ -996,8 +1049,9 @@ func (s *Service) ExecuteStream(ctx context.Context, req ExecuteRequest, callbac // Save result messages for _, msg := range resultMessages { - // Skip user messages (already saved) - if msg.Role == "user" && msg.ToolResult == nil { + // Skip user messages (already saved) — except steered ones, which + // entered the conversation inside the loop and exist nowhere else. + if msg.Role == "user" && msg.ToolResult == nil && !msg.Steered { continue } if msg.Role == "assistant" && strings.TrimSpace(msg.Model) == "" { diff --git a/internal/ai/chat/types.go b/internal/ai/chat/types.go index 9ffcdead0..4dfd8f22a 100644 --- a/internal/ai/chat/types.go +++ b/internal/ai/chat/types.go @@ -94,7 +94,12 @@ type Message struct { ToolCalls []ToolCall `json:"tool_calls"` ToolResult *ToolResult `json:"tool_result,omitempty"` Model string `json:"model,omitempty"` - Timestamp time.Time `json:"timestamp"` + // Steered marks a user message that was injected into a running agentic + // loop at a turn boundary (mid-turn steering) rather than opening a new + // turn. Steered user messages are persisted by the end-of-run save, + // unlike the turn-opening user prompt which is saved before the loop. + Steered bool `json:"steered,omitempty"` + Timestamp time.Time `json:"timestamp"` } func EmptyMessage() Message { @@ -457,6 +462,33 @@ type ErrorData struct { Message string `json:"message"` } +// SteerAppliedData is the data for "steer_applied" events: a steering +// message was injected into the running loop at a turn boundary. The +// originating client reconciles its pending row via ClientMessageID; other +// clients on the same session render Prompt as a new steered user row. +type SteerAppliedData struct { + SessionID string `json:"session_id,omitempty"` + MessageID string `json:"message_id,omitempty"` + ClientMessageID string `json:"client_message_id,omitempty"` + Prompt string `json:"prompt,omitempty"` + Turn int `json:"turn"` +} + +// SessionSteerRequest is the payload for POST /api/ai/sessions/{id}/steer. +type SessionSteerRequest struct { + Prompt string `json:"prompt"` + ClientMessageID string `json:"client_message_id,omitempty"` +} + +// SessionSteerResult reports whether a steering message reached a running +// loop. accepted=false with a reason is a normal outcome (e.g. the run +// finished first); the client falls back to the ordinary queue drain. +type SessionSteerResult struct { + Accepted bool `json:"accepted"` + SessionID string `json:"session_id"` + Reason string `json:"reason,omitempty"` // "no_active_run" | "system_session" | "empty_prompt" | "steer_backlog" +} + // DoneData is the data for "done" events type DoneData struct { SessionID string `json:"session_id,omitempty"` diff --git a/internal/api/ai_handler.go b/internal/api/ai_handler.go index 10e2b02a1..31fe3ddef 100644 --- a/internal/api/ai_handler.go +++ b/internal/api/ai_handler.go @@ -61,6 +61,7 @@ type AIService interface { ForkSession(ctx context.Context, sessionID string) (*chat.Session, error) UndoLastTurn(ctx context.Context, sessionID string, opts chat.SessionTurnUndoOptions) (*chat.SessionTurnUndoResult, error) RedoLastTurn(ctx context.Context, sessionID string) (*chat.SessionTurnRedoResult, error) + SteerSession(ctx context.Context, sessionID string, req chat.SessionSteerRequest) (*chat.SessionSteerResult, error) AnswerQuestion(ctx context.Context, questionID string, answers []chat.QuestionAnswer) error AssistantSurfaceToolContract(ctx context.Context) agentcapabilities.SurfaceToolContract SetAlertProvider(provider chat.AssistantAlertProvider) @@ -3264,6 +3265,43 @@ func (h *AIHandler) HandleRedoLastTurn(w http.ResponseWriter, r *http.Request, s json.NewEncoder(w).Encode(result) } +// HandleSteerSession handles POST /api/ai/sessions/{id}/steer. +// It routes a mid-turn steering message to the session's running agentic +// loop; accepted=false with a reason is a normal outcome and the client +// falls back to the ordinary follow-up queue. +func (h *AIHandler) HandleSteerSession(w http.ResponseWriter, r *http.Request, sessionID string) { + if r.Method != http.MethodPost { + http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) + return + } + ctx := r.Context() + if !h.IsRunning(ctx) { + http.Error(w, "Pulse Assistant is not running", http.StatusServiceUnavailable) + return + } + + svc := h.GetService(ctx) + if svc == nil { + http.Error(w, "Pulse Assistant service not available", http.StatusServiceUnavailable) + return + } + + var steerReq chat.SessionSteerRequest + if err := json.NewDecoder(r.Body).Decode(&steerReq); err != nil { + http.Error(w, "Invalid request body", http.StatusBadRequest) + return + } + + result, err := svc.SteerSession(ctx, sessionID, steerReq) + if err != nil { + http.Error(w, sanitizeErrorForClient(err, "Internal server error"), http.StatusInternalServerError) + return + } + + w.Header().Set("Content-Type", "application/json") + json.NewEncoder(w).Encode(result) +} + // HandleRevert handles POST /api/ai/sessions/{id}/revert // Rejects OpenCode-style file revert requests; Pulse actions use governed history. func (h *AIHandler) HandleRevert(w http.ResponseWriter, r *http.Request, sessionID string) { diff --git a/internal/api/ai_handler_recovery_wiring_test.go b/internal/api/ai_handler_recovery_wiring_test.go index a9dbc5bdf..8a15ba9c3 100644 --- a/internal/api/ai_handler_recovery_wiring_test.go +++ b/internal/api/ai_handler_recovery_wiring_test.go @@ -66,6 +66,9 @@ func (s *capturingAIService) ForkSession(ctx context.Context, sessionID string) func (s *capturingAIService) UndoLastTurn(ctx context.Context, sessionID string, opts chat.SessionTurnUndoOptions) (*chat.SessionTurnUndoResult, error) { return &chat.SessionTurnUndoResult{Success: true, SessionID: sessionID}, nil } +func (s *capturingAIService) SteerSession(ctx context.Context, sessionID string, req chat.SessionSteerRequest) (*chat.SessionSteerResult, error) { + return &chat.SessionSteerResult{Accepted: false, SessionID: sessionID, Reason: "no_active_run"}, nil +} func (s *capturingAIService) RedoLastTurn(ctx context.Context, sessionID string) (*chat.SessionTurnRedoResult, error) { return &chat.SessionTurnRedoResult{Success: true, SessionID: sessionID}, nil } diff --git a/internal/api/ai_handler_test.go b/internal/api/ai_handler_test.go index cfd6a41f8..aa66ddc19 100644 --- a/internal/api/ai_handler_test.go +++ b/internal/api/ai_handler_test.go @@ -197,6 +197,14 @@ func (m *MockAIService) UndoLastTurn(ctx context.Context, sessionID string, opts return args.Get(0).(*chat.SessionTurnUndoResult), args.Error(1) } +func (m *MockAIService) SteerSession(ctx context.Context, sessionID string, req chat.SessionSteerRequest) (*chat.SessionSteerResult, error) { + args := m.Called(ctx, sessionID, req) + if args.Get(0) == nil { + return nil, args.Error(1) + } + return args.Get(0).(*chat.SessionSteerResult), args.Error(1) +} + func (m *MockAIService) RedoLastTurn(ctx context.Context, sessionID string) (*chat.SessionTurnRedoResult, error) { args := m.Called(ctx, sessionID) if args.Get(0) == nil { diff --git a/internal/api/router.go b/internal/api/router.go index ae72c0ffc..5452551da 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -1004,6 +1004,11 @@ func (r *Router) routeAISessions(w http.ResponseWriter, req *http.Request) { return } r.aiHandler.HandleRedoLastTurn(w, req, sessionID) + case "steer": + if !ensureScope(w, req, config.ScopeAIChat) { + return + } + r.aiHandler.HandleSteerSession(w, req, sessionID) case "revert": if !ensureScope(w, req, config.ScopeAIChat) { return diff --git a/internal/api/router_routes_additional_test.go b/internal/api/router_routes_additional_test.go index d2b43ff47..5a0426d6c 100644 --- a/internal/api/router_routes_additional_test.go +++ b/internal/api/router_routes_additional_test.go @@ -252,6 +252,52 @@ func TestRouteAISessions_UndoLastTurnRetryOptions(t *testing.T) { mockSvc.AssertExpectations(t) } +func TestRouteAISessions_SteerSession(t *testing.T) { + mockSvc := &MockAIService{} + mockSvc.On("IsRunning").Return(true) + mockSvc.On("SteerSession", mock.Anything, "session-1", chat.SessionSteerRequest{ + Prompt: "also check pve2", + ClientMessageID: "client-row-1", + }).Return(&chat.SessionSteerResult{ + Accepted: true, + SessionID: "session-1", + }, nil) + + handler := &AIHandler{} + setUnexportedField(t, handler, "defaultService", mockSvc) + + router := &Router{aiHandler: handler} + body := strings.NewReader(`{"prompt":"also check pve2","client_message_id":"client-row-1"}`) + req := httptest.NewRequest(http.MethodPost, "/api/ai/sessions/session-1/steer", body) + rec := httptest.NewRecorder() + + router.routeAISessions(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("expected status %d, got %d", http.StatusOK, rec.Code) + } + if !strings.Contains(rec.Body.String(), `"accepted":true`) { + t.Fatalf("expected accepted result, got %s", rec.Body.String()) + } + mockSvc.AssertExpectations(t) +} + +func TestRouteAISessions_SteerSessionRejectsNonPost(t *testing.T) { + mockSvc := &MockAIService{} + handler := &AIHandler{} + setUnexportedField(t, handler, "defaultService", mockSvc) + + router := &Router{aiHandler: handler} + req := httptest.NewRequest(http.MethodGet, "/api/ai/sessions/session-1/steer", nil) + rec := httptest.NewRecorder() + + router.routeAISessions(rec, req) + + if rec.Code != http.StatusMethodNotAllowed { + t.Fatalf("expected status %d, got %d", http.StatusMethodNotAllowed, rec.Code) + } +} + func TestRouteAISessions_RedoLastTurn(t *testing.T) { mockSvc := &MockAIService{} mockSvc.On("IsRunning").Return(true) diff --git a/scripts/generate-types.go b/scripts/generate-types.go index f1ada8f62..590f3c4ec 100644 --- a/scripts/generate-types.go +++ b/scripts/generate-types.go @@ -52,6 +52,7 @@ func main() { reflect.TypeOf(chat.QuestionData{}), reflect.TypeOf(chat.Question{}), reflect.TypeOf(chat.QuestionOption{}), + reflect.TypeOf(chat.SteerAppliedData{}), reflect.TypeOf(chat.DoneData{}), reflect.TypeOf(chat.ErrorData{}), } @@ -155,6 +156,7 @@ func chatStreamEventUnion() string { // The contract test covers {question_id, questions}; the UI currently expects session_id too. // Keep session_id optional for backward compatibility. buf.WriteString(" | { type: 'question'; data: QuestionData & { session_id?: string } }\n") + buf.WriteString(" | { type: 'steer_applied'; data: SteerAppliedData }\n") buf.WriteString(" | { type: 'done'; data?: DoneData }\n") buf.WriteString(" | { type: 'error'; data: ErrorData };\n") return buf.String()