From b499a0f025431a0513660bfd352883905975ae8d Mon Sep 17 00:00:00 2001 From: Phil Merrell Date: Fri, 17 Jul 2026 15:25:51 -0600 Subject: [PATCH] =?UTF-8?q?feat(sessions):=20static-SK=20write=20path=20?= =?UTF-8?q?=E2=80=94=20born=20static,=20self-migrate,=20no=20rotation=20(i?= =?UTF-8?q?ssue=20#175=20Phase=201b)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Turns on the write side of the static-sort-key migration. Sessions stop encoding lastMessageAt in the sort key, so the row never moves and the ghost-row race that produced "Failed to parse session item" warnings is structurally eliminated for every migrated row. Changed (all resolve the row via GSI, which is SK-scheme-agnostic): - ensure_session_metadata_exists: new sessions born at static SK S#{id} + GSI4 keys, with a real attribute_not_exists(PK) conditional put. The deterministic SK makes the guard meaningful, closing the first-turn duplicate-row race the old timestamped SK made impossible to gate. - update_session_activity: drops the per-turn Phase-B rotation. Static rows update in place (SET GSI4_SK re-positions the sparse recency index — no row move); a still-legacy row does its one-time final rotation to the static SK, carrying any concurrent write. - _store_session_metadata_cloud: static SK; migrate legacy->static on move; SET GSI4 for active, REMOVE for deleted; never un-migrates a static row. - session_service.delete_session: resolves the raw SK via _get_session_by_gsi instead of reconstructing S#ACTIVE#{lastMessageAt}#{id} (which misses migrated rows). Non-rotating soft-delete: SET status=deleted + REMOVE GSI4 in place, or migrate a legacy row to a static tombstone. Drops the S#DELETED# prefix (nothing reads it). The ~10 other writers resolve-then-update-in-place on the current SK and need no change — they already work on a static SK and never rotate. Tests: TestWriteSideMigration (born-static, one-time migrate, no rotation, soft-delete in-place/legacy, end-to-end create->activity->list->delete) plus the real ConditionalCheckFailedException contract (moto raises it for a failed conditional put). Updated three tests that encoded the old rotation contract. Full shared+routes suites: 1689 passed. Co-Authored-By: Claude Opus 4.8 --- .../sessions/services/session_service.py | 86 ++++---- backend/src/apis/shared/sessions/metadata.py | 193 ++++++++++++------ .../tests/shared/test_sessions_metadata.py | 144 +++++++++++-- 3 files changed, 301 insertions(+), 122 deletions(-) diff --git a/backend/src/apis/app_api/sessions/services/session_service.py b/backend/src/apis/app_api/sessions/services/session_service.py index bb0a791b3..02159b6e0 100644 --- a/backend/src/apis/app_api/sessions/services/session_service.py +++ b/backend/src/apis/app_api/sessions/services/session_service.py @@ -221,56 +221,60 @@ async def delete_session(self, user_id: str, session_id: str) -> bool: return False try: - # Get current session via GSI to find its SK - session = await self.get_session(user_id, session_id) - if not session: + # Resolve the raw row (with its actual SK) via GSI. Do NOT reconstruct the + # SK from lastMessageAt — under the static-SK schema (issue #175) a migrated + # row lives at S#{session_id}, not S#ACTIVE#{lastMessageAt}#{id}, so a + # reconstructed key would miss it entirely. + from apis.shared.sessions.metadata import _get_session_by_gsi, _static_session_sk + + existing = await _get_session_by_gsi(session_id, user_id, self.table) + if not existing: logger.info("Session not found for deletion") return False - if session.deleted: + if existing.get('deleted') or existing.get('status') == 'deleted': logger.info("Session already deleted") return True now = datetime.now(timezone.utc) deleted_at = now.isoformat() - - # Build old and new SKs - old_sk = f'S#ACTIVE#{session.last_message_at}#{session_id}' - new_sk = f'S#DELETED#{deleted_at}#{session_id}' pk = f'USER#{user_id}' - - # Build the deleted item with all fields - deleted_item = { - 'PK': pk, - 'SK': new_sk, - 'GSI_PK': f'SESSION#{session_id}', - 'GSI_SK': 'META', - 'sessionId': session_id, - 'userId': user_id, - 'title': session.title or '', - 'status': 'deleted', - 'createdAt': session.created_at, - 'lastMessageAt': session.last_message_at, - 'messageCount': session.message_count or 0, - 'starred': session.starred or False, - 'tags': session.tags or [], - 'deleted': True, - 'deletedAt': deleted_at - } - - # Include preferences if present - # Convert floats to Decimals since DynamoDB high-level API requires Decimal for numbers - if session.preferences: - prefs = session.preferences.model_dump(by_alias=True) - deleted_item['preferences'] = _convert_float_to_decimal(prefs) - - # Use high-level API: put_item + delete_item - # Put new item first, then delete old - if put fails, nothing is lost - # This is simpler and more reliable than transact_write_items - self.table.put_item(Item=deleted_item) - self.table.delete_item( - Key={'PK': pk, 'SK': old_sk} - ) + old_sk = existing['SK'] + target_sk = _static_session_sk(session_id) + + if old_sk == target_sk: + # Already migrated — soft-delete in place: flip status + drop the sparse + # recency keys so the row leaves the active listing. No row move. + self.table.update_item( + Key={'PK': pk, 'SK': target_sk}, + UpdateExpression=( + "SET #s = :d, deleted = :true, deletedAt = :da " + "REMOVE GSI4_PK, GSI4_SK" + ), + ExpressionAttributeNames={'#s': 'status'}, + ExpressionAttributeValues={':d': 'deleted', ':true': True, ':da': deleted_at}, + ) + else: + # Legacy row — migrate to the static tombstone (status=deleted, no GSI4) + # and drop the old row. One-time move; carry existing fields. + deleted_item = { + k: v for k, v in existing.items() + if k not in ('PK', 'SK', 'GSI4_PK', 'GSI4_SK') + } + deleted_item.update({ + 'PK': pk, + 'SK': target_sk, + 'GSI_PK': f'SESSION#{session_id}', + 'GSI_SK': 'META', + 'sessionId': session_id, + 'userId': user_id, + 'status': 'deleted', + 'deleted': True, + 'deletedAt': deleted_at, + }) + # existing came back with Decimals converted to floats — put_item needs Decimal. + self.table.put_item(Item=_convert_float_to_decimal(deleted_item)) + self.table.delete_item(Key={'PK': pk, 'SK': old_sk}) logger.info("Soft-deleted session") diff --git a/backend/src/apis/shared/sessions/metadata.py b/backend/src/apis/shared/sessions/metadata.py index bb47d985e..ea6d6b7db 100644 --- a/backend/src/apis/shared/sessions/metadata.py +++ b/backend/src/apis/shared/sessions/metadata.py @@ -641,14 +641,12 @@ async def _store_session_metadata_cloud( # Convert floats to Decimal for DynamoDB compatibility item = _convert_floats_to_decimal(item) - # Determine SK based on session status + # Static SK (issue #175): identity no longer encodes lastMessageAt or status, + # so the row never moves. active/deleted is the `status` attribute; recency + # lives in the sparse SessionRecencyIndex (GSI4), present only while active. last_message_at = session_metadata.last_message_at or datetime.now(timezone.utc).isoformat() - - if session_metadata.deleted: - deleted_at = session_metadata.deleted_at or datetime.now(timezone.utc).isoformat() - new_sk = f"S#DELETED#{deleted_at}#{session_id}" - else: - new_sk = f"S#ACTIVE#{last_message_at}#{session_id}" + is_active = not session_metadata.deleted + new_sk = _static_session_sk(session_id) # Build primary key pk = f'USER#{user_id}' @@ -656,41 +654,43 @@ async def _store_session_metadata_cloud( # Add GSI keys for direct lookup item['GSI_PK'] = f'SESSION#{session_id}' item['GSI_SK'] = 'META' + # Sparse recency keys — added for active, absent for deleted. + item.update(_recency_gsi_keys(user_id, session_id, last_message_at, is_active)) if existing_session: - # Session exists - check if SK needs to change + # Session exists - check if the (now static) SK needs to change, which + # only happens when migrating a legacy row (S#ACTIVE#…/S#DELETED#…). old_sk = existing_session.get('SK') if old_sk and old_sk != new_sk: - # SK changed (timestamp updated) - need transactional move - # Deep merge existing with new data + # Legacy row → migrate to the static SK. Deep-merge existing onto the + # new item, but drop any stale GSI4 keys so status drives them freshly. merged_item = _deep_merge( - {k: v for k, v in existing_session.items() if k not in ['PK', 'SK']}, + {k: v for k, v in existing_session.items() + if k not in ['PK', 'SK', 'GSI4_PK', 'GSI4_SK']}, item ) merged_item['PK'] = pk merged_item['SK'] = new_sk + if not is_active: + merged_item.pop('GSI4_PK', None) + merged_item.pop('GSI4_SK', None) - # Move session: put new SK first, then delete old SK - # Using high-level Table API (put_item + delete_item) instead of - # transact_write_items to avoid low-level serialization issues - logger.debug(f"🔄 Moving session: old_sk={old_sk[:50]}..., new_sk={new_sk[:50]}...") + # Put new SK first, then delete old — if the put fails the original + # is untouched. This is the row's one-time migration move. + logger.debug(f"🔄 Migrating session to static SK: old_sk={old_sk[:50]}...") try: - # Convert floats to Decimal for DynamoDB compatibility decimal_item = _convert_floats_to_decimal(merged_item) - - # Put new item first — if this fails, original is untouched table.put_item(Item=decimal_item) - # Delete old item table.delete_item(Key={'PK': pk, 'SK': old_sk}) - logger.info(f"💾 Moved session metadata in DynamoDB (SK changed)") + logger.info(f"💾 Migrated session metadata to static SK") except Exception as move_error: - logger.error(f"Session move failed - PK={pk}, old_SK={old_sk}, new_SK={new_sk}") - logger.error(f"Move error: {move_error}") + logger.error(f"Session migration failed - PK={pk}, old_SK={old_sk}, new_SK={new_sk}") + logger.error(f"Migration error: {move_error}") raise else: - # SK unchanged - simple update with deep merge - # Build update expression for partial update + # Already static — in-place update. GSI4 keys are SET when active and + # REMOVEd when the session is (being) soft-deleted. update_expression_parts = [] expression_attribute_names = {} expression_attribute_values = {} @@ -707,17 +707,30 @@ async def _store_session_metadata_cloud( expression_attribute_names[placeholder_name] = key_name expression_attribute_values[placeholder_value] = value + remove_parts = [] + if not is_active: + for gsi_key in ('GSI4_PK', 'GSI4_SK'): + expression_attribute_names[f"#{gsi_key}"] = gsi_key + remove_parts.append(f"#{gsi_key}") + + update_expression = "" if update_expression_parts: update_expression = "SET " + ", ".join(update_expression_parts) - table.update_item( - Key={'PK': pk, 'SK': old_sk}, - UpdateExpression=update_expression, - ExpressionAttributeNames=expression_attribute_names, - ExpressionAttributeValues=expression_attribute_values - ) + if remove_parts: + update_expression += (" " if update_expression else "") + "REMOVE " + ", ".join(remove_parts) + + if update_expression: + kwargs = { + 'Key': {'PK': pk, 'SK': old_sk}, + 'UpdateExpression': update_expression, + 'ExpressionAttributeNames': expression_attribute_names, + } + if expression_attribute_values: + kwargs['ExpressionAttributeValues'] = expression_attribute_values + table.update_item(**kwargs) logger.info(f"💾 Updated session metadata in DynamoDB table {table_name}") else: - # New session - create with put_item + # New session - create with put_item at the static SK item['PK'] = pk item['SK'] = new_sk table.put_item(Item=item) @@ -740,6 +753,28 @@ async def _store_session_metadata_cloud( ) +def _static_session_sk(session_id: str) -> str: + """Target base sort key (issue #175): static — does NOT encode lastMessageAt, + so the row never has to move. One row per session for its whole lifetime.""" + return f"S#{session_id}" + + +def _recency_gsi_keys( + user_id: str, session_id: str, last_message_at: str, is_active: bool +) -> Dict[str, str]: + """SessionRecencyIndex (GSI4) keys for newest-first active-session listing. + + Sparse: present only while the session is active, so soft-delete simply removes + them and the row drops out of the recency list (mirrors DueScheduleIndex/GSI3). + """ + if is_active: + return { + "GSI4_PK": f"USER#{user_id}", + "GSI4_SK": f"{last_message_at}#{session_id}", + } + return {} + + async def ensure_session_metadata_exists(session_id: str, user_id: str) -> bool: """Idempotently create a session metadata row if it doesn't exist yet. @@ -772,11 +807,14 @@ async def ensure_session_metadata_exists(session_id: str, user_id: str) -> bool: try: import boto3 + from botocore.exceptions import ClientError from datetime import datetime, timezone dynamodb = boto3.resource("dynamodb") table = dynamodb.Table(sessions_metadata_table) + # Catch a pre-existing row (legacy S#ACTIVE#… or already-migrated S#{id}) so + # we don't create a second row for the same session. existing = await _get_session_by_gsi(session_id, user_id, table) if existing is not None: return False @@ -784,9 +822,10 @@ async def ensure_session_metadata_exists(session_id: str, user_id: str) -> bool: now = datetime.now(timezone.utc).isoformat() item = { "PK": f"USER#{user_id}", - "SK": f"S#ACTIVE#{now}#{session_id}", + "SK": _static_session_sk(session_id), "GSI_PK": f"SESSION#{session_id}", "GSI_SK": "META", + **_recency_gsi_keys(user_id, session_id, now, is_active=True), "sessionId": session_id, "userId": user_id, "title": "New Conversation", @@ -798,7 +837,18 @@ async def ensure_session_metadata_exists(session_id: str, user_id: str) -> bool: "tags": [], } - table.put_item(Item=item) + # The SK is now deterministic (S#{session_id}), so attribute_not_exists is a + # real idempotency guard (issue #175): two concurrent first turns compute the + # same key and DynamoDB serialises the conditional put — one wins, the other + # gets ConditionalCheckFailed. This closes the first-turn duplicate-row race + # that the old timestamped SK made impossible to guard. + try: + table.put_item(Item=item, ConditionExpression="attribute_not_exists(PK)") + except ClientError as e: + if e.response.get("Error", {}).get("Code") == "ConditionalCheckFailedException": + logger.info(f"Session {session_id} already exists (concurrent create); skipping") + return False + raise logger.info(f"💾 Pre-created session metadata for {session_id}") return True except Exception as e: @@ -998,39 +1048,52 @@ async def update_session_activity( now = datetime.now(timezone.utc).isoformat() pk = f"USER#{user_id}" + target_sk = _static_session_sk(session_id) + gsi4 = _recency_gsi_keys(user_id, session_id, now, is_active=True) + + if old_sk == target_sk: + # Already migrated (issue #175): pure in-place update — lastMessageAt is a + # plain attribute and recency lives in GSI4_SK, so SET-ting GSI4_SK just + # re-positions the index entry. No row move → no SK rotation → the + # ghost-row race is structurally gone. + table.update_item( + Key={"PK": pk, "SK": target_sk}, + UpdateExpression=( + "ADD messageCount :one " + "SET lastMessageAt = :t, preferences = :p, GSI4_PK = :gp, GSI4_SK = :gs" + ), + ExpressionAttributeValues={ + ":one": 1, + ":t": now, + ":p": _convert_floats_to_decimal(merged_prefs), + ":gp": gsi4["GSI4_PK"], + ":gs": gsi4["GSI4_SK"], + }, + ) + logger.info("Updated session activity for %s (in-place, static SK)", session_id) + return True - # Phase A: targeted update of owned attributes on the current SK. - # Disjoint from title, starred, tags, pendingInterrupts. - table.update_item( - Key={"PK": pk, "SK": old_sk}, - UpdateExpression="ADD messageCount :one SET lastMessageAt = :t, preferences = :p", - ExpressionAttributeValues={ - ":one": 1, - ":t": now, - ":p": _convert_floats_to_decimal(merged_prefs), - }, - ) - - # Phase B: SK rotation. lastMessageAt is encoded in the SK for - # recency listing, so a per-turn change forces a row move. Fresh - # read carries any concurrent write (e.g. title-gen) that landed - # between Phase A and now. - new_sk = f"S#ACTIVE#{now}#{session_id}" - if new_sk != old_sk: - fresh_resp = table.get_item(Key={"PK": pk, "SK": old_sk}) - fresh = fresh_resp.get("Item") - if not fresh: - logger.warning( - "update_session_activity: row vanished between Phase A and Phase B for %s", - session_id, - ) - return True - carried = {k: v for k, v in fresh.items() if k not in ("PK", "SK")} - new_item = {"PK": pk, "SK": new_sk, **carried} - table.put_item(Item=new_item) - table.delete_item(Key={"PK": pk, "SK": old_sk}) - - logger.info("Updated session activity for %s (sk_rotated=%s)", session_id, new_sk != old_sk) + # Legacy row — perform the row's FINAL rotation to the static SK, carrying any + # concurrent write (e.g. title-gen) that landed since resolution, and populate + # GSI4. After this the session is static forever and every update is in-place. + fresh_resp = table.get_item(Key={"PK": pk, "SK": old_sk}) + fresh = fresh_resp.get("Item") + if not fresh: + logger.warning( + "update_session_activity: row vanished before migration for %s", session_id + ) + return True + carried = { + k: v for k, v in fresh.items() if k not in ("PK", "SK", "GSI4_PK", "GSI4_SK") + } + carried["lastMessageAt"] = now + carried["messageCount"] = int(fresh.get("messageCount", 0) or 0) + 1 + carried["preferences"] = _convert_floats_to_decimal(merged_prefs) + new_item = {"PK": pk, "SK": target_sk, **carried, **gsi4} + table.put_item(Item=new_item) + table.delete_item(Key={"PK": pk, "SK": old_sk}) + + logger.info("Migrated + updated session activity for %s (legacy -> static SK)", session_id) return True except Exception as e: logger.error("update_session_activity failed for %s: %s", session_id, e, exc_info=True) diff --git a/backend/tests/shared/test_sessions_metadata.py b/backend/tests/shared/test_sessions_metadata.py index f23e5a77e..6ac142456 100644 --- a/backend/tests/shared/test_sessions_metadata.py +++ b/backend/tests/shared/test_sessions_metadata.py @@ -619,24 +619,28 @@ async def test_preserves_assistant_id_in_preferences(self, sessions_metadata_tab assert result.preferences.last_model == "claude-3" @pytest.mark.asyncio - async def test_rotates_sk_to_new_timestamp(self, sessions_metadata_table): - """SK rotation keeps recency listing correct — only one row remains.""" + async def test_activity_updates_in_place_no_rotation(self, sessions_metadata_table): + """Issue #175: activity no longer rotates the SK. One static row (S#{id}); + recency advances via the GSI4 attribute, not a row move.""" from apis.shared.sessions.metadata import ( ensure_session_metadata_exists, update_session_activity, ) await ensure_session_metadata_exists("s1", "u1") items = sessions_metadata_table.scan()["Items"] - s_items = [i for i in items if i["SK"].startswith("S#ACTIVE#")] + s_items = [i for i in items if i.get("GSI_SK") == "META"] assert len(s_items) == 1 - old_sk = s_items[0]["SK"] + assert s_items[0]["SK"] == "S#s1" # static SK, no timestamp + assert s_items[0].get("GSI4_PK") == "USER#u1" # sparse recency key present (active) await update_session_activity(session_id="s1", user_id="u1", last_model="claude-3") items = sessions_metadata_table.scan()["Items"] - s_items_after = [i for i in items if i["SK"].startswith("S#ACTIVE#")] - assert len(s_items_after) == 1 - assert s_items_after[0]["SK"] != old_sk + s_items_after = [i for i in items if i.get("GSI_SK") == "META"] + assert len(s_items_after) == 1 # no duplicate / ghost row + assert s_items_after[0]["SK"] == "S#s1" # SK unchanged — no rotation + assert int(s_items_after[0]["messageCount"]) == 1 + assert s_items_after[0].get("GSI4_PK") == "USER#u1" # still listed (active) @pytest.mark.asyncio async def test_self_heals_when_row_missing(self, sessions_metadata_table): @@ -742,10 +746,9 @@ async def test_user_isolation(self, sessions_metadata_table): class TestEnsureSessionMetadataExists: @pytest.mark.asyncio async def test_repeated_calls_do_not_create_duplicates(self, sessions_metadata_table): - """Regression: each turn calls ensure_session_metadata_exists; the SK - encodes a timestamp, so a put-with-conditional cannot gate creation - and would produce one duplicate row per turn (sidebar duplication bug). - """ + """Regression: each turn calls ensure_session_metadata_exists. With the static + SK (issue #175) the GSI pre-check + conditional put gate creation, so only the + first call creates a row (no sidebar duplication).""" from apis.shared.sessions.metadata import ensure_session_metadata_exists first = await ensure_session_metadata_exists("s1", "u1") @@ -757,13 +760,13 @@ async def test_repeated_calls_do_not_create_duplicates(self, sessions_metadata_t assert third is False items = sessions_metadata_table.scan()["Items"] - s_items = [i for i in items if i["SK"].startswith("S#ACTIVE#") and i.get("sessionId") == "s1"] + s_items = [i for i in items if i["SK"] == "S#s1" and i.get("sessionId") == "s1"] assert len(s_items) == 1 @pytest.mark.asyncio - async def test_survives_sk_rotation(self, sessions_metadata_table): - """After update_session_activity rotates the SK, a subsequent ensure - call must still recognize the session via the GSI and skip the put. + async def test_ensure_idempotent_after_activity(self, sessions_metadata_table): + """After an activity update, a subsequent ensure call must still recognize the + session via the GSI and skip the put — exactly one static row remains. """ from apis.shared.sessions.metadata import ( ensure_session_metadata_exists, @@ -777,7 +780,7 @@ async def test_survives_sk_rotation(self, sessions_metadata_table): assert again is False items = sessions_metadata_table.scan()["Items"] - s_items = [i for i in items if i["SK"].startswith("S#ACTIVE#") and i.get("sessionId") == "s1"] + s_items = [i for i in items if i["SK"] == "S#s1" and i.get("sessionId") == "s1"] assert len(s_items) == 1 @@ -1103,3 +1106,112 @@ def test_returns_finite_float(self): result = _coerce_cost_total({"total": 1.5}) assert isinstance(result, float) assert math.isfinite(result) + + +class TestWriteSideMigration: + """Issue #175 Phase 1b — writes go static, self-migrate legacy rows, maintain GSI4.""" + + @pytest.mark.asyncio + async def test_new_session_born_static(self, sessions_metadata_table): + from apis.shared.sessions.metadata import ensure_session_metadata_exists + assert await ensure_session_metadata_exists("s1", "u1") is True + rows = [i for i in sessions_metadata_table.scan()["Items"] if i.get("GSI_SK") == "META"] + assert len(rows) == 1 + row = rows[0] + assert row["SK"] == "S#s1" # static, no timestamp + assert row["GSI4_PK"] == "USER#u1" + assert row["GSI4_SK"].endswith("#s1") + assert row["status"] == "active" + + @pytest.mark.asyncio + async def test_conditional_put_blocks_duplicate_on_gsi_lag(self, sessions_metadata_table, monkeypatch): + """If the GSI pre-check misses an existing row (eventual-consistency lag), the + deterministic-SK conditional put still prevents a duplicate — moto raises the + real ConditionalCheckFailedException, which ensure swallows.""" + import apis.shared.sessions.metadata as md + _put_migrated_row(sessions_metadata_table, "s1", "2026-01-01T00:00:00Z") + + async def _none(*a, **k): + return None + monkeypatch.setattr(md, "_get_session_by_gsi", _none) + + assert await md.ensure_session_metadata_exists("s1", "u1") is False + rows = [i for i in sessions_metadata_table.scan()["Items"] if i.get("GSI_SK") == "META"] + assert len(rows) == 1 # no duplicate created + + @pytest.mark.asyncio + async def test_activity_migrates_legacy_row_once(self, sessions_metadata_table): + from apis.shared.sessions.metadata import update_session_activity + _put_legacy_row(sessions_metadata_table, "s1", "2026-01-01T00:00:00Z") # messageCount=1 + await update_session_activity(session_id="s1", user_id="u1", last_model="claude-3") + + items = sessions_metadata_table.scan()["Items"] + rows = [i for i in items if i.get("GSI_SK") == "META"] + assert len(rows) == 1 + assert rows[0]["SK"] == "S#s1" # migrated to static + assert not any(i["SK"].startswith("S#ACTIVE#") for i in items) # legacy row gone + assert rows[0]["GSI4_PK"] == "USER#u1" # GSI4 populated + assert int(rows[0]["messageCount"]) == 2 # 1 + 1 + + @pytest.mark.asyncio + async def test_store_migrates_legacy_to_static(self, sessions_metadata_table): + from apis.shared.sessions.metadata import store_session_metadata + _put_legacy_row(sessions_metadata_table, "s1", "2026-01-01T00:00:00Z") + updated = _make_session_metadata("s1", title="Renamed", lastMessageAt="2026-02-02T00:00:00Z") + await store_session_metadata(session_id="s1", user_id="u1", session_metadata=updated) + + items = sessions_metadata_table.scan()["Items"] + rows = [i for i in items if i.get("GSI_SK") == "META"] + assert len(rows) == 1 + assert rows[0]["SK"] == "S#s1" + assert rows[0]["title"] == "Renamed" + assert rows[0]["GSI4_PK"] == "USER#u1" + assert not any(i["SK"].startswith("S#ACTIVE#") for i in items) + + @pytest.mark.asyncio + async def test_soft_delete_static_row_in_place(self, sessions_metadata_table): + from apis.app_api.sessions.services.session_service import SessionService + from apis.shared.sessions.metadata import list_user_sessions + _put_migrated_row(sessions_metadata_table, "s1", "2026-01-01T00:00:00Z") + + assert await SessionService().delete_session("u1", "s1") is True + items = sessions_metadata_table.scan()["Items"] + rows = [i for i in items if i.get("GSI_SK") == "META"] + assert len(rows) == 1 + assert rows[0]["SK"] == "S#s1" # no move + assert rows[0]["status"] == "deleted" + assert "GSI4_PK" not in rows[0] # dropped from recency index + assert not any(i["SK"].startswith("S#DELETED#") for i in items) + sessions, _ = await list_user_sessions("u1") + assert sessions == [] # no longer listed + + @pytest.mark.asyncio + async def test_soft_delete_legacy_row_migrates(self, sessions_metadata_table): + from apis.app_api.sessions.services.session_service import SessionService + from apis.shared.sessions.metadata import list_user_sessions + _put_legacy_row(sessions_metadata_table, "s1", "2026-01-01T00:00:00Z") + + assert await SessionService().delete_session("u1", "s1") is True + items = sessions_metadata_table.scan()["Items"] + rows = [i for i in items if i.get("GSI_SK") == "META"] + assert len(rows) == 1 + assert rows[0]["SK"] == "S#s1" # migrated to static tombstone + assert rows[0]["status"] == "deleted" + assert "GSI4_PK" not in rows[0] + assert not any(i["SK"].startswith("S#ACTIVE#") for i in items) # legacy gone + sessions, _ = await list_user_sessions("u1") + assert sessions == [] + + @pytest.mark.asyncio + async def test_end_to_end_create_activity_list_delete(self, sessions_metadata_table): + from apis.shared.sessions.metadata import ( + ensure_session_metadata_exists, update_session_activity, list_user_sessions, + ) + from apis.app_api.sessions.services.session_service import SessionService + await ensure_session_metadata_exists("s1", "u1") + await update_session_activity(session_id="s1", user_id="u1", last_model="claude-3") + sessions, _ = await list_user_sessions("u1") + assert [s.session_id for s in sessions] == ["s1"] # visible via GSI4 union + await SessionService().delete_session("u1", "s1") + sessions, _ = await list_user_sessions("u1") + assert sessions == []