diff --git a/.github/workflows/gcp-observatory-read-adapter.yml b/.github/workflows/gcp-observatory-read-adapter.yml
new file mode 100644
index 0000000..e4c088f
--- /dev/null
+++ b/.github/workflows/gcp-observatory-read-adapter.yml
@@ -0,0 +1,199 @@
+name: gcp-observatory-read-adapter
+
+on:
+  workflow_dispatch:
+    inputs:
+      action:
+        description: Build only, or deploy the protected staging service and run live receipts
+        required: true
+        default: build_only
+        type: choice
+        options:
+          - build_only
+          - deploy_staging
+
+permissions:
+  contents: read
+  id-token: write
+
+concurrency:
+  group: gcp-observatory-read-adapter-${{ github.ref }}
+  cancel-in-progress: false
+
+env:
+  PROJECT_ID: teleo-501523
+  REGION: europe-west6
+  ZONE: europe-west6-a
+  REPOSITORY: teleo
+  IMAGE_NAME: observatory-read-adapter
+  SERVICE_NAME: observatory-read-adapter-staging
+  RUNTIME_SERVICE_ACCOUNT: sa-observatory-read-adapter@teleo-501523.iam.gserviceaccount.com
+  DB_IAM_USER: sa-observatory-read-adapter@teleo-501523.iam
+  API_KEY_SECRET: observatory-read-api-key-staging
+  WORKLOAD_IDENTITY_PROVIDER: projects/785938879453/locations/global/workloadIdentityPools/github-actions/providers/living-ip-github
+  DEPLOY_SERVICE_ACCOUNT: sa-artifact-builder@teleo-501523.iam.gserviceaccount.com
+
+jobs:
+  build:
+    name: Test, build, and push adapter image
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    outputs:
+      image_ref: ${{ steps.image.outputs.image_ref }}
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.11"
+          cache: pip
+
+      - name: Run focused adapter tests
+        run: |
+          python -m pip install -e '.[dev]'
+          python -m pytest tests/test_observatory_read_adapter.py -q
+          python -m ruff check observatory_read_adapter tests/test_observatory_read_adapter.py
+
+      - id: auth
+        uses: google-github-actions/auth@v3
+        with:
+          workload_identity_provider: ${{ env.WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ env.DEPLOY_SERVICE_ACCOUNT }}
+
+      - uses: google-github-actions/setup-gcloud@v3
+        with:
+          project_id: ${{ env.PROJECT_ID }}
+
+      - name: Configure Artifact Registry
+        run: gcloud auth configure-docker "${REGION}-docker.pkg.dev" --quiet
+
+      - id: image
+        name: Build, smoke, and push immutable image
+        shell: bash
+        run: |
+          set -euo pipefail
+          tag="${GITHUB_SHA::12}"
+          image="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPOSITORY}/${IMAGE_NAME}:${tag}"
+          docker build -f Dockerfile.observatory-read-adapter -t "${image}" .
+          docker run --rm --env PYTHONPYCACHEPREFIX=/tmp/pycache \
+            "${image}" python -m compileall -q /app/observatory_read_adapter
+          docker run --rm "${image}" python -c 'import observatory_read_adapter; print("adapter-import-ok")'
+          docker push "${image}" | tee docker-push.log
+          digest="$(awk '/digest: sha256:/ {print $3}' docker-push.log | tail -1)"
+          test -n "${digest}"
+          image_ref="${image}@${digest}"
+          printf 'image_ref=%s\n' "${image_ref}" >> "${GITHUB_OUTPUT}"
+          printf 'revision=%s\nimage_ref=%s\n' "${GITHUB_SHA}" "${image_ref}" > observatory-adapter-image.txt
+
+      - uses: actions/upload-artifact@v4
+        with:
+          name: observatory-adapter-image
+          path: observatory-adapter-image.txt
+          if-no-files-found: error
+
+  deploy-staging:
+    name: Deploy and prove staging read path
+    if: ${{ inputs.action == 'deploy_staging' }}
+    needs: build
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    steps:
+      - uses: actions/checkout@v4
+
+      - id: auth
+        uses: google-github-actions/auth@v3
+        with:
+          workload_identity_provider: ${{ env.WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ env.DEPLOY_SERVICE_ACCOUNT }}
+
+      - uses: google-github-actions/setup-gcloud@v3
+        with:
+          project_id: ${{ env.PROJECT_ID }}
+
+      - name: Deploy bounded GCP staging service
+        shell: bash
+        env:
+          IMAGE_REF: ${{ needs.build.outputs.image_ref }}
+        run: |
+          set -euo pipefail
+          gcloud run deploy "${SERVICE_NAME}" \
+            --project="${PROJECT_ID}" \
+            --region="${REGION}" \
+            --image="${IMAGE_REF}" \
+            --execution-environment=gen2 \
+            --service-account="${RUNTIME_SERVICE_ACCOUNT}" \
+            --network=teleo-staging-net \
+            --subnet=teleo-staging-europe-west6 \
+            --vpc-egress=private-ranges-only \
+            --ingress=all \
+            --allow-unauthenticated \
+            --port=8080 \
+            --cpu=1 \
+            --memory=512Mi \
+            --concurrency=8 \
+            --max-instances=2 \
+            --timeout=15 \
+            --set-secrets="OBSERVATORY_API_KEY=${API_KEY_SECRET}:latest" \
+            --set-env-vars="GCP_PROJECT_ID=${PROJECT_ID},CLOUD_SQL_INSTANCE=${PROJECT_ID}:${REGION}:teleo-pgvector-standby,DB_NAME=teleo_canonical,DB_IAM_USER=${DB_IAM_USER},DB_AUTHORIZATION_ROLE=kb_observatory_read,SERVICE_REVISION=${GITHUB_SHA}" \
+            --labels="livingip-tier=gcp-staging,livingip-surface=observatory-read-adapter" \
+            --quiet
+
+      - name: Capture positive and negative live receipts
+        shell: bash
+        run: |
+          set -euo pipefail
+          service_url="$(gcloud run services describe "${SERVICE_NAME}" --project="${PROJECT_ID}" --region="${REGION}" --format='value(status.url)')"
+          revision="$(gcloud run services describe "${SERVICE_NAME}" --project="${PROJECT_ID}" --region="${REGION}" --format='value(status.latestReadyRevisionName)')"
+          api_key="$(gcloud secrets versions access latest --secret="${API_KEY_SECRET}" --project="${PROJECT_ID}")"
+          test -n "${service_url}"
+          test -n "${revision}"
+          test -n "${api_key}"
+
+          curl --fail --silent --show-error \
+            --header "X-Api-Key: ${api_key}" \
+            "${service_url}/v1/claims/sample" > positive.json
+          jq -e '
+            .schema == "livingip.observatory-canonical-claim.v1"
+            and .read_only == true
+            and .provenance.database == "teleo_canonical"
+            and .provenance.authorization_role == "kb_observatory_read"
+            and .provenance.transaction_read_only == true
+            and .provenance.write_privileges_denied == true
+            and (.canonical.evidence | length) > 0
+            and .proposal_ledger.distinct_from_canonical == true
+          ' positive.json >/dev/null
+
+          anonymous_status="$(curl --silent --output anonymous.json --write-out '%{http_code}' "${service_url}/v1/claims/sample")"
+          write_status="$(curl --silent --output write.json --write-out '%{http_code}' \
+            --request POST --header "X-Api-Key: ${api_key}" --header 'Content-Type: application/json' \
+            --data '{"status":"applied"}' "${service_url}/v1/claims/sample")"
+          test "${anonymous_status}" = 401
+          test "${write_status}" = 405
+
+          jq -n \
+            --arg service_url "${service_url}" \
+            --arg revision "${revision}" \
+            --arg git_sha "${GITHUB_SHA}" \
+            --argjson positive "$(cat positive.json)" \
+            --arg anonymous_status "${anonymous_status}" \
+            --arg write_status "${write_status}" \
+            '{
+              schema: "livingip.observatory-read-adapter-live-receipt.v1",
+              required_tier: "T3_live_readonly",
+              service_url: $service_url,
+              revision: $revision,
+              git_sha: $git_sha,
+              positive: $positive,
+              negative: {
+                anonymous_http_status: ($anonymous_status | tonumber),
+                authenticated_post_http_status: ($write_status | tonumber)
+              },
+              production_repoint_executed: false
+            }' > observatory-read-adapter-live-receipt.json
+          unset api_key
+
+      - uses: actions/upload-artifact@v4
+        with:
+          name: observatory-read-adapter-live-receipt
+          path: observatory-read-adapter-live-receipt.json
+          if-no-files-found: error
diff --git a/Dockerfile.observatory-read-adapter b/Dockerfile.observatory-read-adapter
new file mode 100644
index 0000000..92c3a1e
--- /dev/null
+++ b/Dockerfile.observatory-read-adapter
@@ -0,0 +1,20 @@
+FROM python:3.11-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+ENV PORT=8080
+
+WORKDIR /app
+
+COPY pyproject.toml README.md /app/
+COPY observatory_read_adapter /app/observatory_read_adapter
+COPY lib /app/lib
+
+RUN pip install --no-cache-dir --upgrade pip \
+  && pip install --no-cache-dir ".[observatory]" \
+  && addgroup --system --gid 10001 adapter \
+  && adduser --system --uid 10001 --ingroup adapter --no-create-home adapter
+
+USER 10001:10001
+
+CMD ["python", "-m", "observatory_read_adapter"]
diff --git a/docs/observatory-read-adapter-gcp-staging.md b/docs/observatory-read-adapter-gcp-staging.md
new file mode 100644
index 0000000..b4d6960
--- /dev/null
+++ b/docs/observatory-read-adapter-gcp-staging.md
@@ -0,0 +1,102 @@
+# GCP Observatory Read Adapter
+
+## Definition of Working
+
+- Working target: an authenticated GCP staging API returns one canonical claim
+  with source/evidence provenance and a separately labeled proposal ledger.
+- Operator path: `GET /v1/claims/sample` or `GET /v1/claims/{uuid}` with the
+  server-only `X-Api-Key` value.
+- Done means: the response names `teleo_canonical`, the dedicated IAM database
+  principal and `kb_observatory_read`, reports a read-only transaction and
+  denied write privileges, and live anonymous/write HTTP attempts return
+  `401`/`405`.
+- Not done: the VPS `teleo-pg`, a public Cloud SQL address, a browser-visible
+  key, a mock-only receipt, or a production Vercel repoint.
+- Required tier: `T3_live_readonly`.
+
+## Existing Connector Boundary
+
+The current `living-ip/livingip-web` main branch reads claims through
+`src/lib/api.ts`. It selects `NEXT_PUBLIC_API_URL` and otherwise falls back to
+`http://77.42.65.182:8081`, then calls the legacy `/api/claims` routes. That is
+the VPS-local/public Observatory path. It is not safe to replace that public
+environment variable with this protected API: the response shape differs and
+the adapter key must never enter a `NEXT_PUBLIC_` variable or browser bundle.
+
+The existing Teleo route `GET /api/kb/claims` is the compatibility reference
+for route-scoped authentication, but its default database/container settings
+still describe the VPS-local process. This service is a separate GCP runtime;
+it does not change Argus or PR #148's Leo runtime files.
+
+## Runtime Contract
+
+- Cloud Run service: `observatory-read-adapter-staging` in `europe-west6`.
+- Direct VPC egress: `teleo-staging-net` / `teleo-staging-europe-west6`.
+- Cloud SQL: `teleo-501523:europe-west6:teleo-pgvector-standby`, private IP.
+- Database: `teleo_canonical` only.
+- Runtime identity: `sa-observatory-read-adapter@teleo-501523.iam.gserviceaccount.com`.
+- IAM database user: `sa-observatory-read-adapter@teleo-501523.iam`.
+- Database authorization role: `kb_observatory_read` (`NOLOGIN`).
+- API authentication: route-scoped key from Secret Manager secret
+  `observatory-read-api-key-staging`; the value is not in Git or deploy logs.
+- No mutation routes exist. Authenticated unsupported methods return `405`.
+
+The Cloud SQL Python Connector uses Application Default Credentials from the
+Cloud Run service identity, automatic IAM database authentication, and private
+IP. Every request starts a read-only transaction and refuses the response if
+the database, IAM principal, authorization membership, required reads, or
+effective write-denial checks drift.
+
+## One-Time Staging Bootstrap
+
+These are staging mutations. Run them only with an authenticated GCP operator;
+they do not route production traffic.
+
+1. Create the dedicated runtime service account and grant only
+   `roles/cloudsql.client` and `roles/cloudsql.instanceUser` in project
+   `teleo-501523`.
+2. Add the service account as a Cloud SQL IAM service-account user on
+   `teleo-pgvector-standby`.
+3. From the existing private GCP VM database-admin path, run
+   `ops/observatory_read_role.sql` against `teleo_canonical` with
+   `OBSERVATORY_DB_IAM_USER=sa-observatory-read-adapter@teleo-501523.iam`.
+   Do not read, copy, or retain the administrator password.
+4. Create `observatory-read-api-key-staging`, add a generated 32-byte-or-longer
+   value through stdin, and grant that secret's accessor role only to the
+   runtime service account and the bounded staging canary identity.
+5. Dispatch `.github/workflows/gcp-observatory-read-adapter.yml` with
+   `action=deploy_staging` from the exact reviewed branch revision.
+
+The workflow builds an immutable image, deploys at most two Cloud Run
+instances, and retains the positive response plus anonymous `401` and
+authenticated POST `405` receipts. It does not modify Vercel.
+
+## Unexecuted Cutover Packet
+
+Production cutover requires a separate exact authorization and a
+`living-ip/livingip-web` change. Do not execute these steps from this lane.
+
+1. Add server-only Vercel secrets `OBSERVATORY_CANONICAL_API_URL` and
+   `OBSERVATORY_CANONICAL_API_KEY` to a protected preview environment. The URL
+   targets the GCP service; neither setting may use `NEXT_PUBLIC_`.
+2. Add a Next.js server route or server action that calls the GCP API, validates
+   `livingip.observatory-canonical-claim.v1`, and maps it to the Observatory UI.
+   The browser calls only the same-origin Next.js route.
+3. Prove preview Deployment Protection, anonymous denial, source/evidence
+   rendering, proposal/live labels, and no browser key exposure.
+4. Obtain exact production-repoint authorization naming the reviewed
+   `livingip-web` commit and Vercel project before changing production values.
+
+## Unexecuted Rollback Packet
+
+1. Revert the authorized `livingip-web` connector commit or disable its
+   server-side feature flag.
+2. Restore the prior production Vercel values and redeploy the last accepted
+   production revision.
+3. Verify `/claims` again uses the prior `/api/claims` contract.
+4. Leave the private Cloud SQL instance and adapter role intact while receipts
+   are reviewed; scale the staging Cloud Run service to zero if isolation is
+   required.
+5. Delete the staging service/role/secret only under a separate cleanup
+   decision. Rollback never adds a Cloud SQL public IP and never redirects the
+   adapter to VPS `teleo-pg`.
diff --git a/docs/reports/observatory-read-adapter/platform-goal-current.json b/docs/reports/observatory-read-adapter/platform-goal-current.json
new file mode 100644
index 0000000..6c60df3
--- /dev/null
+++ b/docs/reports/observatory-read-adapter/platform-goal-current.json
@@ -0,0 +1,12 @@
+{
+  "fallback_goal_path": "goal.md",
+  "platform_goal_after": {
+    "objective": "Complete and verify the private GCP Observatory read adapter defined in this goal file, stopping before production repoint without exact authorization.",
+    "status": "active",
+    "thread_id": "019f6350-69fa-71f3-944b-3fc0065f4fec"
+  },
+  "platform_goal_before": null,
+  "platform_goal_repair_action": "create_goal",
+  "platform_goal_replaced": false,
+  "why_not_replaced": "No prior platform Goal existed; a new active Goal was created from goal.md."
+}
diff --git a/observatory_read_adapter/__init__.py b/observatory_read_adapter/__init__.py
new file mode 100644
index 0000000..c4e4945
--- /dev/null
+++ b/observatory_read_adapter/__init__.py
@@ -0,0 +1,6 @@
+"""Authenticated, read-only Observatory adapter for canonical Teleo claims."""
+
+from .app import create_app
+from .config import Settings
+
+__all__ = ["Settings", "create_app"]
diff --git a/observatory_read_adapter/__main__.py b/observatory_read_adapter/__main__.py
new file mode 100644
index 0000000..61803e2
--- /dev/null
+++ b/observatory_read_adapter/__main__.py
@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+import logging
+
+from aiohttp import web
+
+from .app import create_app
+from .config import Settings
+
+
+def main() -> None:
+    logging.basicConfig(level=logging.INFO)
+    settings = Settings.from_env()
+    web.run_app(create_app(settings), host="0.0.0.0", port=settings.port)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/observatory_read_adapter/app.py b/observatory_read_adapter/app.py
new file mode 100644
index 0000000..baf1c24
--- /dev/null
+++ b/observatory_read_adapter/app.py
@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+import asyncio
+import hmac
+import logging
+import uuid
+from typing import Any
+
+from aiohttp import web
+
+from .config import Settings
+from .db import CloudSqlClaimReader
+
+logger = logging.getLogger("teleo.observatory_read_adapter")
+
+SETTINGS_KEY = web.AppKey("observatory_settings", Settings)
+READER_KEY = web.AppKey("observatory_reader", object)
+PUBLIC_PATHS = frozenset({"/healthz", "/readyz"})
+
+
+def _private_json(payload: dict[str, Any], *, status: int = 200) -> web.Response:
+    response = web.json_response(payload, status=status)
+    response.headers["Cache-Control"] = "private, no-store, max-age=0"
+    response.headers["Pragma"] = "no-cache"
+    response.headers["Vary"] = "X-Api-Key"
+    response.headers["X-Content-Type-Options"] = "nosniff"
+    return response
+
+
+@web.middleware
+async def api_key_auth(request: web.Request, handler: Any) -> web.StreamResponse:
+    if request.path in PUBLIC_PATHS:
+        return await handler(request)
+    settings = request.app[SETTINGS_KEY]
+    provided = request.headers.get("X-Api-Key", "")
+    try:
+        authenticated = hmac.compare_digest(
+            provided.encode("ascii"),
+            settings.api_key.encode("ascii"),
+        )
+    except UnicodeEncodeError:
+        authenticated = False
+    if not authenticated:
+        return _private_json({"error": "unauthorized"}, status=401)
+    return await handler(request)
+
+
+async def healthz(_request: web.Request) -> web.Response:
+    return web.json_response({"status": "ok"})
+
+
+async def readyz(request: web.Request) -> web.Response:
+    reader = request.app[READER_KEY]
+    try:
+        await asyncio.wait_for(asyncio.to_thread(reader.check_ready), timeout=12.0)
+    except Exception as exc:
+        logger.warning("readiness check failed (%s)", type(exc).__name__)
+        return web.json_response({"status": "unavailable"}, status=503)
+    return web.json_response({"status": "ready"})
+
+
+async def get_sample_claim(request: web.Request) -> web.Response:
+    return await _claim_response(request, None)
+
+
+async def get_claim(request: web.Request) -> web.Response:
+    raw_claim_id = request.match_info["claim_id"]
+    try:
+        claim_id = str(uuid.UUID(raw_claim_id))
+    except ValueError:
+        return _private_json({"error": "invalid_claim_id"}, status=400)
+    return await _claim_response(request, claim_id)
+
+
+async def _claim_response(request: web.Request, claim_id: str | None) -> web.Response:
+    reader = request.app[READER_KEY]
+    try:
+        payload = await asyncio.wait_for(
+            asyncio.to_thread(reader.read_claim, claim_id),
+            timeout=12.0,
+        )
+    except Exception as exc:
+        logger.error("canonical claim read failed (%s)", type(exc).__name__)
+        return _private_json({"error": "canonical_claim_unavailable"}, status=503)
+    if payload is None:
+        return _private_json({"error": "claim_not_found"}, status=404)
+    return _private_json(payload)
+
+
+async def _cleanup(app: web.Application) -> None:
+    reader = app[READER_KEY]
+    await asyncio.to_thread(reader.close)
+
+
+def create_app(
+    settings: Settings | None = None,
+    *,
+    reader: Any | None = None,
+) -> web.Application:
+    settings = settings or Settings.from_env()
+    settings.validate()
+    reader = reader or CloudSqlClaimReader(settings)
+    app = web.Application(middlewares=[api_key_auth], client_max_size=16 * 1024)
+    app[SETTINGS_KEY] = settings
+    app[READER_KEY] = reader
+    app.router.add_get("/healthz", healthz)
+    app.router.add_get("/readyz", readyz)
+    app.router.add_get("/v1/claims/sample", get_sample_claim)
+    app.router.add_get("/v1/claims/{claim_id}", get_claim)
+    app.on_cleanup.append(_cleanup)
+    return app
diff --git a/observatory_read_adapter/config.py b/observatory_read_adapter/config.py
new file mode 100644
index 0000000..b01b5ac
--- /dev/null
+++ b/observatory_read_adapter/config.py
@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass, field
+
+EXPECTED_PROJECT = "teleo-501523"
+EXPECTED_REGION = "europe-west6"
+EXPECTED_INSTANCE = "teleo-pgvector-standby"
+EXPECTED_CONNECTION_NAME = f"{EXPECTED_PROJECT}:{EXPECTED_REGION}:{EXPECTED_INSTANCE}"
+EXPECTED_DATABASE = "teleo_canonical"
+EXPECTED_AUTHORIZATION_ROLE = "kb_observatory_read"
+
+
+@dataclass(frozen=True)
+class Settings:
+    project_id: str
+    instance_connection_name: str
+    database: str
+    iam_database_user: str
+    authorization_role: str
+    api_key: str = field(repr=False)
+    service_revision: str = "unknown"
+    port: int = 8080
+
+    @classmethod
+    def from_env(cls) -> Settings:
+        settings = cls(
+            project_id=os.environ.get("GCP_PROJECT_ID", ""),
+            instance_connection_name=os.environ.get("CLOUD_SQL_INSTANCE", ""),
+            database=os.environ.get("DB_NAME", ""),
+            iam_database_user=os.environ.get("DB_IAM_USER", ""),
+            authorization_role=os.environ.get("DB_AUTHORIZATION_ROLE", ""),
+            api_key=os.environ.get("OBSERVATORY_API_KEY", ""),
+            service_revision=os.environ.get("SERVICE_REVISION", "unknown"),
+            port=int(os.environ.get("PORT", "8080")),
+        )
+        settings.validate()
+        return settings
+
+    def validate(self) -> None:
+        if self.project_id != EXPECTED_PROJECT:
+            raise ValueError(f"GCP_PROJECT_ID must be {EXPECTED_PROJECT}")
+        if self.instance_connection_name != EXPECTED_CONNECTION_NAME:
+            raise ValueError(f"CLOUD_SQL_INSTANCE must be {EXPECTED_CONNECTION_NAME}")
+        if self.database != EXPECTED_DATABASE:
+            raise ValueError(f"DB_NAME must be {EXPECTED_DATABASE}")
+        if self.authorization_role != EXPECTED_AUTHORIZATION_ROLE:
+            raise ValueError(f"DB_AUTHORIZATION_ROLE must be {EXPECTED_AUTHORIZATION_ROLE}")
+        if not self.iam_database_user.endswith(f"@{EXPECTED_PROJECT}.iam"):
+            raise ValueError("DB_IAM_USER must be the scoped Cloud SQL service-account user")
+        if any(character.isspace() for character in self.iam_database_user):
+            raise ValueError("DB_IAM_USER must not contain whitespace")
+        if (
+            len(self.api_key) < 32
+            or len(self.api_key) > 512
+            or not self.api_key.isascii()
+            or any(character.isspace() for character in self.api_key)
+        ):
+            raise ValueError("OBSERVATORY_API_KEY must be 32-512 non-whitespace ASCII characters")
+        if not 1 <= self.port <= 65535:
+            raise ValueError("PORT is outside the valid TCP port range")
diff --git a/observatory_read_adapter/db.py b/observatory_read_adapter/db.py
new file mode 100644
index 0000000..8ed89c4
--- /dev/null
+++ b/observatory_read_adapter/db.py
@@ -0,0 +1,315 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+from datetime import date, datetime, timezone
+from decimal import Decimal
+from typing import Any
+
+from .config import Settings
+
+ConnectionFactory = Callable[[], Any]
+
+
+class ReaderContractError(RuntimeError):
+    """The live database session does not match the adapter's read-only contract."""
+
+
+def _json_value(value: Any) -> Any:
+    if isinstance(value, Decimal):
+        return float(value)
+    if isinstance(value, (date, datetime)):
+        return value.isoformat()
+    if isinstance(value, tuple):
+        return list(value)
+    return value
+
+
+def _column_name(column: Any) -> str:
+    return str(column.name) if hasattr(column, "name") else str(column[0])
+
+
+def _row_dict(cursor: Any, row: Any) -> dict[str, Any] | None:
+    if row is None:
+        return None
+    names = [_column_name(column) for column in cursor.description]
+    return {name: _json_value(value) for name, value in zip(names, row, strict=True)}
+
+
+def _rows(cursor: Any) -> list[dict[str, Any]]:
+    names = [_column_name(column) for column in cursor.description]
+    return [
+        {name: _json_value(value) for name, value in zip(names, row, strict=True)}
+        for row in cursor.fetchall()
+    ]
+
+
+class CloudSqlClaimReader:
+    def __init__(self, settings: Settings, *, connection_factory: ConnectionFactory | None = None):
+        self.settings = settings
+        self._connector: Any | None = None
+        if connection_factory is not None:
+            self._connection_factory = connection_factory
+            return
+
+        from google.cloud.sql.connector import Connector, IPTypes
+
+        self._connector = Connector(refresh_strategy="LAZY")
+
+        def connect() -> Any:
+            return self._connector.connect(
+                settings.instance_connection_name,
+                "pg8000",
+                user=settings.iam_database_user,
+                db=settings.database,
+                enable_iam_auth=True,
+                ip_type=IPTypes.PRIVATE,
+            )
+
+        self._connection_factory = connect
+
+    def close(self) -> None:
+        if self._connector is not None:
+            self._connector.close()
+
+    def check_ready(self) -> dict[str, Any]:
+        connection = self._connection_factory()
+        try:
+            cursor = connection.cursor()
+            self._begin_read_only(cursor)
+            session = self._load_session_contract(cursor)
+            connection.rollback()
+            return {
+                "ready": True,
+                "database": session["database"],
+                "authorization_role": self.settings.authorization_role,
+                "transaction_read_only": True,
+            }
+        finally:
+            connection.close()
+
+    def read_claim(self, claim_id: str | None = None) -> dict[str, Any] | None:
+        connection = self._connection_factory()
+        try:
+            cursor = connection.cursor()
+            self._begin_read_only(cursor)
+            session = self._load_session_contract(cursor)
+            claim = self._load_claim(cursor, claim_id)
+            if claim is None:
+                connection.rollback()
+                return None
+
+            evidence = self._load_evidence(cursor, claim["id"])
+            edges = self._load_edges(cursor, claim["id"])
+            proposal_counts = self._load_proposal_counts(cursor)
+            related_proposals = self._load_related_proposals(cursor, claim["id"])
+            recent_proposals = self._load_recent_proposals(cursor)
+            connection.rollback()
+        finally:
+            connection.close()
+
+        generated_at = datetime.now(timezone.utc).isoformat()
+        return {
+            "schema": "livingip.observatory-canonical-claim.v1",
+            "generated_at": generated_at,
+            "read_only": True,
+            "provenance": {
+                "gcp_project": self.settings.project_id,
+                "cloud_sql_instance": self.settings.instance_connection_name,
+                "database": session["database"],
+                "database_principal": session["database_principal"],
+                "authorization_role": self.settings.authorization_role,
+                "transaction_read_only": True,
+                "write_privileges_denied": bool(session["required_writes_denied"]),
+                "service_revision": self.settings.service_revision,
+            },
+            "canonical": {
+                "relation": "public.claims",
+                "claim": claim,
+                "evidence_relation": "public.claim_evidence -> public.sources",
+                "evidence": evidence,
+                "edge_relation": "public.claim_edges",
+                "edges": edges,
+            },
+            "proposal_ledger": {
+                "relation": "kb_stage.kb_proposals",
+                "distinct_from_canonical": True,
+                "status_counts": proposal_counts,
+                "related": related_proposals,
+                "recent": recent_proposals,
+                "semantics": {
+                    "pending_review": "staged only; not canonical",
+                    "approved": "reviewed intent; not canonical until applied_at is set",
+                    "applied": "proposal receipt only; canonical rows remain authoritative",
+                },
+            },
+        }
+
+    @staticmethod
+    def _begin_read_only(cursor: Any) -> None:
+        cursor.execute("set transaction read only")
+        cursor.execute("set local statement_timeout = '5000ms'")
+        cursor.execute("set local lock_timeout = '1000ms'")
+
+    def _load_session_contract(self, cursor: Any) -> dict[str, Any]:
+        cursor.execute(
+            """
+            select current_database() as database,
+                   current_user as database_principal,
+                   current_setting('transaction_read_only') as transaction_read_only,
+                   pg_has_role(current_user, %s, 'member') as has_authorization_role,
+                   has_table_privilege(current_user, 'public.claims', 'select')
+                     and has_table_privilege(current_user, 'public.sources', 'select')
+                     and has_table_privilege(current_user, 'public.claim_evidence', 'select')
+                     and has_table_privilege(current_user, 'public.claim_edges', 'select')
+                     and has_table_privilege(current_user, 'kb_stage.kb_proposals', 'select')
+                     as has_required_reads,
+                   not (
+                     has_table_privilege(current_user, 'public.claims', 'insert,update,delete,truncate')
+                     or has_table_privilege(current_user, 'public.sources', 'insert,update,delete,truncate')
+                     or has_table_privilege(current_user, 'public.claim_evidence', 'insert,update,delete,truncate')
+                     or has_table_privilege(current_user, 'public.claim_edges', 'insert,update,delete,truncate')
+                     or has_table_privilege(current_user, 'kb_stage.kb_proposals', 'insert,update,delete,truncate')
+                   ) as required_writes_denied
+            """,
+            (self.settings.authorization_role,),
+        )
+        session = _row_dict(cursor, cursor.fetchone())
+        if session is None:
+            raise ReaderContractError("database session contract returned no row")
+        if session["database"] != self.settings.database:
+            raise ReaderContractError("database session selected an unexpected database")
+        if session["database_principal"] != self.settings.iam_database_user:
+            raise ReaderContractError("database session selected an unexpected principal")
+        if session["transaction_read_only"] != "on":
+            raise ReaderContractError("database session is not read-only")
+        if not session["has_authorization_role"] or not session["has_required_reads"]:
+            raise ReaderContractError("database principal lacks the exact read authorization")
+        if not session["required_writes_denied"]:
+            raise ReaderContractError("database principal has an unexpected write privilege")
+        return session
+
+    @staticmethod
+    def _load_claim(cursor: Any, claim_id: str | None) -> dict[str, Any] | None:
+        columns = """
+            select c.id::text as id,
+                   c.type::text as type,
+                   left(c.text, 12000) as text,
+                   c.status::text as status,
+                   c.confidence,
+                   coalesce(c.tags, '{}'::text[]) as tags,
+                   c.superseded_by::text as superseded_by,
+                   c.created_at,
+                   c.updated_at
+              from public.claims c
+        """
+        if claim_id is not None:
+            cursor.execute(columns + " where c.id = %s::uuid", (claim_id,))
+            return _row_dict(cursor, cursor.fetchone())
+
+        cursor.execute(
+            columns
+            + """
+             where exists (
+               select 1 from public.claim_evidence ce where ce.claim_id = c.id
+             )
+             order by coalesce(c.updated_at, c.created_at) desc, c.id desc
+             limit 1
+            """
+        )
+        claim = _row_dict(cursor, cursor.fetchone())
+        if claim is not None:
+            return claim
+        cursor.execute(columns + " order by coalesce(c.updated_at, c.created_at) desc, c.id desc limit 1")
+        return _row_dict(cursor, cursor.fetchone())
+
+    @staticmethod
+    def _load_evidence(cursor: Any, claim_id: str) -> list[dict[str, Any]]:
+        cursor.execute(
+            """
+            select ce.id::text as evidence_id,
+                   ce.role::text as role,
+                   ce.weight,
+                   ce.created_at as linked_at,
+                   s.id::text as source_id,
+                   s.source_type::text as source_type,
+                   s.url,
+                   s.storage_path,
+                   left(coalesce(s.excerpt, ''), 2400) as excerpt,
+                   s.hash as content_hash,
+                   s.captured_at,
+                   s.created_at as source_created_at
+              from public.claim_evidence ce
+              join public.sources s on s.id = ce.source_id
+             where ce.claim_id = %s::uuid
+             order by ce.weight desc nulls last, ce.created_at, ce.id
+             limit 30
+            """,
+            (claim_id,),
+        )
+        return _rows(cursor)
+
+    @staticmethod
+    def _load_edges(cursor: Any, claim_id: str) -> list[dict[str, Any]]:
+        cursor.execute(
+            """
+            select e.id::text as edge_id,
+                   case when e.from_claim = %s::uuid then 'outgoing' else 'incoming' end as direction,
+                   e.edge_type::text as edge_type,
+                   e.weight,
+                   case when e.from_claim = %s::uuid then e.to_claim::text else e.from_claim::text end
+                     as connected_claim_id,
+                   e.created_at
+              from public.claim_edges e
+             where e.from_claim = %s::uuid or e.to_claim = %s::uuid
+             order by e.created_at, e.id
+             limit 40
+            """,
+            (claim_id, claim_id, claim_id, claim_id),
+        )
+        return _rows(cursor)
+
+    @staticmethod
+    def _load_proposal_counts(cursor: Any) -> dict[str, int]:
+        cursor.execute(
+            "select status::text as status, count(*)::int as count "
+            "from kb_stage.kb_proposals group by status order by status"
+        )
+        return {str(status): int(count) for status, count in cursor.fetchall()}
+
+    @staticmethod
+    def _load_related_proposals(cursor: Any, claim_id: str) -> list[dict[str, Any]]:
+        cursor.execute(
+            """
+            select p.id::text as id,
+                   p.proposal_type::text as proposal_type,
+                   p.status::text as status,
+                   to_jsonb(p)->>'source_ref' as source_ref,
+                   p.reviewed_at,
+                   p.applied_at,
+                   p.updated_at
+              from kb_stage.kb_proposals p
+             where p.payload::text like %s
+             order by p.updated_at desc, p.id desc
+             limit 20
+            """,
+            (f"%{claim_id}%",),
+        )
+        return _rows(cursor)
+
+    @staticmethod
+    def _load_recent_proposals(cursor: Any) -> list[dict[str, Any]]:
+        cursor.execute(
+            """
+            select p.id::text as id,
+                   p.proposal_type::text as proposal_type,
+                   p.status::text as status,
+                   to_jsonb(p)->>'source_ref' as source_ref,
+                   p.reviewed_at,
+                   p.applied_at,
+                   p.updated_at
+              from kb_stage.kb_proposals p
+             order by p.updated_at desc, p.id desc
+             limit 10
+            """
+        )
+        return _rows(cursor)
diff --git a/ops/observatory_read_role.sql b/ops/observatory_read_role.sql
new file mode 100644
index 0000000..0536237
--- /dev/null
+++ b/ops/observatory_read_role.sql
@@ -0,0 +1,183 @@
+\set ON_ERROR_STOP on
+\getenv observatory_iam_user OBSERVATORY_DB_IAM_USER
+
+-- Provision only after Cloud SQL has created the dedicated IAM database user.
+-- The service-account username omits the .gserviceaccount.com suffix.
+select set_config('observatory.iam_user', :'observatory_iam_user', false);
+
+do $preflight$
+declare
+  principal text := current_setting('observatory.iam_user');
+begin
+  if current_database() <> 'teleo_canonical' then
+    raise exception 'observatory_read_role must run only in teleo_canonical';
+  end if;
+  if principal !~ '^[a-z0-9-]+@teleo-501523[.]iam$' then
+    raise exception 'OBSERVATORY_DB_IAM_USER is not the scoped GCP service-account database user';
+  end if;
+  if not exists (select 1 from pg_catalog.pg_roles where rolname = principal) then
+    raise exception 'Cloud SQL IAM database user does not exist: %', principal;
+  end if;
+end
+$preflight$;
+
+select 'create role kb_observatory_read nologin nosuperuser nocreatedb nocreaterole noinherit noreplication nobypassrls connection limit -1'
+where not exists (select 1 from pg_catalog.pg_roles where rolname = 'kb_observatory_read')
+\gexec
+
+alter role kb_observatory_read
+  with nologin nosuperuser nocreatedb nocreaterole noinherit noreplication nobypassrls connection limit -1
+  password null;
+alter role kb_observatory_read reset all;
+
+revoke all privileges on database teleo_canonical from kb_observatory_read;
+grant connect on database teleo_canonical to kb_observatory_read;
+
+revoke all on schema public, kb_stage from kb_observatory_read;
+revoke all privileges on all tables in schema public, kb_stage from kb_observatory_read;
+revoke all privileges on all sequences in schema public, kb_stage from kb_observatory_read;
+revoke all privileges on all functions in schema public, kb_stage from kb_observatory_read;
+revoke all privileges on all procedures in schema public, kb_stage from kb_observatory_read;
+
+select format(
+         'revoke all privileges (%s) on table %I.%I from kb_observatory_read',
+         pg_catalog.string_agg(pg_catalog.quote_ident(attribute.attname), ', ' order by attribute.attnum),
+         namespace.nspname,
+         relation.relname
+       )
+  from pg_catalog.pg_class relation
+  join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
+  join pg_catalog.pg_attribute attribute on attribute.attrelid = relation.oid
+ where namespace.nspname in ('public', 'kb_stage')
+   and relation.relkind in ('r', 'p', 'v', 'm', 'f')
+   and attribute.attnum > 0
+   and not attribute.attisdropped
+ group by namespace.nspname, relation.relname
+\gexec
+
+grant usage on schema public, kb_stage to kb_observatory_read;
+grant select on
+  public.claims,
+  public.sources,
+  public.claim_evidence,
+  public.claim_edges,
+  kb_stage.kb_proposals
+to kb_observatory_read;
+
+do $membership$
+declare
+  principal text := current_setting('observatory.iam_user');
+  unexpected text;
+begin
+  select pg_catalog.string_agg(granted.rolname, ', ' order by granted.rolname)
+    into unexpected
+    from pg_catalog.pg_auth_members membership
+    join pg_catalog.pg_roles granted on granted.oid = membership.roleid
+    join pg_catalog.pg_roles member on member.oid = membership.member
+   where member.rolname = principal
+     and granted.rolname <> 'kb_observatory_read';
+  if unexpected is not null then
+    raise exception 'dedicated Observatory principal has unexpected role memberships: %', unexpected;
+  end if;
+
+  select pg_catalog.string_agg(member.rolname, ', ' order by member.rolname)
+    into unexpected
+    from pg_catalog.pg_auth_members membership
+    join pg_catalog.pg_roles granted on granted.oid = membership.roleid
+    join pg_catalog.pg_roles member on member.oid = membership.member
+   where granted.rolname = 'kb_observatory_read'
+     and member.rolname <> principal;
+  if unexpected is not null then
+    raise exception 'kb_observatory_read has unexpected members: %', unexpected;
+  end if;
+
+  execute pg_catalog.format('grant kb_observatory_read to %I', principal);
+  execute pg_catalog.format(
+    'alter role %I in database teleo_canonical set default_transaction_read_only = on',
+    principal
+  );
+  execute pg_catalog.format(
+    'alter role %I in database teleo_canonical set statement_timeout = %L',
+    principal,
+    '5s'
+  );
+  execute pg_catalog.format(
+    'alter role %I in database teleo_canonical set lock_timeout = %L',
+    principal,
+    '1s'
+  );
+end
+$membership$;
+
+do $verify$
+declare
+  principal text := current_setting('observatory.iam_user');
+  unexpected text;
+begin
+  if not pg_catalog.pg_has_role(principal, 'kb_observatory_read', 'member') then
+    raise exception 'Observatory principal is not a member of kb_observatory_read';
+  end if;
+  if not exists (
+    select 1 from pg_catalog.pg_roles
+     where rolname = principal and rolinherit and not rolsuper and not rolcreatedb
+       and not rolcreaterole and not rolreplication and not rolbypassrls
+  ) then
+    raise exception 'Observatory principal has unsafe role attributes or cannot inherit read grants';
+  end if;
+
+  select pg_catalog.string_agg(
+           pg_catalog.format('%I.%I:%s', namespace.nspname, relation.relname, privilege.name),
+           ', ' order by namespace.nspname, relation.relname, privilege.name
+         )
+    into unexpected
+    from pg_catalog.pg_class relation
+    join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
+   cross join (values ('INSERT'), ('UPDATE'), ('DELETE'), ('TRUNCATE'), ('REFERENCES'), ('TRIGGER')) privilege(name)
+   where namespace.nspname in ('public', 'kb_stage')
+     and relation.relkind in ('r', 'p', 'v', 'm', 'f')
+     and pg_catalog.has_table_privilege(principal, relation.oid, privilege.name);
+  if unexpected is not null then
+    raise exception 'Observatory principal has effective table writes: %', unexpected;
+  end if;
+
+  select pg_catalog.string_agg(
+           pg_catalog.format('%I.%I', namespace.nspname, relation.relname),
+           ', ' order by namespace.nspname, relation.relname
+         )
+    into unexpected
+    from pg_catalog.pg_class relation
+    join pg_catalog.pg_namespace namespace on namespace.oid = relation.relnamespace
+   where namespace.nspname in ('public', 'kb_stage')
+     and relation.relkind in ('r', 'p', 'v', 'm', 'f')
+     and pg_catalog.has_table_privilege(principal, relation.oid, 'SELECT')
+     and (namespace.nspname, relation.relname) not in (
+       ('public', 'claims'),
+       ('public', 'sources'),
+       ('public', 'claim_evidence'),
+       ('public', 'claim_edges'),
+       ('kb_stage', 'kb_proposals')
+     );
+  if unexpected is not null then
+    raise exception 'Observatory principal can SELECT outside its allowlist: %', unexpected;
+  end if;
+
+  if not (
+    pg_catalog.has_table_privilege(principal, 'public.claims', 'SELECT')
+    and pg_catalog.has_table_privilege(principal, 'public.sources', 'SELECT')
+    and pg_catalog.has_table_privilege(principal, 'public.claim_evidence', 'SELECT')
+    and pg_catalog.has_table_privilege(principal, 'public.claim_edges', 'SELECT')
+    and pg_catalog.has_table_privilege(principal, 'kb_stage.kb_proposals', 'SELECT')
+  ) then
+    raise exception 'Observatory principal is missing a required read grant';
+  end if;
+end
+$verify$;
+
+select jsonb_build_object(
+  'database', current_database(),
+  'authorization_role', 'kb_observatory_read',
+  'database_principal', current_setting('observatory.iam_user'),
+  'required_reads', true,
+  'effective_table_writes_denied', true,
+  'default_transaction_read_only', true
+)::text;
diff --git a/pyproject.toml b/pyproject.toml
index cf64b66..e264213 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -18,9 +18,12 @@ dev = [
     "pytest-asyncio>=0.23",
     "ruff>=0.3",
 ]
+observatory = [
+    "cloud-sql-python-connector[pg8000]>=1.20.4,<1.21",
+]
 
 [tool.setuptools]
-packages = ["lib"]
+packages = ["lib", "observatory_read_adapter"]
 
 [tool.ruff]
 target-version = "py311"
diff --git a/tests/test_observatory_read_adapter.py b/tests/test_observatory_read_adapter.py
new file mode 100644
index 0000000..f259145
--- /dev/null
+++ b/tests/test_observatory_read_adapter.py
@@ -0,0 +1,346 @@
+from __future__ import annotations
+
+import asyncio
+import json
+from pathlib import Path
+
+from aiohttp.test_utils import TestClient, TestServer
+
+from observatory_read_adapter.app import create_app
+from observatory_read_adapter.config import EXPECTED_CONNECTION_NAME, Settings
+from observatory_read_adapter.db import CloudSqlClaimReader
+
+CLAIM_ID = "d0000000-0000-0000-0000-000000000005"
+API_KEY = "observatory-test-api-key-000000000000000000000000"
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def settings(**overrides: object) -> Settings:
+    values = {
+        "project_id": "teleo-501523",
+        "instance_connection_name": EXPECTED_CONNECTION_NAME,
+        "database": "teleo_canonical",
+        "iam_database_user": "sa-observatory-read-adapter@teleo-501523.iam",
+        "authorization_role": "kb_observatory_read",
+        "api_key": API_KEY,
+        "service_revision": "test-revision",
+    }
+    values.update(overrides)
+    return Settings(**values)
+
+
+class FakeReader:
+    def __init__(self) -> None:
+        self.closed = False
+
+    def check_ready(self) -> dict[str, object]:
+        return {
+            "ready": True,
+            "database": "teleo_canonical",
+            "authorization_role": "kb_observatory_read",
+            "transaction_read_only": True,
+        }
+
+    def read_claim(self, claim_id: str | None = None) -> dict[str, object] | None:
+        if claim_id not in (None, CLAIM_ID):
+            return None
+        return {
+            "schema": "livingip.observatory-canonical-claim.v1",
+            "read_only": True,
+            "provenance": {
+                "database": "teleo_canonical",
+                "database_principal": "sa-observatory-read-adapter@teleo-501523.iam",
+                "authorization_role": "kb_observatory_read",
+                "transaction_read_only": True,
+                "write_privileges_denied": True,
+            },
+            "canonical": {
+                "claim": {"id": CLAIM_ID, "status": "open", "text": "Canonical claim"},
+                "evidence": [{"source_id": "e0000000-0000-0000-0000-000000000006"}],
+            },
+            "proposal_ledger": {
+                "distinct_from_canonical": True,
+                "status_counts": {"pending_review": 2, "approved": 1, "applied": 1},
+                "related": [{"status": "approved", "applied_at": None}],
+            },
+        }
+
+    def close(self) -> None:
+        self.closed = True
+
+
+class FakeCursor:
+    def __init__(self) -> None:
+        self.description: list[tuple[str]] = []
+        self.result: list[tuple[object, ...]] = []
+        self.executed: list[tuple[str, tuple[object, ...]]] = []
+
+    def execute(self, sql: str, parameters: tuple[object, ...] = ()) -> None:
+        normalized = " ".join(sql.split()).lower()
+        self.executed.append((normalized, parameters))
+        self.description = []
+        self.result = []
+        if "current_database() as database" in normalized:
+            self.description = [
+                ("database",),
+                ("database_principal",),
+                ("transaction_read_only",),
+                ("has_authorization_role",),
+                ("has_required_reads",),
+                ("required_writes_denied",),
+            ]
+            self.result = [
+                (
+                    "teleo_canonical",
+                    "sa-observatory-read-adapter@teleo-501523.iam",
+                    "on",
+                    True,
+                    True,
+                    True,
+                )
+            ]
+        elif "from public.claims c" in normalized:
+            self.description = [
+                ("id",),
+                ("type",),
+                ("text",),
+                ("status",),
+                ("confidence",),
+                ("tags",),
+                ("superseded_by",),
+                ("created_at",),
+                ("updated_at",),
+            ]
+            self.result = [
+                (
+                    CLAIM_ID,
+                    "strategy",
+                    "Canonical claim",
+                    "open",
+                    0.8,
+                    ["test"],
+                    None,
+                    "2026-07-15T00:00:00+00:00",
+                    "2026-07-15T00:00:00+00:00",
+                )
+            ]
+        elif "from public.claim_evidence ce" in normalized:
+            self.description = [
+                ("evidence_id",),
+                ("role",),
+                ("weight",),
+                ("linked_at",),
+                ("source_id",),
+                ("source_type",),
+                ("url",),
+                ("storage_path",),
+                ("excerpt",),
+                ("content_hash",),
+                ("captured_at",),
+                ("source_created_at",),
+            ]
+            self.result = [
+                (
+                    "e0000000-0000-0000-0000-000000000006",
+                    "grounds",
+                    1.0,
+                    "2026-07-15T00:00:00+00:00",
+                    "f0000000-0000-0000-0000-000000000007",
+                    "document",
+                    "https://example.test/source",
+                    None,
+                    "Source excerpt",
+                    "sha256:test",
+                    "2026-07-15T00:00:00+00:00",
+                    "2026-07-15T00:00:00+00:00",
+                )
+            ]
+        elif "from public.claim_edges e" in normalized:
+            self.description = [
+                ("edge_id",),
+                ("direction",),
+                ("edge_type",),
+                ("weight",),
+                ("connected_claim_id",),
+                ("created_at",),
+            ]
+            self.result = []
+        elif "group by status" in normalized:
+            self.description = [("status",), ("count",)]
+            self.result = [("approved", 1), ("applied", 1), ("pending_review", 2)]
+        elif "where p.payload::text like" in normalized:
+            self.description = [
+                ("id",),
+                ("proposal_type",),
+                ("status",),
+                ("source_ref",),
+                ("reviewed_at",),
+                ("applied_at",),
+                ("updated_at",),
+            ]
+            self.result = [
+                (
+                    "a0000000-0000-0000-0000-000000000001",
+                    "revise_claim",
+                    "approved",
+                    "source:test",
+                    "2026-07-15T00:00:00+00:00",
+                    None,
+                    "2026-07-15T00:00:00+00:00",
+                )
+            ]
+        elif "from kb_stage.kb_proposals p" in normalized:
+            self.description = [
+                ("id",),
+                ("proposal_type",),
+                ("status",),
+                ("source_ref",),
+                ("reviewed_at",),
+                ("applied_at",),
+                ("updated_at",),
+            ]
+            self.result = []
+
+    def fetchone(self) -> tuple[object, ...] | None:
+        return self.result.pop(0) if self.result else None
+
+    def fetchall(self) -> list[tuple[object, ...]]:
+        result, self.result = self.result, []
+        return result
+
+
+class FakeConnection:
+    def __init__(self) -> None:
+        self.cursor_value = FakeCursor()
+        self.rolled_back = False
+        self.closed = False
+
+    def cursor(self) -> FakeCursor:
+        return self.cursor_value
+
+    def rollback(self) -> None:
+        self.rolled_back = True
+
+    def close(self) -> None:
+        self.closed = True
+
+
+def test_settings_fail_closed_on_wrong_database_role_or_short_key() -> None:
+    for overrides in (
+        {"database": "teleo"},
+        {"authorization_role": "leoclean_kb_runtime"},
+        {"api_key": "too-short"},
+        {"instance_connection_name": "other:region:instance"},
+    ):
+        candidate = settings(**overrides)
+        try:
+            candidate.validate()
+        except ValueError:
+            pass
+        else:
+            raise AssertionError(f"unsafe settings accepted: {overrides}")
+
+
+def test_authenticated_claim_response_and_negative_http_paths() -> None:
+    async def exercise() -> None:
+        reader = FakeReader()
+        client = TestClient(TestServer(create_app(settings(), reader=reader)))
+        await client.start_server()
+        try:
+            anonymous = await client.get("/v1/claims/sample")
+            assert anonymous.status == 401
+            assert await anonymous.json() == {"error": "unauthorized"}
+
+            response = await client.get(
+                f"/v1/claims/{CLAIM_ID}",
+                headers={"X-Api-Key": API_KEY},
+            )
+            assert response.status == 200
+            assert response.headers["Cache-Control"] == "private, no-store, max-age=0"
+            payload = await response.json()
+            assert payload["schema"] == "livingip.observatory-canonical-claim.v1"
+            assert payload["provenance"]["database"] == "teleo_canonical"
+            assert payload["provenance"]["write_privileges_denied"] is True
+            assert payload["canonical"]["evidence"]
+            assert payload["proposal_ledger"]["distinct_from_canonical"] is True
+            assert payload["proposal_ledger"]["related"][0]["applied_at"] is None
+            assert API_KEY not in json.dumps(payload)
+
+            write_attempt = await client.post(
+                f"/v1/claims/{CLAIM_ID}",
+                headers={"X-Api-Key": API_KEY},
+                json={"status": "applied"},
+            )
+            assert write_attempt.status == 405
+
+            invalid = await client.get(
+                "/v1/claims/not-a-uuid",
+                headers={"X-Api-Key": API_KEY},
+            )
+            assert invalid.status == 400
+        finally:
+            await client.close()
+        assert reader.closed is True
+
+    asyncio.run(exercise())
+
+
+def test_readiness_checks_database_contract_without_exposing_claims() -> None:
+    async def exercise() -> None:
+        client = TestClient(TestServer(create_app(settings(), reader=FakeReader())))
+        await client.start_server()
+        try:
+            response = await client.get("/readyz")
+            payload = await response.json()
+            assert response.status == 200
+            assert payload == {"status": "ready"}
+        finally:
+            await client.close()
+
+    asyncio.run(exercise())
+
+
+def test_database_reader_returns_provenance_and_keeps_proposals_distinct() -> None:
+    connection = FakeConnection()
+    reader = CloudSqlClaimReader(settings(), connection_factory=lambda: connection)
+
+    payload = reader.read_claim(CLAIM_ID)
+
+    assert payload is not None
+    assert payload["provenance"]["database"] == "teleo_canonical"
+    assert payload["provenance"]["database_principal"] == settings().iam_database_user
+    assert payload["provenance"]["write_privileges_denied"] is True
+    assert payload["canonical"]["claim"]["id"] == CLAIM_ID
+    assert payload["canonical"]["evidence"][0]["content_hash"] == "sha256:test"
+    assert payload["proposal_ledger"]["distinct_from_canonical"] is True
+    assert payload["proposal_ledger"]["related"][0]["status"] == "approved"
+    assert payload["proposal_ledger"]["related"][0]["applied_at"] is None
+    assert connection.rolled_back is True
+    assert connection.closed is True
+    assert any("set transaction read only" in sql for sql, _parameters in connection.cursor_value.executed)
+
+
+def test_database_role_contract_is_select_only_and_iam_scoped() -> None:
+    sql = (ROOT / "ops" / "observatory_read_role.sql").read_text(encoding="utf-8")
+    lowered = sql.lower()
+
+    assert "kb_observatory_read nologin" in lowered
+    assert "default_transaction_read_only = on" in lowered
+    assert "public.claims" in lowered
+    assert "public.sources" in lowered
+    assert "public.claim_evidence" in lowered
+    assert "public.claim_edges" in lowered
+    assert "kb_stage.kb_proposals" in lowered
+    assert "grant select on" in lowered
+    assert "grant insert" not in lowered
+    assert "grant update" not in lowered
+    assert "grant delete" not in lowered
+    assert "effective_table_writes_denied" in lowered
+
+
+def test_container_runs_as_non_root_and_contains_no_password_contract() -> None:
+    dockerfile = (ROOT / "Dockerfile.observatory-read-adapter").read_text(encoding="utf-8")
+
+    assert "USER 10001:10001" in dockerfile
+    assert "PGPASSWORD" not in dockerfile
+    assert "DB_PASS" not in dockerfile
diff --git a/tests/test_observatory_read_role_postgres.py b/tests/test_observatory_read_role_postgres.py
new file mode 100644
index 0000000..be2b155
--- /dev/null
+++ b/tests/test_observatory_read_role_postgres.py
@@ -0,0 +1,182 @@
+from __future__ import annotations
+
+import shutil
+import subprocess
+import time
+import uuid
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+ROLE_SQL = ROOT / "ops" / "observatory_read_role.sql"
+POSTGRES_IMAGE = "postgres:16-alpine"
+DATABASE = "teleo_canonical"
+IAM_USER = "sa-observatory-read-adapter@teleo-501523.iam"
+TEST_PASSWORD = "generated-test-only-password"
+
+BOOTSTRAP_SQL = f"""
+begin;
+create schema kb_stage;
+create role "{IAM_USER}" login inherit password '{TEST_PASSWORD}';
+create table public.claims (id uuid primary key);
+create table public.sources (id uuid primary key);
+create table public.claim_evidence (id uuid primary key);
+create table public.claim_edges (id uuid primary key);
+create table public.private_notes (id uuid primary key);
+create table kb_stage.kb_proposals (id uuid primary key);
+insert into public.claims values ('11111111-1111-1111-1111-111111111111');
+commit;
+"""
+
+
+def run(command: list[str], *, input_text: str | None = None, check: bool = True) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        command,
+        input=input_text,
+        text=True,
+        capture_output=True,
+        check=check,
+        timeout=30,
+    )
+
+
+@pytest.mark.skipif(shutil.which("docker") is None, reason="Docker is required")
+def test_observatory_role_is_read_only_and_exactly_allowlisted() -> None:
+    container = f"teleo-observatory-role-{uuid.uuid4().hex[:10]}"
+    run(
+        [
+            "docker",
+            "run",
+            "--detach",
+            "--rm",
+            "--name",
+            container,
+            "--env",
+            f"POSTGRES_PASSWORD={TEST_PASSWORD}",
+            "--env",
+            f"POSTGRES_DB={DATABASE}",
+            POSTGRES_IMAGE,
+        ]
+    )
+    try:
+        deadline = time.monotonic() + 25
+        while time.monotonic() < deadline:
+            ready = run(
+                ["docker", "exec", container, "pg_isready", "-U", "postgres", "-d", DATABASE],
+                check=False,
+            )
+            if ready.returncode == 0:
+                break
+            time.sleep(0.25)
+        else:
+            raise AssertionError("ephemeral Postgres did not become ready")
+
+        bootstrap: subprocess.CompletedProcess[str] | None = None
+        deadline = time.monotonic() + 15
+        while time.monotonic() < deadline:
+            bootstrap = run(
+                [
+                    "docker",
+                    "exec",
+                    "-i",
+                    container,
+                    "psql",
+                    "--set",
+                    "ON_ERROR_STOP=1",
+                    "-U",
+                    "postgres",
+                    "-d",
+                    DATABASE,
+                ],
+                input_text=BOOTSTRAP_SQL,
+                check=False,
+            )
+            if bootstrap.returncode == 0:
+                break
+            time.sleep(0.25)
+        else:
+            assert bootstrap is not None
+            raise AssertionError(f"ephemeral Postgres bootstrap failed: {bootstrap.stderr[-1000:]}")
+        migration = run(
+            [
+                "docker",
+                "exec",
+                "-e",
+                f"OBSERVATORY_DB_IAM_USER={IAM_USER}",
+                "-i",
+                container,
+                "psql",
+                "-U",
+                "postgres",
+                "-d",
+                DATABASE,
+            ],
+            input_text=ROLE_SQL.read_text(encoding="utf-8"),
+        )
+        assert '"effective_table_writes_denied": true' in migration.stdout
+
+        readback = run(
+            [
+                "docker",
+                "exec",
+                "-e",
+                f"PGPASSWORD={TEST_PASSWORD}",
+                container,
+                "psql",
+                "-h",
+                "127.0.0.1",
+                "-U",
+                IAM_USER,
+                "-d",
+                DATABASE,
+                "-At",
+                "-c",
+                "show default_transaction_read_only; select count(*) from public.claims;",
+            ]
+        )
+        assert readback.stdout.splitlines() == ["on", "1"]
+
+        write_attempt = run(
+            [
+                "docker",
+                "exec",
+                "-e",
+                f"PGPASSWORD={TEST_PASSWORD}",
+                container,
+                "psql",
+                "-h",
+                "127.0.0.1",
+                "-U",
+                IAM_USER,
+                "-d",
+                DATABASE,
+                "-c",
+                "insert into public.claims values ('22222222-2222-2222-2222-222222222222');",
+            ],
+            check=False,
+        )
+        assert write_attempt.returncode != 0
+
+        outside_allowlist = run(
+            [
+                "docker",
+                "exec",
+                "-e",
+                f"PGPASSWORD={TEST_PASSWORD}",
+                container,
+                "psql",
+                "-h",
+                "127.0.0.1",
+                "-U",
+                IAM_USER,
+                "-d",
+                DATABASE,
+                "-c",
+                "select * from public.private_notes;",
+            ],
+            check=False,
+        )
+        assert outside_allowlist.returncode != 0
+    finally:
+        run(["docker", "rm", "--force", container], check=False)