diff --git a/alembic/versions/0003_tenant_contract.py b/alembic/versions/0003_tenant_contract.py
new file mode 100644
index 0000000..bff9cf0
--- /dev/null
+++ b/alembic/versions/0003_tenant_contract.py
@@ -0,0 +1,35 @@
+"""add first-class tenant provenance to audit_events
+
+SAT-1: organization_id is nullable for backward compatibility. Existing
+rows are explicitly classified as unknown, never global. Producers may mark
+an event organization-scoped or explicitly global in tenant_scope.
+"""
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "0003_tenant_contract"
+down_revision: str | None = "0002_integrity_status"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column("audit_events", sa.Column("organization_id", sa.String(length=255), nullable=True))
+    op.add_column(
+        "audit_events",
+        sa.Column("tenant_scope", sa.String(length=16), nullable=False, server_default="unknown"),
+    )
+    op.create_index(
+        "ix_audit_events_org_timestamp_event",
+        "audit_events",
+        ["organization_id", "timestamp", "event_id"],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index("ix_audit_events_org_timestamp_event", table_name="audit_events")
+    op.drop_column("audit_events", "tenant_scope")
+    op.drop_column("audit_events", "organization_id")
diff --git a/audit/decorators.py b/audit/decorators.py
index d275efc..c3a63f6 100644
--- a/audit/decorators.py
+++ b/audit/decorators.py
@@ -1,9 +1,9 @@
 import functools
-import asyncio
+
+from audit.config import AuditConfig
+from audit.context import get_identity, get_trace_id, get_user_id
 from audit.logger import AuditLogger
 from audit.models import AuditEvent
-from audit.config import AuditConfig
-from audit.context import get_trace_id, get_user_id, get_identity
 
 logger = AuditLogger()
 
@@ -26,6 +26,8 @@ async def async_inner(*args, **kwargs):
                     service=AuditConfig.SERVICE_NAME,
                     event_type=event_type,
                     user_id=get_user_id(),
+                    organization_id=str(identity.org_id) if identity and identity.org_id is not None else None,
+                    tenant_scope="organization" if identity and identity.org_id is not None else "unknown",
                     action=action,
                     decision="success",
                     trace_id=get_trace_id(),
@@ -37,4 +39,4 @@ async def async_inner(*args, **kwargs):
 
         return async_inner
 
-    return wrapper
\ No newline at end of file
+    return wrapper
diff --git a/audit/models.py b/audit/models.py
index a1af066..64c7299 100644
--- a/audit/models.py
+++ b/audit/models.py
@@ -1,7 +1,8 @@
-from pydantic import BaseModel, Field
-from typing import Optional, Dict, Any
-from datetime import datetime
 import uuid
+from datetime import datetime
+from typing import Any, Literal
+
+from pydantic import BaseModel, Field, model_validator
 
 
 class AuditEvent(BaseModel):
@@ -14,12 +15,30 @@ class AuditEvent(BaseModel):
     service: str
     event_type: str   # auth, iam, policy, tes
 
-    user_id: Optional[str] = None
+    user_id: str | None = None
+    # First-class tenant provenance. This is intentionally separate from the
+    # arbitrary context object: context is never an authority for tenant
+    # isolation. Legacy payloads default to UNKNOWN, not GLOBAL.
+    organization_id: str | None = None
+    tenant_scope: Literal["organization", "global", "unknown"] = "unknown"
     action: str = ""
-    resource: Optional[str] = None
+    resource: str | None = None
+
+    decision: str | None = None  # allow / deny / success / fail
+    reason: str | None = None
 
-    decision: Optional[str] = None  # allow / deny / success / fail
-    reason: Optional[str] = None
+    trace_id: str | None = None
+    context: dict[str, Any] = Field(default_factory=dict)
 
-    trace_id: Optional[str] = None
-    context: Dict[str, Any] = {}
\ No newline at end of file
+    @model_validator(mode="after")
+    def validate_tenant_scope(self) -> "AuditEvent":
+        # Supplying a first-class organization ID is itself an authoritative
+        # organization-scoped declaration. The explicit discriminator is
+        # still emitted so readers can distinguish legacy UNKNOWN from GLOBAL.
+        if self.organization_id is not None and self.tenant_scope == "unknown":
+            self.tenant_scope = "organization"
+        if self.tenant_scope == "organization" and self.organization_id is None:
+            raise ValueError("organization tenant scope requires organization_id")
+        if self.tenant_scope != "organization" and self.organization_id is not None:
+            raise ValueError("organization_id requires organization tenant scope")
+        return self
diff --git a/consumers/sink.py b/consumers/sink.py
index 7818aab..6f4d3ce 100644
--- a/consumers/sink.py
+++ b/consumers/sink.py
@@ -28,6 +28,8 @@ def write(self, event: dict) -> bool:
             service=event["service"],
             event_type=event["event_type"],
             user_id=event.get("user_id"),
+            organization_id=event.get("organization_id"),
+            tenant_scope=event.get("tenant_scope", "unknown"),
             action=event.get("action", ""),
             resource=event.get("resource"),
             decision=event.get("decision"),
diff --git a/db/models.py b/db/models.py
index 9c436ee..2e175c4 100644
--- a/db/models.py
+++ b/db/models.py
@@ -1,4 +1,4 @@
-from sqlalchemy import JSON, Column, DateTime, String, Text
+from sqlalchemy import JSON, Column, DateTime, Index, String, Text
 from sqlalchemy.sql import func
 
 from db.base import Base
@@ -22,6 +22,10 @@ class AuditEventRecord(Base):
     service = Column(String(255), nullable=False)
     event_type = Column(String(255), nullable=False)
     user_id = Column(String(255), nullable=True)
+    organization_id = Column(String(255), nullable=True)
+    # Legacy rows and events without an authoritative tenant are UNKNOWN.
+    # GLOBAL is only valid when a producer explicitly declares it.
+    tenant_scope = Column(String(16), nullable=False, server_default="unknown")
     action = Column(String(255), nullable=False, default="")
     resource = Column(String(255), nullable=True)
     decision = Column(String(64), nullable=True)
@@ -37,3 +41,7 @@ class AuditEventRecord(Base):
     # new "no signature" rows read identically.
     integrity_status = Column(String(16), nullable=False, server_default="unsigned")
     created_at = Column(DateTime, server_default=func.now(), nullable=False)
+
+    __table_args__ = (
+        Index("ix_audit_events_org_timestamp_event", "organization_id", "timestamp", "event_id"),
+    )
diff --git a/docs/AUDIT_TENANT_CONTRACT_SAT1.md b/docs/AUDIT_TENANT_CONTRACT_SAT1.md
new file mode 100644
index 0000000..c95a0c3
--- /dev/null
+++ b/docs/AUDIT_TENANT_CONTRACT_SAT1.md
@@ -0,0 +1,81 @@
+# SAT-1 — Security Audit Tenant Contract
+
+Status: **implemented; ready for AE-2 authorization work**
+
+## Current lifecycle
+
+Producers create an `AuditEvent`, serialize it once, sign that exact JSON
+string, and publish it to Redis `audit:events`. The worker parses the payload,
+classifies integrity, and persists it to SQL `audit_events`. `GET /audit/events`
+reads the durable table. Redis remains ingestion/backlog, not query history.
+
+## Canonical tenant contract
+
+`organization_id: string | null` is now a first-class event-envelope and
+durable column. It is separate from `context` and is covered by existing
+signing because the logger signs the complete serialized envelope.
+
+`tenant_scope` distinguishes:
+
+| Value | Meaning |
+|---|---|
+| `organization` | `organization_id` is authoritative and present |
+| `global` | Producer explicitly declares a platform/global event |
+| `unknown` | Tenant is unavailable or event predates SAT-1 |
+
+Null `organization_id` never means global. Legacy payloads default to
+`unknown`; the consumer does not infer tenant identity from context, resources,
+URLs, users, traces, or service names.
+
+## Producer coverage matrix
+
+| Producer | Authoritative tenant | Current propagation | Required future change | Risk |
+|---|---|---|---|---|
+| Security Audit native logger/test route | No for platform smoke events; field supported | Envelope model/logger | Explicitly mark global only when deliberate | Low |
+| API Gateway | Verified user/org context exists | Builder omits field | Pass verified org ID | Medium |
+| RAG | Verified IAM `UserContext.org_id` | Context in several paths | Pass top-level field | Medium |
+| TES | Organization context on authenticated runs | Older producer contract | Add verified field | Medium |
+| Workflow Bundles | Varies by path | Producer-specific context | Establish verified propagation | High |
+| LIMS | Request identity varies | Producer-specific context | Add only when verified | High |
+| Model Registry | Verified `UserContext.org_id` | Producer-specific clients | Pass explicit field | Medium |
+| Control Center/other producers | Varies | Producer-specific payloads | Adopt selectively | Medium |
+
+No upstream repositories were modified. Old producers remain readable and
+become `tenant_scope=unknown`.
+
+## Storage and query preparation
+
+Migration `0003_tenant_contract` adds nullable `organization_id` and non-null
+`tenant_scope` with an `unknown` server default. Existing rows are not
+rewritten. Index `ix_audit_events_org_timestamp_event` supports future bounded
+organization/time/ID queries. The query service accepts an internal
+`organization_id` filter, but the existing HTTP route remains platform-admin
+only; SAT-1 does not expose organization-scoped access or trust a query-string
+tenant ID.
+
+## Authorization boundary
+
+Platform-wide access remains verified JWT plus `platform_admin`. Organization-
+scoped access is deferred until the authoritative permission and verified claim
+used to derive caller organization are specified and tested. A browser-supplied
+organization ID must never widen scope.
+
+## Integrity, metadata, retention, and freshness
+
+The tenant fields are included in signed JSON bytes; verification is unchanged.
+Raw signatures and keys remain absent from responses. Legacy arbitrary `context`
+is retained for compatibility but is not tenant identity and needs an explicit
+allowlist before Audit Explorer exposes it. Credentials, tokens, cookies, JWTs,
+authorization headers, API keys, signing material, environment values, private
+paths, and unrestricted context are excluded from the future safe response.
+
+Redis max length is a backlog cap, not retention. Durable retention,
+freshness/as-of timestamps, source-unavailable responses, and safe metadata
+redaction remain undefined and are not invented here.
+
+## AE-2 prerequisites
+
+Expand producer coverage, then define verified caller scope, platform-wide
+override rules, unknown/global handling, bounded filters, safe metadata output,
+and unavailable/error behavior. Do not broaden the existing route based on this
+migration alone.
diff --git a/schemas/audit.py b/schemas/audit.py
index 18a4905..566bbc1 100644
--- a/schemas/audit.py
+++ b/schemas/audit.py
@@ -16,6 +16,8 @@ class AuditEventOut(BaseModel):
     service: str
     event_type: str
     user_id: str | None = None
+    organization_id: str | None = None
+    tenant_scope: str
     action: str
     resource: str | None = None
     decision: str | None = None
diff --git a/services/audit_query_service.py b/services/audit_query_service.py
index 600bcdb..70e308b 100644
--- a/services/audit_query_service.py
+++ b/services/audit_query_service.py
@@ -10,6 +10,7 @@ def list_audit_events(
     page: int,
     page_size: int,
     user_id: str | None = None,
+    organization_id: str | None = None,
     service: str | None = None,
     event_type: str | None = None,
     decision: str | None = None,
@@ -32,6 +33,8 @@ def list_audit_events(
 
     if user_id is not None:
         query = query.filter(AuditEventRecord.user_id == user_id)
+    if organization_id is not None:
+        query = query.filter(AuditEventRecord.organization_id == organization_id)
     if service is not None:
         query = query.filter(AuditEventRecord.service == service)
     if event_type is not None:
diff --git a/tests/test_audit_query_service.py b/tests/test_audit_query_service.py
index 19c9633..71d47fe 100644
--- a/tests/test_audit_query_service.py
+++ b/tests/test_audit_query_service.py
@@ -14,6 +14,7 @@ def _add(db_session, event_id, minutes_offset=0, **overrides):
         service=overrides.get("service", "auth"),
         event_type=overrides.get("event_type", "auth_login"),
         user_id=overrides.get("user_id", "u1"),
+        organization_id=overrides.get("organization_id"),
         action=overrides.get("action", "login"),
         resource=overrides.get("resource"),
         decision=overrides.get("decision", "success"),
@@ -43,6 +44,17 @@ def test_filters_by_user_id(db_session):
     assert [r.event_id for r in rows] == ["e1"]
 
 
+def test_filters_by_organization_id_without_authorizing_scope(db_session):
+    _add(db_session, "e1", organization_id="org-1")
+    _add(db_session, "e2", organization_id="org-2")
+    db_session.commit()
+    rows, total = audit_query_service.list_audit_events(
+        db_session, page=1, page_size=20, organization_id="org-1"
+    )
+    assert total == 1
+    assert [r.event_id for r in rows] == ["e1"]
+
+
 def test_filters_by_service(db_session):
     _add(db_session, "e1", service="auth")
     _add(db_session, "e2", service="policy")
diff --git a/tests/test_decorators.py b/tests/test_decorators.py
index e68d430..e5d3cf7 100644
--- a/tests/test_decorators.py
+++ b/tests/test_decorators.py
@@ -1,6 +1,8 @@
-import pytest
 from unittest.mock import AsyncMock, MagicMock, patch
-from audit.context import trace_id_var, user_id_var, identity_var
+
+import pytest
+
+from audit.context import identity_var, trace_id_var, user_id_var
 
 
 @pytest.fixture(autouse=True)
@@ -224,6 +226,8 @@ async def my_func():
             "verified": True,
         }
     }
+    assert event.organization_id == "7"
+    assert event.tenant_scope == "organization"
 
 
 @pytest.mark.asyncio
diff --git a/tests/test_logger.py b/tests/test_logger.py
index 87fc46c..e3c3e25 100644
--- a/tests/test_logger.py
+++ b/tests/test_logger.py
@@ -9,9 +9,11 @@
 exercise the real serializer construct a real AuditEvent.
 """
 import json
-import pytest
 from datetime import datetime
-from unittest.mock import AsyncMock, MagicMock, patch
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
 from audit.config import AuditConfig
 from audit.models import AuditEvent
 
@@ -45,7 +47,7 @@ async def test_log_writes_to_redis_stream(audit_logger):
     logger, mock_redis = audit_logger
     mock_redis.xadd = AsyncMock()
 
-    event, payload = _serializable_event(service="auth", event_type="auth_login",
+    event, _payload = _serializable_event(service="auth", event_type="auth_login",
                                          user_id="u1", action="login", decision="success")
     await logger.log(event)
 
@@ -88,7 +90,7 @@ async def test_log_event_includes_all_fields(audit_logger):
     logger, mock_redis = audit_logger
     mock_redis.xadd = AsyncMock()
 
-    event, expected = _serializable_event(
+    event, _expected = _serializable_event(
         service="iam",
         event_type="iam_cache_hit",
         user_id="u2",
@@ -172,6 +174,36 @@ async def test_log_real_event_id_survives_serialization(audit_logger):
     assert stored["event_id"] == "fixed-id-123"
 
 
+@pytest.mark.asyncio
+async def test_log_serializes_first_class_tenant_in_signed_payload(audit_logger):
+    logger, mock_redis = audit_logger
+    mock_redis.xadd = AsyncMock()
+    await logger.log(AuditEvent(
+        service="tes", event_type="run", organization_id="org-7",
+        tenant_scope="organization",
+    ))
+    stored = json.loads(mock_redis.xadd.call_args[0][1]["data"])
+    assert stored["organization_id"] == "org-7"
+    assert stored["tenant_scope"] == "organization"
+
+
+@pytest.mark.asyncio
+async def test_tenant_field_is_covered_by_signature(audit_logger):
+    logger, mock_redis = audit_logger
+    mock_redis.xadd = AsyncMock()
+    await logger.log(AuditEvent(
+        service="tes", event_type="run", organization_id="org-7",
+        tenant_scope="organization",
+    ))
+    fields = mock_redis.xadd.call_args[0][1]
+    from audit.signing import verify_audit_event
+    assert verify_audit_event("tes", fields["data"], fields["sig"], AuditConfig.EVENT_SIGNING_SECRET)
+    assert not verify_audit_event(
+        "tes", fields["data"].replace('"org-7"', '"org-8"'), fields["sig"],
+        AuditConfig.EVENT_SIGNING_SECRET,
+    )
+
+
 # ---------------------------------------------------------------------------
 # Error path: failures must never propagate
 # ---------------------------------------------------------------------------
diff --git a/tests/test_migrations.py b/tests/test_migrations.py
index 1f3d1be..ba62d10 100644
--- a/tests/test_migrations.py
+++ b/tests/test_migrations.py
@@ -4,16 +4,18 @@
 """
 from pathlib import Path
 
-from alembic import command
 from alembic.config import Config
 from sqlalchemy import create_engine, inspect
 
+from alembic import command
+
 REPO_ROOT = Path(__file__).resolve().parent.parent
 
 EXPECTED_COLUMNS = {
     "event_id", "timestamp", "service", "event_type", "user_id", "action",
     "resource", "decision", "reason", "trace_id", "context", "created_at",
     "integrity_status",
+    "organization_id", "tenant_scope",
 }
 
 
@@ -86,6 +88,34 @@ def test_integrity_status_column_exists_after_upgrade(tmp_path):
     assert columns["integrity_status"]["nullable"] is False
 
 
+def test_tenant_columns_and_query_index_exist_after_upgrade(tmp_path):
+    db_file = tmp_path / "migration_test.db"
+    cfg = _alembic_config(f"sqlite:///{db_file}")
+    command.upgrade(cfg, "head")
+    engine = create_engine(f"sqlite:///{db_file}")
+    inspector = inspect(engine)
+    columns = {c["name"]: c for c in inspector.get_columns("audit_events")}
+    assert columns["organization_id"]["nullable"] is True
+    assert columns["tenant_scope"]["nullable"] is False
+    indexes = {i["name"] for i in inspector.get_indexes("audit_events")}
+    assert "ix_audit_events_org_timestamp_event" in indexes
+
+
+def test_legacy_rows_are_unknown_after_tenant_migration(tmp_path):
+    db_file = tmp_path / "migration_test.db"
+    cfg = _alembic_config(f"sqlite:///{db_file}")
+    command.upgrade(cfg, "0002_integrity_status")
+    engine = create_engine(f"sqlite:///{db_file}")
+    with engine.begin() as conn:
+        from sqlalchemy import text
+        conn.execute(text("INSERT INTO audit_events (event_id, timestamp, service, event_type, action, context) VALUES ('legacy', '2026-01-01 00:00:00', 'svc', 'test', '', '{}')"))
+    command.upgrade(cfg, "head")
+    with engine.begin() as conn:
+        from sqlalchemy import text
+        row = conn.execute(text("SELECT organization_id, tenant_scope FROM audit_events WHERE event_id='legacy'")).fetchone()
+    assert row == (None, "unknown")
+
+
 def test_existing_rows_backfill_to_unsigned_on_upgrade(tmp_path):
     """The exact PR2 migration guarantee: a row written under 0001 (before
     integrity_status existed at all) must read back as "unsigned" after
diff --git a/tests/test_models.py b/tests/test_models.py
index 9e06b6d..1d43ee6 100644
--- a/tests/test_models.py
+++ b/tests/test_models.py
@@ -11,7 +11,9 @@
 case) and assert they differ -- the check that would have caught the bug.
 """
 import time
-from datetime import datetime
+from datetime import datetime, timezone
+
+import pytest
 
 from audit.models import AuditEvent
 
@@ -35,7 +37,20 @@ def test_explicitly_supplied_event_id_is_respected():
     assert event.event_id == "fixed-id-123"
 
 
+def test_organization_tenant_scope_requires_first_class_id():
+    from pydantic import ValidationError
+    with pytest.raises(ValidationError):
+        AuditEvent(service="svc", event_type="test", tenant_scope="organization")
+
+
+def test_global_and_unknown_are_distinct():
+    global_event = AuditEvent(service="svc", event_type="maintenance", tenant_scope="global")
+    unknown_event = AuditEvent(service="svc", event_type="test")
+    assert global_event.tenant_scope == "global"
+    assert unknown_event.tenant_scope == "unknown"
+
+
 def test_explicitly_supplied_timestamp_is_respected():
-    fixed = datetime(2024, 1, 1, 0, 0, 0)
+    fixed = datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
     event = AuditEvent(service="svc", event_type="test", timestamp=fixed)
     assert event.timestamp == fixed
diff --git a/tests/test_processor.py b/tests/test_processor.py
index 26f01de..d201d3d 100644
--- a/tests/test_processor.py
+++ b/tests/test_processor.py
@@ -1,10 +1,11 @@
 import json
-import pytest
 from datetime import datetime
 
-from audit.models import AuditEvent
-from consumers.processor import process_event, parse_audit_event
+import pytest
+from pydantic import ValidationError
 
+from audit.models import AuditEvent
+from consumers.processor import parse_audit_event, process_event
 
 # ---------------------------------------------------------------------------
 # process_event
@@ -101,6 +102,31 @@ def test_parse_audit_event_preserves_all_fields():
     assert isinstance(event.timestamp, datetime)
 
 
+def test_parse_audit_event_preserves_first_class_tenant():
+    event = parse_audit_event(json.dumps({
+        "event_id": "tenant-1", "timestamp": "2026-01-01T12:00:00",
+        "service": "tes", "event_type": "run", "organization_id": "org-7",
+        "tenant_scope": "organization", "context": {"organization_id": "wrong"},
+    }))
+    assert event.organization_id == "org-7"
+    assert event.tenant_scope == "organization"
+
+
+def test_legacy_event_without_tenant_is_unknown_not_global():
+    event = parse_audit_event(json.dumps({"service": "svc", "event_type": "test"}))
+    assert event.organization_id is None
+    assert event.tenant_scope == "unknown"
+
+
+def test_tenant_is_never_inferred_from_context():
+    event = parse_audit_event(json.dumps({
+        "service": "svc", "event_type": "test",
+        "context": {"organization_id": "org-context-only"},
+    }))
+    assert event.organization_id is None
+    assert event.tenant_scope == "unknown"
+
+
 def test_parse_audit_event_raises_on_invalid_json():
     with pytest.raises(json.JSONDecodeError):
         parse_audit_event("not-json")
@@ -108,5 +134,5 @@ def test_parse_audit_event_raises_on_invalid_json():
 
 def test_parse_audit_event_raises_on_missing_required_fields():
     raw = json.dumps({"user_id": "u1"})  # missing service/event_type
-    with pytest.raises(Exception):
+    with pytest.raises(ValidationError):
         parse_audit_event(raw)
diff --git a/tests/test_routes_audit_events.py b/tests/test_routes_audit_events.py
index 375c415..4a608ad 100644
--- a/tests/test_routes_audit_events.py
+++ b/tests/test_routes_audit_events.py
@@ -2,7 +2,7 @@
 tests via the audit_events_client fixture (real SQLite DB + real FastAPI
 dependency injection, not mocks); SQL-level filter/order/pagination
 correctness is covered separately in tests/test_audit_query_service.py."""
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 import jwt
 
@@ -101,6 +101,7 @@ def test_response_contains_expected_fields(audit_events_client):
     item = body["items"][0]
     assert set(item.keys()) == {
         "event_id", "timestamp", "service", "event_type", "user_id",
+        "organization_id", "tenant_scope",
         "action", "resource", "decision", "reason", "trace_id", "context",
         "created_at", "integrity_status",
     }
@@ -228,6 +229,20 @@ def test_filter_by_integrity_status_via_query_param(audit_events_client):
     assert body["items"][0]["integrity_status"] == "invalid"
 
 
+def test_response_serializes_tenant_fields(audit_events_client):
+    client, sessions = audit_events_client
+    db = sessions()
+    db.add(AuditEventRecord(
+        event_id="org-event", timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc), service="tes",
+        event_type="run", organization_id="org-7", tenant_scope="organization", context={},
+    ))
+    db.commit()
+    db.close()
+    item = client.get("/audit/events", headers=_auth_headers()).json()["items"][0]
+    assert item["organization_id"] == "org-7"
+    assert item["tenant_scope"] == "organization"
+
+
 # ---------------------------------------------------------------------------
 # Existing endpoints unaffected
 # ---------------------------------------------------------------------------
diff --git a/tests/test_sink.py b/tests/test_sink.py
index 950a019..b2276a9 100644
--- a/tests/test_sink.py
+++ b/tests/test_sink.py
@@ -1,16 +1,16 @@
 """PR4.2 regression tests: Sink now persists to audit_events instead of
 printing (consumers/sink.py). Supersedes the print-based Sink tests that
 used to live in tests/test_processor.py."""
-from datetime import datetime
+from datetime import datetime, timezone
 
-from db.models import AuditEventRecord
 from consumers.sink import Sink
+from db.models import AuditEventRecord
 
 
 def _event(event_id="evt-1", **overrides):
     payload = {
         "event_id": event_id,
-        "timestamp": datetime(2026, 1, 1, 12, 0, 0),
+        "timestamp": datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc),
         "service": "auth",
         "event_type": "auth_login",
         "user_id": "u1",
@@ -36,6 +36,20 @@ def test_sink_write_persists_event(db_session):
     assert fetched.user_id == "u1"
 
 
+def test_sink_write_persists_first_class_tenant(db_session):
+    Sink(db_session).write(_event(organization_id="org-7", tenant_scope="organization"))
+    fetched = db_session.get(AuditEventRecord, "evt-1")
+    assert fetched.organization_id == "org-7"
+    assert fetched.tenant_scope == "organization"
+
+
+def test_sink_legacy_event_defaults_to_unknown_tenant(db_session):
+    Sink(db_session).write(_event())
+    fetched = db_session.get(AuditEventRecord, "evt-1")
+    assert fetched.organization_id is None
+    assert fetched.tenant_scope == "unknown"
+
+
 def test_sink_write_preserves_context(db_session):
     sink = Sink(db_session)
     sink.write(_event(context={"a": 1, "b": {"c": 2}}))
@@ -65,7 +79,7 @@ def test_sink_write_handles_optional_fields_missing(db_session):
     sink = Sink(db_session)
     minimal = {
         "event_id": "evt-minimal",
-        "timestamp": datetime(2026, 1, 1),
+        "timestamp": datetime(2026, 1, 1, tzinfo=timezone.utc),
         "service": "svc",
         "event_type": "test",
     }