diff --git a/backend/scripts/backfill_false_interrupted_markers.py b/backend/scripts/backfill_false_interrupted_markers.py new file mode 100644 index 00000000..eebec127 --- /dev/null +++ b/backend/scripts/backfill_false_interrupted_markers.py @@ -0,0 +1,221 @@ +#!/usr/bin/env python3 +"""Clear interrupted-turn markers that no interruption could have produced. + +WHY THIS EXISTS +Until PR #988 the SPA released a session's ``AbortController`` only on the +Stop button, never on normal stream teardown. ``streamingSessionIds()`` reads +that controller to decide which turns a page departure interrupted, so every +completed turn stayed "in flight" for the life of the tab and the next +refresh / tab close / navigation POSTed ``navigated_away`` for it — one +departure marking every session ever streamed in that tab. + +A stale marker costs twice. It shows a "Response interrupted" chip plus a +Continue button on a complete answer (and Continue bills a +``continue_truncated`` turn that resumes an already-finished message), and it +makes the session's NEXT prompt carry a false ```` — +persisted in history, invisible in the UI, telling the model its previous +response was cut off. Markers self-clear only at the start of that session's +next turn (``clear_interrupted_turn``), so a conversation nobody returns to +stays armed indefinitely. + +WHY THE 900s THRESHOLD, AND NOT "THE MARKER LANDED AFTER THE LAST MESSAGE" +A genuine interruption does not always bump ``lastMessageAt``: the +"marker only, no synthetic write" branch of ``_persist_interruption`` (an +interrupted continuation, where the history tail is already an assistant +turn) leaves it at the previous turn. So a modest positive gap is ambiguous +and clearing on it would erase real interruptions. + +What is NOT ambiguous is a gap wider than a turn can live. The SSE stream +times out at 600s, so no turn is still running 15 minutes after its last +message — a departure signal that lands then cannot have interrupted +anything. That is the only claim this script acts on. + +SAFETY +* Dry-run by default; ``--apply`` is required to write. +* Only ``navigated_away`` rows are considered. ``user_stopped`` is the user's + own attested intent and ``connection_lost`` comes from the server's own + backstop — neither is this bug, and neither is touched. +* Each write is conditional on the exact ``lastTurnInterruptedAt`` / + ``lastTurnInterruptReason`` the scan read, so a session that started a new + turn (and was legitimately re-marked) between scan and write is skipped + rather than clobbered. +* Idempotent: a second run finds nothing left to do. + +USAGE + python scripts/backfill_false_interrupted_markers.py \ + --table boisestateai-v2-sessions-metadata --profile prod-ai + # …review the dry-run summary, then: + python scripts/backfill_false_interrupted_markers.py \ + --table boisestateai-v2-sessions-metadata --profile prod-ai --apply +""" + +from __future__ import annotations + +import argparse +import sys +from datetime import datetime, timezone +from typing import Any, Iterator, Optional + +import boto3 +from botocore.exceptions import ClientError + +# The SSE stream times out at 600s. A departure signal that lands more than +# this long after the session's last message cannot have interrupted a live +# turn, whatever the history tail looks like. The extra margin over 600s is +# deliberate slack for clock skew and teardown latency. +DEFAULT_MIN_GAP_SECONDS = 900 + +MARKER_ATTRS = ("lastTurnInterrupted", "lastTurnInterruptReason", "lastTurnInterruptedAt") + + +def _parse_iso(value: Optional[str]) -> Optional[datetime]: + """Parse a stored ISO-8601 timestamp, tolerating a missing offset.""" + if not isinstance(value, str): + return None + try: + parsed = datetime.fromisoformat(value) + except ValueError: + return None + return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) + + +def _scan_marked_rows(table: Any) -> Iterator[dict]: + """Yield every session row carrying an interrupted-turn marker.""" + kwargs: dict[str, Any] = { + "FilterExpression": "attribute_exists(#lti)", + "ProjectionExpression": "#pk, #sk, #gsipk, #lti, #ltr, #ltia, #lma", + "ExpressionAttributeNames": { + "#pk": "PK", + "#sk": "SK", + "#gsipk": "GSI_PK", + "#lti": "lastTurnInterrupted", + "#ltr": "lastTurnInterruptReason", + "#ltia": "lastTurnInterruptedAt", + "#lma": "lastMessageAt", + }, + } + while True: + response = table.scan(**kwargs) + yield from response.get("Items", []) + last_key = response.get("LastEvaluatedKey") + if not last_key: + return + kwargs["ExclusiveStartKey"] = last_key + + +def _select(items: list[dict], min_gap_seconds: int) -> tuple[list[dict], dict[str, int]]: + """Split scanned rows into the provably-false ones and a reason tally.""" + selected: list[dict] = [] + skipped = {"other_reason": 0, "gap_too_small": 0, "unparseable": 0} + + for item in items: + if item.get("lastTurnInterruptReason") != "navigated_away": + skipped["other_reason"] += 1 + continue + marked_at = _parse_iso(item.get("lastTurnInterruptedAt")) + last_message_at = _parse_iso(item.get("lastMessageAt")) + if marked_at is None or last_message_at is None: + skipped["unparseable"] += 1 + continue + gap = (marked_at - last_message_at).total_seconds() + if gap <= min_gap_seconds: + skipped["gap_too_small"] += 1 + continue + selected.append({**item, "_gapSeconds": gap}) + + selected.sort(key=lambda row: row["_gapSeconds"], reverse=True) + return selected, skipped + + +def _clear(table: Any, row: dict) -> str: + """Remove one row's marker. Returns 'cleared', 'raced', or 'error'.""" + try: + table.update_item( + Key={"PK": row["PK"], "SK": row["SK"]}, + UpdateExpression="REMOVE #lti, #ltr, #ltia", + ExpressionAttributeNames={ + "#lti": "lastTurnInterrupted", + "#ltr": "lastTurnInterruptReason", + "#ltia": "lastTurnInterruptedAt", + }, + # The row must still be exactly what the scan saw. A session that + # ran a new turn in between has either cleared the marker itself + # or been re-marked by a real interruption; both must be left + # alone. + ConditionExpression=( + "attribute_exists(#lti) AND #ltia = :ts AND #ltr = :reason" + ), + ExpressionAttributeValues={ + ":ts": row["lastTurnInterruptedAt"], + ":reason": "navigated_away", + }, + ) + return "cleared" + except ClientError as e: + if e.response.get("Error", {}).get("Code") == "ConditionalCheckFailedException": + return "raced" + print(f" ERROR on {row.get('GSI_PK', row['SK'])}: {e}", file=sys.stderr) + return "error" + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + parser.add_argument("--table", required=True, help="Sessions-metadata DynamoDB table name") + parser.add_argument("--profile", default=None, help="AWS profile (default: ambient credentials)") + parser.add_argument("--region", default="us-west-2") + parser.add_argument( + "--min-gap-seconds", + type=int, + default=DEFAULT_MIN_GAP_SECONDS, + help=( + "Only clear markers written more than this long after the session's " + f"last message (default {DEFAULT_MIN_GAP_SECONDS}; must exceed the " + "600s stream timeout to stay provably false)" + ), + ) + parser.add_argument("--apply", action="store_true", help="Write. Without it, dry-run only.") + args = parser.parse_args() + + if args.min_gap_seconds <= 600: + parser.error( + "--min-gap-seconds must exceed the 600s stream timeout; below that a " + "marker can belong to a turn that was genuinely still running." + ) + + session = boto3.Session(profile_name=args.profile, region_name=args.region) + table = session.resource("dynamodb").Table(args.table) + + print(f"Scanning {args.table} ({args.region}) for interrupted-turn markers…") + items = list(_scan_marked_rows(table)) + selected, skipped = _select(items, args.min_gap_seconds) + + print(f"\n rows carrying a marker: {len(items)}") + print(f" left alone — not navigated_away: {skipped['other_reason']}") + print(f" left alone — gap <= {args.min_gap_seconds}s:{'':<9}{skipped['gap_too_small']}") + if skipped["unparseable"]: + print(f" left alone — unparseable dates: {skipped['unparseable']}") + print(f" provably false, to clear: {len(selected)}") + + if selected: + print("\n widest gaps:") + for row in selected[:5]: + days = row["_gapSeconds"] / 86400 + print(f" {row.get('GSI_PK', row['SK'])} +{days:.1f}d after last message") + + if not args.apply: + print("\nDRY RUN — nothing written. Re-run with --apply to clear.") + return 0 + + print(f"\nClearing {len(selected)} markers…") + tally = {"cleared": 0, "raced": 0, "error": 0} + for row in selected: + tally[_clear(table, row)] += 1 + + print(f"\n cleared: {tally['cleared']}") + print(f" skipped (row changed under us): {tally['raced']}") + print(f" errors: {tally['error']}") + return 1 if tally["error"] else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/src/apis/app_api/sessions/routes.py b/backend/src/apis/app_api/sessions/routes.py index 7567b665..69ad4e80 100644 --- a/backend/src/apis/app_api/sessions/routes.py +++ b/backend/src/apis/app_api/sessions/routes.py @@ -685,7 +685,9 @@ async def signal_turn_interrupted_endpoint( user left; they did not reject anything. **Recorded only** — the running turn is deliberately left alone, matching today's behaviour where a refresh lets the turn finish server-side and the reload - offers to continue it. + offers to continue it. Recorded only while the session's + single-flight lease is held: "mid-turn" is a claim this endpoint + verifies rather than takes from the client (see the gate below). Lives on app-api, not inference-api: the AgentCore Runtime data plane only proxies ``/invocations`` + ``/ping``, so a custom inference-api @@ -703,6 +705,29 @@ async def signal_turn_interrupted_endpoint( logger.info("POST /sessions/.../interrupt (reason=%s)", body.reason) try: + # A departure can only interrupt a turn that is actually running. + # The SPA decides that from its own transport state, which has been + # wrong before: a controller left behind after a completed stream + # made every finished turn in the tab eligible, so a later refresh + # marked complete answers as interrupted (a false "Response + # interrupted" chip, and a false interruption note on the session's + # next prompt). The single-flight lease is the server's own answer to + # "is a turn in flight", so assert it here rather than trusting the + # client — old tabs keep running the old SPA long after the fix ships. + # + # `user_stopped` is deliberately NOT gated: the Stop button only + # exists while streaming, and it also arms cancellation below, which + # must reach a turn whose lease read fails for any reason. + if body.reason == "navigated_away": + from apis.shared.sessions.session_lease import is_session_lease_held + + if not await is_session_lease_held(session_id, user_id): + logger.info( + "Ignoring navigated_away for session %s — no turn in flight", + session_id, + ) + return Response(status_code=204) + await set_interrupted_turn( session_id, user_id, diff --git a/backend/tests/routes/test_sessions.py b/backend/tests/routes/test_sessions.py index b16a3891..87896db5 100644 --- a/backend/tests/routes/test_sessions.py +++ b/backend/tests/routes/test_sessions.py @@ -1036,6 +1036,9 @@ def test_returns_204_and_records_navigated_away(self, app, make_user, authentica with patch( "apis.app_api.sessions.routes.set_interrupted_turn", recorder, + ), patch( + "apis.shared.sessions.session_lease.is_session_lease_held", + AsyncMock(return_value=True), ): resp = client.post( "/sessions/sess-001/interrupt", @@ -1050,6 +1053,76 @@ def test_returns_204_and_records_navigated_away(self, app, make_user, authentica source="client_signal", ) + def test_ignores_navigated_away_when_no_turn_is_in_flight( + self, app, make_user, authenticated_client + ): + """A departure can only interrupt a turn that is actually running. + + The SPA decides "is a turn in flight" from its own transport state, + and that has been wrong: a controller left behind after a completed + stream made every finished turn in the tab eligible, so a later + refresh marked complete answers as interrupted. The lease is the + server's own answer to the same question, and old tabs keep running + the old SPA long after a client fix ships — so assert it here. + """ + user = make_user() + client = authenticated_client(app, user) + + recorder = AsyncMock() + with patch( + "apis.app_api.sessions.routes.set_interrupted_turn", + recorder, + ), patch( + "apis.shared.sessions.session_lease.is_session_lease_held", + AsyncMock(return_value=False), + ): + resp = client.post( + "/sessions/sess-001/interrupt", + json={"reason": "navigated_away"}, + ) + + # Still 204: the client never waits on this and a dropped signal is + # not an error it can act on. + assert resp.status_code == 204 + recorder.assert_not_awaited() + + def test_user_stopped_is_not_gated_on_the_lease(self, app, make_user, authenticated_client): + """Stop is never withheld on a lease read. + + The button only exists while a response is streaming, and the same + request arms distributed cancellation — which must still reach a turn + whose lease read fails or fails open (`is_session_lease_held` reports + False for an unconfigured table and for any DynamoDB error). + """ + user = make_user() + client = authenticated_client(app, user) + + recorder = AsyncMock() + cancel = AsyncMock(return_value=True) + with patch( + "apis.app_api.sessions.routes.set_interrupted_turn", + recorder, + ), patch( + "apis.shared.sessions.session_lease.is_session_lease_held", + AsyncMock(return_value=False), + ), patch( + "apis.shared.sessions.session_lease.request_session_cancel", + cancel, + ): + resp = client.post( + "/sessions/sess-001/interrupt", + json={"reason": "user_stopped"}, + ) + + assert resp.status_code == 204 + recorder.assert_awaited_once_with( + "sess-001", + user.user_id, + reason="user_stopped", + source="client_signal", + ) + cancel.assert_awaited_once_with("sess-001", user.user_id) + def test_navigated_away_does_not_cancel_the_turn(self, app, make_user, authenticated_client): """Attribution, not instruction. @@ -1064,6 +1137,11 @@ def test_navigated_away_does_not_cancel_the_turn(self, app, make_user, authentic with patch( "apis.app_api.sessions.routes.set_interrupted_turn", AsyncMock(), + ), patch( + # Held, so the departure is genuinely mid-turn — otherwise this + # would assert nothing beyond the gate above. + "apis.shared.sessions.session_lease.is_session_lease_held", + AsyncMock(return_value=True), ), patch( "apis.shared.sessions.session_lease.request_session_cancel", cancel, diff --git a/backend/tests/test_backfill_false_interrupted_markers.py b/backend/tests/test_backfill_false_interrupted_markers.py new file mode 100644 index 00000000..6628abd2 --- /dev/null +++ b/backend/tests/test_backfill_false_interrupted_markers.py @@ -0,0 +1,194 @@ +"""Tests for the false interrupted-turn marker backfill. + +The risk this guards is asymmetric. Leaving a stale marker behind costs a +spurious "Response interrupted" chip and one false `` on +that session's next prompt; clearing a REAL one destroys the only record that +a turn was cut off, and with it the reload's offer to continue. So the +assertions that matter most are about what the script refuses to touch. + +The 900s threshold is the whole safety argument: a genuine interruption does +not always bump `lastMessageAt` (the "marker only, no synthetic write" branch +of `_persist_interruption`), so a modest gap is ambiguous — but no turn is +still running 15 minutes after its last message, because the stream times out +at 600s. +""" + +from __future__ import annotations + +import importlib.util +import pathlib + +import boto3 +import pytest +from moto import mock_aws + +REGION = "us-east-1" +TABLE = "test-sessions-metadata" + +_SCRIPT = ( + pathlib.Path(__file__).resolve().parents[1] + / "scripts" + / "backfill_false_interrupted_markers.py" +) +_spec = importlib.util.spec_from_file_location("backfill_interrupt_markers", _SCRIPT) +backfill_mod = importlib.util.module_from_spec(_spec) +assert _spec.loader is not None +_spec.loader.exec_module(backfill_mod) + + +@pytest.fixture() +def table(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("AWS_DEFAULT_REGION", REGION) + monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing") + monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing") + with mock_aws(): + ddb = boto3.resource("dynamodb", region_name=REGION) + ddb.create_table( + TableName=TABLE, + KeySchema=[ + {"AttributeName": "PK", "KeyType": "HASH"}, + {"AttributeName": "SK", "KeyType": "RANGE"}, + ], + AttributeDefinitions=[ + {"AttributeName": "PK", "AttributeType": "S"}, + {"AttributeName": "SK", "AttributeType": "S"}, + ], + BillingMode="PAY_PER_REQUEST", + ) + yield ddb.Table(TABLE) + + +def marked_row( + session_id: str, + *, + reason: str = "navigated_away", + last_message_at: str = "2026-09-01T12:00:00+00:00", + marked_at: str = "2026-09-01T13:00:00+00:00", +) -> dict: + return { + "PK": "USER#u1", + "SK": f"S#{session_id}", + "GSI_PK": f"SESSION#{session_id}", + "lastTurnInterrupted": True, + "lastTurnInterruptReason": reason, + "lastTurnInterruptedAt": marked_at, + "lastMessageAt": last_message_at, + "title": "keep me", + } + + +# --------------------------------------------------------------------------- +# Selection +# --------------------------------------------------------------------------- + + +def test_selects_navigated_away_marked_long_after_the_last_message(): + rows = [marked_row("s1")] # +1h + selected, _ = backfill_mod._select(rows, backfill_mod.DEFAULT_MIN_GAP_SECONDS) + assert [r["GSI_PK"] for r in selected] == ["SESSION#s1"] + + +@pytest.mark.parametrize("reason", ["user_stopped", "connection_lost", "unknown"]) +def test_never_selects_a_reason_this_bug_cannot_produce(reason: str): + """Only the client's `navigated_away` path had the stale-controller bug. + + `user_stopped` is the user's own attested intent and `connection_lost` + is the server's own backstop — clearing either would erase a real record. + """ + rows = [marked_row("s1", reason=reason)] + selected, skipped = backfill_mod._select(rows, backfill_mod.DEFAULT_MIN_GAP_SECONDS) + assert selected == [] + assert skipped["other_reason"] == 1 + + +def test_leaves_an_ambiguous_gap_alone(): + """A 10-minute gap is inside the 600s stream timeout. + + An interrupted continuation persists no assistant message, so + `lastMessageAt` stays at the previous turn and a real interruption can + show a positive gap. Below the threshold we cannot tell, so we don't act. + """ + rows = [marked_row("s1", marked_at="2026-09-01T12:10:00+00:00")] + selected, skipped = backfill_mod._select(rows, backfill_mod.DEFAULT_MIN_GAP_SECONDS) + assert selected == [] + assert skipped["gap_too_small"] == 1 + + +def test_leaves_a_marker_that_predates_the_last_message_alone(): + """The shape of a genuine mid-turn departure: the signal lands first, the + turn's partial is persisted after it.""" + rows = [ + marked_row( + "s1", + last_message_at="2026-09-01T12:00:30+00:00", + marked_at="2026-09-01T12:00:00+00:00", + ) + ] + selected, _ = backfill_mod._select(rows, backfill_mod.DEFAULT_MIN_GAP_SECONDS) + assert selected == [] + + +def test_leaves_unparseable_timestamps_alone(): + rows = [marked_row("s1", marked_at="not-a-date")] + selected, skipped = backfill_mod._select(rows, backfill_mod.DEFAULT_MIN_GAP_SECONDS) + assert selected == [] + assert skipped["unparseable"] == 1 + + +# --------------------------------------------------------------------------- +# Writes +# --------------------------------------------------------------------------- + + +def test_clear_removes_only_the_marker_attributes(table): + row = marked_row("s1") + table.put_item(Item=row) + + assert backfill_mod._clear(table, row) == "cleared" + + stored = table.get_item(Key={"PK": row["PK"], "SK": row["SK"]})["Item"] + for attr in backfill_mod.MARKER_ATTRS: + assert attr not in stored + # Everything else on the row survives — this is a targeted REMOVE, not a + # rewrite (a full put_item would drop attributes the script never read). + assert stored["title"] == "keep me" + assert stored["lastMessageAt"] == row["lastMessageAt"] + + +def test_clear_skips_a_row_re_marked_since_the_scan(table): + """A session that ran a new turn between scan and write may have been + legitimately re-marked. The write is conditional on the exact timestamp + the scan read, so it declines rather than clobbering.""" + row = marked_row("s1") + table.put_item(Item=row) + table.update_item( + Key={"PK": row["PK"], "SK": row["SK"]}, + UpdateExpression="SET lastTurnInterruptedAt = :ts", + ExpressionAttributeValues={":ts": "2026-09-02T09:00:00+00:00"}, + ) + + assert backfill_mod._clear(table, row) == "raced" + + stored = table.get_item(Key={"PK": row["PK"], "SK": row["SK"]})["Item"] + assert stored["lastTurnInterrupted"] is True + assert stored["lastTurnInterruptedAt"] == "2026-09-02T09:00:00+00:00" + + +def test_clear_skips_a_row_whose_reason_was_upgraded(table): + """`set_interrupted_turn` lets a stronger reason overwrite a weaker one. + If `user_stopped` landed after our scan, this is no longer our row.""" + row = marked_row("s1") + table.put_item(Item={**row, "lastTurnInterruptReason": "user_stopped"}) + + assert backfill_mod._clear(table, row) == "raced" + stored = table.get_item(Key={"PK": row["PK"], "SK": row["SK"]})["Item"] + assert stored["lastTurnInterruptReason"] == "user_stopped" + + +def test_clear_is_idempotent(table): + row = marked_row("s1") + table.put_item(Item=row) + + assert backfill_mod._clear(table, row) == "cleared" + # A second pass finds nothing to do rather than erroring. + assert backfill_mod._clear(table, row) == "raced" 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 d3c2def6..23ec7c97 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 @@ -29,9 +29,9 @@ describe('ChatHttpService', () => { // and lets cookie auth ride along with the request rather than // attaching a Bearer manually. { provide: BffSessionService, useValue: { csrfHeaders: vi.fn().mockReturnValue({}), handleUnauthorized: vi.fn() } }, - { provide: SessionService, useValue: { currentSession: signal({ sessionId: 's1' }), updateSessionTitleInCache: vi.fn(), getSessionMetadata: vi.fn().mockResolvedValue({}) } }, + { provide: SessionService, useValue: { currentSession: signal({ sessionId: 's1' }), updateSessionTitleInCache: vi.fn(), getSessionMetadata: vi.fn().mockResolvedValue({}), isNewSession: vi.fn().mockReturnValue(false) } }, { provide: StreamParserService, useValue: { getCurrentStreamId: vi.fn().mockReturnValue('stream-1'), parseEventSourceMessage: vi.fn() } }, - { provide: ChatStateService, useValue: { abortRequest: vi.fn(), setChatLoading: vi.fn(), setLastTurnInterrupted: vi.fn(), seedSessionAggregates: vi.fn(), createAbortController: vi.fn().mockReturnValue(new AbortController()), streamingSessionIds: vi.fn().mockReturnValue([]) } }, + { provide: ChatStateService, useValue: { abortRequest: vi.fn(), setChatLoading: vi.fn(), setLastTurnInterrupted: vi.fn(), seedSessionAggregates: vi.fn(), createAbortController: vi.fn().mockReturnValue(new AbortController()), releaseAbortController: vi.fn(), streamingSessionIds: vi.fn().mockReturnValue([]) } }, { provide: MessageMapService, useValue: { endStreaming: vi.fn() } }, { provide: ErrorService, useValue: { handleHttpError: vi.fn(), addError: vi.fn() } }, ], @@ -145,6 +145,28 @@ describe('ChatHttpService', () => { vi.restoreAllMocks(); }); + it('releases the session controller when a stream finishes normally', async () => { + // Without this the controller outlives its stream for the life of the + // tab, so `streamingSessionIds()` keeps reporting a finished turn and the + // next page-hide attributes it to `navigated_away` — a "Response + // interrupted" chip on a complete answer. + const controller = new AbortController(); + chatStateService.createAbortController.mockReturnValue(controller); + vi.spyOn(globalThis, 'fetch').mockResolvedValue( + new Response('event: done\ndata: {}\n\n', { + status: 200, + headers: { 'content-type': 'text/event-stream' }, + }), + ); + + await service.sendChatRequest({ session_id: 's1', message: 'hi' }); + + expect(chatStateService.releaseAbortController).toHaveBeenCalledWith('s1', controller); + // Released, not aborted — the turn completed on its own. + expect(controller.signal.aborted).toBe(false); + vi.restoreAllMocks(); + }); + it('surfaces a soft "Already responding" notice (not a hard error) on a 409 single-flight rejection', async () => { // The inference-api single-flight guard rejects a duplicate turn while the // prior one is still streaming server-side; the BFF relays it as 409. 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 9c31cf91..c1333c2e 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 @@ -134,6 +134,11 @@ export class ChatHttpService { if (!isCurrentStream()) return; this.messageMapService.endStreaming(sessionId); this.chatStateService.setChatLoading(sessionId, false); + // Release the transport handle too. `streamingSessionIds()` reads it + // to decide which turns a page departure interrupted, so a controller + // left behind here makes every completed turn in this tab eligible for + // a `navigated_away` marker on the next refresh / tab close. + this.chatStateService.releaseAbortController(sessionId, abortController); }; try { 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 5380c061..cce75972 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 @@ -205,5 +205,39 @@ describe('ChatStateService', () => { it('abortRequest is a no-op for sessions without an in-flight request', () => { expect(() => service.abortRequest('nope')).not.toThrow(); }); + + it('drops a session from streamingSessionIds once its controller is released', () => { + // The bug this guards: a completed stream used to leave its controller + // behind, so the session looked in-flight for the life of the tab and + // the next page-hide marked its finished turn `navigated_away`. + const a = service.createAbortController('a'); + service.createAbortController('b'); + + expect(service.streamingSessionIds().sort()).toEqual(['a', 'b']); + + service.releaseAbortController('a', a); + + expect(service.streamingSessionIds()).toEqual(['b']); + // Released, not aborted — the stream finished on its own. + expect(a.signal.aborted).toBe(false); + }); + + it('releaseAbortController ignores a controller the session no longer owns', () => { + // A superseded stream's late teardown must not clear the controller of + // the stream that replaced it. + const first = service.createAbortController('a'); + const second = service.createAbortController('a'); + + service.releaseAbortController('a', first); + + expect(service.streamingSessionIds()).toEqual(['a']); + expect(second.signal.aborted).toBe(false); + }); + + it('releaseAbortController is a no-op for an unknown session', () => { + expect(() => + service.releaseAbortController('nope', new AbortController()), + ).not.toThrow(); + }); }); }); 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 df3df963..6f81d303 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 @@ -234,12 +234,35 @@ export class ChatStateService { return controller; } + /** + * Release a session's controller once its stream has finished. + * + * The counterpart to `createAbortController`, and what makes + * `streamingSessionIds()` honest: without it a controller outlives its + * stream for the life of the tab, so every completed turn keeps looking + * in-flight and the page-hide attribution below marks turns that + * finished minutes (or days) earlier as `navigated_away`. That is a + * false "Response interrupted" chip on a complete answer, and a false + * interruption note prepended to the session's next prompt. + * + * Identity-checked: a superseded stream's late teardown must not clear + * the controller of the stream that replaced it (`createAbortController` + * installs the replacement before the old stream's abort unwinds). + */ + releaseAbortController(sessionId: string, controller: AbortController): void { + const state = this.states().get(sessionId); + if (state && state.abortController === controller) { + state.abortController = null; + } + } + /** * Session ids with a stream in flight right now. * * Read at page-hide time to attribute a departure to the turns it * actually interrupted. A live controller is the truthful test: it is - * created per request and nulled on abort, so it tracks the transport + * created per request and released on abort (`abortRequest`) or on + * stream teardown (`releaseAbortController`), so it tracks the transport * rather than the `loading` flag, which other code also drives. */ streamingSessionIds(): string[] {