diff --git a/CHANGELOG.md b/CHANGELOG.md index da6d816a..41cedb11 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,19 @@ All notable changes to this project are documented in this file. Format follows For narrative release notes written for operators and product owners, see [RELEASE_NOTES.md](RELEASE_NOTES.md). +## [1.0.4] - 2026-07-01 + +IAM hotfix restoring AgentCore Memory. Both the App API task role and the AgentCore Runtime execution role were missing `bedrock-agentcore:GetMemory`, which `get_memory_strategies()` needs to resolve strategy IDs β€” breaking the memory dashboard (empty results) and long-term recall (retrieval silently disabled). Ships via the platform (CDK) pipeline; no migration. + +### πŸ› Fixed + +- Memory dashboard (`GET /memory`, `/memory/preferences`, `/memory/facts`) returned empty lists with a 200: `get_memory_strategies()` β†’ `bedrock-agentcore:GetMemory` was denied on the App API task role, so strategy discovery yielded no namespaces and retrieval was skipped (`app-api-iam-grants.ts`) +- Agent stopped recalling long-term memories: the runtime's `_discover_strategy_ids()` makes the same `GetMemory` call to build retrieval namespaces; the runtime execution role also lacked the action, disabling long-term retrieval while `CreateEvent` writes still succeeded (`inference-api-iam-roles.ts`) + +### πŸ—οΈ Infrastructure + +- Added `bedrock-agentcore:GetMemory` to the `AgentCoreMemoryAccess` statement on both the App API Fargate task role (scoped to the memory ARN) and the AgentCore Runtime execution role (scoped to `memory/*`); no other actions changed. Validated against the AWS Service Authorization Reference (`GetMemory` = Read on the `memory` resource type) + ## [1.0.3] - 2026-06-30 Maintenance patch: CI/CD pipeline cleanup, re-enabled path-scoped auto-deploys, and a dependency/CodeQL sweep. No application code or user-facing behavior changes; upgrade in place. diff --git a/README.md b/README.md index 5571891c..d570b3d7 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ **An open-source, production-ready Generative AI platform for institutions** *Built by Boise State University, designed for everyone.* -[![Release](https://img.shields.io/badge/Release-v1.0.3-6366f1?style=flat&logo=github&logoColor=white)](RELEASE_NOTES.md) +[![Release](https://img.shields.io/badge/Release-v1.0.4-6366f1?style=flat&logo=github&logoColor=white)](RELEASE_NOTES.md) [![Nightly](https://github.com/Boise-State-Development/agentcore-public-stack/actions/workflows/nightly.yml/badge.svg)](https://github.com/Boise-State-Development/agentcore-public-stack/actions/workflows/nightly.yml) ![Python](https://img.shields.io/badge/Python-3.13+-3776AB?style=flat&logo=python&logoColor=white) @@ -296,7 +296,7 @@ agentcore-public-stack/ See [RELEASE_NOTES.md](RELEASE_NOTES.md) for the full changelog, including new features, bug fixes, platform upgrades, and deployment notes for each release. -**Current release:** v1.0.3 +**Current release:** v1.0.4 --- diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 4a318580..3408f96f 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -1,3 +1,33 @@ +# Release Notes β€” v1.0.4 + +**Release Date:** July 1, 2026 +**Previous Release:** v1.0.3 (June 30, 2026) + +--- + +> ⚠️ **Upgrading from a beta?** 1.0.4 is an in-place upgrade from any 1.0.x with no migration. Moving from a pre-1.0.0 beta is still the destructive backup β†’ teardown β†’ redeploy β†’ restore migration described in the [1.0.0 notes](#upgrading-an-existing-deployment). Brand-new deployments need none of this. + +--- + +## Highlights + +v1.0.4 is a one-line-per-role IAM hotfix that restores **AgentCore Memory** functionality. Both the App API task role and the AgentCore Runtime execution role granted every memory data-plane action *except* `bedrock-agentcore:GetMemory` β€” the action the SDK's `get_memory_strategies()` call requires to resolve a memory's strategy IDs. Without it, strategy discovery failed silently: the **Settings β†’ memories/preferences page came back empty** (the App API returned empty lists with a 200), and the **agent stopped recalling long-term memories** (the runtime kept writing conversation events but ran with retrieval disabled). Granting `GetMemory` on both roles fixes both symptoms. This is an **infrastructure (IAM) change**, so it deploys via the platform (CDK) pipeline. + +## πŸ› Bug fixes + +- **Memory dashboard showed no memories/preferences.** `GET /memory` calls `get_memory_strategies()` β†’ `bedrock-agentcore:GetMemory`, which the App API task role didn't allow. The call `AccessDenied`, strategy discovery returned no IDs, and the endpoint returned empty `facts`/`preferences` with a 200 β€” so the page rendered blank even though records existed. (`app-api-iam-grants.ts`) +- **Agent didn't recall long-term memories.** At session creation the runtime's `_discover_strategy_ids()` makes the same `GetMemory` call to build its retrieval namespaces; the runtime execution role also lacked the action, so retrieval was silently disabled ("long-term memory retrieval disabled") while event writes (`CreateEvent`) continued to succeed. (`inference-api-iam-roles.ts`) + +## πŸ—οΈ Infrastructure + +- Added `bedrock-agentcore:GetMemory` to the `AgentCoreMemoryAccess` policy statement on **both** roles β€” the App API Fargate task role (scoped to the memory ARN) and the AgentCore Runtime execution role (scoped to `memory/*`). No other action names changed. Verified against the AWS Service Authorization Reference (`GetMemory` is a Read action on the `memory` resource type). + +## πŸš€ Deployment notes + +This is an IAM change on `PlatformStack`, so it ships through the **platform (CDK)** pipeline, not the API-only backend path. After deploy, no data migration is needed and existing memories become visible immediately. Note the App API caches strategy discovery per process (`functools.lru_cache`) β€” a normal deploy rolls the ECS tasks and AgentCore Runtime, so the cache starts fresh; no manual restart required. + +--- + # Release Notes β€” v1.0.3 **Release Date:** June 30, 2026 diff --git a/VERSION b/VERSION index 21e8796a..ee90284c 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -1.0.3 +1.0.4 diff --git a/backend/pyproject.toml b/backend/pyproject.toml index f2bb88ab..e3632cb3 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "agentcore-stack" -version = "1.0.3" +version = "1.0.4" requires-python = ">=3.10" description = "Multi-agent conversational AI system with AWS Bedrock AgentCore" readme = "README.md" diff --git a/backend/uv.lock b/backend/uv.lock index 5af1a6a4..c7ba86b0 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -12,7 +12,7 @@ resolution-markers = [ [[package]] name = "agentcore-stack" -version = "1.0.3" +version = "1.0.4" source = { editable = "." } dependencies = [ { name = "aiofiles" }, diff --git a/frontend/ai.client/package-lock.json b/frontend/ai.client/package-lock.json index d290dec9..fce726e6 100644 --- a/frontend/ai.client/package-lock.json +++ b/frontend/ai.client/package-lock.json @@ -1,12 +1,12 @@ { "name": "ai.client", - "version": "1.0.3", + "version": "1.0.4", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "ai.client", - "version": "1.0.3", + "version": "1.0.4", "dependencies": { "@angular/cdk": "21.2.14", "@angular/common": "21.2.17", diff --git a/frontend/ai.client/package.json b/frontend/ai.client/package.json index ef7f986f..64f88a38 100644 --- a/frontend/ai.client/package.json +++ b/frontend/ai.client/package.json @@ -1,6 +1,6 @@ { "name": "ai.client", - "version": "1.0.3", + "version": "1.0.4", "scripts": { "ng": "ng", "start": "ng serve", diff --git a/frontend/ai.client/src/app/components/sidenav/components/session-list/session-list.html b/frontend/ai.client/src/app/components/sidenav/components/session-list/session-list.html index 532d175f..3672cda7 100644 --- a/frontend/ai.client/src/app/components/sidenav/components/session-list/session-list.html +++ b/frontend/ai.client/src/app/components/sidenav/components/session-list/session-list.html @@ -54,23 +54,39 @@

{{ getSessionTitle(session) }} - - - + + @if (isSessionStreaming(session.sessionId)) { + + + } @else { + + + } diff --git a/frontend/ai.client/src/app/components/sidenav/components/session-list/session-list.spec.ts b/frontend/ai.client/src/app/components/sidenav/components/session-list/session-list.spec.ts index 729b4976..27193768 100644 --- a/frontend/ai.client/src/app/components/sidenav/components/session-list/session-list.spec.ts +++ b/frontend/ai.client/src/app/components/sidenav/components/session-list/session-list.spec.ts @@ -74,4 +74,20 @@ describe('SessionList', () => { component['onSessionClick'](); expect(mockSidenavService.close).toHaveBeenCalled(); }); + + it('reflects per-session streaming state for the in-progress indicator', async () => { + const { ChatStateService } = await import('../../../../session/services/chat/chat-state.service'); + const chatState = TestBed.inject(ChatStateService); + const component = await createComponent(); + + expect(component['isSessionStreaming']('test-session')).toBe(false); + + chatState.setChatLoading('test-session', true); + expect(component['isSessionStreaming']('test-session')).toBe(true); + // Only the streaming conversation shows the indicator. + expect(component['isSessionStreaming']('other-session')).toBe(false); + + chatState.setChatLoading('test-session', false); + expect(component['isSessionStreaming']('test-session')).toBe(false); + }); }); diff --git a/frontend/ai.client/src/app/components/sidenav/components/session-list/session-list.ts b/frontend/ai.client/src/app/components/sidenav/components/session-list/session-list.ts index 56a94410..ee886297 100644 --- a/frontend/ai.client/src/app/components/sidenav/components/session-list/session-list.ts +++ b/frontend/ai.client/src/app/components/sidenav/components/session-list/session-list.ts @@ -8,6 +8,7 @@ import { NgIcon, provideIcons } from '@ng-icons/core'; import { heroChatBubbleLeftRight, heroTrash, heroArrowPath, heroPencilSquare, heroArrowUpOnSquare, heroCloudArrowUp } from '@ng-icons/heroicons/outline'; import { heroEllipsisHorizontalSolid } from '@ng-icons/heroicons/solid'; import { SessionService } from '../../../../session/services/session/session.service'; +import { ChatStateService } from '../../../../session/services/chat/chat-state.service'; import { ShareModalComponent, ShareModalData } from '../../../../session/components/share-modal'; import { ExportDialogComponent, ExportDialogData } from '../../../../session/components/export-dialog'; import { UserService } from '../../../../auth/user.service'; @@ -26,6 +27,7 @@ import { ConfirmationDialogComponent, ConfirmationDialogData } from '../../../co }) export class SessionList { private sessionService = inject(SessionService); + private chatStateService = inject(ChatStateService); private sidenavService = inject(SidenavService); private toastService = inject(ToastService); private dialog = inject(Dialog); @@ -170,6 +172,17 @@ export class SessionList { return sessionId; } + /** + * Whether this conversation has a response streaming right now. Reads the + * same per-session loading state that drives the composer's Stop button, + * so backgrounded conversations (user navigated away mid-stream) show a + * live indicator in the list. Signal-backed β€” the OnPush row re-renders + * when the stream starts or finishes. + */ + protected isSessionStreaming(sessionId: string): boolean { + return this.chatStateService.isSessionLoading(sessionId); + } + /** * Gets the queryParams for a session's routerLink. When the session has * an assistant attached in preferences, we include it in the URL so the diff --git a/frontend/ai.client/src/app/session/components/chat-container/chat-container.component.ts b/frontend/ai.client/src/app/session/components/chat-container/chat-container.component.ts index 7d8a7908..bd5bba7e 100644 --- a/frontend/ai.client/src/app/session/components/chat-container/chat-container.component.ts +++ b/frontend/ai.client/src/app/session/components/chat-container/chat-container.component.ts @@ -169,6 +169,16 @@ export class ChatContainerComponent { return !a.isSharedWithMe; }); + /** + * Anchor the latest user message at the top of the viewport. Exposed for + * the session page's navigation scroll policy (first open of a + * conversation lands on its latest turn, instantly); the composer submit + * path below uses the same anchor with smooth scrolling. + */ + scrollToLastUserMessage(behavior: ScrollBehavior = 'smooth'): void { + this.messageListComponent()?.scrollToLastUserMessage(behavior); + } + // Event handlers onMessageSubmitted(event: { content: string; timestamp: Date; fileUploadIds?: string[] }) { this.messageSubmitted.emit(event); diff --git a/frontend/ai.client/src/app/session/components/message-list/components/streaming-text.component.spec.ts b/frontend/ai.client/src/app/session/components/message-list/components/streaming-text.component.spec.ts new file mode 100644 index 00000000..105e2765 --- /dev/null +++ b/frontend/ai.client/src/app/session/components/message-list/components/streaming-text.component.spec.ts @@ -0,0 +1,71 @@ +import { ComponentFixture, TestBed } from '@angular/core/testing'; +import { describe, it, expect, beforeEach } from 'vitest'; +import { provideMarkdown, MarkdownService } from 'ngx-markdown'; +import { StreamingTextComponent } from './streaming-text.component'; + +describe('StreamingTextComponent', () => { + let fixture: ComponentFixture; + let component: StreamingTextComponent; + + beforeEach(async () => { + TestBed.resetTestingModule(); + await TestBed.configureTestingModule({ + imports: [StreamingTextComponent], + providers: [provideMarkdown()], + }).compileComponents(); + + // Stub render before component creation to prevent unhandled + // rejections from the real KaTeX dependency not being available. + const markdownService = TestBed.inject(MarkdownService); + markdownService.render = () => Promise.resolve(); + + fixture = TestBed.createComponent(StreamingTextComponent); + component = fixture.componentInstance; + }); + + it('shows already-streamed text immediately on mount mid-stream (no replay)', () => { + // Regression: navigating away from a streaming conversation and back + // recreates the message components. The accumulated partial response + // must appear instantly β€” not re-typed from character zero. + const partial = 'The morning mist clung to the ParanΓ‘ wetlands like a thin veil.'; + fixture.componentRef.setInput('text', partial); + fixture.componentRef.setInput('isStreaming', true); + fixture.detectChanges(); + + expect(component.displayedText()).toBe(partial); + }); + + it('shows full text immediately for a non-streaming message', () => { + fixture.componentRef.setInput('text', 'A completed answer.'); + fixture.componentRef.setInput('isStreaming', false); + fixture.detectChanges(); + + expect(component.displayedText()).toBe('A completed answer.'); + }); + + it('animates only text that arrives while mounted', () => { + fixture.componentRef.setInput('text', 'Seeded prefix. '); + fixture.componentRef.setInput('isStreaming', true); + fixture.detectChanges(); + expect(component.displayedText()).toBe('Seeded prefix. '); + + // New delta arrives while mounted β€” the typewriter picks it up from the + // seeded position, so the display never regresses below the prefix. + fixture.componentRef.setInput('text', 'Seeded prefix. And a new sentence.'); + fixture.detectChanges(); + + expect(component.displayedText().startsWith('Seeded prefix. ')).toBe(true); + }); + + it('flushes the full text the moment streaming ends', () => { + fixture.componentRef.setInput('text', 'Partial answer'); + fixture.componentRef.setInput('isStreaming', true); + fixture.detectChanges(); + + fixture.componentRef.setInput('text', 'Partial answer, now complete.'); + fixture.componentRef.setInput('isStreaming', false); + fixture.detectChanges(); + + expect(component.displayedText()).toBe('Partial answer, now complete.'); + }); +}); diff --git a/frontend/ai.client/src/app/session/components/message-list/components/streaming-text.component.ts b/frontend/ai.client/src/app/session/components/message-list/components/streaming-text.component.ts index 15f46602..7f8eaa53 100644 --- a/frontend/ai.client/src/app/session/components/message-list/components/streaming-text.component.ts +++ b/frontend/ai.client/src/app/session/components/message-list/components/streaming-text.component.ts @@ -61,6 +61,16 @@ export class StreamingTextComponent implements OnDestroy { private displayedLength = 0; private lastAnimationTime = 0; + /** + * True once the first effect run has seeded the display state. Text that + * already exists when the component mounts is shown immediately β€” the + * typewriter only animates text that arrives WHILE mounted. Without this, + * navigating away from a streaming conversation and back (which recreates + * the message components) would replay the entire accumulated response + * from character zero instead of picking up from its current state. + */ + private seeded = false; + constructor() { effect(() => { const currentText = this.text(); @@ -72,6 +82,15 @@ export class StreamingTextComponent implements OnDestroy { return; } + if (!this.seeded) { + // First run: catch up to whatever has already streamed (empty for + // a brand-new message, the full partial response on a remount). + this.seeded = true; + this.displayedLength = currentText.length; + this.displayedText.set(currentText); + return; + } + if (streaming && currentText.length > this.displayedLength) { // New text arrived while streaming - animate it this.startAnimation(); diff --git a/frontend/ai.client/src/app/session/components/message-list/components/tool-use/renderers/mcp-app-frame.component.ts b/frontend/ai.client/src/app/session/components/message-list/components/tool-use/renderers/mcp-app-frame.component.ts index fd768718..a8fb6fe7 100644 --- a/frontend/ai.client/src/app/session/components/message-list/components/tool-use/renderers/mcp-app-frame.component.ts +++ b/frontend/ai.client/src/app/session/components/message-list/components/tool-use/renderers/mcp-app-frame.component.ts @@ -832,12 +832,10 @@ export class McpAppFrameComponent implements ToolResultRenderer { args, ), sendMessage: (text) => { - // Mirror the composer's user-turn affordances that a direct - // submitChatRequest() would otherwise skip: show the loading indicator - // and scroll the new user message to the top. The user message is + // submitChatRequest sets the per-session loading state itself; we + // only add the composer's scroll affordance. The user message is // added synchronously inside submitChatRequest, so requesting the // scroll right after means it already exists in the list. - this.chatState.setChatLoading(true); const result = this.chatRequest.submitChatRequest( text, this.conversation.currentSession().sessionId || null, @@ -878,11 +876,15 @@ export class McpAppFrameComponent implements ToolResultRenderer { this.bridge?.notifyDisplayMode('inline'); } - /** Complete tool-call arguments, found by toolUseId in the live stream. */ + /** Complete tool-call arguments, found by toolUseId in the live stream. + * Parser state is per-session; this frame only renders inside the viewed + * conversation, so its stream is the one to search. */ private lookupToolInput(): Record { const id = this.toolUseId(); if (!id) return {}; - for (const msg of this.streamParser.allMessages()) { + const sessionId = this.chatState.viewedSessionId(); + if (!sessionId) return {}; + for (const msg of this.streamParser.allMessagesFor(sessionId)()) { for (const block of msg.content ?? []) { const tu = (block as { toolUse?: { toolUseId?: string; input?: unknown } }) .toolUse; diff --git a/frontend/ai.client/src/app/session/components/message-list/message-list.component.ts b/frontend/ai.client/src/app/session/components/message-list/message-list.component.ts index 55e924c5..142121f3 100644 --- a/frontend/ai.client/src/app/session/components/message-list/message-list.component.ts +++ b/frontend/ai.client/src/app/session/components/message-list/message-list.component.ts @@ -232,24 +232,29 @@ export class MessageListComponent implements OnDestroy { /** * Calculates the height needed for the bottom spacer * This ensures there's enough space for user messages to scroll to the top + * + * Set synchronously (it only reads window.innerHeight, no DOM measurement) + * so the extra scrollable height exists in the same layout pass as the + * messages β€” the navigation scroll restore in ConversationPage relies on + * the full scroll height being available right after render. */ private calculateSpacerHeight(): void { if (!this.isBrowser) return; - // Wait for next frame to ensure DOM is updated - requestAnimationFrame(() => { - const viewportHeight = window.innerHeight; - const spacerHeight = viewportHeight - this.HEADER_HEIGHT; - this.spacerHeight.set(spacerHeight); - }); + const viewportHeight = window.innerHeight; + this.spacerHeight.set(viewportHeight - this.HEADER_HEIGHT); } /** * Scrolls to a specific message by ID * Call this explicitly when user submits a message * Works in both full-page mode (window scroll) and embedded mode (container scroll) + * + * @param behavior 'smooth' for user-visible animation (submit affordance), + * 'auto' for instant positioning (navigation restore β€” animating a jump + * across a whole conversation would be noise). */ - scrollToMessage(messageId: string): void { + scrollToMessage(messageId: string, behavior: ScrollBehavior = 'smooth'): void { if (!this.isBrowser) return; const element = document.getElementById(`message-${messageId}`); @@ -257,7 +262,7 @@ export class MessageListComponent implements OnDestroy { if (this.embeddedMode()) { // In embedded mode, use scrollIntoView which works with any scroll container - element.scrollIntoView({ behavior: 'smooth', block: 'start' }); + element.scrollIntoView({ behavior, block: 'start' }); } else { // In full-page mode, use window scroll with offset for fixed header const elementRect = element.getBoundingClientRect(); @@ -266,7 +271,7 @@ export class MessageListComponent implements OnDestroy { window.scrollTo({ top: absoluteElementTop - offset, - behavior: 'smooth' + behavior }); } } @@ -274,11 +279,11 @@ export class MessageListComponent implements OnDestroy { /** * Scrolls to the last user message */ - scrollToLastUserMessage(): void { + scrollToLastUserMessage(behavior: ScrollBehavior = 'smooth'): void { const msgs = this.messages(); const lastUserMsg = [...msgs].reverse().find(m => m.role === 'user'); if (lastUserMsg) { - this.scrollToMessage(lastUserMsg.id); + this.scrollToMessage(lastUserMsg.id, behavior); } } diff --git a/frontend/ai.client/src/app/session/services/chat/chat-http.service.spec.ts b/frontend/ai.client/src/app/session/services/chat/chat-http.service.spec.ts index 256902aa..28c31987 100644 --- a/frontend/ai.client/src/app/session/services/chat/chat-http.service.spec.ts +++ b/frontend/ai.client/src/app/session/services/chat/chat-http.service.spec.ts @@ -30,9 +30,9 @@ describe('ChatHttpService', () => { // attaching a Bearer manually. { provide: BffSessionService, useValue: { csrfHeaders: vi.fn().mockReturnValue({}), handleUnauthorized: vi.fn() } }, { provide: SessionService, useValue: { currentSession: signal({ sessionId: 's1' }), updateSessionTitleInCache: vi.fn() } }, - { provide: StreamParserService, useValue: {} }, - { provide: ChatStateService, useValue: { isStreaming: signal(false), streamingSessionId: signal(null), abortCurrentRequest: vi.fn(), setChatLoading: vi.fn(), resetState: vi.fn(), getAbortController: vi.fn().mockReturnValue(new AbortController()) } }, - { provide: MessageMapService, useValue: {} }, + { provide: StreamParserService, useValue: { getCurrentStreamId: vi.fn().mockReturnValue('stream-1'), parseEventSourceMessage: vi.fn() } }, + { provide: ChatStateService, useValue: { abortRequest: vi.fn(), setChatLoading: vi.fn(), createAbortController: vi.fn().mockReturnValue(new AbortController()) } }, + { provide: MessageMapService, useValue: { endStreaming: vi.fn() } }, { provide: ErrorService, useValue: { handleHttpError: vi.fn() } }, ], }); @@ -59,10 +59,13 @@ describe('ChatHttpService', () => { expect(result.title).toBe('Generated Title'); }); - it('should cancel chat request', () => { - service.cancelChatRequest(); - expect(chatStateService.abortCurrentRequest).toHaveBeenCalled(); - expect(chatStateService.setChatLoading).toHaveBeenCalledWith(false); - expect(chatStateService.resetState).toHaveBeenCalled(); + it('should cancel only the target session and tear down its streaming state', () => { + const messageMap = TestBed.inject(MessageMapService) as any; + + service.cancelChatRequest('s1'); + + expect(chatStateService.abortRequest).toHaveBeenCalledWith('s1'); + expect(chatStateService.setChatLoading).toHaveBeenCalledWith('s1', false); + expect(messageMap.endStreaming).toHaveBeenCalledWith('s1'); }); }); diff --git a/frontend/ai.client/src/app/session/services/chat/chat-http.service.ts b/frontend/ai.client/src/app/session/services/chat/chat-http.service.ts index 0973dec4..d8318a06 100644 --- a/frontend/ai.client/src/app/session/services/chat/chat-http.service.ts +++ b/frontend/ai.client/src/app/session/services/chat/chat-http.service.ts @@ -58,165 +58,202 @@ export class ChatHttpService { private errorService = inject(ErrorService); async sendChatRequest(requestObject: any): Promise { - const abortController = this.chatStateService.getAbortController(); + const sessionId = requestObject.session_id as string; - // Phase 6c: stream goes through the app-api BFF proxy at - // `${appApiUrl}/chat/stream` (cookie auth) instead of hitting - // inference-api `/invocations` directly with a Bearer token. Same - // SSE protocol on the wire β€” the proxy is transparent. Cookies - // travel because the SPA and `/api/*` are same-origin via - // CloudFront; for local dev the same origin still works because - // the SPA is configured to point at `http://localhost:8000` (the - // app-api directly) and the BFF dual-auth dep accepts Bearer too. - const appApiUrl = this.config.appApiUrl(); - if (!appApiUrl) { - throw new FatalError('App API URL not configured. Please check your configuration.'); - } - const baseUrl = appApiUrl.endsWith('/') ? appApiUrl.slice(0, -1) : appApiUrl; + // Fresh controller per request, keyed by session. Creating it aborts any + // in-flight stream for the SAME session (double-submit guard) without + // touching other sessions' streams. + const abortController = this.chatStateService.createAbortController(sessionId); - // `fetchEventSource` is outside the HttpClient pipeline, so the - // csrfInterceptor never runs against it β€” attach the X-CSRF-Token - // header manually using the same SessionService helper. Returns - // an empty object before bootstrap or in the Bearer rollback path. - const csrfHeaders = this.bffSession.csrfHeaders(); + // Capture this stream's identity. The caller reset the parser for this + // session (startStreaming / beginContinuationStreaming) just before this + // call, so the current stream ID is ours. Events and lifecycle callbacks + // check it before touching state β€” if a newer stream has since reset the + // session, this stream is superseded and must leave everything alone. + const streamId = this.streamParserService.getCurrentStreamId(sessionId); + const isCurrentStream = () => + this.streamParserService.getCurrentStreamId(sessionId) === streamId; + // Sole owner of this stream's teardown (loading flag + streaming state). + // The guard makes it a no-op once a newer stream owns the session, so a + // superseded stream's late close/error can never tear down its + // replacement. Callers must NOT duplicate this cleanup in their catch + // blocks β€” an unguarded duplicate reintroduces exactly that race. + const finalizeStream = () => { + if (!isCurrentStream()) return; + this.messageMapService.endStreaming(sessionId); + this.chatStateService.setChatLoading(sessionId, false); + }; - // Capture into a local so `onopen` (a method-shorthand on the config - // object, not a closure over `this`) can reach the BFF session. - const bffSession = this.bffSession; + try { + // Phase 6c: stream goes through the app-api BFF proxy at + // `${appApiUrl}/chat/stream` (cookie auth) instead of hitting + // inference-api `/invocations` directly with a Bearer token. Same + // SSE protocol on the wire β€” the proxy is transparent. Cookies + // travel because the SPA and `/api/*` are same-origin via + // CloudFront; for local dev the same origin still works because + // the SPA is configured to point at `http://localhost:8000` (the + // app-api directly) and the BFF dual-auth dep accepts Bearer too. + const appApiUrl = this.config.appApiUrl(); + if (!appApiUrl) { + throw new FatalError('App API URL not configured. Please check your configuration.'); + } + const baseUrl = appApiUrl.endsWith('/') ? appApiUrl.slice(0, -1) : appApiUrl; - return fetchEventSource(`${baseUrl}/chat/stream`, { - method: 'POST', - // Send the BFF session cookie (`__Host-bff_session`) on cross-origin - // dev (localhost:4200 β†’ localhost:8000) and on same-origin prod - // (CloudFront). Browsers attach same-origin cookies regardless; - // `include` is the explicit form that also works cross-origin - // when the backend's CORS allows credentials. - credentials: 'include', - headers: { - 'Content-Type': 'application/json', - Accept: 'text/event-stream', - OAuth2CallbackUrl: `${window.location.origin}/oauth-complete`, - ...csrfHeaders, - }, - body: JSON.stringify(requestObject), - signal: abortController.signal, - async onopen(response) { - if (response.ok && response.headers.get('content-type')?.includes('text/event-stream')) { - return; // everything's good - } else if (response.status === 401) { - // BFF session is missing or expired. Bounce to /auth/login β€” - // `handleUnauthorized` is idempotent, so a 401 here that races - // with one from a parallel request only navigates once. - bffSession.handleUnauthorized(); - throw new UnauthorizedError(); - } else if (response.status === 403) { - // Handle forbidden (e.g., usage limit exceeded) - let errorMessage = 'Access forbidden'; + // `fetchEventSource` is outside the HttpClient pipeline, so the + // csrfInterceptor never runs against it β€” attach the X-CSRF-Token + // header manually using the same SessionService helper. Returns + // an empty object before bootstrap or in the Bearer rollback path. + const csrfHeaders = this.bffSession.csrfHeaders(); - try { - const errorData = await response.json(); - if (errorData.error) { - // Structured error from backend - errorMessage = errorData.error.message || errorMessage; - } else if (errorData.message) { - errorMessage = errorData.message; - } - } catch { - // Response not JSON, use default - } + // Capture into a local so `onopen` (a method-shorthand on the config + // object, not a closure over `this`) can reach the BFF session. + const bffSession = this.bffSession; - throw new FatalError(errorMessage); - } else if (response.status >= 400 && response.status < 500 && response.status !== 429) { - // Client-side errors are usually non-retriable - let errorMessage = `Request failed with status ${response.status}`; + return await fetchEventSource(`${baseUrl}/chat/stream`, { + method: 'POST', + // Send the BFF session cookie (`__Host-bff_session`) on cross-origin + // dev (localhost:4200 β†’ localhost:8000) and on same-origin prod + // (CloudFront). Browsers attach same-origin cookies regardless; + // `include` is the explicit form that also works cross-origin + // when the backend's CORS allows credentials. + credentials: 'include', + headers: { + 'Content-Type': 'application/json', + Accept: 'text/event-stream', + OAuth2CallbackUrl: `${window.location.origin}/oauth-complete`, + ...csrfHeaders, + }, + body: JSON.stringify(requestObject), + signal: abortController.signal, + async onopen(response) { + if (response.ok && response.headers.get('content-type')?.includes('text/event-stream')) { + return; // everything's good + } else if (response.status === 401) { + // BFF session is missing or expired. Bounce to /auth/login β€” + // `handleUnauthorized` is idempotent, so a 401 here that races + // with one from a parallel request only navigates once. + bffSession.handleUnauthorized(); + throw new UnauthorizedError(); + } else if (response.status === 403) { + // Handle forbidden (e.g., usage limit exceeded) + let errorMessage = 'Access forbidden'; - try { - const errorData = await response.json(); - if (errorData.error) { - // Structured error from backend - errorMessage = errorData.error.message || errorMessage; - } else if (errorData.message) { - errorMessage = errorData.message; + try { + const errorData = await response.json(); + if (errorData.error) { + // Structured error from backend + errorMessage = errorData.error.message || errorMessage; + } else if (errorData.message) { + errorMessage = errorData.message; + } + } catch { + // Response not JSON, use default } - } catch { - // If response is not JSON, try to get text + + throw new FatalError(errorMessage); + } else if (response.status >= 400 && response.status < 500 && response.status !== 429) { + // Client-side errors are usually non-retriable + let errorMessage = `Request failed with status ${response.status}`; + try { - const errorText = await response.text(); - errorMessage = errorText || errorMessage; + const errorData = await response.json(); + if (errorData.error) { + // Structured error from backend + errorMessage = errorData.error.message || errorMessage; + } else if (errorData.message) { + errorMessage = errorData.message; + } } catch { - // Ignore if we can't read the response + // If response is not JSON, try to get text + try { + const errorText = await response.text(); + errorMessage = errorText || errorMessage; + } catch { + // Ignore if we can't read the response + } } - } - throw new FatalError(errorMessage); - } else { - // Server errors or unexpected status codes (retriable) - const errorMessage = `Server error: ${response.status} ${response.statusText}`; - console.error('RetriableError:', errorMessage); - throw new RetriableError(errorMessage); - } - }, - onmessage: (msg: EventSourceMessage) => { - // Parse the data if it's a string - let parsedData = msg.data; - if (typeof msg.data === 'string') { - try { - parsedData = JSON.parse(msg.data); - } catch (e) { - console.warn('Failed to parse SSE data:', msg.data); - parsedData = msg.data; + throw new FatalError(errorMessage); + } else { + // Server errors or unexpected status codes (retriable) + const errorMessage = `Server error: ${response.status} ${response.statusText}`; + console.error('RetriableError:', errorMessage); + throw new RetriableError(errorMessage); } - } - this.streamParserService.parseEventSourceMessage(msg.event, parsedData); - }, - onclose: () => { - this.messageMapService.endStreaming(); - this.chatStateService.setChatLoading(false); + }, + onmessage: (msg: EventSourceMessage) => { + // Parse the data if it's a string + let parsedData = msg.data; + if (typeof msg.data === 'string') { + try { + parsedData = JSON.parse(msg.data); + } catch (e) { + console.warn('Failed to parse SSE data:', msg.data); + parsedData = msg.data; + } + } + this.streamParserService.parseEventSourceMessage( + sessionId, + msg.event, + parsedData, + streamId, + ); + }, + onclose: () => { + finalizeStream(); - // Title is generated server-side concurrently with the stream - // (see /invocations). Refresh metadata so the sidebar reflects it. - if (this.sessionService.isNewSession(requestObject.session_id)) { - this.refreshTitleFromServer(requestObject.session_id); - } - }, - onerror: (err) => { - this.messageMapService.endStreaming(); - this.chatStateService.setChatLoading(false); + // Title is generated server-side concurrently with the stream + // (see /invocations). Refresh metadata so the sidebar reflects it. + if (this.sessionService.isNewSession(requestObject.session_id)) { + this.refreshTitleFromServer(requestObject.session_id); + } + }, + onerror: (err) => { + finalizeStream(); - // 401 already triggered the redirect β€” skip the toast so it - // doesn't flash before the page tears down. - if (err instanceof UnauthorizedError) { - throw err; - } + // 401 already triggered the redirect β€” skip the toast so it + // doesn't flash before the page tears down. + if (err instanceof UnauthorizedError) { + throw err; + } - // Display error message to user using ErrorService - if (err instanceof FatalError) { - this.errorService.addError('Chat Request Failed', err.message, undefined, undefined); - } else if (err instanceof RetriableError) { - // For retriable errors, show with retry suggestion - this.errorService.addError( - 'Connection Error', - 'A temporary connection error occurred. The request may be retried automatically.', - err.message, - ); - } else { - // Unknown error type - this.errorService.handleNetworkError(err instanceof Error ? err.message : String(err)); - } + // Display error message to user using ErrorService + if (err instanceof FatalError) { + this.errorService.addError('Chat Request Failed', err.message, undefined, undefined); + } else if (err instanceof RetriableError) { + // For retriable errors, show with retry suggestion + this.errorService.addError( + 'Connection Error', + 'A temporary connection error occurred. The request may be retried automatically.', + err.message, + ); + } else { + // Unknown error type + this.errorService.handleNetworkError(err instanceof Error ? err.message : String(err)); + } - throw err; - }, - }); + throw err; + }, + }); + } catch (error) { + // Guarded teardown for failures fetchEventSource surfaces as a + // rejection (and for the pre-flight config throw above). Idempotent + // with the onerror path; no-op if a newer stream owns the session. + finalizeStream(); + throw error; + } } - cancelChatRequest(): void { - // First abort the client-side request - this.chatStateService.abortCurrentRequest(); - - this.chatStateService.setChatLoading(false); - - // Cleanup request-conversation mapping when cancelled - this.chatStateService.resetState(); + /** + * Stop one session's in-flight stream. Only that session is affected β€” + * other conversations streaming concurrently keep going. The aborted + * fetch resolves silently (fetch-event-source calls neither onclose nor + * onerror on abort), so the streaming teardown happens here. + */ + cancelChatRequest(sessionId: string): void { + this.chatStateService.abortRequest(sessionId); + this.messageMapService.endStreaming(sessionId); + this.chatStateService.setChatLoading(sessionId, false); } /** @@ -271,5 +308,4 @@ export class ChatHttpService { throw error; } } - } diff --git a/frontend/ai.client/src/app/session/services/chat/chat-request.service.spec.ts b/frontend/ai.client/src/app/session/services/chat/chat-request.service.spec.ts index c6311026..369ff6aa 100644 --- a/frontend/ai.client/src/app/session/services/chat/chat-request.service.spec.ts +++ b/frontend/ai.client/src/app/session/services/chat/chat-request.service.spec.ts @@ -12,6 +12,8 @@ import { ToolService } from '../../../services/tool/tool.service'; import { SkillService } from '../../../services/skill/skill.service'; import { ChatModeService, ChatMode } from '../../../services/chat-mode/chat-mode.service'; import { FileUploadService } from '../../../services/file-upload'; +import { OAuthConsentService } from '../../../services/oauth-consent/oauth-consent.service'; +import { ToolApprovalService } from '../../../services/tool-approval/tool-approval.service'; describe('ChatRequestService', () => { let service: ChatRequestService; @@ -20,6 +22,10 @@ describe('ChatRequestService', () => { let mockModelService: any; let mockToolService: any; let currentMode: ChatMode; + // Captured from the constructor's setResumeHandler(...) calls so the tests + // can drive the (private) resume paths the way the consent/approval UIs do. + let oauthResumeHandler: ((interruptIds: string[], context?: { sessionId?: string }) => Promise) | null; + let approvalResumeHandler: ((interruptId: string, decision: any, context?: { sessionId?: string }) => Promise) | null; beforeEach(() => { TestBed.resetTestingModule(); @@ -47,8 +53,8 @@ describe('ChatRequestService', () => { ChatRequestService, { provide: ChatHttpService, useValue: mockChatHttpService }, { provide: Router, useValue: mockRouter }, - { provide: ChatStateService, useValue: { setChatLoading: vi.fn(), setLastTurnContinuable: vi.fn(), createNewAbortController: vi.fn() } }, - { provide: MessageMapService, useValue: { addUserMessage: vi.fn(), startStreaming: vi.fn(), beginContinuationStreaming: vi.fn(), endStreaming: vi.fn() } }, + { provide: ChatStateService, useValue: { setChatLoading: vi.fn(), setLastTurnContinuable: vi.fn(), setViewedSession: vi.fn() } }, + { provide: MessageMapService, useValue: { addUserMessage: vi.fn(), startStreaming: vi.fn(), beginContinuationStreaming: vi.fn(), endStreaming: vi.fn(), reloadMessagesForSession: vi.fn().mockResolvedValue(undefined) } }, { provide: SessionService, useValue: { addSessionToCache: vi.fn() } }, { provide: UserService, useValue: { getUser: vi.fn().mockReturnValue({ user_id: 'user1' }) } }, { provide: ModelService, useValue: mockModelService }, @@ -56,6 +62,22 @@ describe('ChatRequestService', () => { { provide: SkillService, useValue: { getEnabledSkillIds: vi.fn().mockReturnValue(['skill_a']) } }, { provide: ChatModeService, useValue: { mode: () => currentMode } }, { provide: FileUploadService, useValue: { getReadyFileById: vi.fn() } }, + { + provide: OAuthConsentService, + useValue: { + setResumeHandler: vi.fn((handler: any) => { + oauthResumeHandler = handler; + }), + }, + }, + { + provide: ToolApprovalService, + useValue: { + setResumeHandler: vi.fn((handler: any) => { + approvalResumeHandler = handler; + }), + }, + }, ], }); service = TestBed.inject(ChatRequestService); @@ -143,6 +165,16 @@ describe('ChatRequestService', () => { ); }); + it('keys loading and viewed-session state to the submitted session', async () => { + const chatState = TestBed.inject(ChatStateService) as any; + + await service.submitChatRequest('Hello', 'session1'); + + expect(chatState.setViewedSession).toHaveBeenCalledWith('session1'); + expect(chatState.setChatLoading).toHaveBeenCalledWith('session1', true); + expect(chatState.setLastTurnContinuable).toHaveBeenCalledWith('session1', false); + }); + it('should throw error when no model selected', async () => { mockModelService.getSelectedModel.mockReturnValue(null); @@ -179,4 +211,60 @@ describe('ChatRequestService', () => { expect(mockChatHttpService.sendChatRequest).not.toHaveBeenCalled(); }); }); + + describe('resumeFromOAuthConsent', () => { + it('pins existing messages (continuation streaming) and reconciles from server', async () => { + const messageMap = TestBed.inject(MessageMapService) as any; + + await oauthResumeHandler!(['int-1'], { sessionId: 'session1' }); + + // The paused tool card must NOT be truncated away: continuation + // streaming pins it as a prefix instead of the normal truncate-to-user sync. + expect(messageMap.beginContinuationStreaming).toHaveBeenCalledWith('session1'); + expect(messageMap.startStreaming).not.toHaveBeenCalled(); + // No new user bubble on a resume turn. + expect(messageMap.addUserMessage).not.toHaveBeenCalled(); + + expect(mockChatHttpService.sendChatRequest).toHaveBeenCalledWith( + expect.objectContaining({ + session_id: 'session1', + message: '', + interrupt_responses: [{ interruptId: 'int-1', response: 'consented' }], + }), + ); + + // The resumed stream can't attach the tool_result live, so we + // reconcile from persisted memory to flip the card to its result. + expect(messageMap.reloadMessagesForSession).toHaveBeenCalledWith('session1'); + }); + + it('is a no-op without interrupt ids', async () => { + await oauthResumeHandler!([], { sessionId: 'session1' }); + expect(mockChatHttpService.sendChatRequest).not.toHaveBeenCalled(); + }); + + it('is a no-op without a session id', async () => { + await oauthResumeHandler!(['int-1'], {}); + expect(mockChatHttpService.sendChatRequest).not.toHaveBeenCalled(); + }); + }); + + describe('resumeFromToolApproval', () => { + it('pins existing messages and reconciles from server after a decision', async () => { + const messageMap = TestBed.inject(MessageMapService) as any; + + await approvalResumeHandler!('int-9', 'approved', { sessionId: 'session1' }); + + expect(messageMap.beginContinuationStreaming).toHaveBeenCalledWith('session1'); + expect(messageMap.startStreaming).not.toHaveBeenCalled(); + expect(mockChatHttpService.sendChatRequest).toHaveBeenCalledWith( + expect.objectContaining({ + session_id: 'session1', + message: '', + interrupt_responses: [{ interruptId: 'int-9', response: 'approved' }], + }), + ); + expect(messageMap.reloadMessagesForSession).toHaveBeenCalledWith('session1'); + }); + }); }); \ No newline at end of file diff --git a/frontend/ai.client/src/app/session/services/chat/chat-request.service.ts b/frontend/ai.client/src/app/session/services/chat/chat-request.service.ts index 3b02a18b..9e7f399d 100644 --- a/frontend/ai.client/src/app/session/services/chat/chat-request.service.ts +++ b/frontend/ai.client/src/app/session/services/chat/chat-request.service.ts @@ -18,7 +18,6 @@ import { ToolApprovalService, } from '../../../services/tool-approval/tool-approval.service'; import { ErrorService } from '../../../services/error/error.service'; -import { StreamParserService } from './stream-parser.service'; import { SystemPromptsService } from '../../../services/system-prompts/system-prompts.service'; import { HttpErrorResponse } from '@angular/common/http'; @@ -46,7 +45,6 @@ export class ChatRequestService implements OnDestroy { private fileUploadService = inject(FileUploadService); private oauthConsentService = inject(OAuthConsentService); private toolApprovalService = inject(ToolApprovalService); - private streamParserService = inject(StreamParserService); private errorService = inject(ErrorService); private systemPromptsService = inject(SystemPromptsService); private router = inject(Router); @@ -74,12 +72,18 @@ export class ChatRequestService implements OnDestroy { ): Promise { // Ensure conversation exists and get its ID // Update URL to reflect current conversation + const isNewSession = !sessionId; + sessionId = sessionId || uuidv4(); + // Any new send (including a "Continue") retires the previous turn's // max_tokens "Continue" affordance immediately, before the stream starts. - this.chatStateService.setLastTurnContinuable(false); + this.chatStateService.setLastTurnContinuable(sessionId, false); - const isNewSession = !sessionId; - sessionId = sessionId || uuidv4(); + // We're about to navigate to this session; point the viewed-session + // facades at it eagerly so the composer's loading state flips before + // the (async) route change lands. + this.chatStateService.setViewedSession(sessionId); + this.chatStateService.setChatLoading(sessionId, true); // If this is a new session, add it to the session cache optimistically // IMPORTANT: This must happen BEFORE navigation to prevent a race condition @@ -111,21 +115,22 @@ export class ChatRequestService implements OnDestroy { // Start streaming for this conversation this.messageMapService.startStreaming(sessionId); - // Build and send request with file upload IDs and assistant ID - const requestObject = this.buildChatRequestObject( - userInput, - sessionId, - fileUploadIds, - assistantId, - ); - try { + // Build and send request with file upload IDs and assistant ID. + // Built inside the try so a synchronous failure (e.g. no model + // selected) still clears this session's loading state. + const requestObject = this.buildChatRequestObject( + userInput, + sessionId, + fileUploadIds, + assistantId, + ); await this.chatHttpService.sendChatRequest(requestObject); } catch (error) { // TODO: Replace with proper logging service // logger.error('Chat request failed', { error, conversationId: sessionId }); - this.chatStateService.setChatLoading(false); - this.messageMapService.endStreaming(); + this.chatStateService.setChatLoading(sessionId, false); + this.messageMapService.endStreaming(sessionId); throw error; // Re-throw to allow caller to handle } } @@ -148,7 +153,7 @@ export class ChatRequestService implements OnDestroy { } // Hide the affordance immediately; retire any stale continuable state. - this.chatStateService.setLastTurnContinuable(false); + this.chatStateService.setLastTurnContinuable(sessionId, false); // Continuation streaming: pins the existing messages (history + // truncated partial + error bubble) as a stable prefix and appends the @@ -157,21 +162,21 @@ export class ChatRequestService implements OnDestroy { // resets the parser (with the correct starting count) so the resumed // stream is treated as a fresh batch. this.messageMapService.beginContinuationStreaming(sessionId); - this.chatStateService.createNewAbortController(); - this.chatStateService.setChatLoading(true); - - // Reuse the normal request shape so the backend rebuilds the same - // model/tools/assistant agent, but with an empty message and the - // continuation flag. No addUserMessage call β†’ no user bubble. - const requestObject = this.buildChatRequestObject('', sessionId, undefined, assistantId); - requestObject['message'] = ''; - requestObject['continue_truncated'] = true; + this.chatStateService.setChatLoading(sessionId, true); try { + // Reuse the normal request shape so the backend rebuilds the same + // model/tools/assistant agent, but with an empty message and the + // continuation flag. No addUserMessage call β†’ no user bubble. Built + // inside the try so a synchronous failure clears loading state. + const requestObject = this.buildChatRequestObject('', sessionId, undefined, assistantId); + requestObject['message'] = ''; + requestObject['continue_truncated'] = true; + await this.chatHttpService.sendChatRequest(requestObject); } catch (error) { - this.chatStateService.setChatLoading(false); - this.messageMapService.endStreaming(); + this.chatStateService.setChatLoading(sessionId, false); + this.messageMapService.endStreaming(sessionId); throw error; } } @@ -288,13 +293,22 @@ export class ChatRequestService implements OnDestroy { return; } - // Reset the parser so the resumed stream is treated as a fresh batch - // of events. Without this, the parser stays in Completed state from - // the prior `done` and ignores everything. - this.streamParserService.reset(sessionId); - this.messageMapService.startStreaming(sessionId); - this.chatStateService.createNewAbortController(); - this.chatStateService.setChatLoading(true); + // A resume turn has NO new user message and the resumed stream does not + // replay the interrupted `tool_use` block (Strands emits only the + // `tool_result` + the final assistant text). Continuation streaming pins + // the existing messages β€” including the assistant message holding the + // paused tool card β€” as a stable prefix and appends the resume after + // them, instead of the normal sync which truncates back to the last user + // message and would discard the tool card. It also resets the parser + // (with the correct starting count) so the resumed stream is a fresh + // batch; without that the parser stays Completed from the prior `done` + // and ignores everything. + // + // Loading is keyed to the resumed session β€” the user may have navigated + // to a different conversation before completing the consent popup, and + // the resume must not hijack that conversation's composer. + this.messageMapService.beginContinuationStreaming(sessionId); + this.chatStateService.setChatLoading(sessionId, true); const resumeRequest: Record = { session_id: sessionId, @@ -313,9 +327,14 @@ export class ChatRequestService implements OnDestroy { try { await this.chatHttpService.sendChatRequest(resumeRequest); + // The live parser could not attach the resumed `tool_result` to the + // paused tool card (its `tool_use` block is in the pinned prefix, not + // in the fresh parser). Reconcile from persisted memory so the card + // flips from "Running…" to its completed result. + await this.messageMapService.reloadMessagesForSession(sessionId); } catch (error) { - this.chatStateService.setChatLoading(false); - this.messageMapService.endStreaming(); + this.chatStateService.setChatLoading(sessionId, false); + this.messageMapService.endStreaming(sessionId); // 400 from the resume route means either the persisted snapshot is // missing/expired, or the agent's `_interrupt_state` doesn't recognize @@ -346,10 +365,13 @@ export class ChatRequestService implements OnDestroy { return; } - this.streamParserService.reset(sessionId); - this.messageMapService.startStreaming(sessionId); - this.chatStateService.createNewAbortController(); - this.chatStateService.setChatLoading(true); + // Same shape as the OAuth resume: no new user message and the resumed + // stream carries only the `tool_result` + final text, not the paused + // `tool_use` block. Pin the existing messages (with the tool card) as a + // prefix and append the resume after them. Loading keyed to the resumed + // session (see resumeFromOAuthConsent). + this.messageMapService.beginContinuationStreaming(sessionId); + this.chatStateService.setChatLoading(sessionId, true); const resumeRequest: Record = { session_id: sessionId, @@ -364,9 +386,12 @@ export class ChatRequestService implements OnDestroy { try { await this.chatHttpService.sendChatRequest(resumeRequest); + // Reconcile from persisted memory so the approved/declined tool card + // shows its result (the live parser can't attach it β€” see above). + await this.messageMapService.reloadMessagesForSession(sessionId); } catch (error) { - this.chatStateService.setChatLoading(false); - this.messageMapService.endStreaming(); + this.chatStateService.setChatLoading(sessionId, false); + this.messageMapService.endStreaming(sessionId); if (this.isExpiredInterruptError(error)) { this.errorService.addError( diff --git a/frontend/ai.client/src/app/session/services/chat/chat-state.service.spec.ts b/frontend/ai.client/src/app/session/services/chat/chat-state.service.spec.ts index 7734e101..68aa5226 100644 --- a/frontend/ai.client/src/app/session/services/chat/chat-state.service.spec.ts +++ b/frontend/ai.client/src/app/session/services/chat/chat-state.service.spec.ts @@ -15,45 +15,99 @@ describe('ChatStateService', () => { TestBed.resetTestingModule(); }); - describe('setChatLoading', () => { - it('should set loading state to true', () => { - service.setChatLoading(true); + describe('viewed-session facades', () => { + it('default to inert values when no session is viewed', () => { + expect(service.viewedSessionId()).toBeNull(); + expect(service.isChatLoading()).toBe(false); + expect(service.currentStopReason()).toBeNull(); + expect(service.lastTurnContinuable()).toBe(false); + expect(service.costDollars()).toBe(0); + expect(service.contextTokens()).toBe(0); + expect(service.contextPct()).toBe(0); + }); + + it('project the viewed session state and follow view changes', () => { + service.setChatLoading('a', true); + service.setStopReason('a', 'end_turn'); + + service.setViewedSession('a'); expect(service.isChatLoading()).toBe(true); + expect(service.currentStopReason()).toBe('end_turn'); + + // Session B has its own untouched state. + service.setViewedSession('b'); + expect(service.isChatLoading()).toBe(false); + expect(service.currentStopReason()).toBeNull(); }); - it('should set loading state to false', () => { - service.setChatLoading(false); + it('react to state created after the session is viewed', () => { + service.setViewedSession('fresh'); expect(service.isChatLoading()).toBe(false); + + service.setChatLoading('fresh', true); + expect(service.isChatLoading()).toBe(true); }); }); - describe('setStopReason', () => { - it('should set stop reason', () => { - service.setStopReason('max_tokens'); - expect(service.currentStopReason()).toBe('max_tokens'); - }); + describe('setChatLoading', () => { + it('is isolated per session', () => { + service.setChatLoading('a', true); + service.setChatLoading('b', true); + service.setChatLoading('b', false); - it('should clear stop reason with null', () => { - service.setStopReason('stop'); - service.setStopReason(null); - expect(service.currentStopReason()).toBeNull(); + expect(service.isSessionLoading('a')).toBe(true); + expect(service.isSessionLoading('b')).toBe(false); }); }); describe('setLastTurnContinuable', () => { - it('defaults to false', () => { - expect(service.lastTurnContinuable()).toBe(false); - }); + it('toggles the continuable flag per session', () => { + service.setViewedSession('a'); + service.setLastTurnContinuable('a', true); + expect(service.lastTurnContinuable()).toBe(true); - it('toggles the continuable flag', () => { - service.setLastTurnContinuable(true); + // Another session's flag doesn't affect the viewed one. + service.setLastTurnContinuable('b', false); expect(service.lastTurnContinuable()).toBe(true); - service.setLastTurnContinuable(false); + service.setLastTurnContinuable('a', false); expect(service.lastTurnContinuable()).toBe(false); }); }); + describe('cost / context aggregates', () => { + it('seeds, accumulates, and isolates per session', () => { + service.setViewedSession('a'); + service.seedSessionAggregates('a', { + totalCost: 1.5, + lastContextTokens: 1000, + contextWindow: 200000, + }); + expect(service.costDollars()).toBe(1.5); + expect(service.contextTokens()).toBe(1000); + expect(service.contextWindowSize()).toBe(200000); + + service.addTurnCost('a', 0.5); + expect(service.costDollars()).toBe(2); + + // A background session's turn cost must not leak into the viewed badge. + service.addTurnCost('b', 10); + expect(service.costDollars()).toBe(2); + + service.setContext('a', 3000, 200000); + expect(service.contextTokens()).toBe(3000); + expect(service.contextPct()).toBeCloseTo(1.5); + }); + + it('ignores non-finite or non-positive turn costs', () => { + service.setViewedSession('a'); + service.addTurnCost('a', NaN); + service.addTurnCost('a', -1); + service.addTurnCost('a', 0); + expect(service.costDollars()).toBe(0); + }); + }); + describe('requestScrollToLastUser', () => { it('starts at 0 and increments the tick on each request', () => { expect(service.scrollToLastUserTick()).toBe(0); @@ -64,50 +118,37 @@ describe('ChatStateService', () => { }); }); - describe('resetState', () => { - it('should reset all state to initial values', () => { - service.setChatLoading(true); - service.setStopReason('stop'); - service.setLastTurnContinuable(true); + describe('abort controllers', () => { + it('creates a fresh controller per session', () => { + const a = service.createAbortController('a'); + const b = service.createAbortController('b'); - service.resetState(); - - expect(service.isChatLoading()).toBe(false); - expect(service.currentStopReason()).toBeNull(); - expect(service.lastTurnContinuable()).toBe(false); + expect(a).toBeInstanceOf(AbortController); + expect(a).not.toBe(b); + expect(a.signal.aborted).toBe(false); + expect(b.signal.aborted).toBe(false); }); - }); - describe('getAbortController', () => { - it('should return current abort controller', () => { - const controller = service.getAbortController(); - expect(controller).toBeInstanceOf(AbortController); - expect(controller.signal.aborted).toBe(false); + it('aborts the previous in-flight controller for the SAME session (double-submit guard)', () => { + const first = service.createAbortController('a'); + const second = service.createAbortController('a'); + + expect(first.signal.aborted).toBe(true); + expect(second.signal.aborted).toBe(false); }); - }); - describe('createNewAbortController', () => { - it('should create and return new abort controller', () => { - const oldController = service.getAbortController(); - const newController = service.createNewAbortController(); - - expect(newController).toBeInstanceOf(AbortController); - expect(newController).not.toBe(oldController); - expect(service.getAbortController()).toBe(newController); + it('abortRequest only aborts the target session', () => { + const a = service.createAbortController('a'); + const b = service.createAbortController('b'); + + service.abortRequest('b'); + + expect(a.signal.aborted).toBe(false); + expect(b.signal.aborted).toBe(true); }); - }); - describe('abortCurrentRequest', () => { - it('should abort current controller and create new one', () => { - const oldController = service.getAbortController(); - - service.abortCurrentRequest(); - - expect(oldController.signal.aborted).toBe(true); - - const newController = service.getAbortController(); - expect(newController).not.toBe(oldController); - expect(newController.signal.aborted).toBe(false); + it('abortRequest is a no-op for sessions without an in-flight request', () => { + expect(() => service.abortRequest('nope')).not.toThrow(); }); }); -}); \ No newline at end of file +}); diff --git a/frontend/ai.client/src/app/session/services/chat/chat-state.service.ts b/frontend/ai.client/src/app/session/services/chat/chat-state.service.ts index 5f994554..8069bc05 100644 --- a/frontend/ai.client/src/app/session/services/chat/chat-state.service.ts +++ b/frontend/ai.client/src/app/session/services/chat/chat-state.service.ts @@ -1,58 +1,85 @@ -import { Injectable, Signal, computed, signal } from '@angular/core'; +import { Injectable, Signal, WritableSignal, computed, signal } from '@angular/core'; + +/** + * Mutable chat state for one conversation. Every stream-scoped flag lives + * here, keyed by session, so two conversations streaming concurrently can + * never clobber each other's loading spinner, cost badge, Stop button, or + * "Continue" affordance. + */ +interface SessionChatState { + loading: WritableSignal; + stopReason: WritableSignal; + lastTurnContinuable: WritableSignal; + costDollars: WritableSignal; + contextTokens: WritableSignal; + contextWindow: WritableSignal; + /** In-flight SSE request controller, one per session. */ + abortController: AbortController | null; +} @Injectable({ providedIn: 'root' }) export class ChatStateService { - private abortController = new AbortController(); - private readonly chatLoading = signal(false); - readonly isChatLoading: Signal = this.chatLoading.asReadonly(); - - // Bumped to ask the message list to scroll the latest user message to the - // top of the viewport. Lets non-composer submit paths (e.g. an MCP App - // widget's ui/message) get the same scroll affordance the composer - // triggers in ChatContainerComponent.onMessageSubmitted. - private readonly scrollToLastUserSignal = signal(0); - readonly scrollToLastUserTick: Signal = this.scrollToLastUserSignal.asReadonly(); + /** + * Per-session state, held in a signal so the viewed-session facades + * below recompute when a session's state is lazily created. + */ + private readonly states = signal>(new Map()); - private readonly stopReason = signal(null); - readonly currentStopReason: Signal = this.stopReason.asReadonly(); + /** + * The session the user is currently looking at. Set by the session page + * on route change (and eagerly by ChatRequestService when it creates a + * new session, so the composer spinner doesn't wait on navigation). + * The readonly facades below project this session's state, which keeps + * existing consumers (composer, cost badge, Continue affordance) + * unchanged while the underlying state is per-session. + */ + private readonly viewedSessionIdSignal = signal(null); + readonly viewedSessionId: Signal = this.viewedSessionIdSignal.asReadonly(); - // True when the most recent turn ended in a recoverable max_tokens - // truncation. Drives the "Continue" affordance on the last assistant - // message. Live-only (not hydrated on reload); set from the stream_error - // event and cleared the moment a new turn starts. - private readonly lastTurnContinuableSignal = signal(false); - readonly lastTurnContinuable: Signal = this.lastTurnContinuableSignal.asReadonly(); + readonly isChatLoading = computed(() => this.viewedState()?.loading() ?? false); + readonly currentStopReason = computed(() => this.viewedState()?.stopReason() ?? null); + readonly lastTurnContinuable = computed(() => this.viewedState()?.lastTurnContinuable() ?? false); // ----- Session-level cost / context aggregates --------------------------- // Drive the cost badge above the composer. Seeded from session metadata // on route change, then incrementally updated via the SSE metadata event // each turn (addTurnCost / setContext). - private readonly costDollarsSignal = signal(0); - readonly costDollars: Signal = this.costDollarsSignal.asReadonly(); - - private readonly contextTokensSignal = signal(0); - readonly contextTokens: Signal = this.contextTokensSignal.asReadonly(); - - private readonly contextWindowSignal = signal(0); - readonly contextWindowSize: Signal = this.contextWindowSignal.asReadonly(); + readonly costDollars = computed(() => this.viewedState()?.costDollars() ?? 0); + readonly contextTokens = computed(() => this.viewedState()?.contextTokens() ?? 0); + readonly contextWindowSize = computed(() => this.viewedState()?.contextWindow() ?? 0); readonly contextPct = computed(() => { - const window = this.contextWindowSignal(); - const tokens = this.contextTokensSignal(); + const window = this.contextWindowSize(); + const tokens = this.contextTokens(); if (!window || window <= 0) return 0; return (tokens / window) * 100; }); + // Bumped to ask the message list to scroll the latest user message to the + // top of the viewport. Lets non-composer submit paths (e.g. an MCP App + // widget's ui/message) get the same scroll affordance the composer + // triggers in ChatContainerComponent.onMessageSubmitted. + private readonly scrollToLastUserSignal = signal(0); + readonly scrollToLastUserTick: Signal = this.scrollToLastUserSignal.asReadonly(); + + /** Point the viewed-session facades at a (possibly null) session. */ + setViewedSession(sessionId: string | null): void { + this.viewedSessionIdSignal.set(sessionId); + } /** - * Sets the chat loading state - * @param loading - Whether the chat is currently loading + * Sets the chat loading state for a session. */ - setChatLoading(loading: boolean): void { - this.chatLoading.set(loading); + setChatLoading(sessionId: string, loading: boolean): void { + this.stateFor(sessionId).loading.set(loading); + } + + /** Whether a specific session is currently streaming (loading). */ + isSessionLoading(sessionId: string): boolean { + return this.states().get(sessionId)?.loading() ?? false; } /** @@ -65,78 +92,100 @@ export class ChatStateService { } /** - * Sets the stop reason for the current message - * @param reason - The stop reason string, or null to clear + * Sets the stop reason for a session's current message. */ - setStopReason(reason: string | null): void { - this.stopReason.set(reason); + setStopReason(sessionId: string, reason: string | null): void { + this.stateFor(sessionId).stopReason.set(reason); } /** - * Marks (or clears) whether the last turn ended in a recoverable + * Marks (or clears) whether a session's last turn ended in a recoverable * max_tokens truncation that the user can continue from. */ - setLastTurnContinuable(continuable: boolean): void { - this.lastTurnContinuableSignal.set(continuable); + setLastTurnContinuable(sessionId: string, continuable: boolean): void { + this.stateFor(sessionId).lastTurnContinuable.set(continuable); } /** - * Seed the session-level cost/context signals from a session metadata - * payload (e.g. when navigating to an existing session). Called BEFORE - * the new metadata loads on route change to clear stale state from the - * previous session. + * Seed a session's cost/context signals from its metadata payload (e.g. + * when navigating to an existing session). */ - seedSessionAggregates(values: { + seedSessionAggregates(sessionId: string, values: { totalCost?: number; lastContextTokens?: number; contextWindow?: number; } = {}): void { - this.costDollarsSignal.set(values.totalCost ?? 0); - this.contextTokensSignal.set(values.lastContextTokens ?? 0); - this.contextWindowSignal.set(values.contextWindow ?? 0); + const state = this.stateFor(sessionId); + state.costDollars.set(values.totalCost ?? 0); + state.contextTokens.set(values.lastContextTokens ?? 0); + state.contextWindow.set(values.contextWindow ?? 0); } - /** Add the cost of a completed turn to the running session total. */ - addTurnCost(amount: number): void { + /** Add the cost of a completed turn to a session's running total. */ + addTurnCost(sessionId: string, amount: number): void { if (!Number.isFinite(amount) || amount <= 0) return; - this.costDollarsSignal.update(prev => prev + amount); + this.stateFor(sessionId).costDollars.update(prev => prev + amount); } - /** Set the most-recent-turn context tokens (and optionally the window). */ - setContext(tokens: number, window?: number): void { + /** Set a session's most-recent-turn context tokens (and optionally the window). */ + setContext(sessionId: string, tokens: number, window?: number): void { + const state = this.stateFor(sessionId); if (Number.isFinite(tokens) && tokens >= 0) { - this.contextTokensSignal.set(tokens); + state.contextTokens.set(tokens); } if (window !== undefined && Number.isFinite(window) && window > 0) { - this.contextWindowSignal.set(window); + state.contextWindow.set(window); } } + // ----- Abort controller management --------------------------------------- + /** - * Resets all state to initial values + * Create a fresh AbortController for a session's outgoing stream. Any + * in-flight request for the SAME session is aborted first (double-submit + * guard); streams for other sessions are untouched. */ - resetState(): void { - this.chatLoading.set(false); - this.stopReason.set(null); - this.lastTurnContinuableSignal.set(false); - this.costDollarsSignal.set(0); - this.contextTokensSignal.set(0); - this.contextWindowSignal.set(0); + createAbortController(sessionId: string): AbortController { + const state = this.stateFor(sessionId); + state.abortController?.abort(); + const controller = new AbortController(); + state.abortController = controller; + return controller; } - // Abort controller management - getAbortController(): AbortController { - return this.abortController; + /** Abort a session's in-flight request (Stop button), if any. */ + abortRequest(sessionId: string): void { + const state = this.states().get(sessionId); + state?.abortController?.abort(); + if (state) { + state.abortController = null; + } } - createNewAbortController(): AbortController { - this.abortController = new AbortController(); - return this.abortController; + private viewedState(): SessionChatState | undefined { + const sessionId = this.viewedSessionIdSignal(); + return sessionId ? this.states().get(sessionId) : undefined; } - abortCurrentRequest(): void { - this.abortController.abort(); - this.abortController = new AbortController(); + /** Get (lazily creating) the state bucket for a session. */ + private stateFor(sessionId: string): SessionChatState { + const existing = this.states().get(sessionId); + if (existing) return existing; + + const created: SessionChatState = { + loading: signal(false), + stopReason: signal(null), + lastTurnContinuable: signal(false), + costDollars: signal(0), + contextTokens: signal(0), + contextWindow: signal(0), + abortController: null, + }; + this.states.update(map => { + const next = new Map(map); + next.set(sessionId, created); + return next; + }); + return created; } } - diff --git a/frontend/ai.client/src/app/session/services/chat/stream-parser.service.spec.ts b/frontend/ai.client/src/app/session/services/chat/stream-parser.service.spec.ts index 933801fb..b0f4c926 100644 --- a/frontend/ai.client/src/app/session/services/chat/stream-parser.service.spec.ts +++ b/frontend/ai.client/src/app/session/services/chat/stream-parser.service.spec.ts @@ -21,7 +21,7 @@ describe('StreamParserService - Citation Handling', () => { ], }); service = TestBed.inject(StreamParserService); - service.reset(); + service.reset('s1'); }); afterEach(() => { @@ -45,13 +45,13 @@ describe('StreamParserService - Citation Handling', () => { }), (citationData: { assistantId: string; documentId: string; fileName: string; text: string }) => { // Reset service for each iteration - service.reset(); + service.reset('s1'); // Parse citation event - service.parseEventSourceMessage('citation', citationData); + service.parseEventSourceMessage('s1', 'citation', citationData); // Get accumulated citations - const citations = service.citations(); + const citations = service.citationsFor('s1')(); // Verify citation was added expect(citations.length).toBe(1); @@ -80,13 +80,13 @@ describe('StreamParserService - Citation Handling', () => { }), (citationData: { documentId: string; fileName: string; text: string }) => { // Reset service for each iteration - service.reset(); + service.reset('s1'); // Parse citation event without assistantId - service.parseEventSourceMessage('citation', citationData); + service.parseEventSourceMessage('s1', 'citation', citationData); // Get accumulated citations - const citations = service.citations(); + const citations = service.citationsFor('s1')(); // Verify citation was NOT added (missing assistantId) expect(citations.length).toBe(0); @@ -110,9 +110,9 @@ describe('StreamParserService - Citation Handling', () => { text: 'Some text', }; - service.parseEventSourceMessage('citation', malformedCitation); + service.parseEventSourceMessage('s1', 'citation', malformedCitation); - const citations = service.citations(); + const citations = service.citationsFor('s1')(); expect(citations.length).toBe(0); }); @@ -124,9 +124,9 @@ describe('StreamParserService - Citation Handling', () => { text: 'Some text', }; - service.parseEventSourceMessage('citation', malformedCitation); + service.parseEventSourceMessage('s1', 'citation', malformedCitation); - const citations = service.citations(); + const citations = service.citationsFor('s1')(); expect(citations.length).toBe(0); }); @@ -138,9 +138,9 @@ describe('StreamParserService - Citation Handling', () => { text: 'Some text', }; - service.parseEventSourceMessage('citation', malformedCitation); + service.parseEventSourceMessage('s1', 'citation', malformedCitation); - const citations = service.citations(); + const citations = service.citationsFor('s1')(); expect(citations.length).toBe(0); }); @@ -152,9 +152,9 @@ describe('StreamParserService - Citation Handling', () => { // text missing }; - service.parseEventSourceMessage('citation', malformedCitation); + service.parseEventSourceMessage('s1', 'citation', malformedCitation); - const citations = service.citations(); + const citations = service.citationsFor('s1')(); expect(citations.length).toBe(0); }); @@ -166,9 +166,9 @@ describe('StreamParserService - Citation Handling', () => { text: 'Some text', }; - service.parseEventSourceMessage('citation', malformedCitation); + service.parseEventSourceMessage('s1', 'citation', malformedCitation); - const citations = service.citations(); + const citations = service.citationsFor('s1')(); expect(citations.length).toBe(0); }); @@ -180,9 +180,9 @@ describe('StreamParserService - Citation Handling', () => { text: 'Some text', }; - service.parseEventSourceMessage('citation', malformedCitation); + service.parseEventSourceMessage('s1', 'citation', malformedCitation); - const citations = service.citations(); + const citations = service.citationsFor('s1')(); expect(citations.length).toBe(0); }); @@ -194,9 +194,9 @@ describe('StreamParserService - Citation Handling', () => { text: 'Some text', }; - service.parseEventSourceMessage('citation', malformedCitation); + service.parseEventSourceMessage('s1', 'citation', malformedCitation); - const citations = service.citations(); + const citations = service.citationsFor('s1')(); expect(citations.length).toBe(0); }); @@ -208,30 +208,30 @@ describe('StreamParserService - Citation Handling', () => { text: { content: 'Some text' }, // object instead of string }; - service.parseEventSourceMessage('citation', malformedCitation); + service.parseEventSourceMessage('s1', 'citation', malformedCitation); - const citations = service.citations(); + const citations = service.citationsFor('s1')(); expect(citations.length).toBe(0); }); it('should skip citation with null data', () => { - service.parseEventSourceMessage('citation', null); + service.parseEventSourceMessage('s1', 'citation', null); - const citations = service.citations(); + const citations = service.citationsFor('s1')(); expect(citations.length).toBe(0); }); it('should skip citation with undefined data', () => { - service.parseEventSourceMessage('citation', undefined); + service.parseEventSourceMessage('s1', 'citation', undefined); - const citations = service.citations(); + const citations = service.citationsFor('s1')(); expect(citations.length).toBe(0); }); it('should skip citation with non-object data', () => { - service.parseEventSourceMessage('citation', 'invalid string data'); + service.parseEventSourceMessage('s1', 'citation', 'invalid string data'); - const citations = service.citations(); + const citations = service.citationsFor('s1')(); expect(citations.length).toBe(0); }); @@ -250,11 +250,11 @@ describe('StreamParserService - Citation Handling', () => { malformedCitations.forEach((malformed) => { expect(() => { - service.parseEventSourceMessage('citation', malformed); + service.parseEventSourceMessage('s1', 'citation', malformed); }).not.toThrow(); }); - const citations = service.citations(); + const citations = service.citationsFor('s1')(); expect(citations.length).toBe(0); }); @@ -264,7 +264,7 @@ describe('StreamParserService - Citation Handling', () => { documentId: 'doc-123', // missing assistantId, fileName and text }; - service.parseEventSourceMessage('citation', malformedCitation); + service.parseEventSourceMessage('s1', 'citation', malformedCitation); // Then, send valid citation const validCitation = { @@ -273,9 +273,9 @@ describe('StreamParserService - Citation Handling', () => { fileName: 'valid.pdf', text: 'Valid text', }; - service.parseEventSourceMessage('citation', validCitation); + service.parseEventSourceMessage('s1', 'citation', validCitation); - const citations = service.citations(); + const citations = service.citationsFor('s1')(); expect(citations.length).toBe(1); expect(citations[0].assistantId).toBe('assistant-1'); expect(citations[0].documentId).toBe('doc-456'); @@ -297,9 +297,9 @@ describe('StreamParserService - Citation Handling', () => { text: 'Some relevant text from the document', }; - service.parseEventSourceMessage('citation', citation); + service.parseEventSourceMessage('s1', 'citation', citation); - const citations = service.citations(); + const citations = service.citationsFor('s1')(); expect(citations.length).toBe(1); expect(citations[0].assistantId).toBe('assistant-1'); expect(citations[0].documentId).toBe('doc-123'); @@ -322,10 +322,10 @@ describe('StreamParserService - Citation Handling', () => { text: 'Text from second document', }; - service.parseEventSourceMessage('citation', citation1); - service.parseEventSourceMessage('citation', citation2); + service.parseEventSourceMessage('s1', 'citation', citation1); + service.parseEventSourceMessage('s1', 'citation', citation2); - const citations = service.citations(); + const citations = service.citationsFor('s1')(); expect(citations.length).toBe(2); expect(citations[0].documentId).toBe('doc-123'); expect(citations[0].assistantId).toBe('assistant-1'); @@ -341,11 +341,11 @@ describe('StreamParserService - Citation Handling', () => { text: 'Some text', }; - service.parseEventSourceMessage('citation', citation); - expect(service.citations().length).toBe(1); + service.parseEventSourceMessage('s1', 'citation', citation); + expect(service.citationsFor('s1')().length).toBe(1); - service.reset(); - expect(service.citations().length).toBe(0); + service.reset('s1'); + expect(service.citationsFor('s1')().length).toBe(0); }); }); }); @@ -366,7 +366,9 @@ describe('StreamParserService - max_tokens Continue affordance', () => { }); service = TestBed.inject(StreamParserService); chatState = TestBed.inject(ChatStateService); - service.reset(); + // The Continue affordance reads through the viewed-session facade. + chatState.setViewedSession('s1'); + service.reset('s1'); }); afterEach(() => { @@ -376,7 +378,7 @@ describe('StreamParserService - max_tokens Continue affordance', () => { it('marks the last turn continuable on a max_tokens stream_error', () => { expect(chatState.lastTurnContinuable()).toBe(false); - service.parseEventSourceMessage('stream_error', { + service.parseEventSourceMessage('s1', 'stream_error', { type: 'stream_error', code: 'max_tokens', message: 'I reached my response-length limit.', @@ -388,7 +390,7 @@ describe('StreamParserService - max_tokens Continue affordance', () => { }); it('does not mark continuable for a non-max_tokens stream_error', () => { - service.parseEventSourceMessage('stream_error', { + service.parseEventSourceMessage('s1', 'stream_error', { type: 'stream_error', code: 'stream_error', message: 'Something went wrong.', @@ -399,7 +401,7 @@ describe('StreamParserService - max_tokens Continue affordance', () => { }); it('retires the affordance when the next assistant turn starts streaming', () => { - service.parseEventSourceMessage('stream_error', { + service.parseEventSourceMessage('s1', 'stream_error', { type: 'stream_error', code: 'max_tokens', message: 'truncated', @@ -408,7 +410,7 @@ describe('StreamParserService - max_tokens Continue affordance', () => { }); expect(chatState.lastTurnContinuable()).toBe(true); - service.parseEventSourceMessage('message_start', { role: 'assistant' }); + service.parseEventSourceMessage('s1', 'message_start', { role: 'assistant' }); expect(chatState.lastTurnContinuable()).toBe(false); }); @@ -417,11 +419,11 @@ describe('StreamParserService - max_tokens Continue affordance', () => { // terminal state (message_start sets currentStreamId; done β†’ // Completed), then the max_tokens stream_error arrives last. It must // still be processed (always-allowed) so Continue appears. - service.parseEventSourceMessage('message_start', { role: 'assistant' }); - service.parseEventSourceMessage('done', null); + service.parseEventSourceMessage('s1', 'message_start', { role: 'assistant' }); + service.parseEventSourceMessage('s1', 'done', null); expect(chatState.lastTurnContinuable()).toBe(false); - service.parseEventSourceMessage('stream_error', { + service.parseEventSourceMessage('s1', 'stream_error', { type: 'stream_error', code: 'max_tokens', message: 'Response length limit reached.', @@ -431,4 +433,161 @@ describe('StreamParserService - max_tokens Continue affordance', () => { expect(chatState.lastTurnContinuable()).toBe(true); }); + + it('keys the affordance to the stream session, not the viewed one', () => { + // A background stream (session s2) truncating must not surface + // Continue on the viewed conversation (s1). + service.reset('s2'); + service.parseEventSourceMessage('s2', 'stream_error', { + type: 'stream_error', + code: 'max_tokens', + message: 'truncated', + recoverable: true, + metadata: { error_kind: 'max_tokens' }, + }); + + expect(chatState.lastTurnContinuable()).toBe(false); + + chatState.setViewedSession('s2'); + expect(chatState.lastTurnContinuable()).toBe(true); + }); +}); + +describe('StreamParserService - concurrent session isolation', () => { + let service: StreamParserService; + + const streamText = (sessionId: string, text: string) => { + service.parseEventSourceMessage(sessionId, 'message_start', { role: 'assistant' }); + service.parseEventSourceMessage(sessionId, 'content_block_delta', { + contentBlockIndex: 0, + text, + }); + }; + + beforeEach(() => { + TestBed.resetTestingModule(); + TestBed.configureTestingModule({ + providers: [ + StreamParserService, + ChatStateService, + ErrorService, + QuotaWarningService, + ], + }); + service = TestBed.inject(StreamParserService); + }); + + afterEach(() => { + TestBed.resetTestingModule(); + }); + + it('parses two interleaved streams into separate per-session state', () => { + service.reset('a', 1); + service.reset('b', 1); + + streamText('a', 'alpha'); + streamText('b', 'bravo'); + service.parseEventSourceMessage('a', 'content_block_delta', { + contentBlockIndex: 0, + text: '-more', + }); + + const aMessages = service.allMessagesFor('a')(); + const bMessages = service.allMessagesFor('b')(); + + expect(aMessages).toHaveLength(1); + expect(aMessages[0].id).toBe('msg-a-1'); + expect(aMessages[0].content).toEqual([{ type: 'text', text: 'alpha-more' }]); + + expect(bMessages).toHaveLength(1); + expect(bMessages[0].id).toBe('msg-b-1'); + expect(bMessages[0].content).toEqual([{ type: 'text', text: 'bravo' }]); + }); + + it('keeps session A streaming after session B resets (route change / new stream)', () => { + service.reset('a'); + streamText('a', 'hello'); + + // Session B starting a stream must not disturb A's in-flight parse. + service.reset('b'); + streamText('b', 'other'); + + service.parseEventSourceMessage('a', 'content_block_delta', { + contentBlockIndex: 0, + text: ' world', + }); + + expect(service.allMessagesFor('a')()[0].content).toEqual([ + { type: 'text', text: 'hello world' }, + ]); + expect(service.streamingMessageIdFor('a')()).toBe('msg-a-0'); + }); + + it('drops late events carrying a superseded stream id (same-session resubmit)', () => { + service.reset('a'); + const staleStreamId = service.getCurrentStreamId('a'); + streamText('a', 'first attempt'); + + // A re-submit resets the session's parser; the old stream's late + // events identify themselves with the captured (now stale) stream id. + service.reset('a'); + expect(service.getCurrentStreamId('a')).not.toBe(staleStreamId); + + service.parseEventSourceMessage( + 'a', + 'message_start', + { role: 'assistant' }, + staleStreamId, + ); + service.parseEventSourceMessage( + 'a', + 'content_block_delta', + { contentBlockIndex: 0, text: 'stale text' }, + staleStreamId, + ); + + expect(service.allMessagesFor('a')()).toHaveLength(0); + + // The replacement stream still parses normally with its own id. + const freshStreamId = service.getCurrentStreamId('a'); + service.parseEventSourceMessage('a', 'message_start', { role: 'assistant' }, freshStreamId); + service.parseEventSourceMessage( + 'a', + 'content_block_delta', + { contentBlockIndex: 0, text: 'fresh text' }, + freshStreamId, + ); + expect(service.allMessagesFor('a')()[0].content).toEqual([ + { type: 'text', text: 'fresh text' }, + ]); + }); + + it('ignores events for a session with no parser state', () => { + expect(() => + service.parseEventSourceMessage('ghost', 'message_start', { role: 'assistant' }), + ).not.toThrow(); + expect(service.allMessagesFor('ghost')()).toEqual([]); + }); + + it('routes cost metadata to the stream session in ChatStateService', () => { + const chatState = TestBed.inject(ChatStateService); + chatState.setViewedSession('a'); + service.reset('a'); + service.reset('b'); + + // Background session B finishes a turn while A is viewed. + service.parseEventSourceMessage('b', 'message_start', { role: 'assistant' }); + service.parseEventSourceMessage('b', 'metadata', { + usage: { inputTokens: 100, outputTokens: 50, totalTokens: 150 }, + cost: 0.25, + contextWindow: 200000, + }); + + // Viewed badge (session A) is untouched... + expect(chatState.costDollars()).toBe(0); + // ...and B's own aggregates carry the turn. + chatState.setViewedSession('b'); + expect(chatState.costDollars()).toBe(0.25); + expect(chatState.contextTokens()).toBe(100); + }); }); diff --git a/frontend/ai.client/src/app/session/services/chat/stream-parser.service.ts b/frontend/ai.client/src/app/session/services/chat/stream-parser.service.ts index e8e1b776..565dbee4 100644 --- a/frontend/ai.client/src/app/session/services/chat/stream-parser.service.ts +++ b/frontend/ai.client/src/app/session/services/chat/stream-parser.service.ts @@ -1,5 +1,5 @@ // services/stream-parser.service.ts -import { Injectable, signal, computed, inject } from '@angular/core'; +import { Injectable, Signal, WritableSignal, signal, computed, inject } from '@angular/core'; import { Message, ContentBlock, Citation } from '../models/message.model'; import { MetadataEvent } from '../models/content-types'; import { ChatStateService } from './chat-state.service'; @@ -62,160 +62,178 @@ export type { ToolProgress }; */ const STREAMING_CONTENT_TOOLS = new Set(['create_artifact', 'update_artifact']); -@Injectable({ - providedIn: 'root', -}) -export class StreamParserService { - private chatStateService = inject(ChatStateService); - private errorService = inject(ErrorService); - private quotaWarningService = inject(QuotaWarningService); - private oauthConsentService = inject(OAuthConsentService); - private toolApprovalService = inject(ToolApprovalService); - private compactionSummary = inject(CompactionSummaryService); - private artifactState = inject(ArtifactStateService); - private mcpAppState = inject(McpAppStateService); +/** + * All mutable parse state for one session's stream. Each session gets its + * own instance so two conversations can stream concurrently without one + * stream's events corrupting the other's message builders or identity. + */ +interface ParserSessionState { + /** Session this state belongs to (used for message IDs and side channels). */ + sessionId: string; - // ========================================================================= - // State Signals - // ========================================================================= + /** Starting message count for ID computation */ + startingMessageCount: number; + + /** + * Current stream ID. Regenerated on every reset β€” a stream captures it at + * start and passes it back with each event so late events from a + * superseded stream (same session, e.g. a re-submit) are dropped. + */ + currentStreamId: string; + + /** Current stream state */ + streamState: StreamState; /** The current message being streamed */ - private currentMessageBuilder = signal(null); + currentMessageBuilder: WritableSignal; /** Completed messages in the current turn (for multi-turn tool use) */ - private completedMessages = signal([]); + completedMessages: WritableSignal; /** Tool progress indicator state */ - private toolProgressSignal = signal({ visible: false }); - public toolProgress = this.toolProgressSignal.asReadonly(); + toolProgress: WritableSignal; /** Error state */ - private errorSignal = signal(null); - public error = this.errorSignal.asReadonly(); + error: WritableSignal; /** Stream completion state */ - private isStreamCompleteSignal = signal(false); - public isStreamComplete = this.isStreamCompleteSignal.asReadonly(); + isStreamComplete: WritableSignal; /** Metadata (usage, metrics) from the stream */ - private metadataSignal = signal(null); + metadata: WritableSignal; /** Pending citations for the next assistant message */ - private pendingCitations = signal([]); - public citations = this.pendingCitations.asReadonly(); - public metadata = this.metadataSignal.asReadonly(); + pendingCitations: WritableSignal; - // ========================================================================= - // Message ID Computation State - // ========================================================================= + /** The current message converted to the final Message format. */ + currentMessage: Signal; - /** Session ID for computing message IDs */ - private sessionId: string | null = null; + /** All messages in this stream (completed + current). */ + allMessages: Signal; - /** Starting message count for ID computation */ - private startingMessageCount: number = 0; + /** The ID of the message currently being streamed, or null. */ + streamingMessageId: Signal; + + /** Callbacks wiring the pure parsing logic to this state's signals. */ + callbacks: StreamParserCallbacks; + + /** Line parser for raw SSE lines */ + lineParser: ReturnType; +} + +@Injectable({ + providedIn: 'root', +}) +export class StreamParserService { + private chatStateService = inject(ChatStateService); + private errorService = inject(ErrorService); + private quotaWarningService = inject(QuotaWarningService); + private oauthConsentService = inject(OAuthConsentService); + private toolApprovalService = inject(ToolApprovalService); + private compactionSummary = inject(CompactionSummaryService); + private artifactState = inject(ArtifactStateService); + private mcpAppState = inject(McpAppStateService); // ========================================================================= - // Computed Signals - Reactive Derived State + // Per-Session State // ========================================================================= /** - * The current message converted to the final Message format. - * Efficiently rebuilds only when the builder changes. + * Parser state per session, held in a signal so the cached per-session + * accessor computeds below re-read when reset() swaps in a fresh state. */ - public currentMessage = computed(() => { - const builder = this.currentMessageBuilder(); - return builder ? this.buildMessage(builder) : null; - }); + private readonly states = signal>(new Map()); - /** - * All messages in the current streaming session (completed + current). - * This is what the UI should bind to for rendering. - */ - public allMessages = computed(() => { - const completed = this.completedMessages(); - const current = this.currentMessage(); - return current ? [...completed, current] : completed; - }); + /** Stable per-session accessor signals (cached so callers can hold them). */ + private readonly allMessagesCache = new Map>(); + private readonly streamingMessageIdCache = new Map>(); + private readonly toolProgressCache = new Map>(); + private readonly citationsCache = new Map>(); + private readonly errorCache = new Map>(); + private readonly isStreamCompleteCache = new Map>(); - /** - * The latest message's text content as a single string. - * Useful for simple text displays. - */ - public currentText = computed(() => { - const message = this.currentMessage(); - if (!message) return ''; - - return message.content - .filter((block) => block.type === 'text' && block.text) - .map((block) => block.text!) - .join(''); - }); + // ========================================================================= + // Public API + // ========================================================================= /** - * Whether we're currently in the middle of a tool use cycle. + * All messages in a session's current streaming batch (completed + + * current). This is what the message-map sync binds to for rendering. + * Returns a stable signal per session; empty until the first reset(). */ - public isToolUseInProgress = computed(() => { - const builder = this.currentMessageBuilder(); - if (!builder) return false; - - return Array.from(builder.contentBlocks.values()).some( - (block) => (block.type === 'toolUse' || block.type === 'tool_use') && !block.isComplete, - ); - }); + allMessagesFor(sessionId: string): Signal { + return this.cachedAccessor(this.allMessagesCache, sessionId, (state) => state.allMessages(), []); + } /** - * The ID of the message currently being streamed, or null if not streaming. + * The ID of the message currently being streamed in a session, or null. * Used by UI components to determine which message should animate. */ - public streamingMessageId = computed(() => { - const builder = this.currentMessageBuilder(); - const isComplete = this.isStreamCompleteSignal(); - - // Return the message ID if we have an active builder and stream is not complete - if (builder && !isComplete) { - return builder.id; - } - return null; - }); - - // ========================================================================= - // Private State - // ========================================================================= + streamingMessageIdFor(sessionId: string): Signal { + return this.cachedAccessor(this.streamingMessageIdCache, sessionId, (state) => state.streamingMessageId(), null); + } - /** Current stream ID - prevents race conditions from overlapping streams */ - private currentStreamId: string | null = null; + /** Tool progress indicator state for a session. */ + toolProgressFor(sessionId: string): Signal { + return this.cachedAccessor(this.toolProgressCache, sessionId, (state) => state.toolProgress(), { visible: false }); + } - /** Current stream state */ - private streamState: StreamState = StreamState.Idle; + /** Pending citations for a session's next assistant message. */ + citationsFor(sessionId: string): Signal { + return this.cachedAccessor(this.citationsCache, sessionId, (state) => state.pendingCitations(), []); + } - /** Line parser for raw SSE lines */ - private lineParser = createStreamLineParser(this.createCallbacks()); + /** Parse-error state for a session. */ + errorFor(sessionId: string): Signal { + return this.cachedAccessor(this.errorCache, sessionId, (state) => state.error(), null); + } - // ========================================================================= - // Public API - // ========================================================================= + /** Stream completion state for a session. */ + isStreamCompleteFor(sessionId: string): Signal { + return this.cachedAccessor(this.isStreamCompleteCache, sessionId, (state) => state.isStreamComplete(), false); + } /** - * Parse an incoming SSE line and update state. + * Parse an incoming SSE line for a session and update its state. * Handles the event: and data: format from SSE. */ - parseSSELine(line: string): void { - // Check if we should process events - if (!this.shouldProcessEvent()) { + parseSSELine(sessionId: string, line: string): void { + const state = this.states().get(sessionId); + if (!state || !this.shouldProcessEvent(state)) { return; } - this.lineParser.parseLine(line); + state.lineParser.parseLine(line); } /** * Parse a pre-parsed EventSourceMessage (from fetch-event-source). + * + * @param sessionId - The session whose stream produced the event + * @param expectedStreamId - The stream ID captured when the request + * started (see {@link getCurrentStreamId}). If the session's parser has + * since been reset for a newer stream, the event is silently dropped β€” + * this is what keeps a superseded stream's late events from corrupting + * its replacement. */ - parseEventSourceMessage(event: string, data: unknown): void { + parseEventSourceMessage( + sessionId: string, + event: string, + data: unknown, + expectedStreamId?: string | null, + ): void { + const state = this.states().get(sessionId); + if (!state) { + return; // No active parser for this session β€” nothing to update. + } + + if (expectedStreamId && state.currentStreamId !== expectedStreamId) { + return; // Stale event from a superseded stream for this session. + } + // Validate inputs if (!event || typeof event !== 'string') { - this.setError('parseEventSourceMessage: event must be a non-empty string'); + this.setError(state, 'parseEventSourceMessage: event must be a non-empty string'); return; } @@ -230,96 +248,185 @@ export class StreamParserService { event === 'error' || event === 'oauth_required' || event === 'stream_error'; - if (!isAlwaysAllowedEvent && !this.shouldProcessEvent()) { + if (!isAlwaysAllowedEvent && !this.shouldProcessEvent(state)) { return; } // Special handling for 'done' event which may have null/undefined data if (data === undefined || data === null) { if (event === 'done') { - processStreamEvent(event, data, this.createCallbacks()); + processStreamEvent(event, data, state.callbacks); return; } - this.setError(`parseEventSourceMessage: data cannot be null/undefined for event '${event}'`); + this.setError(state, `parseEventSourceMessage: data cannot be null/undefined for event '${event}'`); return; } - processStreamEvent(event, data, this.createCallbacks()); + processStreamEvent(event, data, state.callbacks); } /** - * Reset all state for a new conversation/stream. + * Reset a session's state for a new stream. * Generates a new stream ID to prevent race conditions. * - * IMPORTANT: Call this before starting a new stream to prevent - * events from previous streams from interfering. + * IMPORTANT: Call this before starting a new stream so events from that + * session's previous stream (identified by their captured stream ID) are + * dropped instead of interfering. * - * @param sessionId - Session ID for computing predictable message IDs + * @param sessionId - Session the stream belongs to (also used for computing + * predictable message IDs) * @param startingMessageCount - Current message count in the session (for ID computation) */ - reset(sessionId?: string, startingMessageCount?: number): void { - // Generate new stream ID to prevent events from old streams - this.currentStreamId = uuidv4(); - this.streamState = StreamState.Idle; - - // Store session ID and message count for predictable ID generation - this.sessionId = sessionId || null; - this.startingMessageCount = startingMessageCount || 0; - - // Clear all state - this.currentMessageBuilder.set(null); - this.completedMessages.set([]); - this.toolProgressSignal.set({ visible: false }); - this.errorSignal.set(null); - this.isStreamCompleteSignal.set(false); - this.metadataSignal.set(null); - this.pendingCitations.set([]); - - // Reset line parser - this.lineParser.reset(); + reset(sessionId: string, startingMessageCount?: number): void { + const state = this.createState(sessionId, startingMessageCount || 0); + this.states.update((map) => { + const next = new Map(map); + next.set(sessionId, state); + return next; + }); } /** - * Get the current stream ID (for debugging/monitoring). + * Get a session's current stream ID. Captured by ChatHttpService when a + * request starts, and passed back with each event / lifecycle callback to + * detect stale streams. */ - getCurrentStreamId(): string | null { - return this.currentStreamId; + getCurrentStreamId(sessionId: string): string | null { + return this.states().get(sessionId)?.currentStreamId ?? null; } /** - * Get completed messages and clear them (for persisting to backend). + * Get a session's completed messages and clear them. */ - flushCompletedMessages(): Message[] { - const messages = this.completedMessages(); - this.completedMessages.set([]); + flushCompletedMessages(sessionId: string): Message[] { + const state = this.states().get(sessionId); + if (!state) return []; + const messages = state.completedMessages(); + state.completedMessages.set([]); return messages; } + /** + * Drop a session's parser state entirely (e.g. when the session is + * cleared). Cached accessor signals keep working β€” they fall back to + * their empty defaults. + */ + clearSession(sessionId: string): void { + if (!this.states().get(sessionId)) return; + this.states.update((map) => { + const next = new Map(map); + next.delete(sessionId); + return next; + }); + } + + // ========================================================================= + // State Factory + // ========================================================================= + + private createState(sessionId: string, startingMessageCount: number): ParserSessionState { + const currentMessageBuilder = signal(null); + const completedMessages = signal([]); + const isStreamComplete = signal(false); + + const state: ParserSessionState = { + sessionId, + startingMessageCount, + currentStreamId: uuidv4(), + streamState: StreamState.Idle, + currentMessageBuilder, + completedMessages, + toolProgress: signal({ visible: false }), + error: signal(null), + isStreamComplete, + metadata: signal(null), + pendingCitations: signal([]), + currentMessage: computed(() => { + const builder = currentMessageBuilder(); + return builder ? this.buildMessage(state, builder) : null; + }), + allMessages: computed(() => { + const completed = completedMessages(); + const current = state.currentMessage(); + return current ? [...completed, current] : completed; + }), + streamingMessageId: computed(() => { + const builder = currentMessageBuilder(); + const isComplete = isStreamComplete(); + + // Return the message ID if we have an active builder and stream is not complete + if (builder && !isComplete) { + return builder.id; + } + return null; + }), + callbacks: undefined as unknown as StreamParserCallbacks, + lineParser: undefined as unknown as ReturnType, + }; + + state.callbacks = this.createCallbacks(state); + state.lineParser = createStreamLineParser(state.callbacks); + return state; + } + + /** + * Cached per-session accessor: a stable computed that follows the + * session's current state object across resets and falls back to a + * default before the first reset. + */ + private cachedAccessor( + cache: Map>, + sessionId: string, + read: (state: ParserSessionState) => T, + fallback: T, + ): Signal { + let cached = cache.get(sessionId); + if (!cached) { + cached = computed(() => { + const state = this.states().get(sessionId); + return state ? read(state) : fallback; + }); + cache.set(sessionId, cached); + } + return cached; + } + + /** + * Whether this stream's session is the one the user is currently viewing. + * Conversation-view side channels (artifacts panel, MCP App frames, + * compaction badge) are viewed-session-scoped: they reset on route change + * and re-hydrate from the server, so a background stream must not push + * into them while another conversation is on screen. + */ + private isViewedSession(state: ParserSessionState): boolean { + return this.chatStateService.viewedSessionId() === state.sessionId; + } + // ========================================================================= // Callbacks Factory // ========================================================================= /** * Create callbacks for the stream parser core. - * These callbacks wire the pure parsing logic to our state management. + * These callbacks wire the pure parsing logic to one session's state. */ - private createCallbacks(): StreamParserCallbacks { + private createCallbacks(state: ParserSessionState): StreamParserCallbacks { return { - onMessageStart: (data) => this.handleMessageStart(data), - onMessageStop: (data) => this.handleMessageStop(data), - onDone: () => this.handleDone(), + onMessageStart: (data) => this.handleMessageStart(state, data), + onMessageStop: (data) => this.handleMessageStop(state, data), + onDone: () => this.handleDone(state), - onContentBlockStart: (data) => this.handleContentBlockStart(data), - onContentBlockDelta: (data) => this.handleContentBlockDelta(data), - onContentBlockStop: (data) => this.handleContentBlockStop(data), + onContentBlockStart: (data) => this.handleContentBlockStart(state, data), + onContentBlockDelta: (data) => this.handleContentBlockDelta(state, data), + onContentBlockStop: (data) => this.handleContentBlockStop(state, data), - onToolUse: (data) => this.handleToolUseProgress(data), - onToolResult: (data) => this.handleToolResult(data), - onToolProgress: (progress) => this.toolProgressSignal.set(progress), + onToolUse: (data) => this.handleToolUseProgress(state, data), + onToolResult: (data) => this.handleToolResult(state, data), + onToolProgress: (progress) => state.toolProgress.set(progress), - onMetadata: (data) => this.handleMetadata(data), - onReasoning: (data) => this.handleReasoning(data), - onCitation: (data) => this.handleCitation(data), + onMetadata: (data) => this.handleMetadata(state, data), + onReasoning: (data) => this.handleReasoning(state, data), + onCitation: (data) => this.handleCitation(state, data), onQuotaWarning: (data) => this.quotaWarningService.setWarning(data as QuotaWarning), onQuotaExceeded: (data) => this.quotaWarningService.setQuotaExceeded(data as QuotaExceeded), @@ -329,39 +436,39 @@ export class StreamParserService { // assistant message is normally in completedMessages; fall back // to the in-flight builder for tool_use stop reasons that keep // the message active. - const messages = this.allMessages(); - let lastAssistantId: string | undefined; - for (let i = messages.length - 1; i >= 0; i--) { - if (messages[i].role === 'assistant') { - lastAssistantId = messages[i].id; - break; - } - } + const lastAssistantId = this.findLastAssistantId(state); this.oauthConsentService.requestConsent( data.providerId, data.authorizationUrl, data.interruptId, lastAssistantId, - this.sessionId ?? undefined, + state.sessionId, ); }, - onCompaction: (data: CompactionEvent) => this.compactionSummary.recordLive(data), + onCompaction: (data: CompactionEvent) => { + // CompactionSummaryService is viewed-session-scoped (reset on route + // change, reseeded from session metadata) β€” a background stream's + // compaction must not bump the viewed conversation's badge. + if (this.isViewedSession(state)) { + this.compactionSummary.recordLive(data); + } + }, onArtifact: (data: ArtifactEvent) => { + // Viewed-session only: recordLive auto-opens the artifact panel, + // which must never happen for a conversation streaming in the + // background. Navigating back re-hydrates artifacts from the + // server, so the dropped event is recovered there. + if (!this.isViewedSession(state)) { + return; + } // Same post-message_stop timing as oauth_required: the producing // assistant message is the last assistant message in the list. // Anchor live placement to its concrete id β€” the numeric index // only lines up after a reload (it counts the memory tool // messages the folded client message doesn't have). - const messages = this.allMessages(); - let lastAssistantId: string | undefined; - for (let i = messages.length - 1; i >= 0; i--) { - if (messages[i].role === 'assistant') { - lastAssistantId = messages[i].id; - break; - } - } + const lastAssistantId = this.findLastAssistantId(state); this.artifactState.recordLive(data, lastAssistantId); }, @@ -369,8 +476,12 @@ export class StreamParserService { // Inline event (arrives right after its tool_result, mid-stream), // unlike the post-message_stop side channels above β€” just record // it keyed by toolUseId. The tool-use renderer picks it up - // reactively and swaps in the MCP App frame. - this.mcpAppState.recordLive(data); + // reactively and swaps in the MCP App frame. Viewed-session only: + // McpAppStateService is reset on route change, so recording for a + // background conversation would be wiped before it could render. + if (this.isViewedSession(state)) { + this.mcpAppState.recordLive(data); + } }, onToolInputPartial: (data: ToolInputPartialEvent) => { @@ -379,18 +490,13 @@ export class StreamParserService { // the latest healed prefix keyed by toolUseId; the frame relays it to // the App as `ui/notifications/tool-input-partial` for progressive // rendering (e.g. Excalidraw's guided camera tour). - this.mcpAppState.recordPartialInput(data.toolUseId, data.arguments); + if (this.isViewedSession(state)) { + this.mcpAppState.recordPartialInput(data.toolUseId, data.arguments); + } }, onToolApprovalRequired: (data: ToolApprovalRequiredEvent) => { - const messages = this.allMessages(); - let lastAssistantId: string | undefined; - for (let i = messages.length - 1; i >= 0; i--) { - if (messages[i].role === 'assistant') { - lastAssistantId = messages[i].id; - break; - } - } + const lastAssistantId = this.findLastAssistantId(state); this.toolApprovalService.requestApproval({ interruptId: data.interruptId, toolUseId: data.toolUseId, @@ -398,11 +504,11 @@ export class StreamParserService { toolInput: data.toolInput ?? undefined, message: data.message, messageId: lastAssistantId, - sessionId: this.sessionId ?? undefined, + sessionId: state.sessionId, }); }, - onError: (data) => this.handleError(data), + onError: (data) => this.handleError(state, data), onStreamError: (data) => { const streamError = data as ConversationalStreamError; this.errorService.handleConversationalStreamError(streamError); @@ -413,47 +519,52 @@ export class StreamParserService { streamError.code === ErrorCode.MAX_TOKENS || streamError.metadata?.['error_kind'] === 'max_tokens'; if (isMaxTokens) { - this.chatStateService.setLastTurnContinuable(true); + this.chatStateService.setLastTurnContinuable(state.sessionId, true); } }, - onParseError: (message) => this.setError(message), + onParseError: (message) => this.setError(state, message), }; } + private findLastAssistantId(state: ParserSessionState): string | undefined { + const messages = state.allMessages(); + for (let i = messages.length - 1; i >= 0; i--) { + if (messages[i].role === 'assistant') { + return messages[i].id; + } + } + return undefined; + } + // ========================================================================= // Event Handlers // ========================================================================= - private handleMessageStart(data: { role: 'user' | 'assistant' }): void { - // Initialize stream ID if not set - if (!this.currentStreamId) { - this.currentStreamId = uuidv4(); - } - + private handleMessageStart(state: ParserSessionState, data: { role: 'user' | 'assistant' }): void { // Update stream state - this.streamState = StreamState.Streaming; + state.streamState = StreamState.Streaming; // Clear any previous errors - this.errorSignal.set(null); + state.error.set(null); // If there's an existing message, finalize it before starting a new one - const currentBuilder = this.currentMessageBuilder(); + const currentBuilder = state.currentMessageBuilder(); if (currentBuilder) { - this.finalizeCurrentMessage(); + this.finalizeCurrentMessage(state); } // Clear stopReason in ChatStateService - this.chatStateService.setStopReason(null); + this.chatStateService.setStopReason(state.sessionId, null); // A new assistant turn is streaming β€” retire any stale "Continue" // affordance from a previous max_tokens truncation. - this.chatStateService.setLastTurnContinuable(false); + this.chatStateService.setLastTurnContinuable(state.sessionId, false); // Compute predictable message ID - const completedCount = this.completedMessages().length; - const messageIndex = this.startingMessageCount + completedCount; - const computedId = this.sessionId ? `msg-${this.sessionId}-${messageIndex}` : uuidv4(); + const completedCount = state.completedMessages().length; + const messageIndex = state.startingMessageCount + completedCount; + const computedId = `msg-${state.sessionId}-${messageIndex}`; // Create new message builder const builder: MessageBuilder = { @@ -464,24 +575,24 @@ export class StreamParserService { isComplete: false, }; - this.currentMessageBuilder.set(builder); + state.currentMessageBuilder.set(builder); } - private handleContentBlockStart(data: ContentBlockStartEvent): void { - const currentBuilder = this.currentMessageBuilder(); + private handleContentBlockStart(state: ParserSessionState, data: ContentBlockStartEvent): void { + const currentBuilder = state.currentMessageBuilder(); if (!currentBuilder) { - this.setError('content_block_start: received without active message'); + this.setError(state, 'content_block_start: received without active message'); return; } if (currentBuilder.contentBlocks.has(data.contentBlockIndex)) { - this.setError(`content_block_start: block at index ${data.contentBlockIndex} already exists`); + this.setError(state, `content_block_start: block at index ${data.contentBlockIndex} already exists`); return; } const blockType: 'text' | 'tool_use' = data.type === 'tool_use' ? 'tool_use' : 'text'; - this.currentMessageBuilder.update((builder) => { + state.currentMessageBuilder.update((builder) => { if (!builder) return builder; const blockBuilder: ContentBlockBuilder = { @@ -503,7 +614,7 @@ export class StreamParserService { // Show tool progress for tool_use blocks if (blockType === 'tool_use' && data.toolUse) { - this.toolProgressSignal.set({ + state.toolProgress.set({ visible: true, toolName: data.toolUse.name, toolUseId: data.toolUse.toolUseId, @@ -513,16 +624,16 @@ export class StreamParserService { } } - private handleContentBlockDelta(data: ContentBlockDeltaEvent): void { - const currentBuilder = this.currentMessageBuilder(); + private handleContentBlockDelta(state: ParserSessionState, data: ContentBlockDeltaEvent): void { + const currentBuilder = state.currentMessageBuilder(); if (!currentBuilder) { - this.setError('content_block_delta: received without active message'); + this.setError(state, 'content_block_delta: received without active message'); return; } const inferredType = inferContentBlockType(data); - this.currentMessageBuilder.update((builder) => { + state.currentMessageBuilder.update((builder) => { if (!builder) return builder; let block = builder.contentBlocks.get(data.contentBlockIndex); @@ -547,7 +658,7 @@ export class StreamParserService { // Update chunks if (data.text !== undefined) { if (typeof data.text !== 'string') { - this.setError(`content_block_delta: text must be string, got ${typeof data.text}`); + this.setError(state, `content_block_delta: text must be string, got ${typeof data.text}`); return builder; } block.textChunks.push(data.text); @@ -555,7 +666,7 @@ export class StreamParserService { if (data.input !== undefined) { if (typeof data.input !== 'string') { - this.setError(`content_block_delta: input must be string, got ${typeof data.input}`); + this.setError(state, `content_block_delta: input must be string, got ${typeof data.input}`); return builder; } block.inputChunks.push(data.input); @@ -568,19 +679,19 @@ export class StreamParserService { }); } - private handleContentBlockStop(data: { contentBlockIndex: number }): void { - const currentBuilder = this.currentMessageBuilder(); + private handleContentBlockStop(state: ParserSessionState, data: { contentBlockIndex: number }): void { + const currentBuilder = state.currentMessageBuilder(); if (!currentBuilder) { - this.setError('content_block_stop: received without active message'); + this.setError(state, 'content_block_stop: received without active message'); return; } - this.currentMessageBuilder.update((builder) => { + state.currentMessageBuilder.update((builder) => { if (!builder) return builder; const block = builder.contentBlocks.get(data.contentBlockIndex); if (!block) { - this.setError(`content_block_stop: block at index ${data.contentBlockIndex} does not exist`); + this.setError(state, `content_block_stop: block at index ${data.contentBlockIndex} does not exist`); return builder; } @@ -591,7 +702,7 @@ export class StreamParserService { block.isComplete = true; if (block.type === 'tool_use') { - this.toolProgressSignal.set({ visible: false }); + state.toolProgress.set({ visible: false }); } const newBlocks = new Map(builder.contentBlocks); @@ -601,10 +712,10 @@ export class StreamParserService { }); } - private handleToolUseProgress(data: { + private handleToolUseProgress(state: ParserSessionState, data: { tool_use: { name: string; tool_use_id: string; input: string }; }): void { - this.toolProgressSignal.update((progress) => ({ + state.toolProgress.update((progress) => ({ ...progress, visible: true, toolName: data.tool_use.name, @@ -612,14 +723,14 @@ export class StreamParserService { })); } - private handleToolResult(data: ToolResultEventData): void { + private handleToolResult(state: ParserSessionState, data: ToolResultEventData): void { const toolUseId = data.tool_result.toolUseId; const content = data.tool_result.content || []; const status = data.tool_result.status || 'success'; - const currentBuilder = this.currentMessageBuilder(); + const currentBuilder = state.currentMessageBuilder(); if (!currentBuilder) { - this.setError('tool_result: received without active message'); + this.setError(state, 'tool_result: received without active message'); return; } @@ -641,7 +752,7 @@ export class StreamParserService { const resultContent = parseToolResultContent(content); - this.currentMessageBuilder.update((builder) => { + state.currentMessageBuilder.update((builder) => { if (!builder) return builder; const block = builder.contentBlocks.get(foundIndex!); @@ -662,44 +773,51 @@ export class StreamParserService { return { ...builder, contentBlocks: newBlocks }; }); - this.toolProgressSignal.set({ visible: false }); + state.toolProgress.set({ visible: false }); } - private handleMessageStop(data: { stopReason: string }): void { - const currentBuilder = this.currentMessageBuilder(); + private handleMessageStop(state: ParserSessionState, data: { stopReason: string }): void { + const currentBuilder = state.currentMessageBuilder(); if (!currentBuilder) { - this.setError('message_stop: received without active message'); + this.setError(state, 'message_stop: received without active message'); return; } - this.chatStateService.setStopReason(data.stopReason); + this.chatStateService.setStopReason(state.sessionId, data.stopReason); - this.currentMessageBuilder.update((builder) => { + state.currentMessageBuilder.update((builder) => { if (!builder) return builder; return { ...builder, isComplete: true }; }); // If stop reason is tool_use, keep message active for tool result if (data.stopReason !== 'tool_use') { - this.finalizeCurrentMessage(); + this.finalizeCurrentMessage(state); } } - private handleDone(): void { - this.finalizeCurrentMessage(); - this.isStreamCompleteSignal.set(true); - this.toolProgressSignal.set({ visible: false }); - this.streamState = StreamState.Completed; + private handleDone(state: ParserSessionState): void { + this.finalizeCurrentMessage(state); + state.isStreamComplete.set(true); + state.toolProgress.set({ visible: false }); + state.streamState = StreamState.Completed; - // Automatic cleanup after delay + // Automatic cleanup after delay. Guarded on the stream ID so a session + // that started a new stream in the meantime isn't flushed mid-parse. + const streamId = state.currentStreamId; setTimeout(() => { - if (this.streamState === StreamState.Completed) { - this.flushCompletedMessages(); + const current = this.states().get(state.sessionId); + if ( + current === state && + current.currentStreamId === streamId && + current.streamState === StreamState.Completed + ) { + this.flushCompletedMessages(state.sessionId); } }, 5000); } - private handleError(data: unknown): void { + private handleError(state: ParserSessionState, data: unknown): void { let errorMessage = 'Unknown error'; if (data && typeof data === 'object') { @@ -729,23 +847,23 @@ export class StreamParserService { this.errorService.addError('Stream Error', errorMessage); } - this.setError(`Stream error: ${errorMessage}`); + this.setError(state, `Stream error: ${errorMessage}`); } - private handleMetadata(data: MetadataEvent): void { + private handleMetadata(state: ParserSessionState, data: MetadataEvent): void { if (!data.usage && !data.metrics) { return; } - this.metadataSignal.set(data); - this.updateLastCompletedMessageWithMetadata(); + state.metadata.set(data); + this.updateLastCompletedMessageWithMetadata(state); // Drive the session cost + context badge above the composer. // Cost on the wire may be either a number (legacy) or a CostBreakdown // object β€” extract the total either way (matches backend's Union shape). const turnCost = typeof data.cost === 'number' ? data.cost : data.cost?.total ?? 0; if (turnCost > 0) { - this.chatStateService.addTurnCost(turnCost); + this.chatStateService.addTurnCost(state.sessionId, turnCost); } // Only update the context badge from the *final* metadata event β€” @@ -766,21 +884,21 @@ export class StreamParserService { usage.inputTokens + (usage.cacheReadInputTokens ?? 0) + (usage.cacheWriteInputTokens ?? 0); - this.chatStateService.setContext(totalContext, data.contextWindow); + this.chatStateService.setContext(state.sessionId, totalContext, data.contextWindow); } } - private handleReasoning(data: { reasoningText?: string }): void { + private handleReasoning(state: ParserSessionState, data: { reasoningText?: string }): void { if (!data.reasoningText) { return; } - const currentBuilder = this.currentMessageBuilder(); + const currentBuilder = state.currentMessageBuilder(); if (!currentBuilder) { return; } - this.currentMessageBuilder.update((builder) => { + state.currentMessageBuilder.update((builder) => { if (!builder) return builder; // Find or create reasoning block @@ -818,8 +936,8 @@ export class StreamParserService { }); } - private handleCitation(data: Citation): void { - this.pendingCitations.update((citations) => [ + private handleCitation(state: ParserSessionState, data: Citation): void { + state.pendingCitations.update((citations) => [ ...citations, { assistantId: data.assistantId, @@ -834,35 +952,27 @@ export class StreamParserService { // Helper Methods // ========================================================================= - private shouldProcessEvent(): boolean { - if (!this.currentStreamId) { - return true; // Allow first event - } - - if (this.streamState === StreamState.Completed || this.streamState === StreamState.Error) { - return false; - } - - return true; + private shouldProcessEvent(state: ParserSessionState): boolean { + return state.streamState !== StreamState.Completed && state.streamState !== StreamState.Error; } - private setError(message: string): void { - this.errorSignal.set(message); - this.isStreamCompleteSignal.set(true); - this.toolProgressSignal.set({ visible: false }); - this.streamState = StreamState.Error; + private setError(state: ParserSessionState, message: string): void { + state.error.set(message); + state.isStreamComplete.set(true); + state.toolProgress.set({ visible: false }); + state.streamState = StreamState.Error; } - private updateLastCompletedMessageWithMetadata(): void { - const completed = this.completedMessages(); + private updateLastCompletedMessageWithMetadata(state: ParserSessionState): void { + const completed = state.completedMessages(); if (completed.length === 0) return; const lastMessage = completed[completed.length - 1]; - const newMetadata = this.getMetadataForMessage(); + const newMetadata = this.getMetadataForMessage(state); if (!newMetadata) return; if (!lastMessage.metadata) { - this.completedMessages.update((messages) => { + state.completedMessages.update((messages) => { const updated = [...messages]; updated[updated.length - 1] = { ...updated[updated.length - 1], @@ -904,7 +1014,7 @@ export class StreamParserService { (existingBreakdown === undefined && newBreakdown !== undefined); if (needsUpdate) { - this.completedMessages.update((messages) => { + state.completedMessages.update((messages) => { const updated = [...messages]; const existingLatencyObj = existingMetadata['latency'] as Record | undefined; const newLatencyObj = newMetadata['latency'] as Record | undefined; @@ -929,21 +1039,21 @@ export class StreamParserService { // Message Building // ========================================================================= - private buildMessage(builder: MessageBuilder): Message { + private buildMessage(state: ParserSessionState, builder: MessageBuilder): Message { const sortedBlocks = Array.from(builder.contentBlocks.entries()) .sort(([a], [b]) => a - b) - .map(([_, block]) => this.buildContentBlock(block)); + .map(([_, block]) => this.buildContentBlock(state, block)); const message: Message = { id: builder.id, role: builder.role, content: sortedBlocks, createdAt: builder.createdAt, - metadata: this.getMetadataForMessage(), + metadata: this.getMetadataForMessage(state), }; if (builder.role === 'assistant') { - const citations = this.pendingCitations(); + const citations = state.pendingCitations(); if (citations.length > 0) { message.citations = citations; } @@ -952,8 +1062,8 @@ export class StreamParserService { return message; } - private getMetadataForMessage(): Record | null { - const metadataEvent = this.metadataSignal(); + private getMetadataForMessage(state: ParserSessionState): Record | null { + const metadataEvent = state.metadata(); if (!metadataEvent) { return null; } @@ -999,7 +1109,7 @@ export class StreamParserService { return Object.keys(result).length > 0 ? result : null; } - private buildContentBlock(builder: ContentBlockBuilder): ContentBlock { + private buildContentBlock(state: ParserSessionState, builder: ContentBlockBuilder): ContentBlock { // Handle reasoning content blocks if (builder.type === 'reasoningContent') { return { @@ -1024,7 +1134,7 @@ export class StreamParserService { } catch (e) { if (builder.isComplete) { const errorMsg = e instanceof Error ? e.message : 'Unknown JSON parse error'; - this.setError(`Failed to parse tool input JSON for '${builder.toolName}': ${errorMsg}`); + this.setError(state, `Failed to parse tool input JSON for '${builder.toolName}': ${errorMsg}`); } } @@ -1070,20 +1180,20 @@ export class StreamParserService { } as ContentBlock; } - private finalizeCurrentMessage(): void { - const builder = this.currentMessageBuilder(); + private finalizeCurrentMessage(state: ParserSessionState): void { + const builder = state.currentMessageBuilder(); if (!builder) return; - const message = this.buildMessage(builder); + const message = this.buildMessage(state, builder); if (message.content.length > 0) { - this.completedMessages.update((messages) => [...messages, message]); + state.completedMessages.update((messages) => [...messages, message]); } if (builder.role === 'assistant') { - this.pendingCitations.set([]); + state.pendingCitations.set([]); } - this.currentMessageBuilder.set(null); + state.currentMessageBuilder.set(null); } } diff --git a/frontend/ai.client/src/app/session/services/export/export.model.ts b/frontend/ai.client/src/app/session/services/export/export.model.ts index e1d443a4..b102897b 100644 --- a/frontend/ai.client/src/app/session/services/export/export.model.ts +++ b/frontend/ai.client/src/app/session/services/export/export.model.ts @@ -56,9 +56,9 @@ export interface ExportInclude { export const DEFAULT_EXPORT_INCLUDE: ExportInclude = { userMessages: true, assistantMessages: true, - toolCalls: true, - images: true, - citations: true, + toolCalls: false, + images: false, + citations: false, reasoning: false, timestamps: false, }; diff --git a/frontend/ai.client/src/app/session/services/session/message-map.service.spec.ts b/frontend/ai.client/src/app/session/services/session/message-map.service.spec.ts index d84b72c2..123e05ed 100644 --- a/frontend/ai.client/src/app/session/services/session/message-map.service.spec.ts +++ b/frontend/ai.client/src/app/session/services/session/message-map.service.spec.ts @@ -3,6 +3,7 @@ import { HttpTestingController, provideHttpClientTesting } from '@angular/common import { provideHttpClient } from '@angular/common/http'; import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; import { MessageMapService } from './message-map.service'; +import { StreamParserService } from '../chat/stream-parser.service'; import { SessionService } from './session.service'; import { FileUploadService } from '../../../services/file-upload'; import { OAuthConsentService } from '../../../services/oauth-consent/oauth-consent.service'; @@ -98,6 +99,52 @@ describe('MessageMapService', () => { expect(messagesSignal()).toEqual(mockMessages); }); + it('reloadMessagesForSession re-fetches even when messages already exist and does not flip loading state', async () => { + // Seed an existing (stale) message so the load guard would normally skip. + service.addUserMessage('session-reload', 'search'); + + // Server holds the authoritative post-resume state: the assistant's + // tool_use plus its matching tool_result (in a following user message). + const serverMessages = [ + { id: 'msg-reload-0', role: 'user', content: [{ type: 'text', text: 'search' }] }, + { + id: 'msg-reload-1', + role: 'assistant', + content: [ + { type: 'toolUse', toolUse: { toolUseId: 'tu-1', name: 'search_messages', input: {} } }, + ], + }, + { + id: 'msg-reload-2', + role: 'user', + content: [ + { type: 'toolResult', toolResult: { toolUseId: 'tu-1', status: 'success', content: [{ text: 'ok' }] } }, + ], + }, + ]; + mockSessionService.getMessages.mockResolvedValue({ messages: serverMessages }); + + await service.reloadMessagesForSession('session-reload'); + + // Bypassed the "already loaded" guard and hit the API. + expect(mockSessionService.getMessages).toHaveBeenCalledWith('session-reload'); + // Never surfaced the skeleton loading state during a live reconcile. + expect(service.isLoadingSession()).toBe(null); + + // The tool_use card now carries its result (status flipped to complete). + const reloaded = service.getMessagesForSession('session-reload')(); + const assistant = reloaded.find((m) => m.id === 'msg-reload-1')!; + const toolBlock = assistant.content[0] as any; + expect(toolBlock.toolUse.status).toBe('complete'); + expect(toolBlock.toolUse.result).toBeDefined(); + }); + + it('reloadMessagesForSession swallows fetch errors (best-effort reconcile)', async () => { + mockSessionService.getMessages.mockRejectedValue(new Error('API error')); + await expect(service.reloadMessagesForSession('session-reload-err')).resolves.toBeUndefined(); + expect(service.isLoadingSession()).toBe(null); + }); + it('should hydrate pending OAuth interrupts from camelCase wire response', async () => { // Regression: backend serializes with by_alias=True so the wire payload uses // camelCase (pendingInterrupts, interruptId, providerId, ...). If the consumer @@ -148,10 +195,60 @@ describe('MessageMapService', () => { // Verify session exists in map const messagesSignal = service.getMessagesForSession('session-1'); expect(messagesSignal).toBeTruthy(); + expect(service.isStreaming('session-1')).toBe(true); - service.endStreaming(); - // Service should handle end streaming gracefully - expect(service).toBeTruthy(); + service.endStreaming('session-1'); + expect(service.isStreaming('session-1')).toBe(false); + }); + + it('syncs two concurrent streams into their own sessions', () => { + const parser = TestBed.inject(StreamParserService); + + service.addUserMessage('conc-a', 'question A'); + service.addUserMessage('conc-b', 'question B'); + + service.startStreaming('conc-a'); + service.startStreaming('conc-b'); + + // Interleave the two sessions' SSE events, as two live fetches would. + parser.parseEventSourceMessage('conc-a', 'message_start', { role: 'assistant' }); + parser.parseEventSourceMessage('conc-b', 'message_start', { role: 'assistant' }); + parser.parseEventSourceMessage('conc-a', 'content_block_delta', { + contentBlockIndex: 0, + text: 'answer A', + }); + parser.parseEventSourceMessage('conc-b', 'content_block_delta', { + contentBlockIndex: 0, + text: 'answer B', + }); + parser.parseEventSourceMessage('conc-a', 'message_stop', { stopReason: 'end_turn' }); + parser.parseEventSourceMessage('conc-b', 'message_stop', { stopReason: 'end_turn' }); + + // endStreaming performs the final imperative sync for its session only. + service.endStreaming('conc-a'); + service.endStreaming('conc-b'); + + const aMessages = service.getMessagesForSession('conc-a')(); + const bMessages = service.getMessagesForSession('conc-b')(); + + expect(aMessages.map(m => m.role)).toEqual(['user', 'assistant']); + expect(aMessages[1].content).toEqual([{ type: 'text', text: 'answer A' }]); + expect(bMessages.map(m => m.role)).toEqual(['user', 'assistant']); + expect(bMessages[1].content).toEqual([{ type: 'text', text: 'answer B' }]); + }); + + it('endStreaming for one session leaves another session streaming', () => { + service.startStreaming('conc-c'); + service.startStreaming('conc-d'); + + service.endStreaming('conc-c'); + + expect(service.isStreaming('conc-c')).toBe(false); + expect(service.isStreaming('conc-d')).toBe(true); + }); + + it('endStreaming is a no-op for a session without an active stream', () => { + expect(() => service.endStreaming('never-streamed')).not.toThrow(); }); it('should clear session', () => { diff --git a/frontend/ai.client/src/app/session/services/session/message-map.service.ts b/frontend/ai.client/src/app/session/services/session/message-map.service.ts index 5f03506b..6ab3f66f 100644 --- a/frontend/ai.client/src/app/session/services/session/message-map.service.ts +++ b/frontend/ai.client/src/app/session/services/session/message-map.service.ts @@ -1,5 +1,5 @@ // services/message-map.service.ts (updated integration) -import { Injectable, Signal, WritableSignal, signal, effect, inject } from '@angular/core'; +import { EffectRef, Injectable, Injector, Signal, WritableSignal, signal, effect, inject } from '@angular/core'; import { ContentBlock, FileAttachmentData, Message } from '../models/message.model'; import { StreamParserService } from '../chat/stream-parser.service'; import { PendingInterrupt, SessionService } from './session.service'; @@ -20,18 +20,27 @@ interface MessageMap { }) export class MessageMapService { private messageMap = signal({}); - private activeStreamSessionId = signal(null); /** - * Continuation-after-max_tokens state. A "Continue" turn has NO new user - * message, so the normal sync (which truncates back to the last user - * message and replaces everything after it) would discard the truncated - * partial + error bubbles and show only the continuation. In continuation - * mode we instead pin a stable prefix (the messages that existed when the - * continuation started) and append the continuation stream after it. + * Sessions with an in-flight stream. Multiple conversations can stream + * concurrently β€” each one syncs its own parser state into its own map + * entry via a dedicated effect (see {@link createSyncEffect}). */ - private continuationSessionId: string | null = null; - private continuationPrefix: Message[] = []; + private activeStreamSessions = signal>(new Set()); + + /** One live sync effect per streaming session, torn down on endStreaming. */ + private syncEffects = new Map(); + + /** + * Continuation-after-max_tokens state, per session. A "Continue" turn has + * NO new user message, so the normal sync (which truncates back to the + * last user message and replaces everything after it) would discard the + * truncated partial + error bubbles and show only the continuation. In + * continuation mode we instead pin a stable prefix (the messages that + * existed when the continuation started) and append the continuation + * stream after it. + */ + private continuationPrefixes = new Map(); /** * Tracks which session is currently loading messages from the API. @@ -59,29 +68,15 @@ export class MessageMapService { private oauthConsentService = inject(OAuthConsentService); private toolApprovalService = inject(ToolApprovalService); private mcpAppState = inject(McpAppStateService); - - constructor() { - // Reactive effect: automatically sync streaming messages to the message map - effect(() => { - const sessionId = this.activeStreamSessionId(); - const streamMessages = this.streamParser.allMessages(); - - if (sessionId && streamMessages.length > 0) { - this.syncStreamingMessages(sessionId, streamMessages); - } - }); - } + private injector = inject(Injector); /** * Start streaming for a session. * Call this before beginning to parse SSE events. */ startStreaming(sessionId: string): void { - this.activeStreamSessionId.set(sessionId); - // A normal turn uses the standard (truncate-to-last-user) sync. - this.continuationSessionId = null; - this.continuationPrefix = []; + this.continuationPrefixes.delete(sessionId); // Get current message count for this session to enable predictable ID generation const currentMessages = this.messageMap()[sessionId]?.() ?? []; @@ -90,13 +85,7 @@ export class MessageMapService { // Reset stream parser with session context for predictable message IDs this.streamParser.reset(sessionId, messageCount); - // Ensure the session exists in the map - if (!this.messageMap()[sessionId]) { - this.messageMap.update(map => ({ - ...map, - [sessionId]: signal([]) - })); - } + this.beginStreamTracking(sessionId); } /** @@ -108,40 +97,79 @@ export class MessageMapService { * the continuation's message IDs follow the prefix instead of colliding. */ beginContinuationStreaming(sessionId: string): void { - this.activeStreamSessionId.set(sessionId); - const currentMessages = this.messageMap()[sessionId]?.() ?? []; - this.continuationSessionId = sessionId; - this.continuationPrefix = [...currentMessages]; + this.continuationPrefixes.set(sessionId, [...currentMessages]); this.streamParser.reset(sessionId, currentMessages.length); + this.beginStreamTracking(sessionId); + } + + /** + * End streaming for a session. + * Finalizes its messages and clears its streaming state. No-op when the + * session has no active stream (e.g. a superseded stream's late cleanup), + * so it can never tear down a newer stream for the same session. + */ + endStreaming(sessionId: string): void { + if (!this.activeStreamSessions().has(sessionId)) { + return; + } + + // Ensure final messages are synced (continuation merge still active + // here so the final flush keeps the pinned prefix). + const finalMessages = this.streamParser.allMessagesFor(sessionId)(); + if (finalMessages.length > 0) { + this.syncStreamingMessages(sessionId, finalMessages); + } + + this.syncEffects.get(sessionId)?.destroy(); + this.syncEffects.delete(sessionId); + this.continuationPrefixes.delete(sessionId); + this.activeStreamSessions.update(set => { + const next = new Set(set); + next.delete(sessionId); + return next; + }); + } + + /** Whether a session currently has an in-flight stream. */ + isStreaming(sessionId: string): boolean { + return this.activeStreamSessions().has(sessionId); + } + + /** + * Mark a session as streaming, ensure its map entry exists, and attach + * its live parserβ†’map sync effect. Idempotent per session. + */ + private beginStreamTracking(sessionId: string): void { + this.activeStreamSessions.update(set => { + if (set.has(sessionId)) return set; + const next = new Set(set); + next.add(sessionId); + return next; + }); + + // Ensure the session exists in the map if (!this.messageMap()[sessionId]) { this.messageMap.update(map => ({ ...map, [sessionId]: signal([]) })); } - } - /** - * End streaming for the current session. - * Finalizes messages and clears streaming state. - */ - endStreaming(): void { - const sessionId = this.activeStreamSessionId(); - if (sessionId) { - // Ensure final messages are synced (continuation merge still active - // here so the final flush keeps the pinned prefix). - const finalMessages = this.streamParser.allMessages(); - if (finalMessages.length > 0) { - this.syncStreamingMessages(sessionId, finalMessages); - } + // Reactive per-session effect: sync this stream's messages into this + // session's map entry. Each concurrent stream gets its own effect, so + // conversation A keeps updating while the user views conversation B. + if (!this.syncEffects.has(sessionId)) { + const ref = effect(() => { + const streamMessages = this.streamParser.allMessagesFor(sessionId)(); + if (streamMessages.length > 0) { + this.syncStreamingMessages(sessionId, streamMessages); + } + }, { injector: this.injector }); + this.syncEffects.set(sessionId, ref); } - - this.activeStreamSessionId.set(null); - this.continuationSessionId = null; - this.continuationPrefix = []; } /** @@ -264,8 +292,9 @@ export class MessageMapService { // started (history + truncated partial + error bubble) and append the // continuation stream after it. Rebuilt from the stable prefix every // tick so repeated syncs stay idempotent. - if (this.continuationSessionId === sessionId) { - return [...this.continuationPrefix, ...streamMessages]; + const continuationPrefix = this.continuationPrefixes.get(sessionId); + if (continuationPrefix) { + return [...continuationPrefix, ...streamMessages]; } // Find the index of the last user message @@ -304,8 +333,48 @@ export class MessageMapService { return; } - // Set loading state for this session - this._isLoadingSession.set(sessionId); + try { + await this.fetchAndApplyMessages(sessionId, /* showLoading */ true); + } catch (error) { + console.error('Failed to load messages for session:', sessionId, error); + throw error; + } + } + + /** + * Force a re-fetch of a session's messages from the server, replacing the + * in-memory copy with the authoritative persisted state. + * + * Unlike {@link loadMessagesForSession} this bypasses the + * "already loaded" guard and does NOT flip the skeleton loading state, so + * it can reconcile a live thread without a visible flash. It is used after + * an interrupt resume (OAuth consent / tool approval): the resumed stream + * carries only the `tool_result` + the final assistant text β€” Strands does + * not replay the interrupted `tool_use` block β€” so the live parser can't + * attach the result to the paused tool card. Re-reading persisted memory + * (where the backend stored both the `tool_use` and its `tool_result`) + * flips the card from "Running…" to its completed result. + * + * Best-effort: a failure leaves the live-streamed state untouched rather + * than disrupting the UI. + */ + async reloadMessagesForSession(sessionId: string): Promise { + try { + await this.fetchAndApplyMessages(sessionId, /* showLoading */ false); + } catch (error) { + console.error('Failed to reload messages for session:', sessionId, error); + } + } + + /** + * Fetch a session's messages + file metadata, reconstruct tool results and + * file attachments, and replace the message map entry. Shared by the + * initial load and the post-resume reconcile. + */ + private async fetchAndApplyMessages(sessionId: string, showLoading: boolean): Promise { + if (showLoading) { + this._isLoadingSession.set(sessionId); + } try { // Fetch messages and file metadata in parallel @@ -351,12 +420,10 @@ export class MessageMapService { // session.page resets McpAppStateService before this load, so the // non-clobbering seed lands cleanly. this.mcpAppState.seedFromHydration(messagesResponse.uiResources ?? []); - } catch (error) { - console.error('Failed to load messages for session:', sessionId, error); - throw error; } finally { - // Clear loading state - this._isLoadingSession.set(null); + if (showLoading) { + this._isLoadingSession.set(null); + } } } @@ -605,9 +672,12 @@ export class MessageMapService { } /** - * Clear all data for a session. + * Clear all data for a session, including any in-flight stream sync and + * its parser state. */ clearSession(sessionId: string): void { + this.endStreaming(sessionId); + this.streamParser.clearSession(sessionId); this.messageMap.update(map => { const { [sessionId]: _, ...rest } = map; return rest; diff --git a/frontend/ai.client/src/app/session/services/session/scroll-position.service.spec.ts b/frontend/ai.client/src/app/session/services/session/scroll-position.service.spec.ts new file mode 100644 index 00000000..a8325365 --- /dev/null +++ b/frontend/ai.client/src/app/session/services/session/scroll-position.service.spec.ts @@ -0,0 +1,45 @@ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { TestBed } from '@angular/core/testing'; +import { ScrollPositionService } from './scroll-position.service'; + +describe('ScrollPositionService', () => { + let service: ScrollPositionService; + + beforeEach(() => { + TestBed.resetTestingModule(); + TestBed.configureTestingModule({}); + service = TestBed.inject(ScrollPositionService); + }); + + afterEach(() => { + TestBed.resetTestingModule(); + }); + + it('returns undefined for a conversation with no remembered position', () => { + expect(service.get('unknown')).toBeUndefined(); + }); + + it('remembers positions per conversation independently', () => { + service.save('a', 1200); + service.save('b', 0); + + expect(service.get('a')).toBe(1200); + expect(service.get('b')).toBe(0); + }); + + it('overwrites a conversation position on re-save', () => { + service.save('a', 100); + service.save('a', 900); + expect(service.get('a')).toBe(900); + }); + + it('clears a single conversation without touching others', () => { + service.save('a', 100); + service.save('b', 200); + + service.clear('a'); + + expect(service.get('a')).toBeUndefined(); + expect(service.get('b')).toBe(200); + }); +}); diff --git a/frontend/ai.client/src/app/session/services/session/scroll-position.service.ts b/frontend/ai.client/src/app/session/services/session/scroll-position.service.ts new file mode 100644 index 00000000..dfa1e113 --- /dev/null +++ b/frontend/ai.client/src/app/session/services/session/scroll-position.service.ts @@ -0,0 +1,33 @@ +import { Injectable } from '@angular/core'; + +/** + * In-memory scroll position per conversation, for the lifetime of the SPA + * session. Backs the navigation scroll policy (see ConversationPage): + * navigating back to a conversation restores where the user was; a + * conversation with no remembered position (first open, or after a full + * reload) lands on its latest turn instead. + * + * Deliberately not persisted β€” after a reload the message heights, fonts + * and hydrated cards can differ enough that a stored pixel offset points + * somewhere arbitrary, which is worse than the deterministic latest-turn + * anchor. + */ +@Injectable({ providedIn: 'root' }) +export class ScrollPositionService { + private readonly positions = new Map(); + + /** Remember the window scroll offset for a conversation. */ + save(sessionId: string, scrollY: number): void { + this.positions.set(sessionId, scrollY); + } + + /** The remembered offset, or undefined if this conversation has none. */ + get(sessionId: string): number | undefined { + return this.positions.get(sessionId); + } + + /** Drop a conversation's remembered offset. */ + clear(sessionId: string): void { + this.positions.delete(sessionId); + } +} diff --git a/frontend/ai.client/src/app/session/session.page.ts b/frontend/ai.client/src/app/session/session.page.ts index 9a34cd44..f3619c4d 100644 --- a/frontend/ai.client/src/app/session/session.page.ts +++ b/frontend/ai.client/src/app/session/session.page.ts @@ -1,4 +1,5 @@ -import { Component, inject, effect, Signal, signal, computed, OnDestroy } from '@angular/core'; +import { Component, inject, effect, Signal, signal, computed, viewChild, OnDestroy, PLATFORM_ID } from '@angular/core'; +import { isPlatformBrowser } from '@angular/common'; import { ActivatedRoute, Router } from '@angular/router'; import { Subscription } from 'rxjs'; import { v4 as uuidv4 } from 'uuid'; @@ -6,6 +7,7 @@ import { ChatRequestService } from './services/chat/chat-request.service'; import { MessageMapService } from './services/session/message-map.service'; import { Message } from './services/models/message.model'; import { SessionService } from './services/session/session.service'; +import { ScrollPositionService } from './services/session/scroll-position.service'; import { ChatStateService } from './services/chat/chat-state.service'; import { SidenavService } from '../services/sidenav/sidenav.service'; import { HeaderService } from '../services/header/header.service'; @@ -64,6 +66,19 @@ export class ConversationPage implements OnDestroy { private voiceChatService = inject(VoiceChatService); private systemPromptsService = inject(SystemPromptsService); private chatModeService = inject(ChatModeService); + private scrollPositions = inject(ScrollPositionService); + private platformId = inject(PLATFORM_ID); + private isBrowser = isPlatformBrowser(this.platformId); + + private chatContainer = viewChild(ChatContainerComponent); + + /** + * The conversation whose scroll position we're currently tracking. + * Captured into ScrollPositionService the moment the user navigates away + * (route change or component teardown) β€” see the scroll policy note on + * the route subscription. + */ + private lastViewedSessionId: string | null = null; sessionId = signal(null); assistantIdFromQuery = signal(null); @@ -143,7 +158,14 @@ export class ConversationPage implements OnDestroy { readonly sessionConversation = this.sessionService.currentSession; readonly isChatLoading = this.chatStateService.isChatLoading; readonly isLoadingSession = this.messageMapService.isLoadingSession; - readonly streamingMessageId = this.streamParserService.streamingMessageId; + + // Streaming indicator for the conversation on screen. Parser state is + // per-session, so a conversation streaming in the background never + // animates the one being viewed. + readonly streamingMessageId = computed(() => { + const id = this.sessionId(); + return id ? this.streamParserService.streamingMessageIdFor(id)() : null; + }); // Computed signal to check if session has messages readonly hasMessages = computed(() => this.messages().length > 0); @@ -172,6 +194,14 @@ export class ConversationPage implements OnDestroy { }); constructor() { + // Keep the viewed-session facades in ChatStateService (loading, cost + // badge, Continue affordance, Stop button) pointed at the conversation + // on screen. Uses the effective id so a staged session (file attached + // before the first message) counts as viewed too. + effect(() => { + this.chatStateService.setViewedSession(this.effectiveSessionId()); + }); + // Control header visibility based on whether there are messages effect(() => { if (this.hasMessages()) { @@ -226,19 +256,19 @@ export class ConversationPage implements OnDestroy { // session. effect(() => { const session = this.sessionConversation(); - if (!session) return; - this.chatStateService.seedSessionAggregates({ + if (!session?.sessionId) return; + this.chatStateService.seedSessionAggregates(session.sessionId, { totalCost: session.totalCost, lastContextTokens: session.lastContextTokens, contextWindow: session.contextWindow, }); // Refresh-survival for the max_tokens "Continue" affordance: the // truncated partial is already in restored history; this flag is the - // missing piece. Only set true here β€” the route-change reset clears - // it (cross-session safety) and the live stream_error path owns the - // in-turn signal, so we never clobber a live true with stale metadata. + // missing piece. Only set true here β€” the live stream_error path owns + // the in-turn signal, so we never clobber a live true with stale + // metadata. if (session.lastTurnContinuable) { - this.chatStateService.setLastTurnContinuable(true); + this.chatStateService.setLastTurnContinuable(session.sessionId, true); } }); @@ -323,20 +353,35 @@ export class ConversationPage implements OnDestroy { } }); - // Subscribe to route parameter changes + // Subscribe to route parameter changes. + // + // Deliberately does NOT abort an in-flight stream for the session being + // navigated away from: the stream keeps running in the background, + // syncing into its own session's message map (per-session parser + + // per-session sync effect), and the backend completes and persists the + // turn either way. Only the Stop button aborts, and only its own session. + // + // Scroll policy on conversation change: remember where the user was in + // the conversation they're leaving, reset to the top while the next one + // loads (the leftover window offset is meaningless against different + // content), then restore below β€” remembered position if we have one, + // otherwise anchor the latest turn (restoreScrollPosition). this.routeSubscription = this.route.paramMap.subscribe(async params => { const id = params.get('sessionId'); - this.sessionId.set(id); - // Clear stale cost/context badge state BEFORE the new session's - // metadata loads β€” otherwise the previous session's totals briefly - // flash on the badge while the new metadata is in flight. - this.chatStateService.seedSessionAggregates({}); + if (this.isBrowser && this.lastViewedSessionId !== id) { + if (this.lastViewedSessionId) { + this.scrollPositions.save(this.lastViewedSessionId, window.scrollY); + } + window.scrollTo({ top: 0, behavior: 'auto' }); + } + this.lastViewedSessionId = id; + + this.sessionId.set(id); - // Retire any prior session's "Continue" affordance before the new - // session's metadata lands; the seed effect re-sets it from - // metadata.lastTurnContinuable when applicable. - this.chatStateService.setLastTurnContinuable(false); + // Cost/context badge and Continue-affordance state is per-session in + // ChatStateService, and the viewed-session effect above repoints the + // facades β€” no cross-session clearing needed here anymore. // Compaction summary is session-scoped β€” clear before loading the // next session's metadata so the previous session's totals don't @@ -385,6 +430,10 @@ export class ConversationPage implements OnDestroy { // error just means no cards β€” never disrupt the session. this.hydrateArtifacts(id); this.hydrateMcpAppCards(id); + + // Messages are in the map and rendering β€” position the viewport + // per the scroll policy (remembered spot, else latest turn). + this.restoreScrollPosition(id); } else { // No session selected, clear the session metadata this.sessionService.setSessionMetadataId(null); @@ -399,10 +448,48 @@ export class ConversationPage implements OnDestroy { } ngOnDestroy() { + // Leaving the conversation view entirely (e.g. to an admin page, or the + // `/` ↔ `/s/:id` transitions that recreate this component) β€” remember + // where the user was so navigating back restores it. + if (this.isBrowser) { + const id = this.sessionId(); + if (id) { + this.scrollPositions.save(id, window.scrollY); + } + } this.routeSubscription?.unsubscribe(); this.queryParamSubscription?.unsubscribe(); } + /** + * Position the viewport for a freshly navigated-to conversation: + * the remembered offset when the user has been here this SPA session, + * otherwise the latest turn (last user message anchored at top β€” the same + * framing the composer submit uses, so "returning to" and "just asked in" + * a conversation look alike). Instant, never smooth: animating a jump + * across a whole conversation is noise. + * + * Runs two animation frames after the messages landed in the map so the + * message DOM and the bottom spacer have committed and the full scroll + * height exists; bails if the user has already navigated elsewhere. + */ + private restoreScrollPosition(sessionId: string): void { + if (!this.isBrowser) return; + + requestAnimationFrame(() => { + requestAnimationFrame(() => { + if (this.sessionId() !== sessionId) return; + + const saved = this.scrollPositions.get(sessionId); + if (saved !== undefined) { + window.scrollTo({ top: saved, behavior: 'auto' }); + } else { + this.chatContainer()?.scrollToLastUserMessage('auto'); + } + }); + }); + } + /** * Best-effort: pull the session's artifacts and seed the registry so * cards render on load. `seedFromHydration` is non-clobbering, so a @@ -456,8 +543,8 @@ export class ConversationPage implements OnDestroy { const assistantIdToUse = this.assistantIdFromQuery() || this.assistant()?.assistantId || undefined; - // Set loading state before submitting - this.chatStateService.setChatLoading(true); + // Loading state is set inside submitChatRequest once the (possibly + // freshly generated) session id is known β€” it's per-session now. // Submit the chat request with file upload IDs and assistant ID if present this.chatRequestService.submitChatRequest( @@ -511,7 +598,12 @@ export class ConversationPage implements OnDestroy { } onMessageCancelled() { - this.chatHttpService.cancelChatRequest(); + // Stop only the viewed conversation's stream; a conversation streaming + // in the background is unaffected. + const sessionId = this.effectiveSessionId(); + if (sessionId) { + this.chatHttpService.cancelChatRequest(sessionId); + } } /** diff --git a/infrastructure/lib/constructs/app-api/app-api-iam-grants.ts b/infrastructure/lib/constructs/app-api/app-api-iam-grants.ts index 32fce027..2de2a72a 100644 --- a/infrastructure/lib/constructs/app-api/app-api-iam-grants.ts +++ b/infrastructure/lib/constructs/app-api/app-api-iam-grants.ts @@ -403,6 +403,14 @@ export function grantAppApiPermissions(props: AppApiIamGrantsProps): void { // do not exist as IAM actions, so the entire policy was a silent // no-op β€” the App API hit AccessDeniedException on ListEvents. actions: [ + // GetMemory is required by the memory dashboard read path: + // memory_service._get_strategy_namespaces() calls + // MemoryClient.get_memory_strategies(), which invokes GetMemory to + // enumerate the memory's strategy IDs. Without it the call + // AccessDenies, strategy discovery returns (None, None, None), and + // GET /memory returns empty preferences/facts with a 200 β€” the + // memories/preferences page renders blank even though records exist. + 'bedrock-agentcore:GetMemory', 'bedrock-agentcore:CreateEvent', 'bedrock-agentcore:GetEvent', 'bedrock-agentcore:ListEvents', @@ -507,6 +515,7 @@ export function grantAppApiPermissions(props: AppApiIamGrantsProps): void { 'bedrock-agentcore:UpdateGatewayTarget', 'bedrock-agentcore:DeleteGatewayTarget', 'bedrock-agentcore:ListGatewayTargets', + 'bedrock-agentcore:SynchronizeGatewayTargets', // GetGateway resolves the gateway execution-role ARN, which the per- // target Lambda grant names as the invoke principal. 'bedrock-agentcore:GetGateway', diff --git a/infrastructure/lib/constructs/gateway/agentcore-gateway-construct.ts b/infrastructure/lib/constructs/gateway/agentcore-gateway-construct.ts index abcddac7..5c608ac6 100644 --- a/infrastructure/lib/constructs/gateway/agentcore-gateway-construct.ts +++ b/infrastructure/lib/constructs/gateway/agentcore-gateway-construct.ts @@ -62,11 +62,25 @@ export class AgentCoreGatewayConstruct extends Construct { description: 'Execution role for AgentCore Gateway', }); - // NOTE: no standing `lambda:Invoke*` grant. Invoke permission for each - // mcpServer target's Lambda Function URL is granted per-target at - // registration by app-api (lambda:AddPermission naming this role) and - // revoked on delete β€” least-privilege, and admins add servers with no infra - // change. See AgentCoreGatewayTargetAccess on the app-api role. + // Lambda invoke grant for MCP targets. AWS docs require an identity-based + // policy on the gateway service role allowing `lambda:InvokeFunction` on + // target Lambdas, in addition to the resource-based policy added per-target + // at registration (lambda:AddPermission). The resource-policy-only approach + // is not sufficient β€” the Gateway needs both. + // Ref: https://docs.aws.amazon.com/bedrock-agentcore/latest/devguide/gateway-prerequisites-permissions.html + // Scoped to the MCP server naming conventions used by the mcp-servers repo + // (`mcp-*`) and this platform (`${prefix}-mcp-*`). + this.gatewayRole.addToPolicy( + new iam.PolicyStatement({ + sid: 'LambdaInvokeForMcpTargets', + effect: iam.Effect.ALLOW, + actions: ['lambda:InvokeFunction'], + resources: [ + `arn:aws:lambda:${stack.region}:${stack.account}:function:mcp-*`, + `arn:aws:lambda:${stack.region}:${stack.account}:function:${config.projectPrefix}-mcp-*`, + ], + }), + ); this.gatewayRole.addToPolicy( new iam.PolicyStatement({ diff --git a/infrastructure/lib/constructs/inference-api/inference-api-iam-roles.ts b/infrastructure/lib/constructs/inference-api/inference-api-iam-roles.ts index f4318dbf..af0f32fd 100644 --- a/infrastructure/lib/constructs/inference-api/inference-api-iam-roles.ts +++ b/infrastructure/lib/constructs/inference-api/inference-api-iam-roles.ts @@ -301,6 +301,17 @@ export function createRuntimeExecutionRole( // speculative names (CreateMemoryEvent, ListMemoryEvents, // RetrieveMemory) that don't exist as IAM actions. actions: [ + // GetMemory is required for long-term memory RETRIEVAL. At session + // creation the agent calls _discover_strategy_ids() -> + // MemoryClient.get_memory_strategies(), which invokes GetMemory to + // resolve the SEMANTIC / USER_PREFERENCE / SUMMARIZATION strategy IDs + // used to build the retrieval namespaces. Without it that call + // AccessDenies, the retrieval config is left empty, and the agent + // silently runs with "long-term memory retrieval disabled" β€” it keeps + // writing events (CreateEvent) but never recalls stored memories. + // Verified against the AWS Service Authorization Reference: GetMemory + // is a Read action on the `memory` resource type. + 'bedrock-agentcore:GetMemory', 'bedrock-agentcore:CreateEvent', 'bedrock-agentcore:GetEvent', 'bedrock-agentcore:ListEvents', diff --git a/infrastructure/package-lock.json b/infrastructure/package-lock.json index bfc7ce12..0e846d6a 100644 --- a/infrastructure/package-lock.json +++ b/infrastructure/package-lock.json @@ -1,12 +1,12 @@ { "name": "infrastructure", - "version": "1.0.3", + "version": "1.0.4", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "infrastructure", - "version": "1.0.3", + "version": "1.0.4", "dependencies": { "aws-cdk-lib": "2.260.0", "constructs": "10.6.0" diff --git a/infrastructure/package.json b/infrastructure/package.json index 2edef91b..ce53abaf 100644 --- a/infrastructure/package.json +++ b/infrastructure/package.json @@ -1,6 +1,6 @@ { "name": "infrastructure", - "version": "1.0.3", + "version": "1.0.4", "bin": { "infrastructure": "bin/infrastructure.js" },