diff --git a/.changes/observability-unification.added.md b/.changes/observability-unification.added.md
new file mode 100644
index 0000000..edcd789
--- /dev/null
+++ b/.changes/observability-unification.added.md
@@ -0,0 +1 @@
+统一可观测性记录与查询：默认写入内置信息库，可选择 Loki 双向后端；提供精确 Peer 查询/摄取能力、Job 生命周期和操作摘要，并明确外部故障、历史范围与资源关闭责任。
diff --git a/.env.example b/.env.example
index cc8c8c6..dc041cf 100644
--- a/.env.example
+++ b/.env.example
@@ -1,6 +1,6 @@
-# Server Configuration
-HOST=0.0.0.0
-PORT=8000
+# Server Configuration
+HOST=0.0.0.0
+PORT=8000
 PEER_ID=00000000-0000-4000-8000-000000000002
 PEER_NAME=core-py-development
 
@@ -22,20 +22,18 @@ INKCRE_DATABASE_ENVIRONMENT=development
 # Set `true` if your db will scale to 0 when idle
 # This will disable listen to pgsql notifications and enable pool_pre_ping
 DATABASE_SCALE_0=false
-
-# Logging Configuration (Optional)
-# postgresql, logtail
-OBSRV__LOGGING_BACKEND=postgresql 
-# level lower than this will not be sent to backend
-OBSRV__LOGGING_BACKEND_LEVEL=20 
-
-# Logtail As Logging Backend Configuration
-OBSRV__LOGTAIL_SOURCE_TOKEN=your_token_here
-OBSRV__LOGTAIL_HOST=https://in.logtail.com
-
-# PostgreSQL Configuration (Docker Deployment)
-POSTGRES_DB=inkcre
-POSTGRES_USER=postgres
+
+# Ordinary log threshold; backend selection lives in deployment config
+# inkcre.observability.backend / inkcre.observability.backend.v1
+OBSRV__LOGGING_BACKEND_LEVEL=20
+# Private Loki configuration, only used when the deployment selects Loki.
+# OBSRV__LOKI_URL=https://your-loki-host
+# OBSRV__LOKI_HEADERS='{"Authorization":"Basic YOUR_ENCODED_VALUE"}'
+OBSRV__EXTERNAL_CONTENT_ENABLED=false
+
+# PostgreSQL Configuration (Docker Deployment)
+POSTGRES_DB=inkcre
+POSTGRES_USER=postgres
 POSTGRES_PASSWORD=postgres
 POSTGREST_PORT=3000
 CORE_PORT=8000
diff --git a/app/observability.py b/app/observability.py
index 7b31e1e..87edb35 100644
--- a/app/observability.py
+++ b/app/observability.py
@@ -2,6 +2,12 @@
 
 import asyncio
 
+import pydantic
+
+from inkcre_core.context import current_core
+from inkcre_core.libs.obsrv.backends import LokiBackend, PostgreSQLBackend
+from inkcre_core.libs.obsrv.runtime import ObservationRuntime
+
 from opentelemetry.trace import SpanKind
 from starlette.types import ASGIApp, Message, Receive, Scope, Send
 
@@ -10,7 +16,14 @@
   DeploymentConfigService,
 )
 from app.settings import settings
-from inkcre_core.schemas.observability import CONFIG_KEY, CONFIG_SCHEMA, ObservabilityConfig
+from inkcre_core.schemas.observability import (
+  CONFIG_KEY,
+  CONFIG_SCHEMA,
+  ObservabilityConfig,
+  BACKEND_CONFIG_KEY,
+  BACKEND_CONFIG_SCHEMA,
+  ObservabilityBackendConfig,
+)
 from app.version import APPLICATION_VERSION
 from inkcre_core.libs.obsrv.main import get_logger
 from inkcre_core.libs.obsrv.telemetry import is_enabled, operation, start_telemetry
@@ -29,24 +42,67 @@
 )
 
 
+DeploymentConfigManager.register_schema(
+  BACKEND_CONFIG_SCHEMA, ObservabilityBackendConfig, keys=(BACKEND_CONFIG_KEY,)
+)
+
+
 async def initialize_telemetry() -> None:
-  """Read once after database admission; telemetry cannot fail business readiness."""
+  """Bind the selected recording/query backend; OTLP is an independent extra exit."""
+  core = current_core()
+  backend = None
+  selected = "postgresql"
+  error_message = None
+  config = None
+  try:
+    async with asyncio.timeout(5):
+      selection = await DeploymentConfigService.get(BACKEND_CONFIG_KEY)
+      if selection is not None and not isinstance(selection, ObservabilityBackendConfig):
+        raise ValueError("unexpected observability backend schema")
+      selected = selection.backend if selection is not None else "postgresql"
+      if selected == "loki" or settings.obsrv.telemetry_enabled:
+        config = await DeploymentConfigService.get(CONFIG_KEY)
+    if selected == "postgresql":
+      backend = PostgreSQLBackend(core.sessions)
+    else:
+      if not isinstance(config, ObservabilityConfig) or settings.obsrv.loki_url is None:
+        raise ValueError("external backend needs deployment identity and private URL")
+      url = settings.obsrv.loki_url
+      if url.username or url.password or url.query or url.fragment:
+        raise ValueError("private Loki URL cannot contain credentials/query/fragment")
+      headers = pydantic.TypeAdapter(dict[str, str]).validate_json(
+        settings.obsrv.loki_headers.get_secret_value()
+      )
+      backend = LokiBackend(
+        url=str(url), deployment_id=str(config.deployment_id), headers=headers
+      )
+  except Exception:
+    # Missing selection is DB. A failed/invalid selection read is not permission
+    # to silently choose DB or redirect writes from an intended external backend.
+    error_message = "Observability configuration unavailable or incomplete"
+  if core.observability is not None:
+    await core.observability.aclose()
+  core.observability = ObservationRuntime(
+    backend,
+    selected=selected,
+    peer_id=str(core.options.peer_id),
+    error=error_message,
+    allow_external_content=settings.obsrv.external_content_enabled,
+  )
+  core.observability.start()
   if not settings.obsrv.telemetry_enabled:
     return
-  attributes: dict[str, str] = {"inkcre.peer.id": str(settings.peer_id)}
-  endpoints: dict[str, str] = {}
-  try:
-    async with asyncio.timeout(2):
-      config = await DeploymentConfigService.get(CONFIG_KEY)
-    if not isinstance(config, ObservabilityConfig):
-      raise ValueError("missing or unexpected observability schema")
-    attributes["inkcre.deployment.id"] = str(config.deployment_id)
-    if config.otlp_http_endpoints is not None:
-      endpoints = config.otlp_http_endpoints.model_dump(mode="json", exclude_none=True)
-  except Exception as error:
-    # Configuration may contain credentials despite the admitted schema. Never render it.
-    get_logger().warning("Telemetry configuration unavailable (%s)", type(error).__name__)
+  if not isinstance(config, ObservabilityConfig):
+    get_logger().warning("Telemetry configuration unavailable")
     return
+  attributes: dict[str, str] = {
+    "inkcre.peer.id": str(settings.peer_id),
+    "inkcre.deployment.id": str(config.deployment_id),
+  }
+  endpoints: dict[str, str] = {}
+  if config.otlp_http_endpoints is not None:
+    endpoints = config.otlp_http_endpoints.model_dump(mode="json", exclude_none=True)
+
   try:
     start_telemetry(
       enabled=True,
diff --git a/app/routes/observability.py b/app/routes/observability.py
new file mode 100644
index 0000000..73e8037
--- /dev/null
+++ b/app/routes/observability.py
@@ -0,0 +1,60 @@
+"""Authenticated, bounded observability capability; no caller-selected destination."""
+
+import asyncio
+
+import fastapi
+from pydantic import ValidationError
+
+from inkcre_core.business.peer import PeerHTTPInbound
+from inkcre_core.libs.obsrv.backends import ObservabilityUnavailable
+from inkcre_core.libs.obsrv.records import ObservationRequest
+from inkcre_core.libs.obsrv.runtime import current_observability
+
+ROUTER = fastapi.APIRouter(tags=["observability"])
+PEER_INBOUND = PeerHTTPInbound(
+  capability="core.observability.v1", method="POST", path="/observability"
+)
+_MAX_BYTES = 256 * 1024
+
+
+@ROUTER.post("/observability")
+async def observability(request: fastapi.Request):
+  runtime = current_observability()
+  if runtime is None:
+    raise fastapi.HTTPException(503, "Observability is not initialized")
+  if runtime.requests_in_flight >= 4:
+    raise fastapi.HTTPException(429, "Observability request capacity reached")
+  runtime.requests_in_flight += 1
+  try:
+    async with asyncio.timeout(15):
+      payload = bytearray()
+      async for chunk in request.stream():
+        payload.extend(chunk)
+        if len(payload) > _MAX_BYTES:
+          raise fastapi.HTTPException(413, "Observability request exceeds 256 KiB")
+      try:
+        body = ObservationRequest.model_validate_json(payload)
+      except ValidationError as error:
+        # Validation details can contain the original body. Do not reflect log contents.
+        raise fastapi.HTTPException(422, "Invalid observability request") from error
+      if body.expected_backend is not None and body.expected_backend != runtime.selected:
+        raise fastapi.HTTPException(409, "Observability backend selection changed")
+      if body.action == "status":
+        return runtime.status()
+      if body.action == "query":
+        assert body.query is not None
+        try:
+          return await runtime.query(body.query)
+        except ObservabilityUnavailable as error:
+          raise fastapi.HTTPException(
+            503, "Selected observability backend is unavailable"
+          ) from error
+      assert body.records is not None
+      if not runtime.status().recording:
+        raise fastapi.HTTPException(503, "Selected observability backend is not recording")
+      accepted = sum(runtime.emit(record) for record in body.records)
+      return {"accepted": accepted, "dropped": len(body.records) - accepted}
+  except TimeoutError as error:
+    raise fastapi.HTTPException(408, "Observability request timed out") from error
+  finally:
+    runtime.requests_in_flight -= 1
diff --git a/docs/40-deployment/observability.md b/docs/40-deployment/observability.md
index 4a02a56..cb01e15 100644
--- a/docs/40-deployment/observability.md
+++ b/docs/40-deployment/observability.md
@@ -1,10 +1,35 @@
-# 按需启用可观测性
+# 可观测性记录、查询与额外 OTLP 出口
 
-新增遥测缺省关闭。现有 stdout、PostgreSQL 日志、`job.<id>` 关联和 Job 日志查询继续工作；显式配置的 `none` 或 Logtail 也继续生效。开启遥测会增加标准 OTLP 元数据出口，不替换 PG writer，不转发历史日志或任意 Python logger。共享行为由[跨 Peer 可观测性契约](../_shared/20-product-tdd/observability-contract.md)拥有。
+基础观测默认使用内置信息库，Job 生命周期、操作摘要及已配置级别的日志由同一 Core 入口采集。部署选择外部 Loki 后，基础记录写入和查询都使用 Loki；故障不会退回 PostgreSQL。额外 OTLP Trace/指标出口仍由本地开关控制，不是基础查询 API。共享行为由[跨 Peer 可观测性契约](../_shared/20-product-tdd/observability-contract.md)拥有。
+
+## 基础后端与统一查询
+
+公共选择记录为 `configs` 的 key `inkcre.observability.backend`、schema `inkcre.observability.backend.v1`，value 为 `{"backend":"postgresql"}` 或 `{"backend":"loki"}`。缺少选择记录时使用 PostgreSQL；配置读取失败或非法时不启动基础 writer，不擅自当作缺省。它独立于旧 `inkcre.observability.v1`，避免旧 Peer 的严格 schema 拒绝新字段。普通内置使用无需生成额外部署身份。
+
+Loki 是首个双向外部实现，使用 `/loki/api/v1/push` 和 `/loki/api/v1/query_range`，不依赖供应商 SDK。服务端还需要旧身份记录中的 `deployment_id`（用于固定部署流隔离）及私密配置：
+
+```dotenv
+OBSRV__LOKI_URL=https://your-loki-host
+OBSRV__LOKI_HEADERS='{"Authorization":"Basic YOUR_ENCODED_VALUE"}'
+# 默认不将普通日志原文复制到外部；确需启用时先确定访问、保留与删除政策。
+OBSRV__EXTERNAL_CONTENT_ENABLED=false
+```
+
+URL 不带凭据、query 或 fragment；私密 headers 不进入共享 value、浏览器或错误响应。Backend 选择是唯一存储选择源，旧 `OBSRV__LOGGING_BACKEND`/Logtail writer 配置不再选择当前存储；旧 Logtail 配置会给出迁移提示。`OBSRV__LOGGING_BACKEND_LEVEL` 继续控制普通日志级别，最低生命周期事件不依赖该阈值。切换后重启 Peer/重建客户端，历史记录留在旧后端，不自动搬迁、删除或合并查询。
+
+`core.observability.v1` 通过已有 Peer HTTP 协议公告 `POST /observability`，复用 Peer JWT；客户端不能传上游 URL 或认证头。请求 action 为 `status`、`query`、`ingest`。外部委派携带 `expected_backend`，与提供方实例不符返回 409，防止选择改变后误写其它后端。内置消费者可直接通过数据库协议读取，不以 Python Peer 在线为前提。
+
+查询时间窗为 `[since, until)`，最长 31 天、每次 1–500 条。所有 severity、kind、Job、Trace、Peer、operation、文本过滤在后端截断之前执行；文本按字面匹配，不是 LogQL。返回 records、truncated、时间窗及当前 backend，按时间倒序。截断时缩小时间或过滤范围；没有会跳过同刻记录的虚假 cursor。旧 PG 日志保留原 body/attributes 和 `job.<id>` 关联，未产生的历史事件不会补写。
+
+每个 Core 实例最多四个 HTTP 能力请求、15 秒总预算；query 最多四个并发、10 秒预算。摄取最多 100 条/256 KiB，记录说明最多 8192 字符、64 项标量元数据、每项文本最多 1024 字符。状态中的 configured/queryable 代表已具备配置/能力，不是健康探测；dropped 是本实例已知损失，不是部署总数。接受摄取只表示进入队列，不保证持久化。外部错误返回固定诊断，不回显上游响应或密钥。
+
+标准记录 kind 为 log/event/span；span 是耗时/结果操作摘要，不冒充完整 Trace 树。severity_number 继续使用 OTEL 1–24（INFO 9、WARNING 13、ERROR 17）。Job 只有领取成功才记开始，数据库实际关闭后才记终态；失败/超时及失败操作记录 ERROR，取消为 INFO。AI 用量缺失与真零分别保留。选择 Loki 不自动扩大原文权限：默认只保留声明元数据，普通日志 body 替换为固定 `application.log`；显式启用内容另受 owner 的访问和保留政策约束。
+
+SDK 消费者可以创建 `ObservationRuntime(PostgreSQLBackend(core.sessions), selected="postgresql", peer_id=...)` 并赋给 `core.observability`，在自己的异步生命周期中显式调用 `start()`。LokiBackend 接受显式 URL、部署身份和私密 headers；传入的 HTTP client 是借用资源，库不会关闭它。`Core.aclose()` 收束自己的 writer，再关闭自建数据库；多 Core 不共享队列或目标。普通 SDK 导入不启动任何记录任务。
 
 ## 配置与启停
 
-每个 Python Peer 在运行环境或本地 `.env` 设置 `OBSRV__TELEMETRY_ENABLED=true` 才启用。配置 endpoint、认证头或安装 SDK 都不能自动打开它。修改后重启进程；没有后台配置探测或热重载。关闭态不初始化新增 provider、导出线程或队列，也不查询新增共享配置。
+每个 Python Peer 在运行环境或本地 `.env` 设置 `OBSRV__TELEMETRY_ENABLED=true` 才启用。配置 endpoint、认证头或安装 SDK 都不能自动打开它。修改后重启进程；没有后台配置探测或热重载。关闭时不初始化额外 OTLP provider 或导出线程，基础后端仍按上述选择记录和查询。
 
 部署 owner 在数据库已完成初始化后运行 `pdm run python scripts/observability.py`。该命令只在 `configs` 缺少 `inkcre.observability` 时插入一次 UUID；重复运行保留已存在的身份和配置，不能用于重置克隆后的身份。它不会启用任何 Peer。独立 preview、数据库克隆或新 owner 的部署，在启用前显式更换 deployment ID。
 
@@ -12,7 +37,6 @@
 
 ```dotenv
 OBSRV__TELEMETRY_ENABLED=true
-OBSRV__LOGGING_BACKEND=postgresql
 OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=https://your-otlp-host/otlp/v1/traces
 OTEL_EXPORTER_OTLP_LOGS_ENDPOINT=https://your-otlp-host/otlp/v1/logs
 OTEL_EXPORTER_OTLP_METRICS_ENDPOINT=https://your-otlp-host/otlp/v1/metrics
@@ -21,7 +45,7 @@ OTEL_EXPORTER_OTLP_HEADERS="Authorization=Basic%20YOUR_ENCODED_VALUE"
 
 当前出口为标准 OTLP/HTTP protobuf。三个 endpoint 使用完整地址，不自动添加路径；未配置的信号不导出。运行环境和 `.env` 的 per-signal endpoint 优先于共享默认值。认证支持公共 `OTEL_EXPORTER_OTLP_HEADERS` 和优先级更高的 `OTEL_EXPORTER_OTLP_{TRACES,LOGS,METRICS}_HEADERS`。服务名称、版本、运行实例由本实现提供，不读取任意 `OTEL_RESOURCE_ATTRIBUTES`，避免将配置原文引入 Resource。
 
-Python 在数据库准入、Peer 注册之后读取一次共享配置，读取限时两秒；配置缺失、无效或读取失败时暂停新增遥测，不影响业务 readiness。单信号初始化失败只报告不含原始配置的本地诊断，不改变业务 readiness 或 PG 生命周期。配置恢复后需要重启，不自动轮询重试。
+Python 在数据库准入、Peer 注册之后读取一次共享配置，读取限时五秒；基础 backend 选择和额外 OTLP 初始化分别报告失败，不影响业务 readiness。单信号初始化失败只报告不含原始配置的本地诊断，不改变业务 readiness 或 PG 生命周期。配置恢复后需要重启，不自动轮询重试。
 
 Grafana Cloud Free 是当前首个部署目标。`Authorization` 的 Basic 值必须是 `base64(instance ID:token)`，不能在 `Basic%20` 后直接拼原始 `glc_` token。Cloud Portal 的 stack 页面中，使用 OpenTelemetry 的 Configure 获取基础 endpoint 和认证变量；per-signal 地址应追加各自的 `/v1/traces`、`/v1/logs`、`/v1/metrics`，Python 的 Basic 空格使用 `%20`。[Grafana 官方说明](https://grafana.com/docs/grafana-cloud/send-data/otlp/send-data-otlp/)。账号准入、实际免费额度和查询读回必须在目标 stack 单独验收；配置示例不证明云端已通过。不要为本接入启用付费计划或依赖试用专属功能。
 
@@ -58,7 +82,7 @@ GenAI 映射固定到 `semantic-conventions-genai` 修订 `e07f4ebacb08f56db8c4c
 
 基础模式不导出 prompt、答案、工具输入输出、query、provider 配置、任意异常消息或 stack。配置中的模型标识受字符集与长度限制，provider 任意回显 ID/model 不复制；ToolCall ID 使用 SHA-256 摘要关联。最终引用最多导出前 128 项中的有效 ID，同时记录总数和未导出数量；不截断原业务结果。
 
-首批来源事件分别是 `inkcre.retrieval.candidates`、`inkcre.entity.read` 和 `inkcre.agent.query.result`。最后一个事件描述选入 Job 内存 state 的结果，不提前声称最终数据库 close 已成功。当前点位覆盖 Agent 的 retrieve/get_entities/Resolver Tool 与 Agent Query Sink；直接非 Agent 检索、图遍历和 Resolver 内部传递读取尚无完整来源事件覆盖。关联不能证明模型确实依据该实体或答案正确，也不提供长期内容快照。原有 `agent_debug` 与 PG/Logtail 内容行为保持，不能将新出口的元数据策略泛称为整个部署没有原文副本。
+首批来源事件分别是 `inkcre.retrieval.candidates`、`inkcre.entity.read` 和 `inkcre.agent.query.result`。最后一个事件描述选入 Job 内存 state 的结果，不提前声称最终数据库 close 已成功。当前点位覆盖 Agent 的 retrieve/get_entities/Resolver Tool 与 Agent Query Sink；直接非 Agent 检索、图遍历和 Resolver 内部传递读取尚无完整来源事件覆盖。关联不能证明模型确实依据该实体或答案正确，也不提供长期内容快照。原有 `agent_debug` 及历史存储可能包含内容，不能将新外部出口的元数据策略泛称为整个部署没有原文副本。
 
 ## 故障、关闭与迁移后端
 
@@ -66,8 +90,8 @@ GenAI 映射固定到 `semantic-conventions-genai` 修订 `e07f4ebacb08f56db8c4c
 
 Span 和 log 各使用 SDK 的 256 条有界队列、最多 256 条一批、五秒周期；metric 为三十秒 push，不增加 scrape 或唤醒归零实例的采集轮询。HTTP 导出 timeout 采用标准 `OTEL_EXPORTER_OTLP_TIMEOUT`，缺省十秒；按信号的 `OTEL_EXPORTER_OTLP_{TRACES,LOGS,METRICS}_TIMEOUT` 优先，允许有限正数且不超过三十秒。真实 Cloud 首次 TLS 往返超过一秒，固定一秒会误丢弃可用出口的数据；故障实验使用显式一秒配置。源端 SDK warning/导出失败只输出固定本地诊断，不转发 SDK 错误体或凭据。
 
-正常关闭先完成业务资源和 PG writer，再停止接受新记录，并发关闭 trace/log SDK，最后关闭 metrics 以采集前两者的最终导出与丢弃计数。SDK 1.45 的 trace/log worker 等待窗口为三十秒，metrics shutdown 等待参数为对应信号 timeout 加两秒（缺省十二秒）；这些不是网络总时限或交付保证。不先执行一次会忽略 timeout 的 force_flush。平台提前杀进程、请求后冻结或断网时允许丢失尾部记录；使用该平台前需要验证实际终止宽限期，不能延长每个业务请求来掩盖此限制。
+正常关闭先完成业务资源，并发关闭额外 trace/log SDK，最后关闭 metrics；Core 随后停止接受基础记录，最多五秒排空有界队列、两秒关闭后端，再释放数据库。SDK 1.45 的 trace/log worker 等待窗口为三十秒，metrics shutdown 等待参数为对应信号 timeout 加两秒（缺省十二秒）；这些不是网络总时限或交付保证。不先执行一次会忽略 timeout 的 force_flush。平台提前杀进程、请求后冻结或断网时允许丢失尾部记录；使用该平台前需要验证实际终止宽限期，不能延长每个业务请求来掩盖此限制。
 
-更换后端只调整标准出口和服务端认证，重新验证 ID、Links、AI unknown/zero 与聚合；Grafana 的查询、datasource、面板及诊断链接留在部署/消费侧。历史数据、查询和面板不承诺无成本迁移。故障回退是关闭本地遥测开关后重启，PG 无需恢复或迁移，业务 schema 与历史记录保持。
+更换后端只调整标准出口和服务端认证，重新验证 ID、Links、AI unknown/zero 与聚合；Grafana 的查询、datasource、面板及诊断链接留在部署/消费侧。历史数据、查询和面板不承诺无成本迁移。关闭本地 OTLP 开关只关闭额外出口，不改变基础存储。基础存储切回 PostgreSQL 必须显式修改公共选择并重建实例，不能把外部读取失败当作空结果；历史记录不自动迁移。
 
 查询验收使用独立 Viewer service account，经 Grafana 的 datasource query/proxy API 读取 Tempo、Loki、Prometheus；`GRAFANA_URL` 与 `GRAFANA_SERVICE_ACCOUNT_TOKEN` 仅供部署工具使用，不进入应用采集。写入成功必须再核对存储后的 ID、Links、AI 来源/未知值与指标。若本机 Python 没有默认 CA 路径，在进程环境用标准 `OTEL_EXPORTER_OTLP_CERTIFICATE` 指向可信 CA 文件，不能关闭 TLS 校验。
diff --git a/docs/40-deployment/runtime-orchestration.md b/docs/40-deployment/runtime-orchestration.md
index a13a85a..8c91f29 100644
--- a/docs/40-deployment/runtime-orchestration.md
+++ b/docs/40-deployment/runtime-orchestration.md
@@ -135,18 +135,11 @@ Readiness 复用 `app/database_contract` 的同步 psycopg 检查，在 `asyncio
 取消 HTTP 等待不会中断已经运行的 worker，但连接仍由该 worker 的 context manager 关闭。
 业务数据库访问与日志写入均使用原生 async，不把这个例外推广到业务查询。
 
-### 8. PostgreSQL logging has a bounded async lifecycle
+### 8. Core owns a bounded observability lifecycle
 
-同步 logging 调用只捕获 LogModel 和当时的 trace/span，再写入线程安全的有界队列。
-`lifespan` 启动单个异步 writer，每批最多 100 条，在独立事务中写入，绝不加入业务事务。
-队列最多保存 1024 条；满时丢弃 backend 记录并向 stderr 报告，控制台 handler 保持独立。
-数据库写失败时丢弃该批并向 stderr 报告异常类型，不递归调用日志系统。
+同步 logging 和结构化采集都捕获当前 Core 的关联与后端，放入本实例的 1024 条队列；每批最多 100 条，独立于业务事务写入选定 PostgreSQL 或 Loki。失败丢弃并记录无敏感信息的状态，不重试业务、不回退其它存储。只有 App 完成公共选择与私密配置装配后才启动 writer。
 
-关闭时先停止接受新 backend 记录，最多等待五秒排空；超时取消写入并丢弃剩余记录，之后才释放
-连接池。日志仍为 best-effort telemetry，不承诺进程崩溃后的交付。导入模块不启动 writer、不连接数据库。
-
-新增标准遥测在显式启用时，于数据库准入和 Peer 注册后初始化；PG 路径保持独立。
-元数据来源、OTLP 出口、有界关闭和浏览器认证转发见[可观测性配置](observability.md)。
+关闭时先停止接受新记录，最多五秒排空，再限时关闭拥有的后端，随后释放数据库。借用的 session factory/HTTP client 不由后端释放。导入模块不启动 writer、不连接数据库，额外 OTLP 仍独立按需启用。具体配置、查询与历史边界见[可观测性配置](observability.md)。
 
 ### 9. Each Peer may own a scheduler
 
diff --git a/docs/_shared b/docs/_shared
index b333462..33f5a13 160000
--- a/docs/_shared
+++ b/docs/_shared
@@ -1 +1 @@
-Subproject commit b33346225b669a6ac93fcac12a135178a46533c8
+Subproject commit 33f5a13c4ec4ba5b3cad73a0e8f3a097139458de
diff --git a/inkcre_core/context.py b/inkcre_core/context.py
index 5013eb6..e9de455 100644
--- a/inkcre_core/context.py
+++ b/inkcre_core/context.py
@@ -13,6 +13,8 @@
 from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine
 
 if TYPE_CHECKING:
+  from .libs.obsrv.runtime import ObservationRuntime
+  from .libs.obsrv.telemetry import _Runtime
   from .business.job import _Execution
   from .business.peer.contracts import PeerInbound, PeerOutboundFactory
   from .business.source.main import SourceBase
@@ -72,6 +74,8 @@ def __init__(
     self.active_jobs: dict[int, _Execution] = {}
     self.pending_jobs: dict[int, asyncio.Task[bool]] = {}
     self.accepting_jobs = False
+    self.observability: ObservationRuntime | None = None
+    self.telemetry: _Runtime | None = None
     self._closed = False
 
   @contextmanager
@@ -98,6 +102,12 @@ async def aclose(self) -> None:
       self.sources.clear()
       self.inbounds.clear()
       self.outbounds.clear()
+      from .libs.obsrv.telemetry import close_telemetry
+
+      await close_telemetry()
+      if self.observability is not None:
+        await self.observability.aclose()
+        self.observability = None
       if self._owns_engine:
         await self.engine.dispose()
       self._closed = True
diff --git a/inkcre_core/libs/obsrv/backends.py b/inkcre_core/libs/obsrv/backends.py
new file mode 100644
index 0000000..e446cd8
--- /dev/null
+++ b/inkcre_core/libs/obsrv/backends.py
@@ -0,0 +1,277 @@
+"""Explicit PostgreSQL and Loki storage; neither selects an application backend."""
+
+from collections.abc import Sequence
+from datetime import datetime, timezone
+import json
+import re
+from typing import Protocol
+
+import httpx
+from pydantic import BaseModel, ConfigDict
+import sqlalchemy as sa
+from sqlmodel import col
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from .log_record import LogModel
+from .records import (
+  BackendName,
+  EVENT_KEY,
+  ID_KEY,
+  KIND_KEY,
+  ObservationPage,
+  ObservationQuery,
+  ObservationRecord,
+)
+
+
+def _nanoseconds(value: datetime) -> int:
+  delta = value - datetime(1970, 1, 1, tzinfo=timezone.utc)
+  return (delta.days * 86400 + delta.seconds) * 1_000_000_000 + delta.microseconds * 1000
+
+
+class ObservabilityUnavailable(RuntimeError):
+  """Selected backend failed; callers must not reinterpret this as empty results."""
+
+
+class ObservationBackend(Protocol):
+  name: BackendName
+
+  async def write(self, records: Sequence[ObservationRecord]) -> None: ...
+  async def query(self, query: ObservationQuery) -> ObservationPage: ...
+  async def aclose(self) -> None: ...
+
+
+class PostgreSQLBackend:
+  name: BackendName = "postgresql"
+
+  def __init__(self, sessions: async_sessionmaker[AsyncSession]):
+    self.sessions = sessions
+
+  async def write(self, records: Sequence[ObservationRecord]) -> None:
+    async with self.sessions.begin() as session:
+      session.add_all(
+        [
+          LogModel(
+            timestamp=record.timestamp,
+            severity_number=record.severity_number,
+            severity_text=record.severity_text,
+            body=record.body,
+            trace_id=record.trace_id,
+            span_id=record.span_id,
+            attributes={
+              **record.attributes,
+              KIND_KEY: record.kind,
+              EVENT_KEY: record.event_name,
+              ID_KEY: record.id,
+            },
+          )
+          for record in records
+        ]
+      )
+
+  async def query(self, query: ObservationQuery) -> ObservationPage:
+    statement = sa.select(LogModel).where(
+      col(LogModel.timestamp) >= query.since,
+      col(LogModel.timestamp) < query.until,
+    )
+    attrs = col(LogModel.attributes)
+    if query.kind is not None:
+      statement = statement.where(
+        sa.func.coalesce(attrs[KIND_KEY].astext, "log") == query.kind
+      )
+    if query.severity_min is not None:
+      statement = statement.where(col(LogModel.severity_number) >= query.severity_min)
+    if query.job_id is not None:
+      statement = statement.where(
+        sa.or_(
+          attrs["inkcre.job.id"].astext == str(query.job_id),
+          attrs["job_id"].astext == str(query.job_id),
+          col(LogModel.trace_id) == f"job.{query.job_id}",
+        )
+      )
+    if query.trace_id is not None:
+      statement = statement.where(col(LogModel.trace_id) == query.trace_id)
+    for key, value in (
+      ("inkcre.peer.id", str(query.peer_id) if query.peer_id is not None else None),
+      ("inkcre.operation", query.operation),
+    ):
+      if value is not None:
+        statement = statement.where(attrs[key].astext == value)
+    if query.text:
+      escaped = query.text.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+      statement = statement.where(col(LogModel.body).ilike(f"%{escaped}%", escape="\\"))
+    statement = statement.order_by(
+      col(LogModel.timestamp).desc(), col(LogModel.id).desc()
+    ).limit(query.limit + 1)
+    async with self.sessions() as session:
+      rows = (await session.execute(statement)).scalars().all()
+    # Historical PostgreSQL records may contain arbitrary nested logger extras.
+    # Preserve them for display; new ingress is validated by ObservationRecord.
+    records = [
+      ObservationRecord.model_construct(
+        id=str(row.attributes.get(ID_KEY) or f"pg.{row.id}"),
+        timestamp=row.timestamp.astimezone(timezone.utc),
+        kind=row.attributes.get(KIND_KEY, "log"),
+        event_name=row.attributes.get(EVENT_KEY),
+        severity_number=row.severity_number,
+        severity_text=row.severity_text,
+        body=row.body,
+        trace_id=row.trace_id,
+        span_id=row.span_id,
+        attributes={
+          k: v for k, v in row.attributes.items() if k not in (ID_KEY, KIND_KEY, EVENT_KEY)
+        },
+      )
+      for row in rows[: query.limit]
+    ]
+    return ObservationPage(
+      records=records,
+      truncated=len(rows) > query.limit,
+      since=query.since,
+      until=query.until,
+      backend=self.name,
+    )
+
+  async def aclose(self) -> None:
+    # The consumer owns the session factory and engine.
+    pass
+
+
+class _LokiStream(BaseModel):
+  model_config = ConfigDict(extra="ignore")
+  values: list[tuple[str, str]]
+
+
+class _LokiData(BaseModel):
+  model_config = ConfigDict(extra="ignore")
+  resultType: str
+  result: list[_LokiStream]
+
+
+class _LokiResponse(BaseModel):
+  model_config = ConfigDict(extra="ignore")
+  status: str
+  data: _LokiData
+
+
+class LokiBackend:
+  """Loki HTTP API with one deployment stream and private application-owned auth.
+
+  The optional client is borrowed. URL and credentials never enter query payloads
+  or status output. This stores normalized records, not arbitrary OTLP signals.
+  """
+
+  name: BackendName = "loki"
+
+  def __init__(
+    self,
+    *,
+    url: str,
+    deployment_id: str,
+    headers: dict[str, str] | None = None,
+    client: httpx.AsyncClient | None = None,
+  ):
+    self.url = url.rstrip("/")
+    self.deployment_id = deployment_id
+    self._owns_client = client is None
+    self.client = client or httpx.AsyncClient(
+      headers=headers,
+      timeout=10,
+      follow_redirects=False,
+    )
+
+  async def write(self, records: Sequence[ObservationRecord]) -> None:
+    # Flat indexing fields are transport projections; the canonical record is nested.
+    values = []
+    for record in records:
+      projected = {
+        "record": record.model_dump(mode="json"),
+        "kind": record.kind,
+        "severity": record.severity_number,
+        "trace": record.trace_id or "",
+        "peer": record.attributes.get("inkcre.peer.id", ""),
+        "job": str(
+          record.attributes.get("inkcre.job.id", record.attributes.get("job_id", ""))
+        ),
+        "operation": record.attributes.get("inkcre.operation", ""),
+        "body": record.body,
+      }
+      values.append([str(_nanoseconds(record.timestamp)), json.dumps(projected)])
+    response = await self.client.post(
+      self.url + "/loki/api/v1/push",
+      json={
+        "streams": [
+          {
+            "stream": {"service_name": "inkcre", "deployment_id": self.deployment_id},
+            "values": values,
+          }
+        ]
+      },
+    )
+    if response.status_code not in (200, 204):
+      raise ObservabilityUnavailable("Selected observability backend rejected records")
+
+  async def query(self, query: ObservationQuery) -> ObservationPage:
+    selector = (
+      '{service_name="inkcre",deployment_id=' + json.dumps(self.deployment_id) + "} | json"
+    )
+    for key, value in (
+      ("kind", query.kind),
+      ("trace", query.trace_id),
+      ("peer", str(query.peer_id) if query.peer_id is not None else None),
+      ("operation", query.operation),
+    ):
+      if value is not None:
+        selector += f" | {key} = {json.dumps(value)}"
+    if query.job_id is not None:
+      selector += (
+        f" | job = {json.dumps(str(query.job_id))}"
+        f" or trace = {json.dumps(f'job.{query.job_id}')}"
+      )
+    if query.severity_min is not None:
+      selector += f" | severity >= {query.severity_min}"
+    if query.text:
+      # The line contains JSON, so filtering the decoded body avoids matching keys.
+      selector += " | body =~ " + json.dumps("(?is).*" + re.escape(query.text) + ".*")
+    async with self.client.stream(
+      "GET",
+      self.url + "/loki/api/v1/query_range",
+      params={
+        "query": selector,
+        "start": str(_nanoseconds(query.since)),
+        "end": str(_nanoseconds(query.until) - 1),
+        "direction": "backward",
+        "limit": query.limit + 1,
+      },
+    ) as response:
+      if response.status_code != 200:
+        raise ObservabilityUnavailable("Selected observability backend query failed")
+      body = bytearray()
+      async for chunk in response.aiter_bytes():
+        body.extend(chunk)
+        if len(body) > 8 * 1024 * 1024:
+          raise ObservabilityUnavailable(
+            "Selected observability backend response exceeded limit"
+          )
+    result = _LokiResponse.model_validate_json(body)
+    if result.status != "success" or result.data.resultType != "streams":
+      raise ObservabilityUnavailable(
+        "Selected observability backend returned an invalid result"
+      )
+    records = [
+      ObservationRecord.model_validate(json.loads(line)["record"])
+      for stream in result.data.result
+      for _, line in stream.values
+    ]
+    records.sort(key=lambda record: (record.timestamp, record.id), reverse=True)
+    return ObservationPage(
+      records=records[: query.limit],
+      truncated=len(records) > query.limit,
+      since=query.since,
+      until=query.until,
+      backend=self.name,
+    )
+
+  async def aclose(self) -> None:
+    if self._owns_client:
+      await self.client.aclose()
diff --git a/inkcre_core/libs/obsrv/main.py b/inkcre_core/libs/obsrv/main.py
index 72f5b0c..f291b79 100644
--- a/inkcre_core/libs/obsrv/main.py
+++ b/inkcre_core/libs/obsrv/main.py
@@ -29,29 +29,15 @@ def setup_obsrv(config: ObsrvSetting) -> logging.Logger:
   console_handler.setFormatter(console_formatter)
   LOGGER.addHandler(console_handler)
 
-  # Add backend-specific handlers
-  backend = config.logging_backend
-  if backend == "logtail":
-    from .log_handler_logtail import LogtailHandler
-
-    logtail_wrapper = LogtailHandler(
-      source_token=config.logtail_source_token,
-      host=config.logtail_host,
+  # Destination selection belongs to the bound Core, never this global logger.
+  from .runtime import ObservationLogHandler
+
+  LOGGER.addHandler(ObservationLogHandler(level=config.logging_backend_level))
+  if config.logging_backend == "logtail":
+    LOGGER.warning(
+      "Legacy Logtail writer settings no longer select storage; configure "
+      "inkcre.observability.backend and a supported read/write backend"
     )
-    handler = logtail_wrapper.get_handler()
-    if handler:
-      LOGGER.addHandler(handler)
-      LOGGER.info("Logtail logging enabled")
-    else:
-      LOGGER.warning("Logtail logging configured but not available")
-  elif backend == "postgresql":
-    from .log_handler_postgresql import PostgreSQLHandler
-
-    pg_handler = PostgreSQLHandler(level=config.logging_backend_level)
-    LOGGER.addHandler(pg_handler)
-    LOGGER.info("PostgreSQL logging enabled")
-  elif backend not in (None, "", "none"):
-    LOGGER.warning(f"Unknown logging backend: {backend}")
 
   return LOGGER
 
@@ -67,19 +53,11 @@ def get_logger() -> logging.Logger:
 
 
 def start_obsrv() -> None:
-  from .log_handler_postgresql import PostgreSQLHandler
-
-  for handler in LOGGER.handlers:
-    if isinstance(handler, PostgreSQLHandler):
-      handler.start()
+  """Compatibility entry; the App starts its bound runtime after config admission."""
 
 
 async def close_obsrv() -> None:
-  from .log_handler_postgresql import PostgreSQLHandler
   from .telemetry import close_telemetry
 
   await close_telemetry()
-
-  for handler in LOGGER.handlers:
-    if isinstance(handler, PostgreSQLHandler):
-      await handler.aclose()
+  # Core.aclose owns its selected backend and drains it before closing the DB.
diff --git a/inkcre_core/libs/obsrv/records.py b/inkcre_core/libs/obsrv/records.py
new file mode 100644
index 0000000..444f227
--- /dev/null
+++ b/inkcre_core/libs/obsrv/records.py
@@ -0,0 +1,124 @@
+"""Bounded, backend-independent observability records and queries."""
+
+from datetime import datetime, timezone
+from typing import Literal, Self
+import math
+import uuid
+
+from pydantic import (
+  BaseModel,
+  ConfigDict,
+  Field,
+  JsonValue,
+  field_validator,
+  model_validator,
+)
+
+BackendName = Literal["postgresql", "loki"]
+RecordKind = Literal["log", "event", "span"]
+KIND_KEY = "inkcre.record.kind"
+EVENT_KEY = "inkcre.event.name"
+ID_KEY = "inkcre.record.id"
+
+
+class ObservationRecord(BaseModel):
+  model_config = ConfigDict(extra="forbid")
+  id: str = Field(default_factory=lambda: str(uuid.uuid4()), min_length=1, max_length=128)
+  timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))
+  kind: RecordKind = "log"
+  event_name: str | None = Field(default=None, max_length=128)
+  severity_number: int = Field(default=9, ge=1, le=24)
+  severity_text: str = Field(default="INFO", max_length=32)
+  body: str = Field(max_length=8192)
+  trace_id: str | None = Field(default=None, max_length=128)
+  span_id: str | None = Field(default=None, max_length=128)
+  attributes: dict[str, JsonValue] = Field(default_factory=dict, max_length=64)
+
+  @field_validator("timestamp")
+  @classmethod
+  def aware_time(cls, value: datetime) -> datetime:
+    if value.tzinfo is None:
+      raise ValueError("timestamp must include a timezone")
+    return value.astimezone(timezone.utc)
+
+  @field_validator("attributes")
+  @classmethod
+  def bounded_metadata(cls, value: dict[str, JsonValue]) -> dict[str, JsonValue]:
+    for key, item in value.items():
+      if len(key) > 128:
+        raise ValueError("attribute name exceeds 128 characters")
+      values = item if isinstance(item, list) else [item]
+      if len(values) > 32:
+        raise ValueError("attribute array exceeds 32 values")
+      for part in values:
+        if isinstance(part, (dict, list)):
+          raise ValueError("attributes contain only scalar values or scalar arrays")
+        if isinstance(part, str) and len(part) > 1024:
+          raise ValueError("attribute value exceeds 1024 characters")
+        if isinstance(part, float) and not math.isfinite(part):
+          raise ValueError("attribute numbers must be finite")
+    return value
+
+
+class ObservationQuery(BaseModel):
+  model_config = ConfigDict(extra="forbid")
+  since: datetime
+  until: datetime
+  limit: int = Field(default=100, ge=1, le=500)
+  kind: RecordKind | None = None
+  severity_min: int | None = Field(default=None, ge=1, le=24)
+  job_id: int | None = Field(default=None, ge=1)
+  trace_id: str | None = Field(default=None, max_length=128)
+  peer_id: uuid.UUID | None = None
+  operation: str | None = Field(default=None, max_length=128)
+  text: str | None = Field(default=None, max_length=256)
+
+  @field_validator("since", "until")
+  @classmethod
+  def aware_time(cls, value: datetime) -> datetime:
+    if value.tzinfo is None:
+      raise ValueError("query times must include a timezone")
+    return value.astimezone(timezone.utc)
+
+  @model_validator(mode="after")
+  def ordered_window(self) -> Self:
+    if self.since >= self.until:
+      raise ValueError("since must precede until")
+    if (self.until - self.since).total_seconds() > 31 * 86400:
+      raise ValueError("query window cannot exceed 31 days")
+    return self
+
+
+class ObservationPage(BaseModel):
+  records: list[ObservationRecord]
+  truncated: bool
+  since: datetime
+  until: datetime
+  backend: BackendName
+
+
+class ObservationStatus(BaseModel):
+  backend: BackendName
+  configured: bool
+  recording: bool
+  queryable: bool
+  scope: Literal["selected_backend"] = "selected_backend"
+  history: str = "Only the selected backend is queried; prior backend data remains there."
+  dropped: int
+  error: str | None
+
+
+class ObservationRequest(BaseModel):
+  model_config = ConfigDict(extra="forbid")
+  action: Literal["status", "query", "ingest"]
+  expected_backend: BackendName | None = None
+  query: ObservationQuery | None = None
+  records: list[ObservationRecord] | None = Field(default=None, max_length=100)
+
+  @model_validator(mode="after")
+  def action_payload(self) -> Self:
+    if (self.action == "query") != (self.query is not None):
+      raise ValueError("query is required only for the query action")
+    if (self.action == "ingest") != (self.records is not None):
+      raise ValueError("records are required only for the ingest action")
+    return self
diff --git a/inkcre_core/libs/obsrv/runtime.py b/inkcre_core/libs/obsrv/runtime.py
new file mode 100644
index 0000000..fa5d314
--- /dev/null
+++ b/inkcre_core/libs/obsrv/runtime.py
@@ -0,0 +1,278 @@
+"""Instance-owned recording with bounded queues and a selected query backend."""
+
+import asyncio
+from collections.abc import Mapping
+from datetime import datetime, timezone
+import logging
+import uuid
+from queue import Empty, Full, Queue
+from threading import Lock
+from typing import Any
+
+from opentelemetry import trace
+
+from .backends import ObservationBackend, ObservabilityUnavailable
+from .log_record import TRACE_ID, SPAN_ID, ENABLE_LOG_BACKEND, adapt_log_record
+from .records import (
+  BackendName,
+  RecordKind,
+  ObservationQuery,
+  ObservationRecord,
+  ObservationStatus,
+)
+
+# Raw content has a separate permission from selecting an external destination.
+_METADATA = frozenset(
+  {
+    "inkcre.peer.id",
+    "inkcre.job.id",
+    "inkcre.job.status",
+    "inkcre.cron.id",
+    "inkcre.operation",
+    "inkcre.outcome",
+    "inkcre.thread.id",
+    "inkcre.agent.id",
+    "inkcre.tool.id",
+    "inkcre.tool_call.id",
+    "inkcre.resolver.id",
+    "inkcre.block.id",
+    "inkcre.peer.target_id",
+    "error.type",
+    "reason",
+    "outcome",
+    "duration_ms",
+    "input_tokens",
+    "output_tokens",
+    "operation",
+    "job_id",
+    "gen_ai.operation.name",
+    "gen_ai.provider.name",
+    "gen_ai.request.model",
+    "gen_ai.response.model",
+    "gen_ai.usage.input_tokens",
+    "gen_ai.usage.output_tokens",
+    "http.request.method",
+    "http.response.status_code",
+    "http.route",
+    "gen_ai.conversation.id",
+    "gen_ai.tool.name",
+    "inkcre.agent.model_call",
+    "inkcre.agent.model_calls",
+    "inkcre.agent.outcome",
+    "inkcre.agent.tool_call.id_hash",
+    "inkcre.agent.turn",
+    "inkcre.ai.model.id",
+    "inkcre.ai.usage.input.source",
+    "inkcre.ai.usage.output.source",
+    "inkcre.entity.block_ids",
+    "inkcre.entity.read",
+    "inkcre.entity.read.count",
+    "inkcre.entity.read.kind",
+    "inkcre.entity.relation_ids",
+    "inkcre.retrieval.candidate_count",
+    "inkcre.retrieval.candidates",
+    "inkcre.retrieval.method",
+  }
+)
+
+
+def current_observability() -> "ObservationRuntime | None":
+  from inkcre_core.context import _CURRENT
+
+  core = _CURRENT.get(None)
+  return core.observability if core is not None else None
+
+
+class ObservationRuntime:
+  def __init__(
+    self,
+    backend: ObservationBackend | None,
+    *,
+    selected: BackendName,
+    peer_id: str,
+    error: str | None = None,
+    allow_external_content: bool = False,
+  ):
+    self.backend = backend
+    self.selected = selected
+    self.peer_id = peer_id
+    self.error = error
+    self.allow_external_content = allow_external_content
+    self.queue: Queue[ObservationRecord] = Queue(maxsize=1024)
+    self._lock = Lock()
+    self.dropped = 0
+    self._closed = False
+    self.requests_in_flight = 0
+    self._queries = 0
+    self._accepting = False
+    self._task: asyncio.Task | None = None
+
+  def start(self) -> None:
+    if self._closed or self._task is not None:
+      raise RuntimeError("Observability runtime is closed or already started")
+    if self.backend is not None:
+      self._accepting = True
+      self._task = asyncio.create_task(self._write(), name="observability-writer")
+
+  def status(self) -> ObservationStatus:
+    return ObservationStatus(
+      backend=self.selected,
+      configured=self.backend is not None,
+      recording=self._accepting,
+      queryable=self.backend is not None and not self._closed,
+      dropped=self.dropped,
+      error=self.error,
+    )
+
+  def emit(self, record: ObservationRecord) -> bool:
+    if not self._accepting:
+      self.dropped += 1
+      return False
+    attributes = {**record.attributes}
+    if "inkcre.peer.id" not in attributes and len(attributes) == 64:
+      self.dropped += 1
+      self.error = "No metadata capacity for the recording Peer identity"
+      return False
+    attributes.setdefault("inkcre.peer.id", self.peer_id)
+    if self.selected == "loki" and not self.allow_external_content:
+      attributes = {key: value for key, value in attributes.items() if key in _METADATA}
+      record = record.model_copy(
+        update={
+          "body": record.event_name
+          if record.kind != "log" and record.event_name
+          else "application.log",
+          "attributes": attributes,
+        }
+      )
+    else:
+      record = record.model_copy(update={"attributes": attributes})
+    with self._lock:
+      if not self._accepting:
+        self.dropped += 1
+        return False
+      try:
+        self.queue.put_nowait(record)
+      except Full:
+        self.dropped += 1
+        self.error = "Recording queue full; records were dropped"
+        return False
+    return True
+
+  async def _write(self) -> None:
+    assert self.backend is not None
+    while self._accepting or not self.queue.empty():
+      batch = []
+      for _ in range(100):
+        try:
+          batch.append(self.queue.get_nowait())
+        except Empty:
+          break
+      if not batch:
+        await asyncio.sleep(0.1)
+        continue
+      try:
+        await self.backend.write(batch)
+      except Exception:
+        # No recursive logging, credential-bearing errors, retry or fallback store.
+        self.dropped += len(batch)
+        self.error = "Selected backend write failed; records were dropped"
+
+  async def query(self, query: ObservationQuery):
+    if self._closed or self.backend is None:
+      raise ObservabilityUnavailable("Selected observability backend is not configured")
+    if self._queries >= 4:
+      raise ObservabilityUnavailable("Observability query capacity reached")
+    self._queries += 1
+    try:
+      async with asyncio.timeout(10):
+        return await self.backend.query(query)
+    except Exception as error:
+      self.error = "Selected backend query failed"
+      raise ObservabilityUnavailable(self.error) from error
+    finally:
+      self._queries -= 1
+
+  async def aclose(self) -> None:
+    if self._closed:
+      return
+    self._closed = True
+    with self._lock:
+      self._accepting = False
+    if self._task is not None:
+      try:
+        async with asyncio.timeout(5):
+          await self._task
+      except TimeoutError:
+        self.error = "Recording drain timed out; pending records were dropped"
+      finally:
+        self._task = None
+    while not self.queue.empty():
+      self.queue.get_nowait()
+      self.dropped += 1
+    if self.backend is not None:
+      try:
+        async with asyncio.timeout(2):
+          await self.backend.aclose()
+      except Exception:
+        self.error = "Selected backend shutdown did not complete"
+
+
+def record_event(
+  name: str, attributes: Mapping[str, Any], *, kind: RecordKind = "event"
+) -> None:
+  runtime = current_observability()
+  if runtime is None:
+    return
+  span = trace.get_current_span().get_span_context()
+  failure = attributes.get("outcome") in {"error", "timeout"} or attributes.get(
+    "inkcre.job.status"
+  ) in {
+    "failed",
+    "timed_out",
+  }
+  severity = 17 if failure else 9
+  severity_text = "ERROR" if failure else "INFO"
+  try:
+    runtime.emit(
+      ObservationRecord(
+        kind=kind,
+        event_name=name,
+        body=name,
+        severity_number=severity,
+        severity_text=severity_text,
+        trace_id=f"{span.trace_id:032x}" if span.is_valid else TRACE_ID.get(),
+        span_id=f"{span.span_id:016x}" if span.is_valid else SPAN_ID.get(),
+        attributes=dict(attributes),
+      )
+    )
+  except (ValueError, TypeError):
+    runtime.dropped += 1
+    runtime.error = "Invalid observation metadata was dropped"
+
+
+class ObservationLogHandler(logging.Handler):
+  """Resolve the caller's Core at emit time; never capture another instance's sink."""
+
+  def emit(self, record: logging.LogRecord) -> None:
+    runtime = current_observability()
+    if runtime is None or not ENABLE_LOG_BACKEND.get():
+      return
+    try:
+      adapted = adapt_log_record(record)
+      runtime.emit(
+        ObservationRecord(
+          id=str(uuid.uuid4()),
+          timestamp=datetime.fromtimestamp(record.created, timezone.utc),
+          kind="log",
+          event_name=None,
+          severity_number=adapted.severity_number,
+          severity_text=adapted.severity_text,
+          body=adapted.body,
+          trace_id=adapted.trace_id,
+          span_id=adapted.span_id,
+          attributes=adapted.attributes,
+        )
+      )
+    except Exception:
+      runtime.dropped += 1
+      runtime.error = "Application log could not be adapted"
diff --git a/inkcre_core/libs/obsrv/setting.py b/inkcre_core/libs/obsrv/setting.py
index 8e3a334..e9bcfdd 100644
--- a/inkcre_core/libs/obsrv/setting.py
+++ b/inkcre_core/libs/obsrv/setting.py
@@ -2,7 +2,7 @@
 
 from typing import Optional
 
-from pydantic import Field
+from pydantic import Field, SecretStr, AnyHttpUrl
 from pydantic_settings import BaseSettings
 
 
@@ -33,3 +33,14 @@ class ObsrvSetting(BaseSettings):
   logtail_host: Optional[str] = Field(
     default=None, description="Better Stack (Logtail) host URL"
   )
+
+  loki_url: AnyHttpUrl | None = Field(
+    default=None, description="Private Loki base URL for the selected external backend"
+  )
+  loki_headers: SecretStr = Field(
+    default=SecretStr("{}"), description="Private Loki request headers as a JSON object"
+  )
+  external_content_enabled: bool = Field(
+    default=False,
+    description="Explicitly allow original application log content in the external backend",
+  )
diff --git a/inkcre_core/libs/obsrv/telemetry.py b/inkcre_core/libs/obsrv/telemetry.py
index 6343e9d..10f4310 100644
--- a/inkcre_core/libs/obsrv/telemetry.py
+++ b/inkcre_core/libs/obsrv/telemetry.py
@@ -14,6 +14,8 @@
 from urllib.parse import urlsplit
 import uuid
 
+from .runtime import record_event
+
 from opentelemetry.context import Context
 from opentelemetry.trace import INVALID_SPAN, Link, Span, SpanKind, StatusCode, use_span
 from opentelemetry.util.types import AttributeValue
@@ -60,7 +62,11 @@ class _Runtime:
   missing_usage: "Counter | None" = None
 
 
-_runtime: _Runtime | None = None
+def _current_runtime() -> _Runtime | None:
+  from inkcre_core.context import _CURRENT
+
+  core = _CURRENT.get(None)
+  return core.telemetry if core is not None else None
 
 
 def _diagnostic(reason: str) -> None:
@@ -106,12 +112,13 @@ def _endpoint(value: str) -> str:
 
 def is_enabled() -> bool:
   """Whether at least one metadata signal initialized for this runtime."""
-  return _runtime is not None
+  return _current_runtime() is not None
 
 
 def get_export_connection(signal: str) -> tuple[str, dict[str, str], float] | None:
   """Return private endpoint, copied headers and timeout seconds for server forwarding."""
-  connection = _runtime.connections.get(signal) if _runtime is not None else None
+  runtime = _current_runtime()
+  connection = runtime.connections.get(signal) if runtime is not None else None
   return (
     (connection[0], dict(connection[1]), connection[2]) if connection is not None else None
   )
@@ -125,9 +132,11 @@ def start_telemetry(
   endpoints: Mapping[str, str] | None = None,
 ) -> None:
   """Initialize only configured signals; configuration failures never gate readiness."""
-  global _runtime
-  if not enabled or _runtime is not None:
+  if not enabled or _current_runtime() is not None:
     return
+  from inkcre_core.context import current_core
+
+  core = current_core()
 
   _configure_diagnostics()
   try:
@@ -331,7 +340,7 @@ def start_telemetry(
       runtime.connections[signal] = (endpoint, dict(headers), timeout)
 
   if any((runtime.traces, runtime.logs, runtime.metrics)):
-    _runtime = runtime
+    core.telemetry = runtime
   else:
     _diagnostic("no usable per-signal endpoint; telemetry is disabled")
 
@@ -350,8 +359,12 @@ async def close_telemetry() -> None:
   its configured request timeout plus two seconds. These worker wait budgets are
   not transport-wide hard deadlines; freezing or termination can lose records.
   """
-  global _runtime
-  runtime, _runtime = _runtime, None
+  from inkcre_core.context import _CURRENT
+
+  core = _CURRENT.get(None)
+  if core is None:
+    return
+  runtime, core.telemetry = core.telemetry, None
   if runtime is None:
     return
   shutdowns: list[Callable[[], None]] = []
@@ -407,9 +420,28 @@ def operation(
   Exceptions and cancellation propagate unchanged. Their messages, stack traces and
   status descriptions are never recorded by this context manager.
   """
-  runtime = _runtime
+  runtime = _current_runtime()
   if runtime is None:
-    yield Operation(INVALID_SPAN)
+    observation = Operation(INVALID_SPAN)
+    started = time.perf_counter()
+    try:
+      yield observation
+    except BaseException as error:
+      observation.outcome = (
+        "cancelled" if isinstance(error, asyncio.CancelledError) else "error"
+      )
+      raise
+    finally:
+      record_event(
+        name,
+        {
+          **(attributes or {}),
+          "inkcre.operation": name,
+          "outcome": observation.outcome,
+          "duration_ms": (time.perf_counter() - started) * 1000,
+        },
+        kind="span",
+      )
     return
 
   span = INVALID_SPAN
@@ -448,6 +480,16 @@ def operation(
     if observation.outcome != "success":
       _record(partial(span.set_status, StatusCode.ERROR))
     _record(partial(span.set_attribute, "inkcre.outcome", observation.outcome))
+    record_event(
+      name,
+      {
+        **(attributes or {}),
+        "inkcre.operation": name,
+        "outcome": observation.outcome,
+        "duration_ms": (time.perf_counter() - started) * 1000,
+      },
+      kind="span",
+    )
     labels = {"operation": name, "outcome": observation.outcome}
     if runtime.operations is not None:
       _record(partial(runtime.operations.add, 1, labels))
@@ -459,7 +501,8 @@ def operation(
 
 def emit_event(name: str, attributes: Mapping[str, AttributeValue]) -> None:
   """Emit a fixed event name and admitted metadata without a Python logging bridge."""
-  runtime = _runtime
+  record_event(name, attributes)
+  runtime = _current_runtime()
   if runtime is not None and runtime.logger is not None:
     from opentelemetry._logs import SeverityNumber
 
@@ -477,12 +520,20 @@ def record_ai_usage(
   operation: str, input_tokens: int | None, output_tokens: int | None
 ) -> None:
   """Count known nonnegative provider usage and count missing fields separately."""
-  runtime = _runtime
+  known = {
+    kind: value
+    for kind, value in (("input", input_tokens), ("output", output_tokens))
+    if isinstance(value, int) and not isinstance(value, bool) and 0 <= value <= 2**63 - 1
+  }
+  values: dict[str, str | int] = {"inkcre.operation": operation}
+  values.update({kind + "_tokens": value for kind, value in known.items()})
+  record_event("ai.usage", values)
+  runtime = _current_runtime()
   if runtime is None or runtime.tokens is None or runtime.missing_usage is None:
     return
-  for kind, value in (("input", input_tokens), ("output", output_tokens)):
+  for kind in ("input", "output"):
     labels = {"operation": operation, "token.type": kind}
-    if isinstance(value, int) and not isinstance(value, bool) and value >= 0:
-      _record(partial(runtime.tokens.add, value, labels))
+    if kind in known:
+      _record(partial(runtime.tokens.add, known[kind], labels))
     else:
       _record(partial(runtime.missing_usage.add, 1, labels))
diff --git a/inkcre_core/schemas/observability.py b/inkcre_core/schemas/observability.py
index b8e588a..bd35bd3 100644
--- a/inkcre_core/schemas/observability.py
+++ b/inkcre_core/schemas/observability.py
@@ -1,6 +1,7 @@
 """Non-secret deployment telemetry configuration shared by admitted Peers."""
 
 import uuid
+import typing
 import pydantic
 
 CONFIG_KEY = "inkcre.observability"
@@ -35,3 +36,14 @@ class ObservabilityConfig(pydantic.BaseModel):
   diagnostics_url: pydantic.AnyHttpUrl | None = None
 
   _url = pydantic.field_validator("diagnostics_url")(_credential_free_url)
+
+
+BACKEND_CONFIG_KEY = "inkcre.observability.backend"
+BACKEND_CONFIG_SCHEMA = "inkcre.observability.backend.v1"
+
+
+class ObservabilityBackendConfig(pydantic.BaseModel):
+  """Non-secret deployment selection; external credentials remain Peer-local."""
+
+  model_config = pydantic.ConfigDict(extra="forbid")
+  backend: typing.Literal["postgresql", "loki"] = "postgresql"
diff --git a/run.py b/run.py
index 63ca9b9..f429d53 100644
--- a/run.py
+++ b/run.py
@@ -53,6 +53,10 @@
 from app.routes.semantic_retrieval import ROUTER as semantic_retrieval_router
 from app.routes.sink import ROUTER as sink_router
 from app.routes.telemetry import ROUTER as telemetry_router
+from app.routes.observability import (
+  ROUTER as observability_router,
+  PEER_INBOUND as observability_peer_inbound,
+)
 from inkcre_core.business.source import SourceManager
 from inkcre_core.business.cron import CronManager
 from inkcre_core.business.job import JobManager
@@ -151,6 +155,7 @@ async def bootstrap_runtime(app: fastapi.FastAPI) -> None:
     PeerManager.register_inbound(lexical_retrieval_peer_inbound)
     PeerManager.register_inbound(organization_peer_inbound)
     PeerManager.register_inbound(extension_peer_inbound)
+    PeerManager.register_inbound(observability_peer_inbound)
 
   await initialize_telemetry()
 
@@ -385,6 +390,7 @@ async def readiness() -> JSONResponse:
 core_router.include_router(lexical_retrieval_router)
 core_router.include_router(sink_router)
 core_router.include_router(telemetry_router)
+core_router.include_router(observability_router)
 api_app.include_router(core_router)
 
 if __name__ == "__main__":
diff --git a/tests/test_observability_records.py b/tests/test_observability_records.py
new file mode 100644
index 0000000..fccea1c
--- /dev/null
+++ b/tests/test_observability_records.py
@@ -0,0 +1,374 @@
+"""Selected-backend semantics without databases, network or application startup."""
+
+from datetime import datetime, timedelta, timezone
+import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import fastapi
+from fastapi.testclient import TestClient
+import httpx
+import pytest
+from sqlalchemy.dialects import postgresql
+
+from inkcre_core import Core
+from inkcre_core.libs.obsrv.backends import (
+  LokiBackend,
+  PostgreSQLBackend,
+  ObservabilityUnavailable,
+)
+from inkcre_core.libs.obsrv.records import (
+  ObservationQuery,
+  ObservationRecord,
+  ObservationRequest,
+)
+from inkcre_core.libs.obsrv.runtime import ObservationRuntime
+from inkcre_core.libs.obsrv.telemetry import emit_event, operation, record_ai_usage
+
+
+NOW = datetime(2026, 11, 4, 1, 2, 3, 123456, tzinfo=timezone.utc)
+
+
+def test_record_and_query_reject_unbounded_or_ambiguous_input():
+  with pytest.raises(ValueError):
+    ObservationRecord(body="x", attributes={"nested": {"content": "x"}})
+  with pytest.raises(ValueError):
+    ObservationQuery(since=NOW, until=NOW - timedelta(seconds=1))
+  with pytest.raises(ValueError):
+    ObservationQuery(since=NOW, until=NOW + timedelta(days=32))
+  with pytest.raises(ValueError):
+    ObservationRequest(
+      action="ingest", query=ObservationQuery(since=NOW, until=NOW + timedelta(seconds=1))
+    )
+
+
+def test_loki_roundtrip_encodes_filters_and_does_not_leak_auth(async_runner):
+  requests = []
+  record = ObservationRecord(
+    timestamp=NOW,
+    kind="event",
+    event_name="job.closed",
+    body="job.closed",
+    trace_id="job.12",
+    attributes={"inkcre.job.id": 12},
+  )
+  line = None
+
+  def handle(request):
+    nonlocal line
+    requests.append(request)
+    if request.method == "POST":
+      stream = json.loads(request.content)["streams"][0]
+      assert stream["stream"] == {"service_name": "inkcre", "deployment_id": "deployment"}
+      assert stream["values"][0][0] == "1793754123123456000"
+      line = stream["values"][0][1]
+      return httpx.Response(204)
+    assert line is not None
+    query = request.url.params["query"]
+    assert 'job = "12" or trace = "job.12"' in query
+    assert "severity >= 9" in query
+    assert 'kind = "event"' in query
+    # Text is escaped as data in both regexp and LogQL string layers.
+    assert "\\\\." in query
+    assert request.url.params["limit"] == "2"
+    return httpx.Response(
+      200,
+      json={
+        "status": "success",
+        "data": {
+          "resultType": "streams",
+          "result": [
+            {"values": [["1793754123123456000", line], ["1793754123123456000", line]]}
+          ],
+        },
+      },
+    )
+
+  async def scenario():
+    async with httpx.AsyncClient(transport=httpx.MockTransport(handle)) as client:
+      backend = LokiBackend(
+        url="https://logs.example", deployment_id="deployment", client=client
+      )
+      await backend.write([record])
+      page = await backend.query(
+        ObservationQuery(
+          since=NOW - timedelta(seconds=1),
+          until=NOW + timedelta(seconds=1),
+          limit=1,
+          kind="event",
+          severity_min=9,
+          job_id=12,
+          text='a.b\n" | logfmt',
+        )
+      )
+      assert page.records == [record]
+      assert page.truncated and page.backend == "loki"
+      await backend.aclose()
+      assert not client.is_closed  # Borrowed client remains owned by its caller.
+
+  async_runner.run(scenario())
+
+
+def test_external_failure_is_not_empty_or_a_database_fallback(async_runner):
+  async def scenario():
+    async with httpx.AsyncClient(
+      transport=httpx.MockTransport(lambda _: httpx.Response(500, text="private token"))
+    ) as client:
+      backend = LokiBackend(
+        url="https://logs.example", deployment_id="deployment", client=client
+      )
+      runtime = ObservationRuntime(backend, selected="loki", peer_id="peer")
+      runtime.start()
+      assert runtime.emit(
+        ObservationRecord(
+          body="secret content", attributes={"query": "private", "inkcre.job.id": 12}
+        )
+      )
+      with pytest.raises(ObservabilityUnavailable, match="query failed") as error:
+        await runtime.query(ObservationQuery(since=NOW, until=NOW + timedelta(seconds=1)))
+      assert "private" not in str(error.value)
+      await runtime.aclose()
+      assert runtime.dropped == 1
+      assert runtime.status().backend == "loki"
+
+  async_runner.run(scenario())
+
+
+def test_metadata_and_instance_lifetimes_do_not_need_otlp(async_runner, monkeypatch):
+  from inkcre_core.libs.obsrv import telemetry
+
+  monkeypatch.setattr(telemetry, "_current_runtime", lambda: None)
+
+  async def scenario():
+    first = Core("postgresql+psycopg://test:test@127.0.0.1:1/test")
+    second = Core("postgresql+psycopg://test:test@127.0.0.1:1/test")
+    backends = [MagicMock(write=AsyncMock(), aclose=AsyncMock()) for _ in range(2)]
+    for core, backend in zip((first, second), backends):
+      core.observability = ObservationRuntime(
+        backend, selected="loki", peer_id=str(core.options.peer_id)
+      )
+      core.observability.start()
+    assert first.observability is not None
+    assert second.observability is not None
+    with first.bind():
+      emit_event("job.started", {"inkcre.job.id": 12})
+      with operation("job.execute", attributes={"inkcre.job.id": 12}):
+        record_ai_usage("chat", 0, None)
+      first.observability.emit(
+        ObservationRecord(body="secret prompt", attributes={"query": "private"})
+      )
+      with second.bind():
+        emit_event("job.started", {"inkcre.job.id": 24})
+    await first.aclose()
+    first_records = backends[0].write.call_args.args[0]
+    assert [record.event_name for record in first_records] == [
+      "job.started",
+      "ai.usage",
+      "job.execute",
+      None,
+    ]
+    assert first_records[1].attributes["input_tokens"] == 0
+    assert "output_tokens" not in first_records[1].attributes
+    assert first_records[2].attributes["duration_ms"] >= 0
+    assert first_records[3].body == "application.log"
+    assert "query" not in first_records[3].attributes
+    assert second.observability.status().recording
+    await second.aclose()
+    assert backends[1].write.call_args.args[0][0].attributes["inkcre.job.id"] == 24
+
+  async_runner.run(scenario())
+
+
+def test_postgresql_preserves_legacy_job_association_and_filters_before_limit(async_runner):
+  session = MagicMock()
+  session.execute = AsyncMock(return_value=MagicMock())
+  session.execute.return_value.scalars.return_value.all.return_value = [
+    SimpleNamespace(
+      id=8,
+      timestamp=NOW,
+      severity_number=17,
+      severity_text="ERROR",
+      body="historical body",
+      trace_id="job.12",
+      span_id=None,
+      attributes={"legacy": {"nested": True}},
+    )
+  ]
+  sessions = MagicMock()
+  sessions.return_value.__aenter__.return_value = session
+
+  async def scenario():
+    backend = PostgreSQLBackend(sessions)
+    page = await backend.query(
+      ObservationQuery(
+        since=NOW - timedelta(seconds=1),
+        until=NOW + timedelta(seconds=1),
+        job_id=12,
+        text="100%",
+        severity_min=13,
+        limit=1,
+      )
+    )
+    statement = session.execute.call_args.args[0]
+    compiled = statement.compile(dialect=postgresql.dialect())
+    assert "job.12" in compiled.params.values()
+    assert "%100\\%%" in compiled.params.values()
+    assert str(compiled).index("WHERE") < str(compiled).index("LIMIT")
+    assert page.records[0].id == "pg.8"
+    assert page.records[0].kind == "log" and page.records[0].event_name is None
+    assert page.records[0].attributes == {"legacy": {"nested": True}}
+
+  async_runner.run(scenario())
+
+
+def test_route_refuses_stale_backend_and_hides_invalid_body(async_runner):
+  from app.core import CoreMiddleware
+  from app.routes.observability import ROUTER
+
+  core = Core("postgresql+psycopg://test:test@127.0.0.1:1/test")
+  core.observability = ObservationRuntime(None, selected="postgresql", peer_id="peer")
+  app = fastapi.FastAPI()
+  app.add_middleware(CoreMiddleware, core=core)
+  app.include_router(ROUTER)
+  with TestClient(app) as client:
+    response = client.post(
+      "/observability", json={"action": "ingest", "records": [], "expected_backend": "loki"}
+    )
+    assert response.status_code == 409
+    response = client.post("/observability", json={"action": "bad-secret"})
+    assert response.status_code == 422 and "bad-secret" not in response.text
+    response = client.post("/observability", content=b"x" * (256 * 1024 + 1))
+    assert response.status_code == 413
+  async_runner.run(core.aclose())
+
+
+def test_failed_metadata_and_malformed_logging_keep_business_semantics(
+  async_runner, monkeypatch
+):
+  import logging
+  from inkcre_core.libs.obsrv import telemetry
+  from inkcre_core.libs.obsrv.log_record import ENABLE_LOG_BACKEND
+  from inkcre_core.libs.obsrv.runtime import ObservationLogHandler
+
+  monkeypatch.setattr(telemetry, "_current_runtime", lambda: None)
+
+  async def scenario():
+    core = Core("postgresql+psycopg://test:test@127.0.0.1:1/test")
+    backend = MagicMock(write=AsyncMock(), aclose=AsyncMock())
+    runtime = ObservationRuntime(backend, selected="postgresql", peer_id="peer")
+    core.observability = runtime
+    runtime.start()
+    with core.bind():
+      emit_event("job.closed", {"inkcre.job.id": 12, "inkcre.job.status": "failed"})
+      with pytest.raises(RuntimeError, match="business error"):
+        with operation("test.operation"):
+          raise RuntimeError("business error")
+      token = ENABLE_LOG_BACKEND.set(True)
+      try:
+        ObservationLogHandler().emit(
+          logging.LogRecord("inkcre", 20, "", 1, "%d", ("invalid",), None)
+        )
+        ObservationLogHandler().emit(
+          logging.LogRecord("inkcre", 20, "", 1, "x" * 8193, (), None)
+        )
+        nested = logging.LogRecord("inkcre", 20, "", 1, "fixed", (), None)
+        nested.extra_payload = {"unexpected": "nested"}
+        ObservationLogHandler().emit(nested)
+      finally:
+        ENABLE_LOG_BACKEND.reset(token)
+    await core.aclose()
+    records = backend.write.call_args.args[0]
+    assert len(records) == 2
+    assert all(record.severity_number == 17 for record in records)
+    assert runtime.dropped == 3
+
+  async_runner.run(scenario())
+
+
+def test_public_selector_is_authority_and_failed_external_config_never_defaults(
+  async_runner, monkeypatch
+):
+  from app import observability
+  from inkcre_core.schemas.observability import ObservabilityBackendConfig
+
+  monkeypatch.setattr(observability.settings.obsrv, "telemetry_enabled", False)
+  monkeypatch.setattr(observability.settings.obsrv, "loki_url", None)
+
+  async def scenario():
+    core = Core("postgresql+psycopg://test:test@127.0.0.1:1/test")
+    with core.bind():
+      get = AsyncMock(return_value=None)
+      monkeypatch.setattr(observability.DeploymentConfigService, "get", get)
+      await observability.initialize_telemetry()
+      assert core.observability is not None
+      assert core.observability.selected == "postgresql"
+      assert isinstance(core.observability.backend, PostgreSQLBackend)
+      get.assert_awaited_once_with("inkcre.observability.backend")
+      get.side_effect = [ObservabilityBackendConfig(backend="loki"), None]
+      await observability.initialize_telemetry()
+      assert core.observability.selected == "loki"
+      assert core.observability.backend is None
+      assert not core.observability.status().recording
+      get.side_effect = RuntimeError("secret configuration response")
+      await observability.initialize_telemetry()
+      assert core.observability.backend is None
+      assert "secret" not in (core.observability.error or "")
+    await core.aclose()
+
+  async_runner.run(scenario())
+
+
+def test_optional_otlp_connection_belongs_to_bound_core(async_runner):
+  from inkcre_core.libs.obsrv.telemetry import _Runtime, get_export_connection
+
+  first = Core("postgresql+psycopg://test:test@127.0.0.1:1/test")
+  second = Core("postgresql+psycopg://test:test@127.0.0.1:1/test")
+  first.telemetry = _Runtime(connections={"logs": ("https://first.invalid", {}, 1)})
+  second.telemetry = _Runtime(connections={"logs": ("https://second.invalid", {}, 1)})
+
+  async def scenario():
+    with first.bind():
+      connection = get_export_connection("logs")
+      assert connection is not None and connection[0] == "https://first.invalid"
+      with second.bind():
+        connection = get_export_connection("logs")
+        assert connection is not None and connection[0] == "https://second.invalid"
+    await first.aclose()
+    assert first.telemetry is None
+    assert second.telemetry is not None
+    await second.aclose()
+
+  async_runner.run(scenario())
+
+
+def test_capability_bounds_slow_body_and_concurrent_requests(async_runner, monkeypatch):
+  import asyncio
+  from app.routes import observability
+
+  core = Core("postgresql+psycopg://test:test@127.0.0.1:1/test")
+  runtime = ObservationRuntime(None, selected="postgresql", peer_id="peer")
+  core.observability = runtime
+  original_timeout = asyncio.timeout
+  monkeypatch.setattr(
+    observability, "asyncio", SimpleNamespace(timeout=lambda _: original_timeout(0.001))
+  )
+
+  async def receive():
+    await asyncio.Event().wait()
+    return {"type": "http.request", "body": b"", "more_body": False}
+
+  async def scenario():
+    with core.bind():
+      request = fastapi.Request({"type": "http"}, receive=receive)
+      runtime.requests_in_flight = 4
+      with pytest.raises(fastapi.HTTPException) as capacity:
+        await observability.observability(request)
+      assert capacity.value.status_code == 429
+      runtime.requests_in_flight = 0
+      with pytest.raises(fastapi.HTTPException) as timeout:
+        await observability.observability(request)
+      assert timeout.value.status_code == 408
+      assert runtime.requests_in_flight == 0
+    await core.aclose()
+
+  async_runner.run(scenario())