From 34db9a30bda9c9c88ca40844740110f9d6e2a29e Mon Sep 17 00:00:00 2001 From: Phil Merrell Date: Sun, 6 Sep 2026 14:52:12 -0600 Subject: [PATCH 1/2] fix(sessions): stop marking completed turns as interrupted MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A session's AbortController was created per request but released only by the Stop button — never on normal stream teardown. So after any turn that finished on its own, `streamingSessionIds()` kept reporting that session as in-flight for the life of the tab, and the page-hide attribution handler signalled `navigated_away` for it on the next refresh, tab close, or cross-document navigation. One departure marked every session ever streamed in that tab. The result is a "Response interrupted" chip plus a Continue button on a complete answer, and — invisibly — a false `` prepended to that session's next prompt and persisted in its history, telling the model its previous response was cut off and to pick up where it left off. Measured in prod: 1,678 sessions carry the marker and 1,505 of them (90%) received it more than 30s after their last message landed; 1,475 of those are `navigated_away`. Not model-specific and not new — the oldest is 2026-08-21. Two changes: * `releaseAbortController` on stream teardown (identity-checked, so a superseded stream's late close can't clear its replacement's handle). This makes `streamingSessionIds()` mean what its docstring claims. * `POST /sessions/{id}/interrupt` now verifies "mid-turn" against the session's single-flight lease before recording `navigated_away`, rather than taking the client's word for it. Old tabs run the old SPA long after this ships. `user_stopped` is deliberately not gated: the button only exists while streaming, and it also arms distributed cancellation, which must still reach a turn whose lease read fails open. Co-Authored-By: Claude Opus 5 --- backend/src/apis/app_api/sessions/routes.py | 27 ++++++- backend/tests/routes/test_sessions.py | 78 +++++++++++++++++++ .../services/chat/chat-http.service.spec.ts | 26 ++++++- .../services/chat/chat-http.service.ts | 5 ++ .../services/chat/chat-state.service.spec.ts | 34 ++++++++ .../services/chat/chat-state.service.ts | 25 +++++- 6 files changed, 191 insertions(+), 4 deletions(-) diff --git a/backend/src/apis/app_api/sessions/routes.py b/backend/src/apis/app_api/sessions/routes.py index 7567b665a..69ad4e80b 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 b16a3891d..87896db51 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/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 d3c2def6a..23ec7c97c 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 9c31cf913..c1333c2e5 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 5380c061a..cce75972a 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 df3df963e..6f81d303e 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[] { From 2d11086baa539821dd3692c8224c377e07722038 Mon Sep 17 00:00:00 2001 From: Phil Merrell Date: Sun, 6 Sep 2026 14:57:48 -0600 Subject: [PATCH 2/2] chore(sessions): backfill script for false interrupted-turn markers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One-shot cleanup for the markers the stale-controller bug left behind. Each one shows a "Response interrupted" chip plus a Continue button on a complete answer, and puts a false into that session's next prompt — persisted in history, invisible in the UI. Markers self-clear only at the start of that session's next turn, so a conversation nobody returns to stays armed indefinitely. Clears only what is provably false: reason `navigated_away`, marked more than 900s after the session's last message. Gap alone is not enough — an interrupted continuation persists no assistant message, so `lastMessageAt` stays at the previous turn and a real interruption can show a modest positive gap. But the SSE stream times out at 600s, so nothing is still running 15 minutes later. `user_stopped` and `connection_lost` rows are never touched: neither comes from this bug, and clearing one would destroy a real record. Dry-run by default. Each write is conditional on the exact timestamp and reason the scan read, so a session re-marked in between is skipped rather than clobbered; re-running is a no-op. Prod dry-run (2026-09-06): 1,680 marked rows → 787 provably false, 847 left alone as ambiguous, 46 left alone as another reason. Co-Authored-By: Claude Opus 5 --- .../backfill_false_interrupted_markers.py | 221 ++++++++++++++++++ ...test_backfill_false_interrupted_markers.py | 194 +++++++++++++++ 2 files changed, 415 insertions(+) create mode 100644 backend/scripts/backfill_false_interrupted_markers.py create mode 100644 backend/tests/test_backfill_false_interrupted_markers.py diff --git a/backend/scripts/backfill_false_interrupted_markers.py b/backend/scripts/backfill_false_interrupted_markers.py new file mode 100644 index 000000000..eebec127b --- /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/tests/test_backfill_false_interrupted_markers.py b/backend/tests/test_backfill_false_interrupted_markers.py new file mode 100644 index 000000000..6628abd24 --- /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"