diff --git a/backend/src/apis/shared/sessions/metadata.py b/backend/src/apis/shared/sessions/metadata.py
index ea6d6b7d..bc75d0c9 100644
--- a/backend/src/apis/shared/sessions/metadata.py
+++ b/backend/src/apis/shared/sessions/metadata.py
@@ -1840,6 +1840,37 @@ def _encode_list_cursor(session: SessionMetadata) -> str:
     return base64.b64encode(json.dumps(payload).encode('utf-8')).decode('utf-8')
 
 
+# --- Issue #175 Phase 3: contract the dual-scheme read once migration completes ---
+_MIGRATION_MARKER_PK = "MIGRATION#session-sk"
+_MIGRATION_MARKER_SK = "STATE"
+# Memoised once observed True. The marker only ever goes unset -> set (the Phase 2
+# backfill sets it after confirming zero legacy rows), never back, so caching a True
+# result per-process is safe. A False/absent result is NOT cached, so a container
+# that started before the backfill picks up the flip on a later list call.
+_migration_complete_cache = False
+
+
+def _is_session_migration_complete(table) -> bool:
+    """True once the Phase 2 backfill has set the migration-complete marker.
+
+    Gates ``list_user_sessions`` from the dual-scheme union down to a GSI-only read.
+    Fails open (returns False -> keep dual-read) on any error, and stays in dual-read
+    for downstream/forked deployments that haven't run the backfill — so removing the
+    legacy branch here can never blank the sidebar on un-migrated data.
+    """
+    global _migration_complete_cache
+    if _migration_complete_cache:
+        return True
+    try:
+        resp = table.get_item(Key={"PK": _MIGRATION_MARKER_PK, "SK": _MIGRATION_MARKER_SK})
+        if (resp.get("Item") or {}).get("complete"):
+            _migration_complete_cache = True
+            return True
+    except Exception as e:
+        logger.debug("migration marker check failed (staying in dual-read): %s", e)
+    return False
+
+
 async def _list_user_sessions_cloud(
     user_id: str,
     table_name: str,
@@ -1885,25 +1916,13 @@ async def _list_user_sessions_cloud(
         want = (limit + 1) if limit else None
         pk = f'USER#{user_id}'
 
-        # Legacy source: base table, S#ACTIVE# prefix. Resuming after a cursor uses
-        # between('S#ACTIVE#', 'S#ACTIVE#{la}#{sid}') — a single range condition that
-        # keeps the prefix filter while bounding the upper end (inclusive; the exact
-        # cursor row is dropped by the strict-less filter below).
-        if cursor:
-            la, sid = cursor
-            legacy_cond = Key('PK').eq(pk) & Key('SK').between('S#ACTIVE#', f'S#ACTIVE#{la}#{sid}')
-        else:
-            legacy_cond = Key('PK').eq(pk) & Key('SK').begins_with('S#ACTIVE#')
-        legacy_params: Dict[str, Any] = {
-            'KeyConditionExpression': legacy_cond,
-            'ScanIndexForward': False,
-        }
-        if want:
-            legacy_params['Limit'] = want
-        legacy_sessions = _collect_valid_sessions(table, legacy_params, want)
+        # GSI-only once the migration is complete (marker set); otherwise dual-read
+        # the union so un-migrated legacy rows stay visible — for downstream/forked
+        # deployments that haven't run the Phase 2 backfill yet.
+        gsi_only = _is_session_migration_complete(table)
 
         # Migrated source: SessionRecencyIndex GSI. GSI4_SK < '{la}#{sid}' is a clean
-        # strict-less resume. Degrade to legacy-only if the index isn't there yet.
+        # strict-less resume.
         if cursor:
             la, sid = cursor
             gsi_cond = Key('GSI4_PK').eq(pk) & Key('GSI4_SK').lt(f'{la}#{sid}')
@@ -1916,6 +1935,7 @@ async def _list_user_sessions_cloud(
         }
         if want:
             gsi_params['Limit'] = want
+        gsi_failed = False
         try:
             gsi_sessions = _collect_valid_sessions(table, gsi_params, want)
         except ClientError as e:
@@ -1936,9 +1956,32 @@ async def _list_user_sessions_cloud(
                     "session listing", code
                 )
                 gsi_sessions = []
+                gsi_failed = True
             else:
                 raise
 
+        # Legacy source: base table, S#ACTIVE# prefix. Skipped once migration is
+        # complete (GSI-only), UNLESS the GSI query failed — then fall back to legacy
+        # regardless so a transient index error never blanks the list. Resuming after
+        # a cursor uses between('S#ACTIVE#', 'S#ACTIVE#{la}#{sid}') — a single range
+        # condition that keeps the prefix filter while bounding the upper end
+        # (inclusive; the exact cursor row is dropped by the strict-less filter below).
+        if gsi_only and not gsi_failed:
+            legacy_sessions: list[SessionMetadata] = []
+        else:
+            if cursor:
+                la, sid = cursor
+                legacy_cond = Key('PK').eq(pk) & Key('SK').between('S#ACTIVE#', f'S#ACTIVE#{la}#{sid}')
+            else:
+                legacy_cond = Key('PK').eq(pk) & Key('SK').begins_with('S#ACTIVE#')
+            legacy_params: Dict[str, Any] = {
+                'KeyConditionExpression': legacy_cond,
+                'ScanIndexForward': False,
+            }
+            if want:
+                legacy_params['Limit'] = want
+            legacy_sessions = _collect_valid_sessions(table, legacy_params, want)
+
         # Merge: sort by (lastMessageAt, sessionId) descending, drop anything not
         # strictly older than the cursor (removes the inclusive legacy cursor row),
         # dedupe by session_id.
diff --git a/backend/tests/shared/test_sessions_metadata.py b/backend/tests/shared/test_sessions_metadata.py
index 6ac14245..a2065017 100644
--- a/backend/tests/shared/test_sessions_metadata.py
+++ b/backend/tests/shared/test_sessions_metadata.py
@@ -1215,3 +1215,80 @@ async def test_end_to_end_create_activity_list_delete(self, sessions_metadata_ta
         await SessionService().delete_session("u1", "s1")
         sessions, _ = await list_user_sessions("u1")
         assert sessions == []
+
+
+class TestListContractOnMarker:
+    """Issue #175 Phase 3 — list read contracts to GSI-only once the marker is set."""
+
+    @staticmethod
+    def _set_marker(table):
+        table.put_item(Item={"PK": "MIGRATION#session-sk", "SK": "STATE", "complete": True})
+
+    @pytest.mark.asyncio
+    async def test_dual_read_when_marker_absent(self, sessions_metadata_table, monkeypatch):
+        import apis.shared.sessions.metadata as md
+        monkeypatch.setattr(md, "_migration_complete_cache", False)
+        _put_legacy_row(sessions_metadata_table, "leg", "2026-01-02T00:00:00Z")
+        _put_migrated_row(sessions_metadata_table, "mig", "2026-01-01T00:00:00Z")
+
+        sessions, _ = await md.list_user_sessions("u1")
+        assert sorted(s.session_id for s in sessions) == ["leg", "mig"]  # union of both
+
+    @pytest.mark.asyncio
+    async def test_gsi_only_when_marker_set(self, sessions_metadata_table, monkeypatch):
+        import apis.shared.sessions.metadata as md
+        monkeypatch.setattr(md, "_migration_complete_cache", False)
+        _put_legacy_row(sessions_metadata_table, "leg", "2026-01-02T00:00:00Z")
+        _put_migrated_row(sessions_metadata_table, "mig", "2026-01-01T00:00:00Z")
+        self._set_marker(sessions_metadata_table)
+
+        sessions, _ = await md.list_user_sessions("u1")
+        assert [s.session_id for s in sessions] == ["mig"]  # legacy row excluded
+
+    @pytest.mark.asyncio
+    async def test_marker_result_is_memoised(self, sessions_metadata_table, monkeypatch):
+        import apis.shared.sessions.metadata as md
+        monkeypatch.setattr(md, "_migration_complete_cache", False)
+        self._set_marker(sessions_metadata_table)
+        _put_migrated_row(sessions_metadata_table, "mig", "2026-01-01T00:00:00Z")
+
+        await md.list_user_sessions("u1")
+        assert md._migration_complete_cache is True  # cached after first observation
+
+    @pytest.mark.asyncio
+    async def test_gsi_failure_falls_back_to_legacy_even_with_marker(self, sessions_metadata_table, monkeypatch):
+        """Marker set but the GSI query errors → still read legacy so the list never blanks."""
+        import boto3
+        import apis.shared.sessions.metadata as md
+        from botocore.exceptions import ClientError
+
+        monkeypatch.setattr(md, "_migration_complete_cache", False)
+        _put_legacy_row(sessions_metadata_table, "leg", "2026-01-02T00:00:00Z")
+        self._set_marker(sessions_metadata_table)
+        real_table = sessions_metadata_table
+
+        class _GsiFailingTable:
+            def get_item(self, **kw):
+                return real_table.get_item(**kw)  # marker lookup goes through
+
+            def query(self, **kw):
+                if kw.get("IndexName") == "SessionRecencyIndex":
+                    raise ClientError(
+                        {"Error": {"Code": "ValidationException",
+                                   "Message": "The table does not have the specified index: SessionRecencyIndex"}},
+                        "Query",
+                    )
+                return real_table.query(**kw)
+
+        class _FakeResource:
+            def Table(self, _name):
+                return _GsiFailingTable()
+
+        real_resource = boto3.resource
+        monkeypatch.setattr(
+            boto3, "resource",
+            lambda svc, **kw: _FakeResource() if svc == "dynamodb" else real_resource(svc, **kw),
+        )
+
+        sessions, _ = await md.list_user_sessions("u1")
+        assert [s.session_id for s in sessions] == ["leg"]  # fell back to legacy, not blank