security: scope activity journal by principal (Closes #1462)

src/main.py

@@ -496,6 +496,15 @@ async def _upstream_identity() -> tuple[int, str]:
     return principal_id, principal_login
 
 
+async def _security_principal_id(request: Request) -> int:
+    session = getattr(request.state, "dashboard_session", None)
+    principal_id = getattr(session, "principal_id", None)
+    if isinstance(principal_id, int) and not isinstance(principal_id, bool) and principal_id > 0:
+        return principal_id
+    principal_id, _login = await _upstream_identity()
+    return principal_id
+
+
 class DashboardSignIn(BaseModel):
     access_token: str = Field(min_length=1, max_length=1_024)
     device_label: str = Field(default="This device", min_length=1, max_length=64)
@@ -1929,6 +1938,7 @@ async def decide_human_gate(
         target=gate_id,
     )
     login = await _human_gate_login(request)
+    principal_id = int(login.partition(":")[0])
     journal = _security_event_store()
     operation_id = None
     try:
@@ -1939,6 +1949,7 @@ async def decide_human_gate(
                 operation_id = await asyncio.to_thread(
                     journal.reserve,
                     "human_gate_decision",
+                    principal_id=principal_id,
                     method=payload.decision,
                     target=gate_id,
                 )
@@ -2062,6 +2073,7 @@ async def sign_in(payload: DashboardSignIn, request: Request, response: Response
         await asyncio.to_thread(
             _security_event_store().record,
             "sign_in",
+            principal_id=principal_id,
             method="token",
             device_label=payload.device_label,
             target="dashboard",
@@ -2099,12 +2111,16 @@ async def sign_in(payload: DashboardSignIn, request: Request, response: Response
 
 @app.get("/api/v1/security-events")
 async def list_security_events(
+    request: Request,
     limit: int = Query(default=25, ge=1, le=100),
     cursor: int | None = Query(default=None, ge=1),
 ):
     try:
         page = await asyncio.to_thread(
-            _security_event_store().list, limit=limit, cursor=cursor
+            _security_event_store().list,
+            principal_id=await _security_principal_id(request),
+            limit=limit,
+            cursor=cursor,
         )
         authentication_alerts = await asyncio.to_thread(
             _login_attempt_store().list_alerts, limit=24
@@ -2270,6 +2286,7 @@ async def verify_passkey_registration(payload: PasskeyCeremony, request: Request
         operation_id = await asyncio.to_thread(
             journal.reserve,
             "passkey_enrolled",
+            principal_id=await _security_principal_id(request),
             method="passkey",
             device_label=current.device_label,
             target="passkey",
@@ -2381,6 +2398,7 @@ async def revoke_enrolled_passkey(
         operation_id = await asyncio.to_thread(
             journal.reserve,
             "passkey_revoked",
+            principal_id=await _security_principal_id(request),
             device_label=credential.device_label,
             target="passkey",
         )
@@ -2556,6 +2574,7 @@ async def verify_passkey_authorization(
             await asyncio.to_thread(
                 _security_event_store().record,
                 "passkey_counter_anomaly",
+                principal_id=await _security_principal_id(request),
                 method="passkey",
                 device_label=stored.device_label,
                 target=f"{payload.action}:{payload.target}",
@@ -2667,6 +2686,7 @@ async def verify_passkey_authentication(
             await asyncio.to_thread(
                 _security_event_store().record,
                 "passkey_counter_anomaly",
+                principal_id=stored.principal_id,
                 method="passkey",
                 device_label=stored.device_label,
                 target="sign_in:dashboard",
@@ -2705,6 +2725,7 @@ async def verify_passkey_authentication(
         await asyncio.to_thread(
             _security_event_store().record,
             "sign_in",
+            principal_id=stored.principal_id,
             method="passkey",
             device_label=stored.device_label,
             target="dashboard",
@@ -3806,6 +3827,7 @@ async def save_today_recap_and_log_time(
                         operation_id = await asyncio.to_thread(
                             journal.reserve,
                             "gitea_time_logged",
+                            principal_id=await _security_principal_id(request),
                             target=f"{repository}#{number}",
                         )
                     except SecurityEventStoreError:
@@ -3906,6 +3928,7 @@ async def sign_out(request: Request, response: Response):
         operation_id = await asyncio.to_thread(
             journal.reserve,
             "sign_out",
+            principal_id=await _security_principal_id(request),
             target="current_device",
         )
     except SecurityEventStoreError:
@@ -4016,6 +4039,7 @@ async def revoke_active_device(
         operation_id = await asyncio.to_thread(
             journal.reserve,
             "device_revoked",
+            principal_id=await _security_principal_id(request),
             device_label=target.device_label,
             target="device",
         )
@@ -4068,6 +4092,7 @@ async def sign_out_all_devices(
         operation_id = await asyncio.to_thread(
             journal.reserve,
             "all_sessions_revoked",
+            principal_id=await _security_principal_id(request),
             target="all_devices",
         )
     except SecurityEventStoreError:
@@ -6641,6 +6666,7 @@ async def _delete_conversation_comment(
         operation_id = await asyncio.to_thread(
             journal.reserve,
             "comment_deleted",
+            principal_id=await _security_principal_id(request),
             target=target,
         )
     except SecurityEventStoreError:
@@ -7096,6 +7122,7 @@ async def close_assigned_issue(
         operation_id = await asyncio.to_thread(
             journal.reserve,
             "issue_closed",
+            principal_id=await _security_principal_id(request),
             target=target,
         )
     except HTTPException:
@@ -8001,6 +8028,7 @@ async def merge_assigned_pull(
         operation_id = await asyncio.to_thread(
             journal.reserve,
             "pull_merged",
+            principal_id=await _security_principal_id(request),
             target=f"{repository}#{number}",
         )
     except SecurityEventStoreError:
@@ -8092,6 +8120,7 @@ async def delete_merged_source_branch(
         operation_id = await asyncio.to_thread(
             journal.reserve,
             "source_branch_deleted",
+            principal_id=await _security_principal_id(request),
             target=f"{repository}#{number}@{submission.expected_head_sha}",
         )
     except SecurityEventStoreError:
@@ -8279,7 +8308,10 @@ async def prepare_release_rollback(
     journal = _security_event_store()
     try:
         operation_id = await asyncio.to_thread(
-            journal.reserve, "release_rollback_prepared", target=target
+            journal.reserve,
+            "release_rollback_prepared",
+            principal_id=await _security_principal_id(request),
+            target=target,
         )
     except SecurityEventStoreError:
         return JSONResponse(
@@ -8377,6 +8409,7 @@ async def submit_review(
                         if submission.decision == "approve"
                         else "pull_review_changes_requested"
                     ),
+                    principal_id=await _security_principal_id(request),
                     target=f"{repository}#{number}",
                 )
             except SecurityEventStoreError:

src/security_event_store.py

@@ -76,6 +76,7 @@ class SecurityEventStore:
                 CREATE TABLE IF NOT EXISTS security_events (
                     id INTEGER PRIMARY KEY AUTOINCREMENT,
                     payload TEXT,
+                    principal_id INTEGER,
                     created_at INTEGER NOT NULL,
                     status TEXT NOT NULL DEFAULT 'completed',
                     operation_id TEXT
@@ -93,6 +94,10 @@ class SecurityEventStore:
                 connection.execute(
                     "ALTER TABLE security_events ADD COLUMN operation_id TEXT"
                 )
+            if "principal_id" not in columns:
+                connection.execute(
+                    "ALTER TABLE security_events ADD COLUMN principal_id INTEGER"
+                )
             plaintext_columns = {"kind", "method", "device_label", "target"}
             if "payload" not in columns or plaintext_columns.intersection(columns):
                 self._migrate_plaintext(
@@ -102,6 +107,10 @@ class SecurityEventStore:
                 "CREATE INDEX IF NOT EXISTS security_events_created "
                 "ON security_events(created_at DESC, id DESC)"
             )
+            connection.execute(
+                "CREATE INDEX IF NOT EXISTS security_events_principal "
+                "ON security_events(principal_id, id DESC)"
+            )
             connection.execute(
                 "CREATE UNIQUE INDEX IF NOT EXISTS security_events_operation "
                 "ON security_events(operation_id) WHERE operation_id IS NOT NULL"
@@ -126,6 +135,7 @@ class SecurityEventStore:
             CREATE TABLE security_events_encrypted (
                 id INTEGER PRIMARY KEY AUTOINCREMENT,
                 payload TEXT,
+                principal_id INTEGER,
                 created_at INTEGER NOT NULL,
                 status TEXT NOT NULL DEFAULT 'completed',
                 operation_id TEXT
@@ -145,7 +155,8 @@ class SecurityEventStore:
         ) in rows:
             connection.execute(
                 "INSERT INTO security_events_encrypted "
-                "(id, payload, created_at, status, operation_id) VALUES (?, ?, ?, ?, ?)",
+                "(id, payload, principal_id, created_at, status, operation_id) "
+                "VALUES (?, ?, NULL, ?, ?, ?)",
                 (
                     event_id,
                     payload
@@ -195,6 +206,7 @@ class SecurityEventStore:
         self,
         kind: str,
         *,
+        principal_id: int,
         method: str | None = None,
         device_label: str | None = None,
         target: str | None = None,
@@ -203,8 +215,9 @@ class SecurityEventStore:
         try:
             with self._connect() as connection:
                 cursor = connection.execute(
-                    "INSERT INTO security_events(created_at, status) VALUES (?, 'completed')",
+                    "INSERT INTO security_events(principal_id, created_at, status) "
-                    (now,),
+                    "VALUES (?, ?, 'completed')",
+                    (principal_id, now),
                 )
                 event_id = cursor.lastrowid
                 payload = self._seal_event(
@@ -224,6 +237,7 @@ class SecurityEventStore:
         self,
         kind: str,
         *,
+        principal_id: int,
         method: str | None = None,
         device_label: str | None = None,
         target: str | None = None,
@@ -233,9 +247,9 @@ class SecurityEventStore:
         try:
             with self._connect() as connection:
                 cursor = connection.execute(
-                    "INSERT INTO security_events(created_at, status, operation_id) "
+                    "INSERT INTO security_events(principal_id, created_at, status, operation_id) "
-                    "VALUES (?, 'pending', ?)",
+                    "VALUES (?, ?, 'pending', ?)",
-                    (now, operation_id),
+                    (principal_id, now, operation_id),
                 )
                 event_id = cursor.lastrowid
                 connection.execute(
@@ -282,12 +296,14 @@ class SecurityEventStore:
                 "Security activity is temporarily unavailable"
             ) from exc
 
-    def list(self, *, limit: int = 50, cursor: int | None = None) -> SecurityEventPage:
+    def list(
+        self, *, principal_id: int, limit: int = 50, cursor: int | None = None
+    ) -> SecurityEventPage:
         bounded_limit = min(100, max(1, limit))
-        parameters: list[int] = []
+        parameters: list[int] = [principal_id]
-        where = ""
+        where = "WHERE principal_id = ?"
         if cursor is not None:
-            where = "WHERE id < ?"
+            where += " AND id < ?"
             parameters.append(cursor)
         parameters.append(bounded_limit + 1)
         try:

tests/conftest.py

@@ -2,6 +2,8 @@
 
 import os
 
+import pytest
+
 
 os.environ.setdefault(
     "STACKCHAIN_PRIVATE_STATE_ENCRYPTION_KEY",
@@ -10,4 +12,15 @@ os.environ.setdefault(
 os.environ.setdefault(
     "STACKCHAIN_PUSH_STATE_ENCRYPTION_KEY",
     "cHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHBwcHA=",
-)
+)
+
+
+@pytest.fixture(autouse=True)
+def stable_upstream_identity(monkeypatch):
+    """Keep API tests off the network when security ownership resolves identity."""
+    from src import main
+
+    async def current_user():
+        return {"id": 42, "login": "timmy"}
+
+    monkeypatch.setattr(main, "current_user", current_user)

tests/test_dashboard_auth.py

@@ -135,7 +135,8 @@ async def test_source_branch_deletion_accepts_exact_one_time_fresh_authorization
         def record(self, *_args, **_kwargs):
             pass
 
-        def reserve(self, kind, *, target):
+        def reserve(self, kind, *, principal_id, target):
+            assert principal_id == 42
             lifecycle.append(("reserve", kind, target))
             return "cleanup-operation"
 
@@ -415,7 +416,7 @@ async def test_stale_nonzero_passkey_counter_denies_sign_in_and_records_anomaly(
             },
         )
 
-    events = main._security_event_store().list(limit=10).events
+    events = main._security_event_store().list(principal_id=42, limit=10).events
     assert denied.status_code == 401
     assert "stackchain_session=" not in denied.headers.get("set-cookie", "")
     assert [event.kind for event in events] == ["passkey_counter_anomaly"]
@@ -470,7 +471,7 @@ async def test_stale_nonzero_passkey_counter_denies_fresh_authorization(
             headers=headers,
         )
 
-    events = main._security_event_store().list(limit=10).events
+    events = main._security_event_store().list(principal_id=42, limit=10).events
     assert options.status_code == 200
     assert denied.status_code == 401
     assert "grant" not in denied.json()

tests/test_human_gate_api.py

@@ -211,7 +211,7 @@ async def test_successful_decision_records_one_completed_privacy_safe_security_e
             headers={"Idempotency-Key": "decision-audited"},
         )
 
-    events = journal.list().events
+    events = journal.list(principal_id=1).events
     assert decided.status_code == 201
     assert [
         {"kind": event.kind, "method": event.method, "target": event.target, "status": event.status}
@@ -282,7 +282,7 @@ async def test_rejected_decision_discards_its_pending_security_event(
         )
 
     assert rejected.status_code == 409
-    assert journal.list().events == []
+    assert journal.list(principal_id=1).events == []
 
 
 @pytest.mark.anyio

tests/test_private_state.py

@@ -107,7 +107,9 @@ def test_eager_private_stores_share_the_private_filesystem_boundary(tmp_path, bu
 @pytest.mark.parametrize(
     "exercise",
     [
-        lambda path: SecurityEventStore(path, clock=lambda: 1).record("sign_in"),
+        lambda path: SecurityEventStore(path, clock=lambda: 1).record(
+            "sign_in", principal_id=42
+        ),
         lambda path: LoginAttemptStore(
             path, clock=lambda: 1, max_failures=3, window_seconds=60
         ).record_failure("203.0.113.10"),

tests/test_pull_api.py

@@ -1599,7 +1599,7 @@ async def test_assigned_pull_merge_reports_success_and_retains_pending_audit_whe
     lifecycle = []
 
     class InterruptedJournal:
-        def reserve(self, kind, *, target):
+        def reserve(self, kind, *, principal_id, target):
             lifecycle.append(("reserve", kind, target))
             return "merge-operation"
 
@@ -1640,7 +1640,7 @@ async def test_assigned_pull_merge_reconciles_acceptance_before_timeout(monkeypa
     merged = False
 
     class LifecycleJournal:
-        def reserve(self, kind, *, target):
+        def reserve(self, kind, *, principal_id, target):
             calls.append(("reserve", kind, target))
             return "merge-operation"
 
@@ -1782,7 +1782,7 @@ async def test_assigned_pull_merge_discards_audit_reservation_after_definite_rej
     lifecycle = []
 
     class LifecycleJournal:
-        def reserve(self, kind, *, target):
+        def reserve(self, kind, *, principal_id, target):
             lifecycle.append(("reserve", kind, target))
             return "merge-operation"
 
@@ -2240,7 +2240,7 @@ async def test_source_branch_cleanup_endpoint_audits_and_deletes_the_exact_merge
     lifecycle = []
 
     class Journal:
-        def reserve(self, kind, *, target):
+        def reserve(self, kind, *, principal_id, target):
             lifecycle.append(("reserve", kind, target))
             return "branch-cleanup"
 
@@ -2740,7 +2740,7 @@ async def test_release_rollback_endpoint_requires_failed_exact_commit_and_audits
     lifecycle = []
 
     class Journal:
-        def reserve(self, kind, *, target):
+        def reserve(self, kind, *, principal_id, target):
             lifecycle.append(("reserve", kind, target))
             return "rollback-operation"
 

tests/test_security_activity.py

@@ -66,6 +66,32 @@ async def test_authenticated_security_activity_lists_private_sign_in_history(sec
     assert b"stackchain_session" not in persisted
 
 
+@pytest.mark.anyio
+async def test_security_activity_does_not_follow_a_previous_upstream_identity(
+    security_access, monkeypatch
+):
+    transport = httpx.ASGITransport(app=main.app)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as timmy:
+        await timmy.post(
+            "/api/v1/session",
+            json={"access_token": "correct horse battery staple", "device_label": "Timmy phone"},
+        )
+
+    async def other_user():
+        return {"id": 84, "login": "other"}
+
+    monkeypatch.setattr(main, "current_user", other_user)
+    async with httpx.AsyncClient(transport=transport, base_url="https://test") as other:
+        signed_in = await other.post(
+            "/api/v1/session",
+            json={"access_token": "correct horse battery staple", "device_label": "Other phone"},
+        )
+        activity = await other.get("/api/v1/security-events")
+
+    assert signed_in.status_code == 200
+    assert [event["device_label"] for event in activity.json()["events"]] == ["Other phone"]
+
+
 @pytest.mark.anyio
 async def test_tampered_security_activity_returns_no_partial_history(security_access):
     transport = httpx.ASGITransport(app=main.app, raise_app_exceptions=False)
@@ -211,7 +237,7 @@ async def test_passkey_enrollment_registry_failure_discards_reserved_event(
 
     events = SecurityEventStore(
         security_access / "security.sqlite3", clock=lambda: 0
-    ).list(limit=10).events
+    ).list(principal_id=42, limit=10).events
     assert enrolled.status_code == 503
     assert enrolled.json() == {"detail": "Passkey registry is temporarily unavailable"}
     assert all(event.kind != "passkey_enrolled" for event in events)
@@ -596,7 +622,8 @@ async def test_comment_deletion_journal_tracks_failed_and_confirmed_outcomes(
         return {"id": 42, "deleted": True}
 
     class Journal:
-        def reserve(self, kind, *, target):
+        def reserve(self, kind, *, principal_id, target):
+            assert principal_id == 42
             journal_calls.append(("reserve", kind, target))
             return "operation-42"
 
@@ -873,7 +900,7 @@ async def test_sign_out_is_journaled_before_the_session_is_removed(security_acce
 
     events = SecurityEventStore(
         security_access / "security.sqlite3", clock=lambda: 0
-    ).list(limit=10).events
+    ).list(principal_id=42, limit=10).events
     assert signed_out.status_code == 200
     assert [(event.kind, event.device_label) for event in events[:2]] == [
         ("sign_out", None),
@@ -974,7 +1001,7 @@ async def test_sign_out_all_is_journaled_after_sessions_are_removed(security_acc
 
     events = SecurityEventStore(
         security_access / "security.sqlite3", clock=lambda: 0
-    ).list(limit=10).events
+    ).list(principal_id=42, limit=10).events
     assert response.status_code == 200
     assert events[0].kind == "all_sessions_revoked"
     assert events[0].target == "all_devices"

tests/test_security_event_store.py

@@ -12,7 +12,7 @@ def test_security_event_schema_exposes_no_plaintext_metadata_columns(tmp_path):
     store = SecurityEventStore(
         tmp_path / "security.sqlite3", clock=lambda: 1_000, encryption_key=PRIVATE_KEY
     )
-    store.record("issue_closed", method="passkey", target="private/repo#42")
+    store.record("issue_closed", principal_id=42, method="passkey", target="private/repo#42")
 
     with sqlite3.connect(store.path) as connection:
         columns = {
@@ -22,6 +22,7 @@ def test_security_event_schema_exposes_no_plaintext_metadata_columns(tmp_path):
     assert columns == {
         "id",
         "payload",
+        "principal_id",
         "created_at",
         "status",
         "operation_id",
@@ -38,7 +39,7 @@ def test_security_event_payload_is_encrypted_at_rest_and_survives_restart(tmp_pa
     }
     store = SecurityEventStore(path, clock=lambda: 1_000, encryption_key=PRIVATE_KEY)
 
-    store.record(**canaries)
+    store.record(principal_id=42, **canaries)
 
     with sqlite3.connect(path) as connection:
         payload = connection.execute(
@@ -48,13 +49,13 @@ def test_security_event_payload_is_encrypted_at_rest_and_survives_restart(tmp_pa
     database_bytes = path.read_bytes()
     assert all(value.encode() not in database_bytes for value in canaries.values())
     reopened = SecurityEventStore(path, clock=lambda: 1_001, encryption_key=PRIVATE_KEY)
-    event = reopened.list(limit=10).events[0]
+    event = reopened.list(principal_id=42, limit=10).events[0]
     assert (event.kind, event.method, event.device_label, event.target) == tuple(
         canaries.values()
     )
 
 
-def test_legacy_security_events_migrate_without_changing_journal_semantics(tmp_path):
+def test_legacy_security_events_migrate_encrypted_but_remain_unattributed(tmp_path):
     path = tmp_path / "security.sqlite3"
     with sqlite3.connect(path) as connection:
         connection.executescript(
@@ -80,17 +81,8 @@ def test_legacy_security_events_migrate_without_changing_journal_semantics(tmp_p
         )
 
     store = SecurityEventStore(path, clock=lambda: 1_000, encryption_key=PRIVATE_KEY)
-    page = store.list(limit=1)
+    assert store.list(principal_id=42, limit=10).events == []
-    older = store.list(limit=10, cursor=page.next_cursor)
-
-    assert [(event.id, event.kind, event.created_at, event.status) for event in page.events] == [
-        (9, "legacy_issue_closed_canary", 901, "pending")
-    ]
-    assert [(event.id, event.kind, event.created_at, event.status) for event in older.events] == [
-        (7, "legacy_sign_in_canary", 900, "completed")
-    ]
     store.finalize("operation-9")
-    assert store.list(limit=1).events[0].status == "completed"
     with sqlite3.connect(path) as connection:
         rows = connection.execute(
             "SELECT id, payload FROM security_events ORDER BY id"
@@ -105,12 +97,12 @@ def test_legacy_security_events_migrate_without_changing_journal_semantics(tmp_p
 def test_wrong_key_or_tampered_security_activity_fails_closed(tmp_path):
     path = tmp_path / "security.sqlite3"
     store = SecurityEventStore(path, clock=lambda: 1_000, encryption_key=PRIVATE_KEY)
-    store.record("issue_closed", target="private/repo#42")
+    store.record("issue_closed", principal_id=42, target="private/repo#42")
 
     with pytest.raises(SecurityEventStoreError, match="temporarily unavailable"):
         SecurityEventStore(
             path, clock=lambda: 1_001, encryption_key=b"x" * 32
-        ).list(limit=10)
+        ).list(principal_id=42, limit=10)
 
     with sqlite3.connect(path) as connection:
         payload = connection.execute(
@@ -122,7 +114,7 @@ def test_wrong_key_or_tampered_security_activity_fails_closed(tmp_path):
             (payload[:-1] + replacement,),
         )
     with pytest.raises(SecurityEventStoreError, match="temporarily unavailable"):
-        store.list(limit=10)
+        store.list(principal_id=42, limit=10)
 
 
 def test_missing_security_activity_encryption_key_fails_with_store_error(
@@ -140,20 +132,21 @@ def test_security_events_are_private_bounded_and_reverse_chronological(tmp_path)
 
     store.record(
         "sign_in",
+        principal_id=42,
         method="token",
         device_label="  Timmy   Phone  " + "x" * 80,
         target="dashboard",
     )
     now[0] += 1
-    store.record("device_revoked", device_label="Old phone", target="device")
+    store.record("device_revoked", principal_id=42, device_label="Old phone", target="device")
 
-    page = store.list(limit=1)
+    page = store.list(principal_id=42, limit=1)
     assert [(event.kind, event.device_label, event.target) for event in page.events] == [
         ("device_revoked", "Old phone", "device")
     ]
     assert page.next_cursor is not None
 
-    older = store.list(limit=10, cursor=page.next_cursor)
+    older = store.list(principal_id=42, limit=10, cursor=page.next_cursor)
     assert older.events[0].kind == "sign_in"
     assert older.events[0].method == "token"
     assert older.events[0].device_label == ("Timmy Phone " + "x" * 52)
@@ -162,10 +155,29 @@ def test_security_events_are_private_bounded_and_reverse_chronological(tmp_path)
         row[1] for row in sqlite3.connect(store.path).execute("PRAGMA table_info(security_events)")
     }
     assert columns == {
-        "id", "payload", "created_at", "status", "operation_id",
+        "id", "payload", "principal_id", "created_at", "status", "operation_id",
     }
 
 
+def test_security_events_are_isolated_by_principal_across_cursor_pages(tmp_path):
+    now = [1_000]
+    store = SecurityEventStore(
+        tmp_path / "security.sqlite3", clock=lambda: now[0], encryption_key=PRIVATE_KEY
+    )
+    store.record("sign_in", principal_id=42, device_label="Timmy phone")
+    now[0] += 1
+    store.record("sign_in", principal_id=84, device_label="Other phone")
+    now[0] += 1
+    store.record("device_revoked", principal_id=42, device_label="Old Timmy phone")
+
+    first = store.list(principal_id=42, limit=1)
+    second = store.list(principal_id=42, limit=10, cursor=first.next_cursor)
+
+    assert [event.device_label for event in first.events] == ["Old Timmy phone"]
+    assert [event.device_label for event in second.events] == ["Timmy phone"]
+    assert all(event.device_label != "Other phone" for event in first.events + second.events)
+
+
 def test_security_event_retention_prunes_age_and_count(tmp_path):
     now = [0]
     store = SecurityEventStore(
@@ -176,32 +188,32 @@ def test_security_event_retention_prunes_age_and_count(tmp_path):
     )
     for index in range(4):
         now[0] = index
-        store.record("sign_in", method="token", device_label=f"Device {index}")
+        store.record("sign_in", principal_id=42, method="token", device_label=f"Device {index}")
 
-    assert [event.device_label for event in store.list(limit=10).events] == [
+    assert [event.device_label for event in store.list(principal_id=42, limit=10).events] == [
         "Device 3", "Device 2", "Device 1"
     ]
 
     now[0] = 20
-    store.record("sign_out", device_label="Current")
+    store.record("sign_out", principal_id=42, device_label="Current")
-    assert [event.kind for event in store.list(limit=10).events] == ["sign_out"]
+    assert [event.kind for event in store.list(principal_id=42, limit=10).events] == ["sign_out"]
 
 
 def test_security_event_reservation_is_durable_until_finalized(tmp_path):
     store = SecurityEventStore(tmp_path / "security.sqlite3", clock=lambda: 1_000)
 
-    operation_id = store.reserve("issue_closed", target="stackchain/api#7")
+    operation_id = store.reserve("issue_closed", principal_id=42, target="stackchain/api#7")
 
     pending = SecurityEventStore(
         tmp_path / "security.sqlite3", clock=lambda: 1_001
-    ).list(limit=10).events
+    ).list(principal_id=42, limit=10).events
     assert [(event.kind, event.target, event.status) for event in pending] == [
         ("issue_closed", "stackchain/api#7", "pending")
     ]
 
     store.finalize(operation_id)
 
-    completed = store.list(limit=10).events
+    completed = store.list(principal_id=42, limit=10).events
     assert [(event.kind, event.target, event.status) for event in completed] == [
         ("issue_closed", "stackchain/api#7", "completed")
     ]
@@ -209,8 +221,8 @@ def test_security_event_reservation_is_durable_until_finalized(tmp_path):
 
 def test_failed_operation_can_discard_its_pending_reservation(tmp_path):
     store = SecurityEventStore(tmp_path / "security.sqlite3", clock=lambda: 1_000)
-    operation_id = store.reserve("issue_closed", target="stackchain/api#7")
+    operation_id = store.reserve("issue_closed", principal_id=42, target="stackchain/api#7")
 
     store.discard(operation_id)
 
-    assert store.list(limit=10).events == []
+    assert store.list(principal_id=42, limit=10).events == []