fix: restrict legacy encounter sessions to their creator

backend/python/api/combat.py

@@ -62,6 +62,7 @@ def start_combat():
 
         combat_sessions[session_id] = {
             'id': session_id,
+            'owner_id': g.user['id'],
             'attacker_gang_id': attacker_gang_id,
             'target_block_id': target_block_id,
             'target_snapshot_id': target_snapshot_id,
@@ -128,11 +129,10 @@ def submit_combat_turn():
         session_id = data.get('session_id')
         action = data.get('action', {})
 
-        if session_id not in combat_sessions:
+        session = combat_sessions.get(session_id)
+        if session is None or session.get('owner_id') != g.user['id']:
             return jsonify({'error': 'Combat session not found'}), 404
 
-        session = combat_sessions[session_id]
-
         if session['status'] != 'active':
             return jsonify({'error': 'Combat session not active'}), 400
 
@@ -196,11 +196,10 @@ def submit_combat_turn():
 @require_auth
 def get_combat_session(session_id: str):
     """Get combat session state"""
-    if session_id not in combat_sessions:
+    session = combat_sessions.get(session_id)
+    if session is None or session.get('owner_id') != g.user['id']:
         return jsonify({'error': 'Combat session not found'}), 404
 
-    session = combat_sessions[session_id]
-
     return jsonify({
         'session_id': session_id,
         'status': session['status'],

backend/python/api/driveby.py

@@ -62,6 +62,7 @@ def start_driveby():
 
         driveby_sessions[session_id] = {
             'id': session_id,
+            'owner_id': g.user['id'],
             'target_block_id': target_block_id,
             'target_snapshot_id': target_snapshot.snapshot_id,
             'vehicle': vehicle,
@@ -115,11 +116,10 @@ def execute_driveby():
         session_id = data.get('session_id')
         target_tile = data.get('target_tile', {})
 
-        if session_id not in driveby_sessions:
+        session = driveby_sessions.get(session_id)
+        if session is None or session.get('owner_id') != g.user['id']:
             return jsonify({'error': 'Session not found'}), 404
 
-        session = driveby_sessions[session_id]
-
         if session['shots_fired'] >= session['max_shots']:
             return jsonify({'error': 'No shots remaining'}), 400
 
@@ -178,11 +178,10 @@ def complete_driveby(session_id: str):
         Final stats and rewards
     """
     try:
-        if session_id not in driveby_sessions:
+        session = driveby_sessions.get(session_id)
+        if session is None or session.get('owner_id') != g.user['id']:
             return jsonify({'error': 'Session not found'}), 404
 
-        session = driveby_sessions[session_id]
-
         # Calculate escape success (based on vehicle speed and heat)
         escape_chance = 0.7 + (session['vehicle'].get('speed', 50) / 200.0)
         rng = random.Random(f"{session['seed']}-escape")

backend/python/tests/test_encounter_session_ownership.py

@@ -0,0 +1,210 @@
+"""Request-level encounter ownership contracts with synthetic verified users."""
+
+import copy
+import os
+import sys
+from unittest.mock import Mock
+
+import pytest
+from flask import Flask
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
+
+import api.combat as combat
+import api.driveby as driveby
+from middleware.auth import DEV_USER
+from services.block_state_engine import BlockSnapshot, MemberSnapshot, TileSnapshot
+
+
+USER_A = {'id': 'synthetic-user-a', 'role': 'authenticated'}
+USER_B = {'id': 'synthetic-user-b', 'role': 'authenticated'}
+HEADERS_A = {'Authorization': 'Bearer synthetic-token-a'}
+HEADERS_B = {'Authorization': 'Bearer synthetic-token-b'}
+
+
+@pytest.fixture
+def encounter_app(monkeypatch):
+    monkeypatch.setattr(combat, 'combat_sessions', {})
+    monkeypatch.setattr(driveby, 'driveby_sessions', {})
+    users = {'synthetic-token-a': USER_A, 'synthetic-token-b': USER_B}
+    monkeypatch.setattr(
+        'middleware.auth._verify_token',
+        lambda token: copy.deepcopy(users.get(token)),
+    )
+    snapshot = BlockSnapshot(
+        block_id='synthetic-block',
+        snapshot_id='synthetic-snapshot',
+        snapshot_version=1,
+        created_at='2026-10-15T00:00:00',
+        address='1208 Fictional Way',
+        city='Fictional City',
+        bbox=[0, 0, 1, 1],
+        center=[0.5, 0.5],
+        grid_width=2,
+        grid_height=2,
+        tiles=[[
+            TileSnapshot(
+                x=x,
+                y=y,
+                tile_type='sidewalk',
+                terrain_bonus={'cover': 0},
+                member_id='synthetic-defender' if (x, y) == (1, 1) else None,
+            )
+            for x in range(2)
+        ] for y in range(2)],
+        members=[MemberSnapshot(
+            id='synthetic-defender',
+            role='shooter',
+            stats={'health': 100, 'defense': 0},
+            position={'x': 1, 'y': 1},
+            loadout={},
+        )],
+        heat_level=0,
+        income_rate=0,
+        defense_rating=0,
+        fortification_level=0,
+        seed='synthetic-seed',
+    )
+    engine = Mock()
+    engine.create_snapshot.return_value = snapshot.snapshot_id
+    engine.get_block_snapshot.return_value = snapshot
+    monkeypatch.setattr(combat, 'get_block_state_engine', lambda: engine)
+    monkeypatch.setattr(driveby, 'get_block_state_engine', lambda: engine)
+
+    app = Flask(__name__)
+    app.config.update(TESTING=True, SUPABASE_URL='https://synthetic.invalid')
+    app.register_blueprint(combat.combat_bp)
+    app.register_blueprint(driveby.driveby_bp)
+    return app, engine
+
+
+def start_session(client, kind, headers):
+    payload = {
+        'target_block_id': 'synthetic-block',
+        # The caller cannot choose the authenticated session owner.
+        'owner_id': USER_B['id'],
+    }
+    if kind == 'combat':
+        payload.update(
+            attacker_gang_id='synthetic-gang',
+            attacker_members=['synthetic-attacker'],
+        )
+    else:
+        payload.update(shooters=['synthetic-attacker'], vehicle={'speed': 50})
+    response = client.post(f'/api/{kind}/start', json=payload, headers=headers)
+    assert response.status_code == 201, response.get_json()
+    return response.get_json()['session_id']
+
+
+def session_store(kind):
+    return combat.combat_sessions if kind == 'combat' else driveby.driveby_sessions
+
+
+def call_session_route(client, route, session_id, headers):
+    if route == 'combat-read':
+        return client.get(f'/api/combat/{session_id}', headers=headers)
+    if route == 'combat-turn':
+        return client.post('/api/combat/turn', headers=headers, json={
+            'session_id': session_id,
+            'action': {'type': 'retreat'},
+        })
+    if route == 'driveby-shoot':
+        return client.post('/api/driveby/shoot', headers=headers, json={
+            'session_id': session_id,
+            'target_tile': {'x': 1, 'y': 1},
+        })
+    return client.post(f'/api/driveby/{session_id}/complete', headers=headers)
+
+
+ROUTES = ['combat-read', 'combat-turn', 'driveby-shoot', 'driveby-complete']
+
+
+@pytest.mark.parametrize('kind', ['combat', 'driveby'])
+def test_start_binds_session_to_verified_caller(encounter_app, kind):
+    app, _ = encounter_app
+    session_id = start_session(app.test_client(), kind, HEADERS_A)
+
+    assert session_store(kind)[session_id]['owner_id'] == USER_A['id']
+
+
+@pytest.mark.parametrize('route', ROUTES)
+@pytest.mark.parametrize('case', ['foreign', 'missing', 'ownerless'])
+def test_unavailable_sessions_share_404_without_work_or_mutation(encounter_app, route, case):
+    app, engine = encounter_app
+    client = app.test_client()
+    kind = route.split('-')[0]
+    session_id = start_session(client, kind, HEADERS_A)
+    store = session_store(kind)
+    if case == 'ownerless':
+        store[session_id].pop('owner_id', None)
+    before = copy.deepcopy(store)
+    engine.reset_mock()
+
+    missing = call_session_route(client, route, 'missing-session', HEADERS_A)
+    response = call_session_route(
+        client,
+        route,
+        'another-missing-session' if case == 'missing' else session_id,
+        HEADERS_B if case == 'foreign' else HEADERS_A,
+    )
+
+    assert missing.status_code == response.status_code == 404
+    assert response.get_json() == missing.get_json()
+    assert store == before
+    engine.get_block_snapshot.assert_not_called()
+    engine.create_snapshot.assert_not_called()
+
+
+@pytest.mark.parametrize('route', ROUTES)
+def test_verified_owner_can_continue_the_session(encounter_app, route):
+    app, _ = encounter_app
+    client = app.test_client()
+    kind = route.split('-')[0]
+    session_id = start_session(client, kind, HEADERS_A)
+
+    response = call_session_route(client, route, session_id, HEADERS_A)
+
+    assert response.status_code == 200, response.get_json()
+    if route == 'combat-read':
+        assert response.get_json()['current_turn'] == 0
+    elif route == 'combat-turn':
+        assert response.get_json()['status'] == 'defender_victory'
+        assert session_store(kind)[session_id]['current_turn'] == 1
+    elif route == 'driveby-shoot':
+        assert response.get_json()['shots_remaining'] == 2
+        assert session_store(kind)[session_id]['shots_fired'] == 1
+    else:
+        assert response.get_json()['completed'] is True
+        assert session_id not in session_store(kind)
+
+
+@pytest.mark.parametrize('route', ROUTES)
+def test_unverified_request_cannot_access_session(encounter_app, route):
+    app, engine = encounter_app
+    client = app.test_client()
+    kind = route.split('-')[0]
+    session_id = start_session(client, kind, HEADERS_A)
+    before = copy.deepcopy(session_store(kind))
+    engine.reset_mock()
+
+    response = call_session_route(client, route, session_id, {
+        'Authorization': 'Bearer synthetic-invalid-token',
+    })
+
+    assert response.status_code == 401
+    assert session_store(kind) == before
+    engine.get_block_snapshot.assert_not_called()
+
+
+@pytest.mark.parametrize('kind', ['combat', 'driveby'])
+def test_offline_dev_owner_can_start_and_continue(encounter_app, kind):
+    app, _ = encounter_app
+    app.config['SUPABASE_URL'] = None
+    client = app.test_client()
+    session_id = start_session(client, kind, {})
+
+    assert session_store(kind)[session_id]['owner_id'] == DEV_USER['id']
+    routes = ['combat-read', 'combat-turn'] if kind == 'combat' else ['driveby-shoot', 'driveby-complete']
+    for route in routes:
+        response = call_session_route(client, route, session_id, {})
+        assert response.status_code == 200, response.get_json()

docs/PROJECT_LOG.md

@@ -69,6 +69,12 @@ Payments/monetization P0s are separately listed in `docs/MVP_STATUS_AND_DEV_PLAN
 
 ## Log
 
+### 2026-10-14 — Creator-bound legacy HTTP encounter sessions
+
+- Record the authenticated creator when the existing Flask combat/drive-by session starts. Check that creator before existing-session access, preserving ordinary creator behavior and offline development flow. Session-not-found responses stay consistent; no combat calculation, roster or economy rule changes.
+- Regression-first: 12 failed / 12 passed on unchanged `9eef5aa`; corrected request-level suite passes 24 cases. Full backend passes 131 tests with no warnings. The combined local frontend/backend candidate also passed 1,069 frontend tests / 4 existing skips and all build/type/lint/asset gates; this PR itself does not change frontend behavior.
+- Evidence: `docs/evidence/2026-10-14-session-ownership/README.md`. Legacy in-memory HTTP sessions are not the canonical solo renderer or a live PvP implementation. Durable sessions, wider action validation, live #175 proof and release readiness remain separate. No schema, RLS, credential, scheduler, production or deployment change. Exact published-head CI and integration-owner review remain required.
+
 ### 2026-10-01 — Backend timestamps are timezone-aware UTC (#170)
 
 - Replaced all 17 `datetime.utcnow()` call sites in the live backend runtime with one shared helper, `backend/python/utils/utc.py` (`utc_now`, `to_iso_utc`, `utc_now_iso`). Python 3.12 deprecates `utcnow()` and the required CI runs the full backend suite, so the debt was visible on every run: the audited baseline emitted 153 warnings, all of them this single warning. The suite now runs with 0 warnings.

docs/evidence/2026-10-14-session-ownership/README.md

@@ -0,0 +1,22 @@
+# Legacy HTTP encounter session ownership
+
+Source base: `9eef5aac28a82275d7a2e1596e8658b167953ed7`. The two runtime source files and request-level regression test are byte-identical to independently reviewed local commit `37dddc1e1f672bb34ca12145f88bdd2499fb9d15` (original worker implementation `39c32abdd7b5acf986249d4004f4b4d3bab113f8`).
+
+## Contract and checks
+
+Session creation derives its creator from the server-verified caller, not a submitted owner field. Existing-session reads and mutations check that creator before session-derived work. Missing, foreign and ownerless sessions use the existing not-found response. Ordinary creator and offline development paths remain usable.
+
+The regression suite uses two synthetic verified identities through the real Flask blueprints with local snapshot mocks. It checks server-derived creator binding, access denial without snapshot work/state mutation, ordinary creator continuation, invalid authentication and offline dev behavior. No real user, token, database or network access is used.
+
+- Unchanged base with the new tests: 12 failed / 12 passed.
+- Corrected focused suite: 24 passed.
+- Full backend: 131 passed, zero warnings; root independently inspected the diff and re-ran the complete backend gate.
+- Complete local #204/#205 plus both reviewed repairs: 1,069 frontend tests / 4 existing skips; 131 backend tests; type/lint/build/assets/packages passed. This is supporting compatibility evidence, not a claim that other drafts are merged.
+
+## Scope and limits
+
+Only `api/combat.py`, `api/driveby.py`, the new ownership test, this text record and a new project-log entry change. There is no frontend renderer or result-writer change. The legacy sessions remain in memory; this PR does not add a match server, durable results, roster authorization or complete command validation. Live authenticated staging (#175), gameplay integrity and human PvP require their separate proofs.
+
+No browser behavior changes in this backend-only PR. Its request-level creator/denial sequence is the relevant direct integration proof. No migration, RLS, secret, scheduler or deployment action was performed.
+
+Revert this focused PR for source rollback; restarting the process discards these existing in-memory sessions. No database data or schema rollback is needed. Integration is independent of #204/#205/#202/#169, with append-only project-log reconciliation required.