diff --git a/.bumpversion.toml b/.bumpversion.toml
index 5a40649c..fc34248e 100644
--- a/.bumpversion.toml
+++ b/.bumpversion.toml
@@ -3,7 +3,7 @@
 # https://peps.python.org/pep-0440/
 
 [tool.bumpversion]
-    current_version = "1.0.2.dev4"
+    current_version = "1.0.2.dev7"
     parse = """(?x)
     (?P<major>0|[1-9]\\d*)\\.
     (?P<minor>0|[1-9]\\d*)\\.
diff --git a/.gitignore b/.gitignore
index 85b3c458..b3549b32 100644
--- a/.gitignore
+++ b/.gitignore
@@ -192,6 +192,7 @@ certs/
 docker-compose.override.yml
 
 CLAUDE.md
+/docker/init_pycharm_helpers.sh
 
 # Claude Code (share .claude/agents + .claude/skills; keep local settings out)
 .claude/settings.local.json
\ No newline at end of file
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index e8fe2b7d..074a4c0c 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -53,7 +53,7 @@ repos:
         name: pytest
         # --build avoids stale images; TEST_MARKER env is honored by entrypoint-test.sh
         # (CLI args after the service name are ignored), so set the marker via env.
-        entry: docker compose run --rm -T --build -e "TEST_MARKER=not integration" tests
+        entry: docker compose -f docker-compose.yml run --rm -T --build -e "TEST_MARKER=not integration" tests
         language: system
         pass_filenames: false
         types: [python]
diff --git a/docs/changelog/1.0.2.dev6.md b/docs/changelog/1.0.2.dev6.md
new file mode 100644
index 00000000..0369be38
--- /dev/null
+++ b/docs/changelog/1.0.2.dev6.md
@@ -0,0 +1,203 @@
+# 1.0.2.dev5 → 1.0.2.dev6 — Record Visibility & Context Scopes (incl. cross-owner USERS / ORGANIZATIONS)
+
+## Summary
+
+This changelog documents the storage work delivered across the **1.0.2.dev5 → 1.0.2.dev6** line. Two related
+capabilities landed:
+
+1. **Record visibility** — every storage record now carries a read-access scope (`Visibility`) that is independent of
+   write ownership. You can tag a record `PUBLIC` / `PRIVATE` / `INTERNAL` on write and filter by it on list.
+2. **Context scopes** — the old string `scope: Literal["mission", "setup"]`
+   argument on the storage service was replaced by a typed `Context`
+   enum, and the enum was extended with two **read-only cross-owner** scopes,
+   `USERS` and `ORGANIZATIONS`. `dev6` completes the client so those scopes are actually emitted on the wire — a kin can
+   now list records shared by other kins of the same user/organization.
+
+The concrete owner id for the cross-owner scopes is **resolved server-side**
+from request metadata; the client only sends the context *kind*.
+
+> **Requirement:** these features need `agentic-mesh-protocol` with the
+> `visibility` fields and the `CONTEXT_USERS` / `CONTEXT_ORGANIZATIONS` enum
+> values (shipped in the proto ≥ `1.0.1.dev4`). It is pulled in transitively by
+> this SDK version.
+
+## What changed
+
+### `Context` — the owner/scope of an operation (replaces `scope`)
+
+```python
+
+from digitalkin.models.services.services import Context
+
+
+class ContextStorage(Enum):
+    UNSPECIFIED = "unspecified"
+    MISSIONS = "missions"  # this mission (default)
+    SETUP_VERSIONS = "setup_versions"  # this setup version (shared across missions)
+    USERS = "users"  # read-only: all kins of the same user
+    ORGANIZATIONS = "organizations"  # read-only: all kins of the same organization
+```
+
+- `MISSIONS` (default) and `SETUP_VERSIONS` are **read/write** owner contexts.
+- `USERS` and `ORGANIZATIONS` are **read-only, list-only** cross-owner scopes. The strategy holds no user/org id; it
+  sends only the kind and the storage service resolves the concrete id from the `x-user-id` / `x-organization-id`
+  request metadata.
+
+Every public storage method now takes `context: ContextStorage` instead of the old `scope: str`:
+
+| Method              | Signature (relevant args)                                                                                                            |
+|---------------------|--------------------------------------------------------------------------------------------------------------------------------------|
+| `store`             | `store(collection, record_id, data, data_type=DataType.OUTPUT, context=ContextStorage.MISSIONS, visibility=Visibility.UNSPECIFIED)`  |
+| `read`              | `read(collection, record_id, context=ContextStorage.MISSIONS)`                                                                       |
+| `update`            | `update(collection, record_id, data, context=ContextStorage.MISSIONS, visibility=Visibility.UNSPECIFIED)`                            |
+| `remove`            | `remove(collection, record_id, context=ContextStorage.MISSIONS)`                                                                     |
+| `list`              | `list(collection, context=ContextStorage.MISSIONS, visibilities=None)`                                                               |
+| `remove_collection` | `remove_collection(collection, context=ContextStorage.MISSIONS)`                                                                     |
+| `upsert`            | `upsert(collection, record_id, data, data_type=DataType.OUTPUT, context=ContextStorage.MISSIONS, visibility=Visibility.UNSPECIFIED)` |
+
+### `Visibility` — read-access scope of a record
+
+```python
+from digitalkin.models.services.storage import Visibility
+
+class Visibility(Enum):
+    UNSPECIFIED = 0   # let the storage service apply its default
+    PUBLIC      = 1
+    PRIVATE     = 2
+    INTERNAL    = 3
+```
+
+- The integer values **mirror the storage proto** exactly.
+- Ownership (who may *edit*) stays keyed on the record's `context`; `Visibility`
+  only governs who may *read* it.
+- `StorageRecord` gained a `visibility: Visibility` field (default
+  `UNSPECIFIED`), populated from the wire on read.
+- `visibility=UNSPECIFIED` is the proto default (`0`) and is wire-identical to not setting it, so the storage service
+  applies its own default.
+
+### Cross-owner wire mapping (completed in dev6)
+
+`GrpcStorage._context_enum` now maps the resolved context to the right wire enum, including the new cross-owner kinds:
+
+- `setup_versions:…` → `CONTEXT_SETUP_VERSIONS`
+- `users:` → `CONTEXT_USERS`
+- `organizations:` → `CONTEXT_ORGANIZATIONS`
+- otherwise → `CONTEXT_MISSIONS`
+
+and `StorageStrategy._resolve_context` returns a kind-only marker (`users:` /
+`organizations:`) for the cross-owner scopes, since the concrete id is resolved server-side.
+
+> **Local `DefaultStorage`** has no cross-owner data model, so listing under
+> `USERS` / `ORGANIZATIONS` returns `[]` in local/dev mode. Cross-owner reads
+> are a remote (`GrpcStorage`) capability.
+
+## How to use
+
+All examples assume you have a storage strategy (e.g. `context.storage` inside a trigger handler).
+
+### Write a record with a visibility
+
+```python
+from digitalkin.models.services.storage import Visibility
+from digitalkin.models.services.services import Context
+
+# Readable by every kin of the same user, owned by this mission
+await storage.store(
+    "reports",
+    "q3-summary",
+    {"title": "Q3", "body": "..."},
+    visibility=Visibility.PUBLIC,
+)
+
+# Persist under the setup version (survives across missions), keep it internal
+await storage.upsert(
+    "shared_config",
+    "defaults",
+    {"lang": "fr"},
+    context=Context.SETUP_VERSIONS,
+    visibility=Visibility.INTERNAL,
+)
+```
+
+### Change a record's visibility later
+
+```python
+# UNSPECIFIED leaves the current visibility unchanged
+await storage.update("reports", "q3-summary", {"title": "Q3", "body": "..."},
+                     visibility=Visibility.PRIVATE)
+```
+
+### List and filter by visibility
+
+```python
+# All readable records in this mission
+records = await storage.list("reports")
+
+# Only PUBLIC + INTERNAL records
+records = await storage.list(
+    "reports",
+    visibilities=[Visibility.PUBLIC, Visibility.INTERNAL],
+)
+
+for r in records:
+    print(r.record_id, r.visibility.name, r.context)
+```
+
+### Cross-owner reads (discover data produced by other kins of the same user)
+
+```python
+# Records other kins of the SAME USER created and shared, subject to visibility.
+# The server resolves the concrete user id from the request metadata.
+records = await storage.list(
+    "reports",
+    context=ContextStorage.USERS,
+    visibilities=[Visibility.PUBLIC],
+)
+
+# Same, but across the whole organization
+records = await storage.list("reports", context=ContextStorage.ORGANIZATIONS)
+```
+
+> Cross-owner scopes are **read-only**: use them with `list` only. `store` /
+> `update` / `remove` always target the owning `MISSIONS` / `SETUP_VERSIONS`
+> context.
+
+## Migration
+
+- **`scope=` → `context=`**: replace every `scope="mission"` /
+  `scope="setup"` string argument with `context=ContextStorage.MISSIONS` /
+  `context=ContextStorage.SETUP_VERSIONS`. The parameter was renamed and retyped from a `str` literal to the
+  `Context` enum, so passing the old string raises `TypeError`.
+- **`data_type`**: pass the `DataType` enum (e.g. `DataType.OUTPUT`), not a string — `data_type="OUTPUT"` no longer
+  works.
+- **New optional args**: `visibility` (on `store`/`update`/`upsert`) and
+  `visibilities` (on `list`) are optional; omit them to keep the previous behaviour (server default visibility, no
+  visibility filter).
+- **No change to `read` / `remove` semantics** beyond the `scope` → `context`
+  rename.
+
+Minimal before/after:
+
+```python
+# before (<= 1.0.0a0)
+await storage.list("reports", scope="setup")
+await storage.store("reports", "r1", data, data_type="OUTPUT")
+
+# after (>= 1.0.2.dev6)
+from digitalkin.models.services.storage import DataType
+from digitalkin.models.services.services import Context
+
+await storage.list("reports", context=Context.SETUP_VERSIONS)
+await storage.store("reports", "r1", data, data_type=DataType.OUTPUT)
+```
+
+## Verification
+
+Storage regression coverage lives in `tests/services/storage/`:
+
+- `test_grpc_storage.py` — round-trips `visibility` on `store`/`update`, the
+  `visibilities` filter on `list`, and
+  `test_list_cross_owner_context_and_visibilities` locks the wire mapping (`USERS → CONTEXT_USERS`,
+  `ORGANIZATIONS → CONTEXT_ORGANIZATIONS`).
+- `test_storage_strategy_locks.py` — per-record lock keys use the resolved context string, so locks are created and
+  cleaned up under the right owner.
diff --git a/docs/changelog/1.0.2.dev7.md b/docs/changelog/1.0.2.dev7.md
new file mode 100644
index 00000000..5182eb4f
--- /dev/null
+++ b/docs/changelog/1.0.2.dev7.md
@@ -0,0 +1,115 @@
+# 1.0.2.dev6 → 1.0.2.dev7 — Filesystem Context Scopes (mission/setup/user/organization)
+
+## Summary
+
+This release brings the **filesystem** service in line with the storage context model shipped in `dev6`. The string
+`context: Literal["mission", "setup"]`
+argument is replaced by a typed `Context` enum, extended with the two **read-only cross-owner** scopes `USERS` and
+`ORGANIZATIONS`. A file produced by one kin can now be read by another kin of the same user/organization, subject to
+server-side access control.
+
+A small consistency fix also lands on the **storage** side: `ContextStorage.UNSPECIFIED`
+now maps to the unspecified wire enum instead of being silently treated as MISSIONS, matching the filesystem behaviour.
+
+Only the context *kind* is sent on the wire — no id is transmitted by the client. The concrete owner (mission / setup /
+user / organization) is resolved server-side from the request context.
+
+> **Requirement:** needs `agentic-mesh-protocol` with the filesystem
+> `CONTEXT_USERS` / `CONTEXT_ORGANIZATIONS` enum values. It is pulled in
+> transitively by this SDK version.
+
+## What changed
+
+### `Context` — the owner/scope of a filesystem operation (replaces `scope`/`context` strings)
+
+```python
+
+from digitalkin.models.services.services import Context
+
+
+class ContextFile(Enum):
+  UNSPECIFIED = "unspecified"
+  MISSIONS = "mission"  # this mission (default)
+  SETUP = "setup"  # this setup version
+  USERS = "user"  # read-only: all kins of the same user
+  ORGANIZATIONS = "organization"  # read-only: all kins of the same organization
+```
+
+- `MISSIONS` (default) and `SETUP` are the read/write owner contexts.
+- `USERS` and `ORGANIZATIONS` are **read-only cross-owner** scopes (use them on reads: `get_file` / `get_files`).
+- The enum values are singular strings, so Pydantic still coerces legacy string contexts on `FileFilter` (e.g.
+  `FileFilter(context="setup")`).
+
+Read methods now take `context: ContextFile`:
+
+| Method      | Signature (relevant args)                                                             |
+|-------------|---------------------------------------------------------------------------------------|
+| `get_file`  | `get_file(file_id, context=ContextFile.MISSIONS, *, include_content=False)`           |
+| `get_files` | `get_files(filters, ...)` where `filters.context: ContextFile = ContextFile.MISSIONS` |
+
+`_context_enum` maps every kind to its wire enum, including
+`CONTEXT_USERS` / `CONTEXT_ORGANIZATIONS` / `CONTEXT_UNSPECIFIED`.
+
+### Writes stay owner-scoped
+
+`upload_files` / `update_file` / `delete_files` remain mission-scoped, exactly as before. Cross-owner scopes are
+read-only — you cannot write into another kin's user/organization space.
+
+### Storage: UNSPECIFIED consistency fix
+
+`ContextStorage.UNSPECIFIED` now resolves to the `unspecified:` kind marker and maps to `CONTEXT_UNSPECIFIED` on the
+wire (server applies its default), instead of silently becoming `CONTEXT_MISSIONS`. Public callers are unaffected — the
+default context stays `MISSIONS`.
+
+## How to use
+
+```python
+
+from digitalkin.models.services.services import Context
+from digitalkin.services.filesystem.filesystem_strategy import FileFilter
+
+# Read a file owned by the current mission (default)
+record = await filesystem.get_file(file_id, include_content=True)
+
+# Read a file from the setup-version scope
+record = await filesystem.get_file(file_id, context=Context.SETUP)
+
+# Cross-owner: list files shared by other kins of the same user.
+# The server resolves the concrete user id; no id is sent by the client.
+records, total = await filesystem.get_files(
+  FileFilter(context=Context.USERS, prefix="reports/"),
+)
+
+# Same across the whole organization
+records, total = await filesystem.get_files(FileFilter(context=Context.ORGANIZATIONS))
+```
+
+## Migration
+
+- **`context="mission"` / `context="setup"` → `Context`**: pass
+  `ContextFile.MISSIONS` / `ContextFile.SETUP` to `get_file`. For `FileFilter`, the legacy strings still validate
+  (Pydantic coerces them by value), but prefer the enum for clarity.
+- **No change to write calls** (`upload_files` / `update_file` / `delete_files`).
+- Import the enum from `digitalkin.models.services.filesystem`.
+
+Minimal before/after:
+
+```python
+# before
+await filesystem.get_file(file_id, context="setup")
+await filesystem.get_files(FileFilter(context="mission", prefix="x/"))
+
+# after
+from digitalkin.models.services.services import Context
+
+await filesystem.get_file(file_id, context=Context.SETUP)
+await filesystem.get_files(FileFilter(context=Context.MISSIONS, prefix="x/"))
+```
+
+## Verification
+
+- `tests/services/filesystem/test_grpc_filesystem.py::TestContextScopes` locks the wire mapping for all kinds —
+  `MISSIONS`, `SETUP`, `USERS`, `ORGANIZATIONS`,
+  `UNSPECIFIED` — on both `get_files` and `get_file`.
+- `tests/services/storage/test_grpc_storage.py::TestListData` covers
+  `UNSPECIFIED → CONTEXT_UNSPECIFIED` alongside the cross-owner storage scopes.
diff --git a/pyproject.toml b/pyproject.toml
index 0f8cf6ed..3589687d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -37,7 +37,7 @@
         "pydantic>=2.12.4",
         "redis[hiredis]>=7.4.0,<9",
     ]
-    version = "1.0.2.dev4"
+    version = "1.0.2.dev7"
 
     [project.optional-dependencies]
         agno = [ "agno>=2.6" ]
diff --git a/src/digitalkin/__version__.py b/src/digitalkin/__version__.py
index 9bc785d0..bab33813 100644
--- a/src/digitalkin/__version__.py
+++ b/src/digitalkin/__version__.py
@@ -5,4 +5,4 @@
 try:
     __version__ = version("digitalkin")
 except PackageNotFoundError:
-    __version__ = "1.0.2.dev4"
+    __version__ = "1.0.2.dev7"
diff --git a/src/digitalkin/grpc_servers/_base_server.py b/src/digitalkin/grpc_servers/_base_server.py
index 7834d17b..5b149cea 100644
--- a/src/digitalkin/grpc_servers/_base_server.py
+++ b/src/digitalkin/grpc_servers/_base_server.py
@@ -3,6 +3,7 @@
 import abc
 import asyncio
 import os
+import sys
 
 from digitalkin.models.settings.profiling import get_profiling_settings
 
@@ -196,10 +197,16 @@ def _create_server(self) -> GrpcServer:
         try:  # noqa: PLW0717
             grpc_compression = get_server_settings().grpc.compression.to_grpc()
 
-            try:
-                cpu_count = len(os.sched_getaffinity(0))
-                logger.info("vCPU count: %d", cpu_count)
-            except (AttributeError, OSError):
+            # sched_getaffinity is Linux-only; the sys.platform guard lets mypy skip
+            # it as unreachable on other platforms without a type: ignore.
+            if sys.platform == "linux":
+                try:
+                    cpu_count = len(os.sched_getaffinity(0))
+                    logger.info("vCPU count: %d", cpu_count)
+                except OSError:
+                    cpu_count = os.cpu_count() or 1
+                    logger.info("CPU count: %d", cpu_count)
+            else:
                 cpu_count = os.cpu_count() or 1
                 logger.info("CPU count: %d", cpu_count)
 
diff --git a/src/digitalkin/mixins/storage_mixin.py b/src/digitalkin/mixins/storage_mixin.py
index 12a91012..c35fbe1b 100644
--- a/src/digitalkin/mixins/storage_mixin.py
+++ b/src/digitalkin/mixins/storage_mixin.py
@@ -3,6 +3,7 @@
 from typing import Any, Literal
 
 from digitalkin.models.module.module_context import ModuleContext
+from digitalkin.models.services.storage import DataType
 from digitalkin.services.storage.storage_strategy import StorageRecord
 
 
@@ -36,7 +37,7 @@ async def store_storage(
         Raises:
             StorageServiceError: If storage operation fails
         """
-        return await context.storage.store(collection, record_id, data, data_type=data_type)
+        return await context.storage.store(collection, record_id, data, data_type=DataType[data_type])
 
     @staticmethod
     async def read_storage(context: ModuleContext, collection: str, record_id: str) -> StorageRecord | None:
@@ -101,4 +102,4 @@ async def upsert_storage(
         Raises:
             StorageServiceError: If upsert operation fails
         """
-        return await context.storage.upsert(collection, record_id, data, data_type=data_type)
+        return await context.storage.upsert(collection, record_id, data, data_type=DataType[data_type])
diff --git a/src/digitalkin/models/services/registry.py b/src/digitalkin/models/services/registry.py
index 3aeffee1..a1da0fea 100644
--- a/src/digitalkin/models/services/registry.py
+++ b/src/digitalkin/models/services/registry.py
@@ -18,7 +18,11 @@ class RegistryModuleStatus(str, Enum):
 
 
 class RegistryModuleType(str, Enum):
-    """Module type in the registry."""
+    """Module type in the registry.
+
+    Member names mirror the proto ``ModuleType`` enum (minus the ``MODULE_TYPE_``
+    prefix): they are looked up by name from the wire value, so they must match.
+    """
 
     UNSPECIFIED = "unspecified"
     ARCHETYPE = "archetype"
diff --git a/src/digitalkin/models/services/services.py b/src/digitalkin/models/services/services.py
index cdc057a0..316be0de 100644
--- a/src/digitalkin/models/services/services.py
+++ b/src/digitalkin/models/services/services.py
@@ -8,3 +8,19 @@ class ServicesMode(str, Enum):
 
     LOCAL = "local"
     REMOTE = "remote"
+
+
+class Context(Enum):
+    """Owner/scope of a file in the filesystem service.
+
+    Mirrors the filesystem proto context kinds. MISSIONS/SETUP are the read/write
+    owner contexts this strategy operates on; USERS/ORGANIZATIONS are read-only
+    cross-owner scopes whose concrete id is resolved server-side from the request
+    metadata (the client sends only the kind).
+    """
+
+    UNSPECIFIED = "unspecified"
+    MISSIONS = "mission"
+    SETUP = "setup"
+    USERS = "user"
+    ORGANIZATIONS = "organization"
diff --git a/src/digitalkin/models/services/storage.py b/src/digitalkin/models/services/storage.py
index fa447254..da3be7ff 100644
--- a/src/digitalkin/models/services/storage.py
+++ b/src/digitalkin/models/services/storage.py
@@ -51,3 +51,16 @@ class DataType(Enum):
     VIEW = "VIEW"
     LOGS = "LOGS"
     OTHER = "OTHER"
+
+
+class Visibility(Enum):
+    """Read-access scope of a record, mirroring the storage proto kinds by name.
+
+    Ownership (who may edit) stays keyed on the record context; this only governs
+    who may read. UNSPECIFIED lets the storage service apply its server-side default.
+    """
+
+    UNSPECIFIED = "unspecified"
+    PUBLIC = "public"
+    PRIVATE = "private"
+    INTERNAL = "internal"
diff --git a/src/digitalkin/services/filesystem/default_filesystem.py b/src/digitalkin/services/filesystem/default_filesystem.py
index 956d07e2..c8d2d803 100644
--- a/src/digitalkin/services/filesystem/default_filesystem.py
+++ b/src/digitalkin/services/filesystem/default_filesystem.py
@@ -9,6 +9,7 @@
 from anyio import Path as AsyncPath
 
 from digitalkin.logger import logger
+from digitalkin.models.services.services import Context
 from digitalkin.services.filesystem.exceptions import FilesystemServiceError
 from digitalkin.services.filesystem.filesystem_strategy import (
     FileFilter,
@@ -206,7 +207,7 @@ async def get_files(
     async def get_file(
         self,
         file_id: str,
-        context: Literal["mission", "setup"] = "mission",  # noqa: ARG002
+        context: Context = Context.MISSIONS,  # noqa: ARG002
         *,
         include_content: bool = False,
     ) -> FilesystemRecord:
diff --git a/src/digitalkin/services/filesystem/filesystem_strategy.py b/src/digitalkin/services/filesystem/filesystem_strategy.py
index 952b6e4b..ec98fc8a 100644
--- a/src/digitalkin/services/filesystem/filesystem_strategy.py
+++ b/src/digitalkin/services/filesystem/filesystem_strategy.py
@@ -6,6 +6,7 @@
 
 from pydantic import BaseModel, Field
 
+from digitalkin.models.services.services import Context
 from digitalkin.services.base_strategy import BaseStrategy
 
 
@@ -29,8 +30,9 @@ class FilesystemRecord(BaseModel):
 class FileFilter(BaseModel):
     """Filter criteria for querying files."""
 
-    context: Literal["mission", "setup"] = Field(
-        default="mission", description="The context of the files (mission or setup)"
+    context: Context = Field(
+        default=Context.MISSIONS,
+        description="The context of the files: mission/setup (owner) or user/organization (read-only cross-owner)",
     )
     names: list[str] | None = Field(default=None, description="Filter by file names (exact matches)")
     file_ids: list[str] | None = Field(default=None, description="Filter by file IDs")
@@ -129,7 +131,7 @@ async def upload_files(
     async def get_file(
         self,
         file_id: str,
-        context: Literal["mission", "setup"] = "mission",
+        context: Context = Context.MISSIONS,
         *,
         include_content: bool = False,
     ) -> FilesystemRecord:
@@ -141,7 +143,7 @@ async def get_file(
 
         Args:
             file_id: The ID of the file to be retrieved
-            context: The context of the files (mission or setup)
+            context: The context of the file (mission/setup, or user/organization for cross-owner reads)
             include_content: Whether to include file content in response
 
         Returns:
diff --git a/src/digitalkin/services/filesystem/grpc_filesystem.py b/src/digitalkin/services/filesystem/grpc_filesystem.py
index 7fd158b9..e75221e6 100644
--- a/src/digitalkin/services/filesystem/grpc_filesystem.py
+++ b/src/digitalkin/services/filesystem/grpc_filesystem.py
@@ -10,6 +10,7 @@
 from digitalkin.grpc_servers.utils.grpc_error_handler import GrpcErrorHandlerMixin
 from digitalkin.logger import logger
 from digitalkin.models.grpc_servers.models import ClientConfig
+from digitalkin.models.services.services import Context
 from digitalkin.services.filesystem.exceptions import FilesystemServiceError
 from digitalkin.services.filesystem.filesystem_strategy import (
     FileFilter,
@@ -80,26 +81,32 @@ def _file_proto_to_data(file: filesystem_pb2.File) -> FilesystemRecord:
         )
 
     @staticmethod
-    def _context_enum(context: str) -> filesystem_pb2.ContextFile:
-        """Map a scope literal to the wire's context-kind enum.
+    def _context_enum(context: Context) -> filesystem_pb2.ContextFile:
+        """Map a context kind to the wire's context-kind enum.
 
         Since dev4 the request carries only the kind; the concrete id is resolved
-        server-side from the x-mission-id / x-setup-id task metadata stamped by
-        ``RequestIdClientInterceptor``.
+        server-side from the request metadata stamped by ``RequestIdClientInterceptor``.
+        USERS/ORGANIZATIONS are read-only cross-owner scopes — only the kind is sent;
+        the server derives the owning user/organization from the request context (no id
+        is transmitted by the client).
 
         Args:
-            context: The scope literal ("mission" or "setup").
+            context: The context kind.
 
         Returns:
-            The matching ``ContextFile`` enum value, ``CONTEXT_UNSPECIFIED`` otherwise.
+            The matching ``ContextFile`` wire enum, ``CONTEXT_UNSPECIFIED`` otherwise.
         """
         # TODO(validate): remove after prod validation
         # [VALIDATE CTXENUM] server resolves the concrete id from metadata
         match context:
-            case "setup":
+            case Context.SETUP:
                 return filesystem_pb2.CONTEXT_SETUP
-            case "mission":
+            case Context.MISSIONS:
                 return filesystem_pb2.CONTEXT_MISSIONS
+            case Context.USERS:
+                return filesystem_pb2.CONTEXT_USERS
+            case Context.ORGANIZATIONS:
+                return filesystem_pb2.CONTEXT_ORGANIZATIONS
         return filesystem_pb2.CONTEXT_UNSPECIFIED
 
     def _filter_to_proto(self, filters: FileFilter) -> filesystem_pb2.FileFilter:
@@ -188,7 +195,7 @@ async def upload_files(
     async def get_file(
         self,
         file_id: str,
-        context: Literal["mission", "setup"] = "mission",
+        context: Context = Context.MISSIONS,
         *,
         include_content: bool = False,
     ) -> FilesystemRecord:
@@ -196,7 +203,7 @@ async def get_file(
 
         Args:
             file_id: The ID of the file to be retrieved
-            context: The context of the files (mission or setup)
+            context: The context of the file (mission/setup, or user/organization for cross-owner reads)
             include_content: Whether to include file content in response
 
         Returns:
diff --git a/src/digitalkin/services/storage/default_storage.py b/src/digitalkin/services/storage/default_storage.py
index d6fdec09..48fcb727 100644
--- a/src/digitalkin/services/storage/default_storage.py
+++ b/src/digitalkin/services/storage/default_storage.py
@@ -9,7 +9,7 @@
 from pydantic import BaseModel
 
 from digitalkin.logger import logger
-from digitalkin.models.services.storage import DataType
+from digitalkin.models.services.storage import DataType, Visibility
 from digitalkin.services.storage.storage_strategy import (
     StorageRecord,
     StorageStrategy,
@@ -67,6 +67,7 @@ def _load_from_file(self) -> dict[str, StorageRecord]:
                     record_id=rd["record_id"],
                     data=data_model,
                     data_type=DataType[rd["data_type"]],
+                    visibility=Visibility[rd["visibility"]] if rd.get("visibility") else Visibility.UNSPECIFIED,
                     creation_date=datetime.datetime.fromisoformat(rd["creation_date"])
                     if rd.get("creation_date")
                     else None,
@@ -97,6 +98,7 @@ def _save_to_file(self) -> None:
                         "collection": record.collection,
                         "record_id": record.record_id,
                         "data_type": record.data_type.name,
+                        "visibility": record.visibility.name,
                         "data": record.data.model_dump(),
                         "creation_date": record.creation_date.isoformat() if record.creation_date else None,
                         "update_date": record.update_date.isoformat() if record.update_date else None,
@@ -141,21 +143,29 @@ async def _read(self, collection: str, record_id: str, context: str) -> StorageR
         Args:
             collection: The unique name to retrieve data for
             record_id: The unique ID of the record
-            context: Owner context scoping the lookup.
+            context: Resolved owner context scoping the lookup.
 
         Returns:
             StorageRecord: The corresponding record
         """
         return self.storage.get(self._key(context, collection, record_id))
 
-    async def _update(self, collection: str, record_id: str, data: BaseModel, context: str) -> StorageRecord | None:
+    async def _update(
+        self,
+        collection: str,
+        record_id: str,
+        data: BaseModel,
+        context: str,
+        visibility: Visibility = Visibility.UNSPECIFIED,
+    ) -> StorageRecord | None:
         """Update a record in the database scoped to a specific context.
 
         Args:
             collection: The unique name to retrieve data for
             record_id: The unique ID of the record
             data: The data to modify
-            context: Owner context scoping the update.
+            context: Resolved owner context scoping the update.
+            visibility: New read-access scope; UNSPECIFIED leaves it unchanged.
 
         Returns:
             StorageRecord: The modified record
@@ -165,6 +175,8 @@ async def _update(self, collection: str, record_id: str, data: BaseModel, contex
         if not rec:
             return None
         rec.data = data
+        if visibility is not Visibility.UNSPECIFIED:
+            rec.visibility = visibility
         rec.update_date = datetime.datetime.now(datetime.timezone.utc)
         self._save_to_file()
         logger.debug("Modified %s", key)
@@ -176,7 +188,7 @@ async def _remove(self, collection: str, record_id: str, context: str) -> bool:
         Args:
             collection: The unique name to retrieve data for
             record_id: The unique ID of the record
-            context: Owner context scoping the deletion.
+            context: Resolved owner context scoping the deletion.
 
         Returns:
             bool: True if the record was removed, False otherwise
@@ -189,25 +201,32 @@ async def _remove(self, collection: str, record_id: str, context: str) -> bool:
         logger.debug("Removed %s", key)
         return True
 
-    async def _list(self, collection: str, context: str) -> list[StorageRecord]:
+    async def _list(
+        self, collection: str, context: str, visibilities: list[Visibility] | None = None
+    ) -> list[StorageRecord]:
         """List records in a collection scoped to a specific context.
 
         Args:
             collection: The unique name to retrieve data for
-            context: Owner context scoping the listing.
+            context: Resolved owner context scoping the listing.
+            visibilities: Optional read-access scopes to filter by (None = no filter).
 
         Returns:
             A list of storage records
         """
         prefix = f"{context}|{collection}:"
-        return [r for k, r in self.storage.items() if k.startswith(prefix)]
+        records = [r for k, r in self.storage.items() if k.startswith(prefix)]
+        if visibilities:
+            allowed = set(visibilities)
+            records = [r for r in records if r.visibility in allowed]
+        return records
 
     async def _remove_collection(self, collection: str, context: str) -> bool:
         """Wipe a collection scoped to a specific context.
 
         Args:
             collection: The unique name to retrieve data for
-            context: Owner context scoping the wipe.
+            context: Resolved owner context scoping the wipe.
 
         Returns:
             bool: True if the collection was removed, False otherwise
diff --git a/src/digitalkin/services/storage/grpc_storage.py b/src/digitalkin/services/storage/grpc_storage.py
index 89cb30fb..88fa189e 100644
--- a/src/digitalkin/services/storage/grpc_storage.py
+++ b/src/digitalkin/services/storage/grpc_storage.py
@@ -8,7 +8,8 @@
 from digitalkin.grpc_servers.utils.grpc_client_wrapper import GrpcClientWrapper
 from digitalkin.logger import logger
 from digitalkin.models.grpc_servers.models import ClientConfig
-from digitalkin.models.services.storage import DataType
+from digitalkin.models.services.services import Context
+from digitalkin.models.services.storage import DataType, Visibility
 from digitalkin.services.storage.exceptions import StorageServiceError
 from digitalkin.services.storage.storage_strategy import (
     StorageRecord,
@@ -39,24 +40,52 @@ def _is_circuit_open(error: Exception) -> bool:
         return isinstance(error.__cause__, CircuitOpenError)
 
     def _context_enum(self, context: str) -> data_pb2.ContextStorage:
-        """Map a resolved context id-string to the wire's context-kind enum.
+        """Map a resolved context string to the wire's context-kind enum.
 
         Since dev4 the request carries only the kind; the concrete id is resolved
-        server-side from the x-mission-id / x-setup-id task metadata stamped by
-        ``RequestIdClientInterceptor``.
+        server-side from the request metadata stamped by ``RequestIdClientInterceptor``.
+        USERS/ORGANIZATIONS are read-only cross-owner scopes — only the kind is sent;
+        the server derives the owning user/organization from the request context (no id
+        is transmitted by the client).
 
         Args:
-            context: The resolved context id (``self.mission_id`` or ``self.setup_version_id``).
+            context: The resolved context string from ``_resolve_context``.
 
         Returns:
-            ``CONTEXT_SETUP_VERSIONS`` for the setup-version scope, else ``CONTEXT_MISSIONS``.
+            The matching ``CONTEXT_*`` wire enum.
         """
         # TODO(validate): remove after prod validation
         # [VALIDATE CTXENUM] server resolves the concrete id (incl. setup->current version) from metadata
         if context == self.setup_version_id or context.startswith("setup_versions:"):
             return data_pb2.CONTEXT_SETUP_VERSIONS
+        if context.startswith(f"{Context.USERS.value}:"):
+            return data_pb2.CONTEXT_USERS
+        if context.startswith(f"{Context.ORGANIZATIONS.value}:"):
+            return data_pb2.CONTEXT_ORGANIZATIONS
+        if context.startswith(f"{Context.UNSPECIFIED.value}:"):
+            return data_pb2.CONTEXT_UNSPECIFIED
         return data_pb2.CONTEXT_MISSIONS
 
+    @staticmethod
+    def _visibility_enum(visibility: Visibility) -> data_pb2.Visibility:
+        """Map an SDK ``Visibility`` to its storage-proto wire enum.
+
+        Args:
+            visibility: The SDK visibility level.
+
+        Returns:
+            The matching ``VISIBILITY_*`` wire enum (``VISIBILITY_UNSPECIFIED`` by default).
+        """
+        match visibility:
+            case Visibility.PUBLIC:
+                return data_pb2.VISIBILITY_PUBLIC
+            case Visibility.PRIVATE:
+                return data_pb2.VISIBILITY_PRIVATE
+            case Visibility.INTERNAL:
+                return data_pb2.VISIBILITY_INTERNAL
+            case _:
+                return data_pb2.VISIBILITY_UNSPECIFIED
+
     def _build_record_from_proto(self, proto: data_pb2.StorageRecord) -> StorageRecord:
         """Convert a protobuf StorageRecord message into our Pydantic model.
 
@@ -74,6 +103,7 @@ def _build_record_from_proto(self, proto: data_pb2.StorageRecord) -> StorageReco
         coll = proto.collection
         rid = proto.record_id
         dtype = DataType[data_pb2.DataType.Name(proto.data_type)]
+        visibility = Visibility[data_pb2.Visibility.Name(proto.visibility).removeprefix("VISIBILITY_")]
 
         # Selective deserialization: only the nested Struct payload
         payload = ProtoUtils.proto_to_dict(proto.data) if proto.HasField("data") else {}
@@ -89,6 +119,7 @@ def _build_record_from_proto(self, proto: data_pb2.StorageRecord) -> StorageReco
             record_id=rid,
             data=validated,
             data_type=dtype,
+            visibility=visibility,
             creation_date=creation_date,
             update_date=update_date,
         )
@@ -127,16 +158,17 @@ async def _store(self, record: StorageRecord) -> StorageRecord:
             StorageServiceError: If there is an error while storing the record
         """
         logger.debug("debug:_store collection=%s id=%s", record.collection, record.record_id)
+        data_struct = Struct()
+        data_struct.update(record.data.model_dump())
+        req = data_pb2.StoreRecordRequest(
+            data=data_struct,
+            context=self._context_enum(record.context),
+            collection=record.collection,
+            record_id=record.record_id,
+            data_type=record.data_type.name,
+            visibility=self._visibility_enum(record.visibility),
+        )
         try:
-            data_struct = Struct()
-            data_struct.update(record.data.model_dump())
-            req = data_pb2.StoreRecordRequest(
-                data=data_struct,
-                context=self._context_enum(record.context),
-                collection=record.collection,
-                record_id=record.record_id,
-                data_type=record.data_type.name,
-            )
             resp = await self.exec_grpc_query("StoreRecord", req)
             return self._build_record_from_proto(resp.stored_data)
         except PermissionDeniedError:
@@ -179,18 +211,13 @@ async def _read(self, collection: str, record_id: str, context: str) -> StorageR
                 logger.info("gRPC ReadRecord failed for %s:%s: %s", collection, record_id, e)
             return None
 
-        try:
-            return self._build_record_from_proto(resp.stored_data)
-        except Exception:
-            logger.warning("Invalid record data for %s:%s in ReadRecord", collection, record_id, exc_info=True)
-            return None
-
     async def _update(
         self,
         collection: str,
         record_id: str,
         data: BaseModel,
         context: str,
+        visibility: Visibility = Visibility.UNSPECIFIED,
     ) -> StorageRecord | None:
         """Overwrite a document via gRPC scoped to a specific context.
 
@@ -201,15 +228,16 @@ async def _update(
             PermissionDeniedError: If the service rejects the call with PERMISSION_DENIED.
         """
         logger.debug("debug:_update context=%s collection=%s id=%s", context, collection, record_id)
+        struct = Struct()
+        struct.update(data.model_dump())
+        req = data_pb2.UpdateRecordRequest(
+            data=struct,
+            context=self._context_enum(context),
+            collection=collection,
+            record_id=record_id,
+            visibility=self._visibility_enum(visibility),
+        )
         try:
-            struct = Struct()
-            struct.update(data.model_dump())
-            req = data_pb2.UpdateRecordRequest(
-                data=struct,
-                context=self._context_enum(context),
-                collection=collection,
-                record_id=record_id,
-            )
             resp = await self.exec_grpc_query("UpdateRecord", req)
             return self._build_record_from_proto(resp.stored_data)
         except PermissionDeniedError:
@@ -252,7 +280,9 @@ async def _remove(self, collection: str, record_id: str, context: str) -> bool:
             return False
         return True
 
-    async def _list(self, collection: str, context: str) -> list[StorageRecord]:
+    async def _list(
+        self, collection: str, context: str, visibilities: list[Visibility] | None = None
+    ) -> list[StorageRecord]:
         """List all documents in a collection via gRPC scoped to a specific context.
 
         Returns:
@@ -267,6 +297,8 @@ async def _list(self, collection: str, context: str) -> list[StorageRecord]:
                 context=self._context_enum(context),
                 collection=collection,
             )
+            if visibilities:
+                req.visibilities.extend(self._visibility_enum(v) for v in visibilities)
             resp = await self.exec_grpc_query("ListRecords", req)
         except PermissionDeniedError:
             # TODO(validate): remove after prod validation
diff --git a/src/digitalkin/services/storage/storage_strategy.py b/src/digitalkin/services/storage/storage_strategy.py
index 1f86fbe0..152d5448 100644
--- a/src/digitalkin/services/storage/storage_strategy.py
+++ b/src/digitalkin/services/storage/storage_strategy.py
@@ -8,8 +8,8 @@
 
 from pydantic import BaseModel, Field
 
-from digitalkin.logger import logger
-from digitalkin.models.services.storage import DataType
+from digitalkin.models.services.services import Context
+from digitalkin.models.services.storage import DataType, Visibility
 from digitalkin.services.base_strategy import BaseStrategy
 from digitalkin.services.storage.exceptions import StorageServiceError
 
@@ -21,14 +21,15 @@ class StorageRecord(BaseModel):
     collection: str = Field(..., description="Logical collection name")
     record_id: str = Field(..., description="Unique ID of this record in its collection")
     data_type: DataType = Field(default=DataType.OUTPUT, description="Category of the data of this record")
+    visibility: Visibility = Field(
+        default=Visibility.UNSPECIFIED,
+        description="Read-access scope of this record (UNSPECIFIED = storage-service default)",
+    )
     data: BaseModel = Field(..., description="The typed payload of this record")
     creation_date: datetime.datetime | None = Field(default=None, description="When this record was first created")
     update_date: datetime.datetime | None = Field(default=None, description="When this record was last modified")
 
 
-Scope = Literal["mission", "setup"]
-
-
 class StorageStrategy(BaseStrategy, ABC):
     """Define CRUD + list/remove-collection against a collection/record store.
 
@@ -37,14 +38,35 @@ class StorageStrategy(BaseStrategy, ABC):
     (setup-version scope). Both attributes are expected to already contain the
     full prefix (`missions:<id>` / `setup_versions:<id>`).
 
-    Public methods accept `scope: Literal["mission", "setup"]` (default
-    `"mission"`); internally we resolve it to the matching context string and
-    pass that to the abstract `_store/_read/_update/_remove/_list/_remove_collection`.
+    Public methods accept a `context: Context` kind (default `Context.MISSIONS`);
+    internally we resolve it to the matching context string and pass that to the
+    abstract `_store/_read/_update/_remove/_list/_remove_collection`.
+    `Context.USERS`/`Context.ORGANIZATIONS` are read-only cross-owner scopes usable only for listing.
     """
 
-    def _resolve_context(self, scope: Scope) -> str:
-        """Return the context string for the given scope."""
-        return self.mission_id if scope == "mission" else self.setup_version_id
+    def _resolve_context(self, context: Context) -> str:
+        """Resolve a context kind to its storage context string.
+
+        MISSIONS/SETUP map to the owner contexts this strategy was built with.
+        USERS/ORGANIZATIONS (read-only cross-owner) and UNSPECIFIED hold no concrete
+        id here, so they return a kind-only marker (`user:`, `organization:`,
+        `unspecified:`); the storage service resolves the id — or applies its default
+        for UNSPECIFIED — server-side from the request metadata.
+
+        Args:
+            context: The context kind to resolve.
+
+        Returns:
+            The context string: `missions:<id>`, `setup_versions:<id>`, or the
+            kind marker `user:` / `organization:` / `unspecified:`.
+        """
+        match context:
+            case Context.MISSIONS:
+                return self.mission_id
+            case Context.SETUP:
+                return self.setup_version_id
+            case _:
+                return f"{context.value}:"
 
     def _validate_data(self, collection: str, data: dict[str, Any]) -> BaseModel:
         """Validate data against the model schema for the given key.
@@ -77,6 +99,7 @@ def _create_storage_record(
         validated_data: BaseModel,
         data_type: DataType,
         context: str,
+        visibility: Visibility,
     ) -> StorageRecord:
         """Create a storage record stamped with the given context.
 
@@ -86,6 +109,7 @@ def _create_storage_record(
             validated_data: The validated data model
             data_type: The type of data
             context: Owner context to stamp on the record (mission or setup-version).
+            visibility: Read-access scope for the record.
 
         Returns:
             A complete storage record with metadata
@@ -96,6 +120,7 @@ def _create_storage_record(
             record_id=record_id,
             data=validated_data,
             data_type=data_type,
+            visibility=visibility,
         )
 
     @staticmethod
@@ -120,14 +145,21 @@ async def _read(self, collection: str, record_id: str, context: str) -> StorageR
         Args:
             collection: The unique name to retrieve data for
             record_id: The unique ID of the record
-            context: Owner context (e.g. `missions:<mission_id>` or `setup_versions:<setup_version_id>`).
+            context: Resolved owner context (e.g. `missions:<mission_id>` or `setup_versions:<setup_version_id>`).
 
         Returns:
             A storage record with validated data
         """
 
     @abstractmethod
-    async def _update(self, collection: str, record_id: str, data: BaseModel, context: str) -> StorageRecord | None:
+    async def _update(
+        self,
+        collection: str,
+        record_id: str,
+        data: BaseModel,
+        context: str,
+        visibility: Visibility = Visibility.UNSPECIFIED,
+    ) -> StorageRecord | None:
         """Overwrite an existing record's payload scoped to a specific context.
 
         Args:
@@ -135,6 +167,7 @@ async def _update(self, collection: str, record_id: str, data: BaseModel, contex
             record_id: The unique ID of the record
             data: The new data to store
             context: Owner context for the record being updated.
+            visibility: New read-access scope; UNSPECIFIED leaves it unchanged.
 
         Returns:
             StorageRecord: The modified record
@@ -154,12 +187,15 @@ async def _remove(self, collection: str, record_id: str, context: str) -> bool:
         """
 
     @abstractmethod
-    async def _list(self, collection: str, context: str) -> list[StorageRecord]:
+    async def _list(
+        self, collection: str, context: str, visibilities: list[Visibility] | None = None
+    ) -> list[StorageRecord]:
         """List all records in a collection scoped to a specific context.
 
         Args:
             collection: The unique name for the record type
             context: Owner context filter.
+            visibilities: Optional read-access scopes to filter by (None = no filter).
 
         Returns:
             A list of storage records
@@ -201,7 +237,7 @@ def _record_lock(self, context: str, collection: str, record_id: str) -> asyncio
         """Get or create an asyncio.Lock for a specific record under a given context.
 
         Args:
-            context: Owner context the record lives under
+            context: Resolved owner context string the record lives under.
             collection: The collection name
             record_id: The record ID
 
@@ -215,8 +251,9 @@ async def store(
         collection: str,
         record_id: str | None,
         data: dict[str, Any],
-        data_type: Literal["OUTPUT", "VIEW", "LOGS", "OTHER"] = "OUTPUT",
-        scope: Scope = "mission",
+        data_type: DataType = DataType.OUTPUT,
+        context: Context = Context.MISSIONS,
+        visibility: Visibility = Visibility.UNSPECIFIED,
     ) -> StorageRecord:
         """Store a new record in the storage.
 
@@ -225,8 +262,9 @@ async def store(
             record_id: The unique ID for the record (optional)
             data: The data to store
             data_type: The type of data being stored (default: OUTPUT)
-            scope: "mission" (default) writes under the current mission context;
+            context: "mission" (default) writes under the current mission context;
                 "setup" writes under the setup-version context.
+            visibility: Read-access scope for the record (UNSPECIFIED = server default).
 
         Returns:
             The ID of the created record
@@ -234,38 +272,39 @@ async def store(
         Raises:
             ValueError: If the data type is invalid or if validation fails
         """
-        if not self._is_valid_data_type_name(data_type):
+        if not self._is_valid_data_type_name(data_type.value):
             msg = f"Invalid data type '{data_type}'. Must be one of {list(DataType.__members__.keys())}"
             raise ValueError(msg)
         record_id = record_id or uuid4().hex
-        data_type_enum = DataType[data_type]
-        context = self._resolve_context(scope)
         validated_data = self._validate_data(collection, data)
-        record = self._create_storage_record(collection, record_id, validated_data, data_type_enum, context)
-        async with self._record_lock(context, collection, record_id):
+        record = self._create_storage_record(
+            collection, record_id, validated_data, data_type, self._resolve_context(context), visibility
+        )
+        async with self._record_lock(record.context, collection, record_id):
             return await self._store(record)
 
-    async def read(self, collection: str, record_id: str, scope: Scope = "mission") -> StorageRecord | None:
+    async def read(self, collection: str, record_id: str, context: Context = Context.MISSIONS) -> StorageRecord | None:
         """Get a record by key under the given scope.
 
         Args:
             collection: The unique name to retrieve data for
             record_id: The unique ID of the record
-            scope: Which context to read from (default: "mission").
+            context: Which context to read from (default: "mission").
 
         Returns:
             The matching record if it exists, otherwise None.
         """
-        context = self._resolve_context(scope)
-        async with self._record_lock(context, collection, record_id):
-            return await self._read(collection, record_id, context)
+        ctx = self._resolve_context(context)
+        async with self._record_lock(ctx, collection, record_id):
+            return await self._read(collection, record_id, ctx)
 
     async def update(
         self,
         collection: str,
         record_id: str,
         data: dict[str, Any],
-        scope: Scope = "mission",
+        context: Context = Context.MISSIONS,
+        visibility: Visibility = Visibility.UNSPECIFIED,
     ) -> StorageRecord | None:
         """Validate & overwrite an existing record under the given scope.
 
@@ -273,60 +312,68 @@ async def update(
             collection: The unique name for the record type
             record_id: The unique ID of the record
             data: The new data to store
-            scope: Which context the record lives under (default: "mission").
+            context: Which context the record lives under (default: "mission").
+            visibility: New read-access scope; UNSPECIFIED leaves it unchanged.
 
         Returns:
             StorageRecord: The modified record
         """
         validated_data = self._validate_data(collection, data)
-        context = self._resolve_context(scope)
-        async with self._record_lock(context, collection, record_id):
-            return await self._update(collection, record_id, validated_data, context)
+        ctx = self._resolve_context(context)
+        async with self._record_lock(ctx, collection, record_id):
+            return await self._update(collection, record_id, validated_data, ctx, visibility)
 
-    async def remove(self, collection: str, record_id: str, scope: Scope = "mission") -> bool:
+    async def remove(self, collection: str, record_id: str, context: Context = Context.MISSIONS) -> bool:
         """Delete a record from the storage under the given scope.
 
         Args:
             collection: The unique name for the record type
             record_id: The unique ID of the record
-            scope: Which context the record lives under (default: "mission").
+            context: Which context the record lives under (default: "mission").
 
         Returns:
             True if the deletion was successful, False otherwise
         """
-        context = self._resolve_context(scope)
-        async with self._record_lock(context, collection, record_id):
-            result = await self._remove(collection, record_id, context)
+        ctx = self._resolve_context(context)
+        async with self._record_lock(ctx, collection, record_id):
+            result = await self._remove(collection, record_id, ctx)
         if result:
-            self._record_locks.pop(f"{context}|{collection}:{record_id}", None)
+            self._record_locks.pop(f"{ctx}|{collection}:{record_id}", None)
         return result
 
-    async def list(self, collection: str, scope: Scope = "mission") -> list[StorageRecord]:
+    async def list(
+        self,
+        collection: str,
+        context: Context = Context.MISSIONS,
+        visibilities: list[Visibility] | None = None,
+    ) -> list[StorageRecord]:
         """Get all records in a collection under the given scope.
 
         Args:
             collection: The unique name for the record type
-            scope: Which context to list (default: "mission").
+            context: Which context to list (default: "mission"). "user"/"organization"
+                list across an owner and require `owner_id`.
+            visibilities: Optional read-access scopes to filter by (None = no filter).
 
         Returns:
             A list of storage records under the resolved context.
         """
-        return await self._list(collection, self._resolve_context(scope))
+        return await self._list(collection, self._resolve_context(context), visibilities)
 
-    async def remove_collection(self, collection: str, scope: Scope = "mission") -> bool:
+    async def remove_collection(self, collection: str, context: Context = Context.MISSIONS) -> bool:
         """Wipe a collection clean under the given scope.
 
         Args:
             collection: The unique name for the record type
-            scope: Which context the records live under (default: "mission").
+            context: Which context the records live under (default: "mission").
 
         Returns:
             True if the deletion was successful, False otherwise
         """
-        context = self._resolve_context(scope)
-        result = await self._remove_collection(collection, context)
+        ctx = self._resolve_context(context)
+        result = await self._remove_collection(collection, ctx)
         if result:
-            prefix = f"{context}|{collection}:"
+            prefix = f"{ctx}|{collection}:"
             for key in [k for k in self._record_locks if k.startswith(prefix)]:
                 self._record_locks.pop(key, None)
         return result
@@ -336,8 +383,9 @@ async def upsert(
         collection: str,
         record_id: str,
         data: dict[str, Any],
-        data_type: Literal["OUTPUT", "VIEW", "LOGS", "OTHER"] = "OUTPUT",
-        scope: Scope = "mission",
+        data_type: DataType = DataType.OUTPUT,
+        context: Context = Context.MISSIONS,
+        visibility: Visibility = Visibility.UNSPECIFIED,
     ) -> StorageRecord:
         """Insert or update a record atomically under the given scope.
 
@@ -350,7 +398,8 @@ async def upsert(
             record_id: The unique ID for the record
             data: The data to store
             data_type: The type of data being stored (default: OUTPUT)
-            scope: Which context to upsert under (default: "mission").
+            context: Which context to upsert under (default: "mission").
+            visibility: Read-access scope for the record (UNSPECIFIED = server default).
 
         Returns:
             The created or updated storage record
@@ -359,18 +408,17 @@ async def upsert(
             ValueError: If the data type is invalid or if validation fails
             StorageServiceError: If update of an existing record fails unexpectedly
         """
-        if not self._is_valid_data_type_name(data_type):
+        if not self._is_valid_data_type_name(data_type.value):
             msg = f"Invalid data type '{data_type}'. Must be one of {list(DataType.__members__.keys())}"
             raise ValueError(msg)
-        data_type_enum = DataType[data_type]
-        context = self._resolve_context(scope)
         validated_data = self._validate_data(collection, data)
-        async with self._record_lock(context, collection, record_id):
-            if await self._read(collection, record_id, context):
-                updated = await self._update(collection, record_id, validated_data, context)
+        ctx = self._resolve_context(context)
+        async with self._record_lock(ctx, collection, record_id):
+            if await self._read(collection, record_id, ctx):
+                updated = await self._update(collection, record_id, validated_data, ctx, visibility)
                 if updated is None:
                     msg = f"Update failed for existing record '{collection}:{record_id}'"
                     raise StorageServiceError(msg)
                 return updated
-            record = self._create_storage_record(collection, record_id, validated_data, data_type_enum, context)
+            record = self._create_storage_record(collection, record_id, validated_data, data_type, ctx, visibility)
             return await self._store(record)
diff --git a/tests/services/filesystem/test_grpc_filesystem.py b/tests/services/filesystem/test_grpc_filesystem.py
index 5878ed73..d779b7a9 100644
--- a/tests/services/filesystem/test_grpc_filesystem.py
+++ b/tests/services/filesystem/test_grpc_filesystem.py
@@ -5,6 +5,8 @@
 import secrets
 import string
 import types
+from typing import TYPE_CHECKING
+from unittest.mock import AsyncMock
 
 import grpc
 import grpc_testing
@@ -16,8 +18,14 @@
 )
 from google.protobuf import struct_pb2
 from grpc.framework.foundation import logging_pool
+from hypothesis import given
+from hypothesis import strategies as st
+from mock_filesystem_servicer import MockFilesystemServicer
+from tests.fixtures.grpc_fixtures import FakeContext
 
+from digitalkin.grpc_servers.exceptions import PermissionDeniedError
 from digitalkin.models.grpc_servers.models import ClientConfig
+from digitalkin.models.services.services import Context
 from digitalkin.models.settings.utils.channel import ControlFlow, SecurityMode
 from digitalkin.services.filesystem.exceptions import FilesystemServiceError
 from digitalkin.services.filesystem.filesystem_strategy import (
@@ -26,8 +34,9 @@
     UploadFileData,
 )
 from digitalkin.services.filesystem.grpc_filesystem import GrpcFilesystem
-from mock_filesystem_servicer import MockFilesystemServicer
-from tests.fixtures.grpc_fixtures import FakeContext
+
+if TYPE_CHECKING:
+    from digitalkin.models.services import services
 
 service_instance = MockFilesystemServicer()
 service_name = filesystem_service_pb2.DESCRIPTOR.services_by_name["FilesystemService"]
@@ -91,7 +100,7 @@ def client(test_channel: grpc_testing.Channel) -> GrpcFilesystem:
     # Override the channel and stub to use our test channel
     client.stub = filesystem_service_pb2_grpc.FilesystemServiceStub(test_channel)
 
-    async def _test_exec_grpc_query(self, query_endpoint, request):
+    async def _test_exec_grpc_query(self, query_endpoint, request) -> object:
         response = getattr(self.stub, query_endpoint)(request)
         return await response if asyncio.iscoroutine(response) else response
 
@@ -1066,3 +1075,117 @@ def test_file_status_handling(
 #     """
 #
 # Add regression tests below as bugs are discovered and fixed.
+
+
+class TestContextScopes:
+    """Tests that the ContextFile kind maps to the right wire enum."""
+
+    @pytest.mark.parametrize(
+        ("context", "wire"),
+        [
+            (Context.MISSIONS, filesystem_pb2.CONTEXT_MISSIONS),
+            (Context.SETUP, filesystem_pb2.CONTEXT_SETUP),
+            (Context.USERS, filesystem_pb2.CONTEXT_USERS),
+            (Context.ORGANIZATIONS, filesystem_pb2.CONTEXT_ORGANIZATIONS),
+            (Context.UNSPECIFIED, filesystem_pb2.CONTEXT_UNSPECIFIED),
+        ],
+    )
+    def test_get_files_forwards_context_kind(
+        self,
+        context: Context,
+        wire: "services.Context",
+        client: GrpcFilesystem,
+        test_channel: grpc_testing.Channel,
+        mock_servicer: MockFilesystemServicer,
+    ) -> None:
+        """get_files emits the matching context kind on the request and its filter.
+
+        Covers the cross-owner scopes USERS / ORGANIZATIONS added alongside the
+        existing MISSIONS / SETUP; the concrete owner id is resolved server-side.
+        """
+        future = client_execution_thread_pool.submit(
+            asyncio.run,
+            client.get_files(FileFilter(context=context)),
+        )
+
+        method_desc = service_name.methods_by_name["GetFiles"]
+        _, request, rpc = test_channel.take_unary_unary(method_desc)
+
+        assert request.context == wire
+        assert request.filters.context == wire
+
+        rpc.send_initial_metadata(())
+        rpc.terminate(mock_servicer.GetFiles(request, FakeContext()), (), grpc.StatusCode.OK, "")
+
+        files, _total = future.result(timeout=5.0)
+        assert isinstance(files, list)
+
+    def test_get_file_forwards_cross_owner_context(
+        self,
+        client: GrpcFilesystem,
+        test_channel: grpc_testing.Channel,
+        mock_servicer: MockFilesystemServicer,
+    ) -> None:
+        """get_file under USERS emits CONTEXT_USERS on the wire."""
+        future = client_execution_thread_pool.submit(
+            asyncio.run,
+            client.get_file("file_x", context=Context.USERS),
+        )
+
+        method_desc = service_name.methods_by_name["GetFile"]
+        _, request, rpc = test_channel.take_unary_unary(method_desc)
+
+        assert request.context == filesystem_pb2.CONTEXT_USERS
+
+        rpc.send_initial_metadata(())
+        rpc.terminate(mock_servicer.GetFile(request, FakeContext()), (), grpc.StatusCode.OK, "")
+        assert isinstance(future.result(timeout=5.0), FilesystemRecord)
+
+
+class TestContextEnumContract:
+    """SDK ``Context`` kind -> filesystem wire enum (``_context_enum``)."""
+
+    _WIRE = (
+        (Context.MISSIONS, filesystem_pb2.CONTEXT_MISSIONS),
+        (Context.SETUP, filesystem_pb2.CONTEXT_SETUP),
+        (Context.USERS, filesystem_pb2.CONTEXT_USERS),
+        (Context.ORGANIZATIONS, filesystem_pb2.CONTEXT_ORGANIZATIONS),
+        (Context.UNSPECIFIED, filesystem_pb2.CONTEXT_UNSPECIFIED),
+    )
+
+    @pytest.mark.unit
+    @pytest.mark.contract
+    @pytest.mark.parametrize(("ctx", "wire"), _WIRE)
+    def test_context_enum_maps_to_wire(self, ctx: "Context", wire: int) -> None:
+        """Each Context kind maps to its filesystem proto ``CONTEXT_*`` constant."""
+        assert GrpcFilesystem._context_enum(ctx) == wire
+
+    @pytest.mark.property
+    @given(ctx=st.sampled_from(list(Context)))
+    def test_context_enum_is_total(self, ctx: "Context") -> None:
+        """Every Context kind maps to a defined filesystem wire enum (never crashes)."""
+        assert GrpcFilesystem._context_enum(ctx) in {wire for _, wire in self._WIRE}
+
+
+class TestFilesystemRefusalAndFailures:
+    """Refusals (PERMISSION_DENIED) propagate; other gRPC failures wrap as FilesystemServiceError."""
+
+    @pytest.mark.grpc
+    @pytest.mark.regression
+    async def test_permission_denied_propagates(self, client: GrpcFilesystem) -> None:
+        """Authz refusals are re-raised as-is, never masked as a service error."""
+        client.exec_grpc_query = AsyncMock(side_effect=PermissionDeniedError("denied"))  # type: ignore[method-assign]
+        with pytest.raises(PermissionDeniedError):
+            await client.get_file("f")
+        with pytest.raises(PermissionDeniedError):
+            await client.get_files(FileFilter())
+
+    @pytest.mark.grpc
+    @pytest.mark.chaos
+    async def test_grpc_failure_wrapped(self, client: GrpcFilesystem) -> None:
+        """A generic gRPC failure surfaces as FilesystemServiceError on reads."""
+        client.exec_grpc_query = AsyncMock(side_effect=RuntimeError("boom"))  # type: ignore[method-assign]
+        with pytest.raises(FilesystemServiceError):
+            await client.get_file("f")
+        with pytest.raises(FilesystemServiceError):
+            await client.get_files(FileFilter())
diff --git a/tests/services/storage/test_grpc_storage.py b/tests/services/storage/test_grpc_storage.py
index 433a2091..e95da6f6 100644
--- a/tests/services/storage/test_grpc_storage.py
+++ b/tests/services/storage/test_grpc_storage.py
@@ -17,6 +17,9 @@
 import grpc_testing
 import pytest
 from agentic_mesh_protocol.storage.v1 import data_pb2, storage_service_pb2, storage_service_pb2_grpc
+from google.protobuf.struct_pb2 import Struct
+from hypothesis import given
+from hypothesis import strategies as st
 from pydantic import BaseModel, Field
 from tests.fixtures.grpc_fixtures import AsyncStubWrapper, FakeContext
 from tests.services.storage.mock_storage_servicer import MockStorageServicer
@@ -25,7 +28,8 @@
 from digitalkin.grpc_servers.utils.circuit_breaker import CircuitBreaker
 from digitalkin.models.grpc_servers.circuit_breaker import CBState
 from digitalkin.models.grpc_servers.models import ClientConfig
-from digitalkin.models.services.storage import DataType
+from digitalkin.models.services.services import Context
+from digitalkin.models.services.storage import DataType, Visibility
 from digitalkin.models.settings.grpc_client import get_circuit_breaker_settings, get_grpc_client_settings
 from digitalkin.services.storage.exceptions import StorageServiceError
 from digitalkin.services.storage.grpc_storage import GrpcStorage
@@ -71,7 +75,7 @@ class LogDataModel(BaseModel):
 def thread_pool():
     """Create thread pool and ensure cleanup.
 
-    Returns:
+    Yields:
         ThreadPoolExecutor instance
     """
     pool = futures.ThreadPoolExecutor(max_workers=1)
@@ -316,7 +320,7 @@ def test_store_record_with_output_type(
 
         method_desc = storage_service_pb2.DESCRIPTOR.services_by_name["StorageService"].methods_by_name["StoreRecord"]
 
-        future = thread_pool.submit(asyncio.run, client.store(collection, record_id, data, data_type="OUTPUT"))
+        future = thread_pool.submit(asyncio.run, client.store(collection, record_id, data, data_type=DataType.OUTPUT))
 
         _, request, rpc = test_channel.take_unary_unary(method_desc)
 
@@ -357,7 +361,7 @@ def test_store_record_with_logs_type(
 
         method_desc = storage_service_pb2.DESCRIPTOR.services_by_name["StorageService"].methods_by_name["StoreRecord"]
 
-        future = thread_pool.submit(asyncio.run, client.store(collection, record_id, data, data_type="LOGS"))
+        future = thread_pool.submit(asyncio.run, client.store(collection, record_id, data, data_type=DataType.LOGS))
 
         _, request, rpc = test_channel.take_unary_unary(method_desc)
 
@@ -393,7 +397,7 @@ def test_store_record_with_view_type(
 
         method_desc = storage_service_pb2.DESCRIPTOR.services_by_name["StorageService"].methods_by_name["StoreRecord"]
 
-        future = thread_pool.submit(asyncio.run, client.store(collection, record_id, data, data_type="VIEW"))
+        future = thread_pool.submit(asyncio.run, client.store(collection, record_id, data, data_type=DataType.VIEW))
 
         _, request, rpc = test_channel.take_unary_unary(method_desc)
 
@@ -429,7 +433,7 @@ def test_store_record_with_other_type(
 
         method_desc = storage_service_pb2.DESCRIPTOR.services_by_name["StorageService"].methods_by_name["StoreRecord"]
 
-        future = thread_pool.submit(asyncio.run, client.store(collection, record_id, data, data_type="OTHER"))
+        future = thread_pool.submit(asyncio.run, client.store(collection, record_id, data, data_type=DataType.OTHER))
 
         _, request, rpc = test_channel.take_unary_unary(method_desc)
 
@@ -1177,6 +1181,48 @@ def test_list_records_success(
         values = sorted([r.data.value for r in results])
         assert values == [100, 200, 300]
 
+    @pytest.mark.grpc
+    @pytest.mark.integration
+    @pytest.mark.smoke
+    def test_list_cross_owner_context_and_visibilities(
+        self,
+        client: GrpcStorage,
+        test_channel: grpc_testing.Channel,
+        mock_servicer: MockStorageServicer,
+        thread_pool: futures.ThreadPoolExecutor,
+    ) -> None:
+        """List under USERS/ORGANIZATIONS maps to the cross-owner wire enum.
+
+        Verifies:
+        - context=USERS -> CONTEXT_USERS, context=ORGANIZATIONS -> CONTEXT_ORGANIZATIONS
+        - the visibilities filter is forwarded on the wire
+        """
+        collection = "test_collection"
+        list_method_desc = storage_service_pb2.DESCRIPTOR.services_by_name["StorageService"].methods_by_name[
+            "ListRecords"
+        ]
+
+        for scope_context, wire in (
+            (Context.USERS, data_pb2.CONTEXT_USERS),
+            (Context.ORGANIZATIONS, data_pb2.CONTEXT_ORGANIZATIONS),
+            (Context.UNSPECIFIED, data_pb2.CONTEXT_UNSPECIFIED),
+        ):
+            list_future = thread_pool.submit(
+                asyncio.run,
+                client.list(collection, context=scope_context, visibilities=[Visibility.PUBLIC, Visibility.INTERNAL]),
+            )
+            _, list_request, list_rpc = test_channel.take_unary_unary(list_method_desc)
+
+            assert list_request.context == wire
+            assert list_request.collection == collection
+            assert list(list_request.visibilities) == [data_pb2.VISIBILITY_PUBLIC, data_pb2.VISIBILITY_INTERNAL]
+
+            list_context = FakeContext()
+            list_response = mock_servicer.ListRecords(list_request, list_context)
+            list_rpc.send_initial_metadata(())
+            list_rpc.terminate(list_response, (), grpc.StatusCode.OK, "")
+            assert isinstance(list_future.result(timeout=1.0), list)
+
     @pytest.mark.grpc
     @pytest.mark.integration
     @pytest.mark.edge_case
@@ -1520,7 +1566,10 @@ def _open_storage_breaker(monkeypatch: pytest.MonkeyPatch) -> None:
     @pytest.mark.grpc
     @pytest.mark.unit
     async def test_store_logs_quietly_when_circuit_open(
-        self, client: GrpcStorage, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture,
+        self,
+        client: GrpcStorage,
+        monkeypatch: pytest.MonkeyPatch,
+        caplog: pytest.LogCaptureFixture,
     ) -> None:
         """Open-circuit StoreRecord raises but logs at DEBUG (no stack trace)."""
         self._open_storage_breaker(monkeypatch)
@@ -1543,7 +1592,10 @@ async def test_store_logs_quietly_when_circuit_open(
     @pytest.mark.grpc
     @pytest.mark.unit
     async def test_read_logs_quietly_when_circuit_open(
-        self, client: GrpcStorage, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture,
+        self,
+        client: GrpcStorage,
+        monkeypatch: pytest.MonkeyPatch,
+        caplog: pytest.LogCaptureFixture,
     ) -> None:
         """Open-circuit ReadRecord returns None and logs at DEBUG only."""
         self._open_storage_breaker(monkeypatch)
@@ -1623,3 +1675,183 @@ def test_unavailable_opens_breaker(
 
         assert result is None
         assert CircuitBreaker.get_or_create("StorageService").state == CBState.OPEN
+
+
+# ============================================================================
+# Enum coverage: Visibility & Context (new SDK enums used by the storage service)
+# ============================================================================
+
+
+class TestVisibilityEnumMapping:
+    """SDK ``Visibility`` <-> storage-proto wire enum, both directions."""
+
+    _WIRE = (
+        (Visibility.UNSPECIFIED, data_pb2.VISIBILITY_UNSPECIFIED),
+        (Visibility.PUBLIC, data_pb2.VISIBILITY_PUBLIC),
+        (Visibility.PRIVATE, data_pb2.VISIBILITY_PRIVATE),
+        (Visibility.INTERNAL, data_pb2.VISIBILITY_INTERNAL),
+    )
+
+    @pytest.mark.unit
+    @pytest.mark.parametrize(("vis", "wire"), _WIRE)
+    def test_visibility_enum_maps_to_wire(self, vis: Visibility, wire: int) -> None:
+        """Each SDK visibility maps to its proto ``VISIBILITY_*`` constant."""
+        assert GrpcStorage._visibility_enum(vis) == wire
+
+    @pytest.mark.contract
+    def test_sdk_visibility_names_mirror_proto(self) -> None:
+        """Every SDK Visibility has a matching ``VISIBILITY_<NAME>`` in the proto."""
+        proto_names = {v.name for v in data_pb2.Visibility.DESCRIPTOR.values}
+        assert {f"VISIBILITY_{v.name}" for v in Visibility} <= proto_names
+
+    @pytest.mark.contract
+    def test_visibility_values_are_lowercase_string_names(self) -> None:
+        """Visibility values are intentionally strings mirroring the member name."""
+        for v in Visibility:
+            assert isinstance(v.value, str)
+            assert v.value == v.name.lower()
+
+    @pytest.mark.regression
+    def test_visibility_enum_avoids_uncallable_proto_wrapper(self) -> None:
+        """``data_pb2.Visibility(...)`` is not callable at runtime; the mapper must not rely on it."""
+        with pytest.raises(TypeError):
+            data_pb2.Visibility(1)
+        assert GrpcStorage._visibility_enum(Visibility.PUBLIC) == data_pb2.VISIBILITY_PUBLIC
+
+    @pytest.mark.property
+    @given(vis=st.sampled_from(list(Visibility)))
+    def test_visibility_round_trips_through_wire(self, vis: Visibility) -> None:
+        """Write mapping -> proto -> read mapping recovers the same member."""
+        wire = GrpcStorage._visibility_enum(vis)
+        name = data_pb2.Visibility.Name(wire).removeprefix("VISIBILITY_")
+        assert Visibility[name] is vis
+
+    @pytest.mark.validation
+    def test_unknown_visibility_name_is_rejected(self) -> None:
+        """Name-based lookup (used by the tools) rejects unknown levels."""
+        with pytest.raises(KeyError):
+            _ = Visibility["BOGUS"]
+
+
+class TestVisibilityWire:
+    """Visibility on the wire: sent on store, reconstructed on read."""
+
+    @pytest.mark.grpc
+    @pytest.mark.smoke
+    @pytest.mark.parametrize(
+        "vis", [Visibility.PUBLIC, Visibility.PRIVATE, Visibility.INTERNAL, Visibility.UNSPECIFIED]
+    )
+    def test_store_puts_visibility_on_request(
+        self,
+        vis: Visibility,
+        client: GrpcStorage,
+        test_channel: grpc_testing.Channel,
+        mock_servicer: MockStorageServicer,
+        thread_pool: futures.ThreadPoolExecutor,
+    ) -> None:
+        """A store carries the chosen visibility as the proto wire enum."""
+        data = {"mission_id": MISSION_ID, "name": "vis", "value": 1}
+        method_desc = storage_service_pb2.DESCRIPTOR.services_by_name["StorageService"].methods_by_name["StoreRecord"]
+
+        future = thread_pool.submit(asyncio.run, client.store("test_collection", "vis_rec", data, visibility=vis))
+        _, request, rpc = test_channel.take_unary_unary(method_desc)
+
+        assert request.visibility == GrpcStorage._visibility_enum(vis)
+
+        rpc.send_initial_metadata(())
+        rpc.terminate(mock_servicer.StoreRecord(request, FakeContext()), (), grpc.StatusCode.OK, "")
+        assert future.result(timeout=1.0) is not None
+
+    @pytest.mark.unit
+    @pytest.mark.parametrize("vis", list(Visibility))
+    def test_build_record_reads_visibility_from_wire(self, client: GrpcStorage, vis: Visibility) -> None:
+        """Reading a record reconstructs the SDK visibility from the proto int (string-valued enum)."""
+        struct = Struct()
+        struct.update({"mission_id": MISSION_ID, "name": "vis", "value": 1})
+        proto = data_pb2.StorageRecord(
+            context=MISSION_ID,
+            collection="test_collection",
+            record_id="r",
+            data=struct,
+            data_type=data_pb2.OUTPUT,
+            visibility=GrpcStorage._visibility_enum(vis),
+        )
+        assert client._build_record_from_proto(proto).visibility is vis
+
+    @pytest.mark.edge_case
+    def test_unknown_wire_visibility_is_skipped(self, client: GrpcStorage) -> None:
+        """An out-of-range wire visibility makes the record skipped rather than crash a whole list."""
+        struct = Struct()
+        struct.update({"mission_id": MISSION_ID, "name": "vis", "value": 1})
+        proto = data_pb2.StorageRecord(
+            context=MISSION_ID,
+            collection="test_collection",
+            record_id="r",
+            data=struct,
+            data_type=data_pb2.OUTPUT,
+            visibility=99,
+        )
+        assert client._build_record_or_skip(proto) is None
+
+
+class TestContextWireMapping:
+    """SDK ``Context`` kind -> storage wire enum (via ``_resolve_context`` + ``_context_enum``)."""
+
+    @pytest.mark.contract
+    @pytest.mark.parametrize(
+        ("ctx", "wire"),
+        [
+            (Context.MISSIONS, data_pb2.CONTEXT_MISSIONS),
+            (Context.SETUP, data_pb2.CONTEXT_SETUP_VERSIONS),
+            (Context.USERS, data_pb2.CONTEXT_USERS),
+            (Context.ORGANIZATIONS, data_pb2.CONTEXT_ORGANIZATIONS),
+            (Context.UNSPECIFIED, data_pb2.CONTEXT_UNSPECIFIED),
+        ],
+    )
+    def test_context_resolves_to_wire(self, client: GrpcStorage, ctx: Context, wire: int) -> None:
+        """Each Context kind resolves + maps to the expected wire enum."""
+        assert client._context_enum(client._resolve_context(ctx)) == wire
+
+    @pytest.mark.regression
+    def test_cross_owner_markers_are_singular(self, client: GrpcStorage) -> None:
+        """Unified Context values are singular; kind-only markers must match them."""
+        assert client._resolve_context(Context.USERS) == "user:"
+        assert client._resolve_context(Context.ORGANIZATIONS) == "organization:"
+        assert client._resolve_context(Context.UNSPECIFIED) == "unspecified:"
+
+
+class TestStorageRefusalAndFailures:
+    """Refusals (PERMISSION_DENIED) propagate; other gRPC failures degrade gracefully."""
+
+    @pytest.mark.grpc
+    @pytest.mark.regression
+    async def test_permission_denied_propagates_on_all_ops(self, client: GrpcStorage) -> None:
+        """Authz refusals are never swallowed — every op re-raises PermissionDeniedError."""
+        client.exec_grpc_query = AsyncMock(side_effect=PermissionDeniedError("denied"))  # type: ignore[method-assign]
+        data = {"mission_id": MISSION_ID, "name": "x", "value": 1}
+        with pytest.raises(PermissionDeniedError):
+            await client.store("test_collection", "r", data)
+        with pytest.raises(PermissionDeniedError):
+            await client.read("test_collection", "r")
+        with pytest.raises(PermissionDeniedError):
+            await client.update("test_collection", "r", data)
+        with pytest.raises(PermissionDeniedError):
+            await client.remove("test_collection", "r")
+        with pytest.raises(PermissionDeniedError):
+            await client.list("test_collection")
+        with pytest.raises(PermissionDeniedError):
+            await client.remove_collection("test_collection")
+
+    @pytest.mark.grpc
+    @pytest.mark.chaos
+    async def test_grpc_failure_degrades_gracefully(self, client: GrpcStorage) -> None:
+        """A generic gRPC failure raises on writes-that-must-confirm and returns empty/false on best-effort reads."""
+        client.exec_grpc_query = AsyncMock(side_effect=RuntimeError("boom"))  # type: ignore[method-assign]
+        data = {"mission_id": MISSION_ID, "name": "x", "value": 1}
+        with pytest.raises(StorageServiceError):
+            await client.store("test_collection", "r", data)
+        assert await client.read("test_collection", "r") is None
+        assert await client.update("test_collection", "r", data) is None
+        assert await client.remove("test_collection", "r") is False
+        assert await client.list("test_collection") == []
+        assert await client.remove_collection("test_collection") is False
diff --git a/uv.lock b/uv.lock
index 343e2fab..a4cb36e5 100644
--- a/uv.lock
+++ b/uv.lock
@@ -715,7 +715,7 @@ wheels = [
 
 [[package]]
 name = "digitalkin"
-version = "1.0.2.dev2"
+version = "1.0.2.dev7"
 source = { editable = "." }
 dependencies = [
     { name = "ag-ui-protocol" },
@@ -907,7 +907,7 @@ name = "exceptiongroup"
 version = "1.3.1"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
-    { name = "typing-extensions", marker = "python_full_version < '3.12'" },
+    { name = "typing-extensions" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/50/79/66800aadf48771f6b62f7eb014e352e5d06856655206165d775e675a02c9/exceptiongroup-1.3.1.tar.gz", hash = "sha256:8b412432c6055b0b7d14c310000ae93352ed6754f70fa8f7c34141f91c4e3219", size = 30371, upload-time = "2025-11-21T23:01:54.787Z" }
 wheels = [
@@ -1549,7 +1549,7 @@ name = "importlib-metadata"
 version = "9.0.0"
 source = { registry = "https://pypi.org/simple" }
 dependencies = [
-    { name = "zipp", marker = "python_full_version < '3.12'" },
+    { name = "zipp" },
 ]
 sdist = { url = "https://files.pythonhosted.org/packages/a9/01/15bb152d77b21318514a96f43af312635eb2500c96b55398d020c93d86ea/importlib_metadata-9.0.0.tar.gz", hash = "sha256:a4f57ab599e6a2e3016d7595cfd72eb4661a5106e787a95bcc90c7105b831efc", size = 56405, upload-time = "2026-03-20T06:42:56.999Z" }
 wheels = [