diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 368baf9e..a416a89f 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -43,6 +43,16 @@ jobs:
       - name: Validate observability IaC
         run: bash deploy/observability/scripts/validate-iac.sh
 
+      - name: Verify local log pipeline and trace correlation
+        timeout-minutes: 10
+        run: python3 deploy/observability/scripts/verify-log-pipeline.py
+
+      - name: Upload log pipeline verification
+        uses: actions/upload-artifact@v4
+        with:
+          name: observability-log-pipeline
+          path: build/reports/observability/log-pipeline.json
+
   backend-build-test:
     name: Backend Build & Test (Unit + Integration)
     runs-on: ubuntu-latest
diff --git a/deploy/docker-compose.ec2.yml.example b/deploy/docker-compose.ec2.yml.example
index 9a9b1449..5f4f04be 100644
--- a/deploy/docker-compose.ec2.yml.example
+++ b/deploy/docker-compose.ec2.yml.example
@@ -3,6 +3,15 @@ services:
     image: ${IMAGE_REPOSITORY}:${IMAGE_TAG}
     container_name: beach-complex-backend
     restart: unless-stopped
+    labels:
+      beach.logs: backend
+    logging:
+      driver: json-file
+      options:
+        max-size: "50m"
+        max-file: "5"
+        compress: "false"
+        labels: "beach.logs"
     env_file:
       - ${APP_RUNTIME_ENV_FILE:-.env}
     ports:
diff --git a/deploy/observability/README.md b/deploy/observability/README.md
new file mode 100644
index 00000000..dd2ff9cb
--- /dev/null
+++ b/deploy/observability/README.md
@@ -0,0 +1,182 @@
+# 관측 로그 파이프라인
+
+앱 서버의 Docker `json-file` 로그를 Alloy가 읽어 관측 서버 Loki로 보내고, Grafana에서 로그와 Tempo Trace를 연결한다.
+Trace는 기존 경로대로 애플리케이션에서 Tempo로 직접 전송한다.
+
+```text
+앱 EC2: Backend stdout → Docker json-file → Alloy ── TCP 3100 ──→ 관측 EC2: Loki
+         └────────────────── OTLP 4317/4318 ──────────────────→ Tempo
+                                                                 ↑
+                                                            Grafana 조회
+```
+
+구현 범위는 [이슈 #233](https://github.com/Beach-complex/Beach_complex/issues/233)의 PR4다. 서버 적용과 실제 앱 검증은 아직 수행하지 않았다.
+
+- 검증 명령만 필요하면 [verify-local.txt](verify-local.txt)를 사용한다.
+- 서버에 적용할 때는 아래의 서버 적용 절차와 운영 제약을 함께 확인한다.
+
+## 로컬 검증
+
+필요 도구: Docker Engine + Compose v2 이상, Python 3, Terraform 1.15.8, cloud-init.
+Terraform 테스트는 mock AWS provider를 사용하므로 AWS 자격증명이나 실제 리소스가 필요하지 않다.
+
+기존 `beach.logs=backend` 컨테이너가 있으면 중지된 컨테이너도 포함하여 통합 테스트가 실행을 거부한다.
+이 경우 앱 컨테이너를 건드리지 않고 별도의 테스트용 Docker 데몬에서 실행한다.
+
+저장소 루트에서 다음 명령을 실행한다.
+
+```bash
+bash deploy/observability/scripts/validate-iac.sh
+python3 deploy/observability/scripts/verify-log-pipeline.py
+```
+
+### 설정 검사
+
+`validate-iac.sh`는 Terraform fmt/init/validate와 mock plan 3건(dev/staging/prod), cloud-init schema 및 EC2 user data 16KB 제한,
+Compose 병합, Loki/Alloy 실제 이미지의 설정 검증을 수행한다.
+
+### 통합 테스트
+
+`verify-log-pipeline.py`는 임의의 localhost 포트, 전용 Compose 프로젝트와 볼륨을 만든다. 합성 JSON 로그와 OTLP span으로 다음을 확인하고 자체 리소스를 정리한다.
+
+- 실제 Docker json-file → Alloy → Loki 수집, 원본 JSON 보존 및 traceId 검색
+- 다른 컨테이너 제외, `/series` 기준 인덱스 라벨 4개만 존재
+- Grafana 데이터 소스의 변수·정규식·연결 상태, Grafana 프록시 API를 통한 로그 ↔ Trace 조회
+- Alloy 정상 재시작 후 저장된 파일 지문·읽기 위치 사용 및 중지 중 쌓인 로그 수집
+- 두 컨테이너의 stdout/stderr에서 조각이 교차하는 40KB JSON의 원문 보존과 traceId 검색
+- Alloy 중지 중 회전된 로그 110건과 파일 경계에 걸친 90KB JSON 복구, 재시작 후 중복 전송 검사
+- backend 컨테이너 교체 시 새 로그 경로 발견, 짧은 Loki 중단 후 전송 재시도
+- Loki 컨테이너 재생성 후 기존 데이터 조회 및 실제 적용된 보존 기간 설정
+
+보고서는 `build/reports/observability/log-pipeline.json`에 생성하며 CI에서도 같은 검증을 실행하도록 구성했다.
+
+### 서버에서 별도로 확인할 항목
+
+로컬 검증은 합성 데이터와 Docker 볼륨을 사용한다. 실제 EC2/EBS 부팅·재부팅, 브라우저 클릭,
+운영 앱의 로그와 span 연결, 보존 기간 경과에 따른 실제 삭제는 서버 적용 후 확인할 항목이다.
+24시간 운영 검증, 대시보드·알림·SLI/SLO 및 Terraform 원격 상태 저장소는 #233에서 분리한 후속 범위다.
+
+## 구성과 저장 정책
+
+| 구성 | 파일 / 동작 |
+|---|---|
+| Loki | `compose/loki/local-config.yaml`, 단일 프로세스, TSDB v13 + 파일 시스템 저장소 |
+| 저장소 | 관측 EBS의 `/opt/beach-observability/data/loki` → `/loki`; chunk/index/WAL/compactor 상태 포함 |
+| 보존 기간 | Terraform `env`: dev 3일, staging 7일, prod 14일; compactor 활성화, 삭제 지연 2시간 |
+| 접근 | 관측 SG TCP 3100은 앱 SG만 허용. 인증 없는 Loki를 인터넷에 직접 공개하지 않는다. |
+| 수집 | `app-agent/alloy/config.alloy`, Docker 로그 디렉터리 읽기 전용; Docker socket 사용 없음 |
+| 수집 대상 | Docker logging attribute `beach.logs=backend`인 컨테이너만 통과 |
+| 재배포 | 기본 200ms 주기로 새 로그 파일을 탐색. 컨테이너 ID를 설정에 고정하지 않는다. |
+| 읽기 위치 | 파일 지문과 읽기 위치를 앱 서버 `alloy-data` Docker 볼륨에 저장. `down -v`로 삭제하지 않는다. |
+| 회전 | backend Docker 로그 파일당 50MB, 최대 5개, 압축하지 않음 |
+| 인덱스 라벨 | `app=beach-complex`, `env`, `service=backend`, `host`만 유지 |
+| 식별자 | traceId/spanId/requestId/userId/notificationId/outboxEventId는 JSON 본문에 유지; 검색 인덱스용 라벨로 만들지 않는다. |
+
+## 서버 적용 절차
+
+### 1. 관측 서버의 변경 사항 확인
+
+관측 서버 Terraform은 검증된 EBS 마운트 아래에 Loki 데이터 디렉터리(UID/GID 10001)를 만들고,
+설정 및 환경별 `LOKI_RETENTION_PERIOD`를 배치한다. 기존 마운트 가드를 유지한다.
+현재 모듈의 `user_data_replace_on_change = true` 때문에 이 변경의 apply는 관측 EC2 교체를 유발할 수 있다.
+적용 전 실제 state 기준 plan의 교체 대상과 EBS 보존 여부를 확인해야 한다.
+
+### 2. 앱 서버의 환경 변수 설정
+
+`app-agent/.env.example`을 참고해 `app-agent/.env`를 작성한다.
+
+```dotenv
+LOKI_URL=http://<관측-서버-private-ip>:3100/loki/api/v1/push
+APP_ENVIRONMENT=dev
+LOG_HOST=beach-app-ec2
+DOCKER_LOG_ROOT=/var/lib/docker/containers
+```
+
+`LOG_HOST`에는 재배포 후에도 유지되는 서버 식별자를 사용한다.
+`DOCKER_LOG_ROOT`는 Docker 데이터 경로 아래의 `containers` 디렉터리다.
+앱 서버에서 `docker info --format '{{.DockerRootDir}}'`로 데이터 경로를 확인한 뒤 `/containers`를 붙인다.
+
+### 3. 앱 로그 설정 적용 및 Alloy 기동
+
+실제 앱 Compose의 서비스가 `app`인지 확인한 뒤 `backend-logging.override.yml`을 함께 사용한다.
+서비스명이 다르면 override의 서비스 키를 실제 이름으로 맞춘다.
+로그 속성은 컨테이너 생성 때 적용되므로 기존 컨테이너에는 재생성이 필요하다.
+
+```bash
+# 앱 서버, 실제 기존 Compose 파일·환경 파일을 그대로 지정해서 병합 결과를 먼저 확인한다.
+docker compose -f <기존-compose.yml> -f <경로>/app-agent/backend-logging.override.yml config
+# 이후 배포 단계에서만 app을 재생성하고 Alloy를 기동한다.
+docker compose -f <기존-compose.yml> -f <경로>/app-agent/backend-logging.override.yml up -d --no-deps app
+docker compose --env-file <경로>/app-agent/.env -f <경로>/app-agent/docker-compose.yml up -d
+```
+
+지속 배포에 사용하는 Compose에도 logging 설정이 남아 있어야 한다. 저장소의 `deploy/docker-compose.ec2.yml.example`에도 같은 설정을 추가했다.
+`APP_ENVIRONMENT`는 Loki 라벨이며 Spring profile을 변경하지 않는다. 현재 Logback의 prod profile이 NDJSON을 출력한다.
+JSON 검색을 확인할 때는 실제 앱 출력 형식을 먼저 확인하고, 수집 때문에 애플리케이션 profile을 임의로 바꾸지 않는다.
+
+### 4. Grafana에서 로그와 Trace 조회
+
+Grafana Explore → Loki에서 다음을 조회한다. `env`는 적용한 환경에 맞춘다.
+
+```logql
+{app="beach-complex", env="dev", service="backend"} | json
+{app="beach-complex", env="dev", service="backend"} | json | traceId="<32자리 trace ID>"
+```
+
+로그의 `View trace` 링크는 Tempo를 가리킨다. JSON `traceId`와 현재 개발용 텍스트의 `traceId=...`를 지원한다.
+Tempo의 Trace to logs는 span 시간 전후 5분에서 동일 traceId 문자열을 검색한다.
+Trace가 샘플링되지 않았거나 Tempo 보존 기간이 지났으면 해당 로그의 링크 대상은 없을 수 있다.
+
+## 운영 제약
+
+- **전송 보장:** 읽기 위치 저장과 Loki 전송 완료는 하나의 작업으로 처리되지 않는다. 짧은 장애에는 재시도하지만, 장시간 장애·강제 종료·회전 한도 초과 시 무손실이나 정확히 한 번의 전송을 보장하지 않는다.
+
+- **긴 로그 재조립:** 조각은 메모리에만 저장한다. 첫 조각을 읽은 뒤 5초 안에 한 줄이 완성되지 않거나 재조립 도중 Alloy가 종료되면 원문 보존을 보장하지 않는다.
+
+- **로그 크기와 회전:** 한 줄은 256KiB 이내로 유지한다. 수집 대상은 최대 5개 파일이므로 Docker의 보존 개수와 압축하지 않는 설정도 유지한다.
+
+- **읽기 위치 보존:** `alloy-data` 볼륨을 삭제하면 저장된 읽기 위치도 사라진다. `docker compose down -v`를 사용하지 않는다.
+
+- **이전 초안에서 전환:** 기존 `loki.source.file`의 positions 파일은 자동 이관되지 않는다. 새 수집기를 적용하면 남아 있는 로그를 다시 읽을 수 있다.
+
+- **Loki 보존 기간:** 보존 기간은 디스크 용량 상한이 아니다. 오래된 데이터는 compactor 실행 주기와 삭제 지연 이후에 삭제된다.
+
+## 구현 상세
+
+<details>
+<summary>Alloy 실행 권한, 파일 식별 방식과 정렬 우회 설정</summary>
+
+### 실행 권한
+
+Alloy는 root 소유 Docker 파일을 읽기 위해 UID 0으로 실행하지만, capability를 모두 제거하고 `no-new-privileges`를 사용한다.
+빈 상태 볼륨에 이미지 내부 UID 소유권이 복사되지 않도록 `nocopy: true`를 지정한다.
+이는 rootful Docker의 기본 `json-file` 배치를 기준으로 한다. rootless Docker나 user namespace remapping 환경은 별도 권한 검증이 필요하다.
+
+### 수집기 버전과 로그 처리
+
+`otelcol.receiver.filelog`와 `otelcol.storage.file`은 고정한 Alloy v1.19.2에서 공개 미리보기 기능이다.
+Compose와 설정 검사에 `--stability.level=public-preview`를 명시했다. 업그레이드할 때 같은 회귀 테스트를 실행한다.
+
+파일 지문은 파일 내용 일부를 비교해 같은 파일인지 식별하는 값이다. Docker 레코드 끝의 시각까지 비교하도록 크기를 128KiB로 설정했다. 긴 로그 조각의 앞부분이 같아도 다른 파일로 구분한다.
+파일 지문으로 이름이 바뀐 파일을 추적하고 `.log.4`부터 `.log`까지 차례로 읽는다.
+Docker가 나눈 한 줄은 컨테이너 ID와 stdout/stderr를 기준으로 묶어, 줄바꿈이 나올 때까지 구분자 없이 합친다.
+
+### 정렬 옵션의 우회 설정
+
+이 버전의 Alloy 정렬 설정은 `lexicographic`을 받지만 내부 Stanza는 `alphabetical`을 요구한다.
+이를 피하려고 `operators`의 `file_input`에 정렬을 직접 설정했다. 바깥쪽 입력은 사용하지 않는 경로를 가리킨다.
+두 입력이 같은 파일을 읽지 않으며, 실제 입력은 같은 file storage에 읽기 위치를 저장한다.
+이 우회 설정은 해당 버전의 매핑이 수정되면 제거할 수 있다.
+
+</details>
+
+## 참고 자료
+
+- [ADR-014: Loki 및 보존·라벨 정책](https://github.com/Beach-complex/beach_complex_docs/blob/main/adr/ADR-014-logging-storage-backend.md)
+- [ADR-015: Alloy 및 file tail](https://github.com/Beach-complex/beach_complex_docs/blob/main/adr/ADR-015-log-collector.md)
+- [Alloy filelog receiver](https://grafana.com/docs/alloy/latest/reference/components/otelcol/otelcol.receiver.filelog/)
+- [Alloy file storage](https://grafana.com/docs/alloy/latest/reference/components/otelcol/otelcol.storage.file/)
+- [Stanza 재조립 설정](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/v0.158.0/pkg/stanza/docs/operators/recombine.md)
+- [Alloy Loki write 및 재시도](https://grafana.com/docs/alloy/latest/reference/components/loki/loki.write/)
+- [Loki retention](https://grafana.com/docs/loki/latest/operations/storage/retention/)
+- [Grafana Tempo와 로그 연결](https://grafana.com/docs/grafana/latest/datasources/tempo/configure-tempo-data-source/)
diff --git a/deploy/observability/app-agent/.env.example b/deploy/observability/app-agent/.env.example
new file mode 100644
index 00000000..4a7c6ec7
--- /dev/null
+++ b/deploy/observability/app-agent/.env.example
@@ -0,0 +1,5 @@
+LOKI_URL=http://<observability-private-ip>:3100/loki/api/v1/push
+APP_ENVIRONMENT=dev
+LOG_HOST=beach-app-ec2
+# Set to DockerRootDir/containers when Docker uses a non-default data root.
+DOCKER_LOG_ROOT=/var/lib/docker/containers
diff --git a/deploy/observability/app-agent/alloy/config.alloy b/deploy/observability/app-agent/alloy/config.alloy
new file mode 100644
index 00000000..7bed5db9
--- /dev/null
+++ b/deploy/observability/app-agent/alloy/config.alloy
@@ -0,0 +1,114 @@
+logging {
+	level = "info"
+}
+
+// Fingerprints identify files even after Docker renames them during rotation.
+// Both components are public preview in the pinned Alloy v1.19.2 image.
+otelcol.storage.file "docker" {
+	directory        = "/var/lib/alloy/docker-offsets"
+	create_directory = true
+	fsync            = true
+}
+
+otelcol.receiver.filelog "docker" {
+	// v1.19.2's ordering wrapper accepts "lexicographic", whereas its bundled
+	// stanza reader requires "alphabetical". Configure the file_input operator
+	// directly until that upstream mapping is fixed. The wrapper input is idle.
+	include   = ["/var/log/docker/__unused_wrapper_input__"]
+	storage   = otelcol.storage.file.docker.handler
+	operators = [
+		{
+			type      = "json_parser",
+			id        = "parse_docker",
+			output    = "select_backend",
+			on_error  = "drop",
+			timestamp = {
+				parse_from  = "attributes.time",
+				layout_type = "gotime",
+				layout      = "2006-01-02T15:04:05.999999999Z07:00",
+			},
+		},
+		{
+			type     = "file_input",
+			id       = "docker_files",
+			output   = "parse_docker",
+			include  = ["/var/log/docker/*/*-json.log", "/var/log/docker/*/*-json.log.[1-4]"],
+			start_at = "beginning",
+			// Include Docker timestamps after escaped 16 KiB chunks in the fingerprint.
+			// A 1 KiB prefix can be identical across containers with repetitive payloads.
+			fingerprint_size  = "128KiB",
+			include_file_path = true,
+			include_file_name = false,
+			// Oldest file first, one reader at a time, across all five files.
+			max_concurrent_files = 1,
+			max_batches          = 0,
+			ordering_criteria    = {
+				regex    = "/[^/]+/[^/]+-json\\.log(?P<rotation>\\.[1-4])?$",
+				group_by = "/([^/]+)/[^/]+-json\\.log",
+				top_n    = 5,
+				sort_by  = [{sort_type = "alphabetical", regex_key = "rotation", ascending = false}],
+			},
+			attributes = {
+				app                     = "beach-complex",
+				env                     = sys.env("APP_ENVIRONMENT"),
+				service                 = "backend",
+				host                    = sys.env("LOG_HOST"),
+				"loki.attribute.labels" = "app,env,service,host",
+				"loki.format"           = "raw",
+			},
+		},
+		{
+			type = "filter",
+			id   = "select_backend",
+			expr = "attributes.attrs == nil or attributes.attrs[\"beach.logs\"] != \"backend\"",
+		},
+		{
+			type       = "regex_parser",
+			parse_from = "attributes[\"log.file.path\"]",
+			regex      = "/(?P<container>[^/]+)/[^/]+-json.log",
+			on_error   = "drop",
+		},
+		{
+			type  = "add",
+			field = "attributes.source",
+			value = "EXPR(attributes.container + \"/\" + attributes.stream)",
+		},
+		{
+			type               = "recombine",
+			combine_field      = "attributes.log",
+			combine_with       = "",
+			is_last_entry      = "attributes.log endsWith \"\\n\"",
+			source_identifier  = "attributes.source",
+			overwrite_with     = "oldest",
+			force_flush_period = "5s",
+			max_log_size       = 262144,
+		},
+		{
+			type = "move",
+			from = "attributes.log",
+			to   = "body",
+		},
+	]
+
+	output {
+		logs = [otelcol.exporter.loki.backend.input]
+	}
+}
+
+otelcol.exporter.loki "backend" {
+	forward_to = [loki.process.backend.receiver]
+}
+
+loki.process "backend" {
+	forward_to = [loki.write.observability.receiver]
+	// Preserve app JSON; IDs and Docker paths never become indexed labels.
+	stage.label_keep {
+		values = ["app", "env", "service", "host"]
+	}
+}
+
+loki.write "observability" {
+	endpoint {
+		url = sys.env("LOKI_URL")
+	}
+}
diff --git a/deploy/observability/app-agent/backend-logging.override.yml b/deploy/observability/app-agent/backend-logging.override.yml
new file mode 100644
index 00000000..68e00e33
--- /dev/null
+++ b/deploy/observability/app-agent/backend-logging.override.yml
@@ -0,0 +1,12 @@
+# Merge with the app server's existing Compose file. Service name must be app.
+services:
+  app:
+    labels:
+      beach.logs: backend
+    logging:
+      driver: json-file
+      options:
+        max-size: "50m"
+        max-file: "5"
+        compress: "false"
+        labels: "beach.logs"
diff --git a/deploy/observability/app-agent/docker-compose.yml b/deploy/observability/app-agent/docker-compose.yml
new file mode 100644
index 00000000..22547f11
--- /dev/null
+++ b/deploy/observability/app-agent/docker-compose.yml
@@ -0,0 +1,32 @@
+services:
+  alloy:
+    image: grafana/alloy:v1.19.2
+    restart: unless-stopped
+    # Docker log files are root-owned; no Docker socket or host write access.
+    user: "0:0"
+    command:
+      - run
+      - --stability.level=public-preview
+      - --server.http.listen-addr=127.0.0.1:12345
+      - --storage.path=/var/lib/alloy
+      - /etc/alloy/config.alloy
+    environment:
+      LOKI_URL: ${LOKI_URL:?Set the observability private Loki push URL}
+      APP_ENVIRONMENT: ${APP_ENVIRONMENT:?Set dev, staging or prod}
+      LOG_HOST: ${LOG_HOST:?Set a stable application host identifier}
+    volumes:
+      - ./alloy/config.alloy:/etc/alloy/config.alloy:ro
+      - ${DOCKER_LOG_ROOT:-/var/lib/docker/containers}:/var/log/docker:ro
+      # Keep the empty volume root-owned instead of copying image UID ownership.
+      - type: volume
+        source: alloy-data
+        target: /var/lib/alloy
+        volume:
+          nocopy: true
+    security_opt:
+      - no-new-privileges:true
+    cap_drop:
+      - ALL
+
+volumes:
+  alloy-data:
diff --git a/deploy/observability/compose/docker-compose.yml b/deploy/observability/compose/docker-compose.yml
index aae74802..11472548 100644
--- a/deploy/observability/compose/docker-compose.yml
+++ b/deploy/observability/compose/docker-compose.yml
@@ -1,4 +1,20 @@
 services:
+  loki:
+    image: grafana/loki:3.7.8
+    container_name: beach-loki
+    restart: unless-stopped
+    command:
+      - -config.file=/etc/loki/local-config.yaml
+      - -config.expand-env=true
+    environment:
+      LOKI_RETENTION_PERIOD: ${LOKI_RETENTION_PERIOD:-72h}
+    volumes:
+      - ./loki/local-config.yaml:/etc/loki/local-config.yaml:ro
+      - ../data/loki:/loki
+    # Terraform permits TCP 3100 only from the application server SG.
+    ports:
+      - 3100:3100
+
   prometheus:
     image: prom/prometheus:v3.13.2
     container_name: beach-prometheus
@@ -46,6 +62,7 @@ services:
     depends_on:
       - prometheus
       - tempo
+      - loki
     environment:
       GF_USERS_ALLOW_SIGN_UP: "false"
     volumes:
diff --git a/deploy/observability/compose/grafana/provisioning/datasources/datasources.yml b/deploy/observability/compose/grafana/provisioning/datasources/datasources.yml
index 777168da..e78c4dd7 100644
--- a/deploy/observability/compose/grafana/provisioning/datasources/datasources.yml
+++ b/deploy/observability/compose/grafana/provisioning/datasources/datasources.yml
@@ -11,6 +11,21 @@ datasources:
     isDefault: true
     editable: false
 
+  - name: Loki
+    uid: loki
+    type: loki
+    access: proxy
+    url: http://loki:3100
+    isDefault: false
+    editable: false
+    jsonData:
+      derivedFields:
+        - name: TraceID
+          datasourceUid: tempo
+          matcherRegex: '(?:"traceId"\s*:\s*"|traceId=)([a-f0-9]{32})'
+          url: '$${__value.raw}'
+          urlDisplayLabel: View trace
+
   - name: Tempo
     uid: tempo
     type: tempo
@@ -18,3 +33,10 @@ datasources:
     url: http://tempo:3200
     isDefault: false
     editable: false
+    jsonData:
+      tracesToLogsV2:
+        datasourceUid: loki
+        spanStartTimeShift: '-5m'
+        spanEndTimeShift: '5m'
+        customQuery: true
+        query: '{app="beach-complex", service="backend"} |= "$${__span.traceId}"'
diff --git a/deploy/observability/compose/loki/local-config.yaml b/deploy/observability/compose/loki/local-config.yaml
new file mode 100644
index 00000000..12920b3c
--- /dev/null
+++ b/deploy/observability/compose/loki/local-config.yaml
@@ -0,0 +1,46 @@
+auth_enabled: false
+
+server:
+  http_listen_port: 3100
+
+common:
+  instance_addr: 127.0.0.1
+  path_prefix: /loki
+  storage:
+    filesystem:
+      chunks_directory: /loki/chunks
+      rules_directory: /loki/rules
+  replication_factor: 1
+  ring:
+    kvstore:
+      store: inmemory
+
+schema_config:
+  configs:
+    - from: 2024-01-01
+      store: tsdb
+      object_store: filesystem
+      schema: v13
+      index:
+        prefix: index_
+        period: 24h
+
+ingester:
+  wal:
+    enabled: true
+    dir: /loki/wal
+
+compactor:
+  working_directory: /loki/compactor
+  retention_enabled: true
+  retention_delete_delay: 2h
+  delete_request_store: filesystem
+
+limits_config:
+  # ADR-014: dev 3d, staging 7d, prod 14d; cloud-init supplies the environment.
+  retention_period: ${LOKI_RETENTION_PERIOD:-72h}
+  # Do not implicitly add service_name to the explicit four-label contract.
+  discover_service_name: []
+
+analytics:
+  reporting_enabled: false
diff --git a/deploy/observability/scripts/fixtures/log-emitter.py b/deploy/observability/scripts/fixtures/log-emitter.py
new file mode 100644
index 00000000..06807c55
--- /dev/null
+++ b/deploy/observability/scripts/fixtures/log-emitter.py
@@ -0,0 +1,23 @@
+"""Write synthetic logs through PID 1; /partial holds back the newline."""
+import sys
+
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+
+class Handler(BaseHTTPRequestHandler):
+    def do_POST(self):
+        line = self.rfile.read(int(self.headers['Content-Length'])).decode()
+        stream = sys.stderr if self.path.startswith('/stderr') else sys.stdout
+        print(line, end='' if self.path.endswith('/partial') else '\n', file=stream, flush=True)
+        self.send_response(204)
+        self.end_headers()
+
+    def do_GET(self):
+        self.send_response(200)
+        self.end_headers()
+
+    def log_message(self, *_):
+        pass
+
+
+HTTPServer(('0.0.0.0', 8000), Handler).serve_forever()
diff --git a/deploy/observability/scripts/validate-iac.sh b/deploy/observability/scripts/validate-iac.sh
index 8a132f5b..2acbcdc1 100644
--- a/deploy/observability/scripts/validate-iac.sh
+++ b/deploy/observability/scripts/validate-iac.sh
@@ -42,6 +42,8 @@ validate_terraform() {
     -lockfile=readonly \
     -no-color
   terraform -chdir="$terraform_environment" validate -no-color
+  # Mock provider: verifies SG and retention contracts without contacting AWS.
+  terraform -chdir="$terraform_environment" test -no-color
 }
 
 render_cloud_init() {
@@ -59,6 +61,8 @@ base64gzip(templatefile("${path.module}/../../modules/observability_ec2/cloud-in
     )
   ),
   tempo_config_base64gzip        = base64gzip(file("${path.module}/../../../compose/tempo/tempo.yaml")),
+  loki_config_base64gzip         = base64gzip(file("${path.module}/../../../compose/loki/local-config.yaml")),
+  loki_retention_period          = lookup({ dev = "72h", staging = "168h", prod = "336h" }, var.env),
   grafana_datasources_base64gzip = base64gzip(file("${path.module}/../../../compose/grafana/provisioning/datasources/datasources.yml"))
 }))
 EOF
@@ -96,7 +100,24 @@ EOF
 }
 
 validate_compose() {
+  local loki_image alloy_image
   docker compose -f "$compose_dir/docker-compose.yml" config --quiet
+  LOKI_URL=http://loki:3100/loki/api/v1/push APP_ENVIRONMENT=dev LOG_HOST=validation \
+    docker compose -f deploy/observability/app-agent/docker-compose.yml config --quiet
+  IMAGE_REPOSITORY=local/backend IMAGE_TAG=validation APP_RUNTIME_ENV_FILE=/dev/null \
+    docker compose -f deploy/docker-compose.ec2.yml.example \
+    -f deploy/observability/app-agent/backend-logging.override.yml config --quiet
+  loki_image="$(docker compose -f "$compose_dir/docker-compose.yml" config --format json \
+    | python3 -c 'import json,sys; print(json.load(sys.stdin)["services"]["loki"]["image"])')"
+  alloy_image="$(LOKI_URL=http://loki:3100/loki/api/v1/push APP_ENVIRONMENT=dev LOG_HOST=validation \
+    docker compose -f deploy/observability/app-agent/docker-compose.yml config --images)"
+  docker run --rm \
+    --volume "$compose_dir/loki/local-config.yaml:/etc/loki/local-config.yaml:ro" "$loki_image" \
+    -config.file=/etc/loki/local-config.yaml -config.expand-env=true -verify-config=true
+  LOKI_URL=http://loki:3100/loki/api/v1/push APP_ENVIRONMENT=dev LOG_HOST=validation \
+    docker run --rm --env LOKI_URL --env APP_ENVIRONMENT --env LOG_HOST \
+      --volume "$PWD/deploy/observability/app-agent/alloy/config.alloy:/etc/alloy/config.alloy:ro" \
+      "$alloy_image" validate --stability.level=public-preview /etc/alloy/config.alloy
 }
 
 main() {
diff --git a/deploy/observability/scripts/verify-log-pipeline.py b/deploy/observability/scripts/verify-log-pipeline.py
new file mode 100644
index 00000000..e49bc671
--- /dev/null
+++ b/deploy/observability/scripts/verify-log-pipeline.py
@@ -0,0 +1,310 @@
+#!/usr/bin/env python3
+"""Local Docker E2E: real json-file -> Alloy -> Loki -> Grafana/Tempo.
+
+Uses repository configs and isolated named volumes, network and random loopback
+ports. Never calls AWS or deploys the backend. Only its own project is removed.
+"""
+import base64
+import copy
+import datetime
+import json
+import os
+from pathlib import Path
+import re
+import secrets
+import subprocess
+import tempfile
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+
+ROOT = Path(__file__).resolve().parents[3]
+OBS = ROOT / 'deploy/observability'
+PROJECT = 'beach-logs-test-' + secrets.token_hex(4)
+REPORT = ROOT / 'build/reports/observability/log-pipeline.json'
+ENV = dict(os.environ, LOKI_URL='http://loki:3100/loki/api/v1/push',
+           APP_ENVIRONMENT='dev', LOG_HOST=PROJECT, LOKI_RETENTION_PERIOD='72h')
+PASSWORD = secrets.token_hex(20)
+checks = []
+
+
+def run(*args, **kwargs):
+    return subprocess.check_output(args, text=True, env=ENV, **kwargs).strip()
+
+
+def request(url, payload=None, headers=None):
+    req = urllib.request.Request(url, data=payload, headers=headers or {})
+    with urllib.request.urlopen(req, timeout=8) as response:
+        return response.read().decode()
+
+
+def wait_for(description, check, timeout=90):
+    deadline = time.monotonic() + timeout
+    last = None
+    while time.monotonic() < deadline:
+        try:
+            value = check()
+            if value:
+                return value
+        except (urllib.error.URLError, TimeoutError, ConnectionError, json.JSONDecodeError) as exc:
+            last = str(exc)
+        time.sleep(1)
+    raise AssertionError(f'Timed out: {description}; last error: {last}')
+
+
+def passed(message):
+    checks.append(message)
+    print('PASS ' + message, flush=True)
+
+
+def main():
+    REPORT.unlink(missing_ok=True)
+    # Refuse to mix another backend's logs into the test Loki instance.
+    if run('docker', 'ps', '-aq', '--filter', 'label=beach.logs=backend'):
+        raise RuntimeError('A labelled backend exists; use an isolated Docker daemon for this test.')
+    docker_logs = run('docker', 'info', '--format', '{{.DockerRootDir}}') + '/containers'
+    base = json.loads(run('docker', 'compose', '-f', str(OBS / 'compose/docker-compose.yml'),
+                          'config', '--format', 'json'))
+    agent = json.loads(run('docker', 'compose', '-f', str(OBS / 'app-agent/docker-compose.yml'),
+                           'config', '--format', 'json'))
+    services = {n: copy.deepcopy(base['services'][n]) for n in ['loki', 'tempo', 'grafana']}
+    services['alloy'] = agent['services']['alloy']
+    volumes = {}
+    # Keep actual command, image, read-only config mounts and env configuration.
+    # Replace only production storage, host ports and global container names.
+    for name, service in services.items():
+        service.pop('container_name', None)
+        service.pop('depends_on', None)
+        service['restart'] = 'no'
+        for port in service.get('ports', []):
+            port.update(host_ip='127.0.0.1', published='0')
+        for mount in service.get('volumes', []):
+            if mount['target'] == '/var/log/docker':
+                mount['source'] = docker_logs
+            elif not mount.get('read_only', False):
+                volume = name + '-data'
+                volumes[volume] = {}
+                mount.clear()
+                mount.update(type='volume', source=volume,
+                             target={'loki': '/loki', 'tempo': '/var/tempo',
+                                     'grafana': '/var/lib/grafana', 'alloy': '/var/lib/alloy'}[name])
+                if name == 'alloy':
+                    mount['volume'] = {'nocopy': True}
+        if name == 'alloy':
+            service['environment']['LOG_HOST'] = PROJECT
+            # Expose collector metrics only on a random loopback port in this test.
+            service['command'] = [arg.replace('127.0.0.1:12345', '0.0.0.0:12345')
+                                  for arg in service['command']]
+            service['ports'] = [{'target': 12345, 'published': '0', 'host_ip': '127.0.0.1'}]
+    services['grafana']['environment'].update(GF_SECURITY_ADMIN_PASSWORD=PASSWORD,
+                                               GF_ANALYTICS_REPORTING_ENABLED='false')
+    logging_contract = json.loads(run('docker', 'compose', '-f',
+        str(OBS / 'app-agent/backend-logging.override.yml'), 'config', '--no-consistency', '--format', 'json'))['services']['app']
+    for name, label in [('emitter', 'backend'), ('emitter2', 'backend'), ('ignored', 'unrelated')]:
+        services[name] = {
+            'image': 'python:3.12-slim', 'command': ['python3', '-u', '/fixture.py'],
+            'volumes': [{'type': 'bind', 'source': str(OBS / 'scripts/fixtures/log-emitter.py'),
+                         'target': '/fixture.py', 'read_only': True}],
+            'ports': [{'target': 8000, 'published': '0', 'host_ip': '127.0.0.1'}],
+            'labels': {'beach.logs': label}, 'logging': copy.deepcopy(logging_contract['logging']),
+        }
+    # Exercise real rotation quickly while retaining the production five-file policy.
+    for name in ['emitter', 'emitter2', 'ignored']:
+        services[name]['logging']['options']['max-size'] = '64k'
+    document = {'services': services, 'volumes': volumes}
+    with tempfile.TemporaryDirectory(prefix=PROJECT) as temp:
+        config = Path(temp) / 'compose.json'
+        config.write_text(json.dumps(document))
+        os.chmod(config, 0o600)
+
+        def compose(*args):
+            return run('docker', 'compose', '-p', PROJECT, '-f', str(config), *args)
+
+        def address(service, port):
+            return 'http://' + compose('port', service, str(port))
+
+        try:
+            compose('up', '-d')
+            loki, tempo, grafana = address('loki', 3100), address('tempo', 3200), address('grafana', 3000)
+            emitter, ignored = address('emitter', 8000), address('ignored', 8000)
+            wait_for('Loki readiness', lambda: 'ready' in request(loki + '/ready'))
+            wait_for('Tempo readiness', lambda: 'ready' in request(tempo + '/ready'))
+            wait_for('Grafana readiness', lambda: json.loads(request(grafana + '/api/health'))['database'] == 'ok')
+            wait_for('log emitters', lambda: request(emitter) == '' and request(ignored) == '')
+            passed('Local Loki, Tempo, Grafana and Docker log emitters ready')
+
+            def query(expression):
+                qs = urllib.parse.urlencode({'query': expression, 'limit': '1000'})
+                return json.loads(request(loki + '/loki/api/v1/query_range?' + qs))['data']['result']
+
+            selector = '{app="beach-complex",env="dev",service="backend",host="' + PROJECT + '"}'
+
+            def entries(marker):
+                return [v for stream in query(selector + ' |= "' + marker + '"') for v in stream['values']]
+
+            trace_id, span_id = secrets.token_hex(16), secrets.token_hex(8)
+            record = {'message': 'pr4-first-' + PROJECT, 'traceId': trace_id,
+                      'spanId': span_id, 'requestId': 'synthetic-request', 'userId': 'synthetic-user',
+                      'notificationId': 'synthetic-notification', 'outboxEventId': 'synthetic-outbox'}
+            line = json.dumps(record, separators=(',', ':'))
+            request(emitter, line.encode())
+            found = wait_for('Docker -> Alloy -> Loki', lambda: entries(record['message']))
+            assert found[0][1].rstrip('\n') == line, 'Original app JSON was altered'
+            parsed = query(selector + ' | json | traceId="' + trace_id + '"')
+            assert parsed, 'traceId JSON filter did not work'
+            # Query results also contain structured metadata (e.g. detected_level).
+            # The series endpoint reports the actual indexed stream labels.
+            series_qs = urllib.parse.urlencode({'match[]': selector})
+            series = json.loads(request(loki + '/loki/api/v1/series?' + series_qs))['data']
+            assert series, 'No indexed series found'
+            for labels in series:
+                assert set(labels) == {'app', 'env', 'service', 'host'}, labels
+            request(ignored, json.dumps({'message': 'must-not-collect-' + PROJECT}).encode())
+            # Send a backend barrier after the negative fixture, then let batching finish.
+            request(emitter, json.dumps({'message': 'barrier-' + PROJECT}).encode())
+            wait_for('backend barrier', lambda: entries('barrier-' + PROJECT))
+            time.sleep(6)
+            assert not entries('must-not-collect-' + PROJECT)
+            passed('Real Docker JSON is preserved; unrelated logs excluded; only four low-cardinality labels')
+
+            now = time.time_ns()
+            span = {'resourceSpans': [{'resource': {'attributes': [
+                {'key': 'service.name', 'value': {'stringValue': 'beach-complex'}}]},
+                'scopeSpans': [{'scope': {'name': 'pr4-local-smoke'}, 'spans': [{
+                    'traceId': trace_id, 'spanId': span_id, 'name': 'synthetic-log-correlation',
+                    'kind': 2, 'startTimeUnixNano': str(now - 1000000), 'endTimeUnixNano': str(now),
+                }]}]}]}
+            request(address('tempo', 4318) + '/v1/traces', json.dumps(span).encode(),
+                    {'Content-Type': 'application/json'})
+            auth = {'Authorization': 'Basic ' + base64.b64encode(('admin:' + PASSWORD).encode()).decode()}
+
+            def grafana_get(path):
+                return json.loads(request(grafana + path, headers=auth))
+
+            ds = grafana_get('/api/datasources/uid/loki')
+            field = ds['jsonData']['derivedFields'][0]
+            assert field['datasourceUid'] == 'tempo'
+            assert field['url'] == '${__value.raw}', field
+            assert re.search(field['matcherRegex'], line).group(1) == trace_id
+            assert re.search(field['matcherRegex'], '[traceId=' + trace_id + ',spanId=x]').group(1) == trace_id
+            tempo_ds = grafana_get('/api/datasources/uid/tempo')
+            reverse = tempo_ds['jsonData']['tracesToLogsV2']
+            assert reverse['datasourceUid'] == 'loki'
+            assert '${__span.traceId}' in reverse['query']
+            wait_for('trace through Grafana proxy', lambda: grafana_get('/api/datasources/proxy/uid/tempo/api/traces/' + trace_id))
+            health = wait_for('Grafana Loki datasource', lambda: grafana_get('/api/datasources/uid/loki/health'))
+            assert health['status'] == 'OK', health
+            reverse_query = reverse['query'].replace('${__span.traceId}', trace_id)
+            proxy_query = urllib.parse.urlencode({'query': reverse_query})
+            assert grafana_get('/api/datasources/proxy/uid/loki/loki/api/v1/query_range?' + proxy_query)['data']['result']
+            passed('Grafana datasource health and provisioned log-to-trace / trace-to-log targets resolve')
+
+            # Graceful stop persists offsets; logs produced while stopped must be read on restart.
+            compose('stop', 'alloy')
+            request(emitter, json.dumps({'message': 'while-stopped-' + PROJECT}).encode())
+            compose('start', 'alloy')
+            wait_for('Alloy offset resume', lambda: entries('while-stopped-' + PROJECT))
+            offsets = compose('exec', '-T', 'alloy', 'sh', '-c',
+                              'find /var/lib/alloy/docker-offsets -type f -size +0c')
+            assert offsets, 'No persistent file fingerprint/offset database'
+            assert len(entries(record['message'])) == len(found)
+            passed('Alloy persisted file fingerprints/offsets and resumed logs written during its stop')
+
+            # #263: a 40 KB JSON record spans multiple Docker 16 KiB records.
+            emitter2 = address('emitter2', 8000)
+            wait_for('second backend', lambda: request(emitter2) == '')
+            long_lines = []
+            for index, endpoint in enumerate([emitter, emitter + '/stderr', emitter2, emitter2 + '/stderr']):
+                payload = {'message': f'long-{index}-' + PROJECT,
+                           'padding': 'x' * 40000, 'traceId': secrets.token_hex(16)}
+                original = json.dumps(payload, separators=(',', ':'))
+                long_lines.append((endpoint, payload, original))
+                request(endpoint + '/partial', original[:20000].encode())
+            # Let the first fragment from every source reach Alloy before finishing.
+            time.sleep(1)
+            for endpoint, payload, original in long_lines:
+                request(endpoint, original[20000:].encode())
+            for endpoint, payload, original in long_lines:
+                received = wait_for('complete long JSON', lambda: entries(payload['message']))
+                assert len(received) == 1 and received[0][1] == original + '\n', [len(v[1]) for v in received]
+                assert query(selector + ' | json | traceId="' + payload['traceId'] + '"')
+                assert re.search(field['matcherRegex'], received[0][1]).group(1) == payload['traceId']
+            for name in ['emitter', 'emitter2']:
+                container_id = compose('ps', '-q', name)
+                compose('exec', '-T', 'alloy', 'test', '-s',
+                        f'/var/log/docker/{container_id}/{container_id}-json.log.1')
+            passed('40 KB JSON preserved during live rotation with interleaved stdout/stderr from two containers')
+
+            # #263: all these records remain in Docker's five files while Alloy is stopped.
+            compose('stop', 'alloy')
+            rotation_marker = 'rotation-' + PROJECT
+            expected_rotation = []
+            for number in range(110):
+                original = json.dumps({'message': rotation_marker, 'number': number,
+                                       'padding': 'r' * 1000}, separators=(',', ':'))
+                expected_rotation.append(original + '\n')
+                request(emitter, original.encode())
+            retained = compose('logs', '--no-log-prefix', 'emitter')
+            assert retained.count(rotation_marker) == 110, 'Fixture exceeded Docker retention'
+            # Force a long record across a rotation boundary while the reader is offline.
+            offline_long = json.dumps({'message': 'offline-long-' + PROJECT,
+                                       'padding': 'z' * 90000, 'traceId': secrets.token_hex(16)},
+                                      separators=(',', ':'))
+            request(emitter, offline_long.encode())
+            assert compose('logs', '--no-log-prefix', 'emitter').count(rotation_marker) == 110
+            compose('start', 'alloy')
+            recovered = wait_for('all 110 rotated records',
+                                 lambda: (rows if len(rows := entries(rotation_marker)) == 110 else None))
+            assert sorted(v[1] for v in recovered) == sorted(expected_rotation)
+            complete = wait_for('long JSON across rotated files', lambda: entries('offline-long-' + PROJECT))
+            assert len(complete) == 1 and complete[0][1] == offline_long + '\n', [len(v[1]) for v in complete]
+            # Recreate the collector using the same volume and check that completed files are not replayed.
+            compose('up', '-d', '--force-recreate', '--no-deps', 'alloy')
+            request(emitter, json.dumps({'message': 'rotation-resume-' + PROJECT}).encode())
+            wait_for('resume after rotation', lambda: entries('rotation-resume-' + PROJECT))
+            assert len(entries(rotation_marker)) == 110
+            metrics = request(address('alloy', 12345) + '/metrics')
+            sent = sum(float(value) for value in re.findall(
+                r'^loki_write_sent_entries_total\{[^\n]*\} ([0-9.e+]+)$', metrics, re.MULTILINE))
+            assert sent == 1, f'Restarted Alloy replayed previously read records: {sent} sends'
+            passed('All 110 retained rotated records and a 90 KB JSON spanning files recovered after Alloy restart')
+
+            compose('up', '-d', '--force-recreate', '--no-deps', 'emitter')
+            emitter = address('emitter', 8000)
+            wait_for('recreated backend emitter', lambda: request(emitter) == '')
+            request(emitter, json.dumps({'message': 'new-container-' + PROJECT}).encode())
+            wait_for('new Docker container ID discovery', lambda: entries('new-container-' + PROJECT))
+            passed('Alloy discovers a recreated backend without a hardcoded container ID')
+
+            compose('stop', 'loki')
+            request(emitter, json.dumps({'message': 'loki-outage-' + PROJECT}).encode())
+            time.sleep(6)
+            compose('start', 'loki')
+            loki = address('loki', 3100)
+            wait_for('Loki outage recovery', lambda: entries('loki-outage-' + PROJECT))
+            passed('Alloy retries delivery across a short Loki outage')
+
+            compose('up', '-d', '--force-recreate', '--no-deps', 'loki')
+            loki = address('loki', 3100)
+            wait_for('recreated Loki readiness', lambda: 'ready' in request(loki + '/ready'))
+            wait_for('Loki persistent log data', lambda: entries(record['message']))
+            config_text = request(loki + '/config')
+            assert re.search(r'retention_period: (72h|3d)', config_text), 'Dev retention was not applied'
+            assert 'retention_enabled: true' in config_text
+            passed('Loki data survives container recreation; compactor retention active at 72h')
+            REPORT.parent.mkdir(parents=True, exist_ok=True)
+            REPORT.write_text(json.dumps({'checked_at_utc': datetime.datetime.now(datetime.timezone.utc).isoformat(),
+                'scope': 'isolated local Docker; synthetic app logs and OTLP trace; not EC2 deployment or browser UI',
+                'checks': checks, 'images': {k: v['image'] for k, v in services.items()},
+                'production_deployed': False}, ensure_ascii=False, indent=2) + '\n')
+            print('Report: ' + str(REPORT), flush=True)
+        except Exception:
+            print(compose('logs', '--no-color', '--tail', '35', 'alloy', 'loki', 'grafana'), flush=True)
+            raise
+        finally:
+            compose('down', '--volumes', '--remove-orphans')
+
+
+if __name__ == '__main__':
+    main()
diff --git a/deploy/observability/terraform/environments/dev/tests/log-pipeline.tftest.hcl b/deploy/observability/terraform/environments/dev/tests/log-pipeline.tftest.hcl
new file mode 100644
index 00000000..3b91d4b0
--- /dev/null
+++ b/deploy/observability/terraform/environments/dev/tests/log-pipeline.tftest.hcl
@@ -0,0 +1,76 @@
+mock_provider "aws" {
+  mock_data "aws_subnet" {
+    defaults = {
+      availability_zone = "us-east-1a"
+      vpc_id            = "vpc-0123456789abcdef0"
+    }
+  }
+}
+
+variables {
+  instance_name                = "local-test"
+  project_name                 = "beach-complex"
+  app_server_security_group_id = "sg-0123456789abcdef0"
+  app_server_private_ip        = "10.0.1.200"
+  subnet_id                    = "subnet-0123456789abcdef0"
+  ami_id                       = "ami-0123456789abcdef0"
+  instance_type                = "t3.small"
+  volume_size_gb               = 30
+}
+
+run "dev_logs" {
+  command = plan
+  module {
+    source = "../../modules/observability_ec2"
+  }
+  variables {
+    env = "dev"
+  }
+  assert {
+    condition = (
+      aws_vpc_security_group_ingress_rule.loki.from_port == 3100 &&
+      aws_vpc_security_group_ingress_rule.loki.to_port == 3100 &&
+      aws_vpc_security_group_ingress_rule.loki.ip_protocol == "tcp" &&
+      aws_vpc_security_group_ingress_rule.loki.referenced_security_group_id == var.app_server_security_group_id &&
+      aws_vpc_security_group_ingress_rule.loki.cidr_ipv4 == null &&
+      aws_vpc_security_group_ingress_rule.loki.cidr_ipv6 == null
+    )
+    error_message = "Loki must accept TCP 3100 only from the app security group."
+  }
+  assert {
+    condition     = local.loki_retention_period == "72h" && strcontains(local.cloud_init_plan_probe, "LOKI_RETENTION_PERIOD=72h")
+    error_message = "Dev bootstrap must set the ADR-014 three-day retention."
+  }
+  assert {
+    condition     = local.cloud_init_plan_probe_bytes <= 16384
+    error_message = "The new Loki asset must fit EC2 user data."
+  }
+}
+
+run "staging_retention" {
+  command = plan
+  module {
+    source = "../../modules/observability_ec2"
+  }
+  variables {
+    env = "staging"
+  }
+  assert {
+    condition     = strcontains(local.cloud_init_plan_probe, "LOKI_RETENTION_PERIOD=168h")
+    error_message = "Staging bootstrap must set seven-day retention."
+  }
+}
+
+run "prod_retention" {
+  command = plan
+  module {
+    source = "../../modules/observability_ec2"
+  }
+  variables {
+    env = "prod"
+  }
+  assert {
+    condition     = strcontains(local.cloud_init_plan_probe, "LOKI_RETENTION_PERIOD=336h")
+    error_message = "Prod bootstrap must set fourteen-day retention."
+  }
+}
diff --git a/deploy/observability/terraform/modules/observability_ec2/cloud-init.yml.tftpl b/deploy/observability/terraform/modules/observability_ec2/cloud-init.yml.tftpl
index 0e67b7fa..677b83c8 100644
--- a/deploy/observability/terraform/modules/observability_ec2/cloud-init.yml.tftpl
+++ b/deploy/observability/terraform/modules/observability_ec2/cloud-init.yml.tftpl
@@ -50,6 +50,18 @@ write_files:
     encoding: gzip+base64
     content: ${grafana_datasources_base64gzip}
 
+  - path: /run/beach-observability/loki.yaml
+    owner: root:root
+    permissions: "0644"
+    encoding: gzip+base64
+    content: ${loki_config_base64gzip}
+
+  - path: /run/beach-observability/observability.env
+    owner: root:root
+    permissions: "0644"
+    content: |
+      LOKI_RETENTION_PERIOD=${loki_retention_period}
+
   - path: /etc/systemd/system/docker.service.d/observability-mount.conf
     owner: root:root
     permissions: "0644"
@@ -217,14 +229,18 @@ write_files:
       install -d -m 0755 \
         "$compose_directory/prometheus" \
         "$compose_directory/tempo" \
+        "$compose_directory/loki" \
         "$compose_directory/grafana/provisioning/datasources"
       install -d -m 0755 -o 65534 -g 65534 "$data_directory/prometheus"
       install -d -m 0755 -o 10001 -g 10001 "$data_directory/tempo"
+      install -d -m 0755 -o 10001 -g 10001 "$data_directory/loki"
       install -d -m 0755 -o 472 -g 0 "$data_directory/grafana"
 
       install_asset "$staging_directory/docker-compose.yml" "$compose_directory/docker-compose.yml"
       install_asset "$staging_directory/prometheus.yml" "$compose_directory/prometheus/prometheus.yml"
       install_asset "$staging_directory/tempo.yaml" "$compose_directory/tempo/tempo.yaml"
+      install_asset "$staging_directory/loki.yaml" "$compose_directory/loki/local-config.yaml"
+      install_asset "$staging_directory/observability.env" "$compose_directory/.env"
       install_asset "$staging_directory/grafana-datasources.yml" "$compose_directory/grafana/provisioning/datasources/datasources.yml"
 
       docker compose --project-directory "$compose_directory" config --quiet
diff --git a/deploy/observability/terraform/modules/observability_ec2/main.tf b/deploy/observability/terraform/modules/observability_ec2/main.tf
index b79899da..5ba2ab3c 100644
--- a/deploy/observability/terraform/modules/observability_ec2/main.tf
+++ b/deploy/observability/terraform/modules/observability_ec2/main.tf
@@ -25,6 +25,10 @@ locals {
   tempo_config_base64gzip = base64gzip(
     file("${path.module}/../../../compose/tempo/tempo.yaml")
   )
+  loki_config_base64gzip = base64gzip(
+    file("${path.module}/../../../compose/loki/local-config.yaml")
+  )
+  loki_retention_period = lookup({ dev = "72h", staging = "168h", prod = "336h" }, var.env)
   grafana_datasources_base64gzip = base64gzip(
     file("${path.module}/../../../compose/grafana/provisioning/datasources/datasources.yml")
   )
@@ -39,6 +43,8 @@ locals {
     compose_base64gzip             = local.compose_base64gzip
     prometheus_config_base64gzip   = local.prometheus_config_base64gzip
     tempo_config_base64gzip        = local.tempo_config_base64gzip
+    loki_config_base64gzip         = local.loki_config_base64gzip
+    loki_retention_period          = local.loki_retention_period
     grafana_datasources_base64gzip = local.grafana_datasources_base64gzip
   })
 
@@ -52,6 +58,8 @@ locals {
     compose_base64gzip             = local.compose_base64gzip
     prometheus_config_base64gzip   = local.prometheus_config_base64gzip
     tempo_config_base64gzip        = local.tempo_config_base64gzip
+    loki_config_base64gzip         = local.loki_config_base64gzip
+    loki_retention_period          = local.loki_retention_period
     grafana_datasources_base64gzip = local.grafana_datasources_base64gzip
   })
 
@@ -115,6 +123,15 @@ resource "aws_vpc_security_group_ingress_rule" "otlp_http" {
   to_port                      = 4318
 }
 
+resource "aws_vpc_security_group_ingress_rule" "loki" {
+  security_group_id            = aws_security_group.this.id
+  referenced_security_group_id = var.app_server_security_group_id
+  description                  = "Loki log ingestion from the application server Alloy"
+  from_port                    = 3100
+  ip_protocol                  = "tcp"
+  to_port                      = 3100
+}
+
 resource "aws_vpc_security_group_egress_rule" "all" {
   security_group_id = aws_security_group.this.id
   description       = "Outbound access for package installation and service communication"
diff --git a/deploy/observability/verify-local.txt b/deploy/observability/verify-local.txt
new file mode 100644
index 00000000..3d20d971
--- /dev/null
+++ b/deploy/observability/verify-local.txt
@@ -0,0 +1,21 @@
+로그 파이프라인 로컬 검증
+
+준비
+- 저장소 루트에서 실행한다.
+- Docker Engine + Compose v2 이상, Python 3, Terraform 1.15.8, cloud-init이 필요하다.
+- AWS 자격증명이나 실제 AWS 리소스는 필요하지 않다.
+- 기존 beach.logs=backend 컨테이너가 있으면 중지 상태여도 통합 테스트가 중단된다.
+  이 경우 별도의 테스트용 Docker 데몬에서 실행한다.
+
+1. 설정 검사: Terraform mock plan 3건, cloud-init, Compose, Loki/Alloy
+bash deploy/observability/scripts/validate-iac.sh
+
+2. 통합 테스트: Docker 로그 수집, 회전 복구, 긴 JSON 재조립, Grafana/Tempo 연결
+python3 deploy/observability/scripts/verify-log-pipeline.py
+
+3. 결과 확인
+cat build/reports/observability/log-pipeline.json
+
+통합 테스트가 만든 컨테이너와 볼륨은 종료 시 자동으로 정리한다.
+서버 적용과 실제 앱 검증은 별도로 수행한다.
+자세한 검증 범위와 적용 절차: deploy/observability/README.md