feat(education): establish operational independence contracts

docs/education/migration/issues/EDU-015-operational-independence.md

@@ -1,9 +1,9 @@
 # EDU-015 — Operational independence and legacy exit
 
-- **Status:** blocked
+- **Status:** implemented-contracts
 - **Type:** integration and deployment program
 - **Phase:** 6
-- **Blockers:** EDU-011, EDU-013, EDU-014 child decisions, all temporary-adapter owners and exit plans
+- **Blockers:** production deployment evidence and implementation of each future EDU-011 worker/scanner workload
 
 ## Outcome
 
@@ -18,6 +18,17 @@ The target backend runs its selected education capabilities without depending on
 - Reconcile migrated data and operational runbooks.
 - Prove deployment, startup, Flyway, and core user flows.
 
+## Delivered contracts
+
+- `/admin-api/education/operations/health` exposes the runtime dependency registry plus durable Worker/Scanner and open dead-letter summaries.
+- `JAVA_READ` performs a target-only PostgreSQL schema readiness proof; Scalar is explicitly `NOT_SELECTED` in that mode.
+- `SCALAR_READ` is marked legacy, required only when selected, uses `FAIL_CLOSED`, and records success/failure telemetry without exposing URL or credentials.
+- V4170 owns durable `education_operational_component` and tenant-scoped `education_dead_letter` contracts without retaining business payloads.
+- `EducationTenantContextPropagation` re-establishes and restores tenant context for executor tasks.
+- `tools/education-target-smoke/java-read-readiness.sh` asserts target-only provider selection against a running target deployment.
+
+Production migration and deployment evidence remain release activities and are not claimed by this change.
+
 ## Acceptance criteria
 
 - [ ] Every temporary legacy dependency has an owner, telemetry, failure policy, and exit date.

docs/education/pilot-acceptance-runbook.md

@@ -13,18 +13,19 @@ yudao:
     catalog-read-enabled: true
     practice-write-enabled: true
     pilot-tenant-ids: [<pilot-tenant-id>]
-    catalog-mode: SCALAR_READ
+    catalog-mode: JAVA_READ
     scalar:
-      enabled: true
-      base-url: ${EDUCATION_SCALAR_BASE_URL}
-      token: ${EDUCATION_SCALAR_TOKEN}
+      enabled: false
+      owner: <required-only-when-scalar-read>
+      exit-date: <yyyy-MM-dd>
 ```
 
 要求：
 
 - `pilot-tenant-ids` 在 Pilot 环境必须显式配置，不能使用空列表。
-- Scalar token 只能通过密钥管理或环境变量注入，不写入仓库、日志或测试报告。
-- 发布前调用管理端 `/admin-api/education/capability`，核对模块、题库读取、练习写入和 Pilot 租户数量。
+- Pilot 默认以 `JAVA_READ` 启动，只依赖目标 PostgreSQL；不得启动旧 NestJS API、旧 Worker、Supabase 或旧资产扫描服务作为前置条件。
+- 仅在有明确负责人、告警、故障策略和退出日期的兼容窗口内切换 `SCALAR_READ`。Scalar token 只能通过密钥管理或环境变量注入。
+- 发布前调用管理端 `/admin-api/education/capability` 和 `/admin-api/education/operations/health`；后者必须显示 `java-read-postgresql=UP`、`scalar-catalog=NOT_SELECTED`。
 
 ## 3. 发布步骤
 
@@ -32,9 +33,10 @@ yudao:
 2. 使用 Server 配置的 PostgreSQL Flyway 执行 migrate 和 validate，检查版本、脚本、checksum 与 success；不得手工应用 Education SQL 或执行 rollback SQL。
 3. 先以 `catalog-read-enabled=false`、`practice-write-enabled=false` 部署应用。
 4. 验证 System、Infra、Member 基础 smoke。
-5. 仅对 Pilot 租户开启题库读取，完成 Scalar 只读 smoke。
+5. 仅对 Pilot 租户开启题库读取，完成 JAVA_READ 只读 smoke，并运行 `tools/education-target-smoke/java-read-readiness.sh`。
 6. 对 Pilot 租户开启练习写入，完成会话、答案、交卷、报告、错题和收藏 smoke。
-7. 观察错误率、延迟和数据库写入后再扩大租户列表。
+7. 若部署 Worker 或 Scanner，先确认其持续写入 `education_operational_component`，且 `/admin-api/education/operations/health` 无 `DOWN` 组件和未处理死信。
+8. 观察错误率、延迟和数据库写入后再扩大租户列表。
 
 ## 4. Smoke 清单
 
@@ -60,7 +62,17 @@ yudao:
 
 ## 5. 故障与回滚
 
-### Scalar 故障
+### JAVA_READ / PostgreSQL 故障
+
+1. 设置 `catalog-read-enabled=false`，停止新的目录和题目读取；不得静默切回 Scalar。
+2. 保持 `enabled=true`，使已有会话、报告、错题和收藏仍可访问。
+3. 检查 `/admin-api/education/operations/health` 的 `java-read-postgresql` 结果和 Flyway 历史。
+4. 如需冻结新写入，再设置 `practice-write-enabled=false`。
+5. 通过应用回滚或更高版本 Flyway 前滚修复，不执行 `flyway clean` 或手工回滚 SQL。
+
+### Scalar 兼容窗口故障
+
+仅当部署明确选择 `SCALAR_READ` 时适用：
 
 1. 设置 `catalog-read-enabled=false`，停止新的 Scalar 读取。
 2. 保持 `enabled=true`，使已有会话、报告、错题和收藏仍可访问。
@@ -88,7 +100,10 @@ yudao:
 
 发布窗口至少观察：
 
-- Scalar 请求成功率、4xx/5xx/timeout、P95/P99 延迟；
+- `/admin-api/education/operations/health` 的必需依赖、Worker/Scanner 心跳和 open dead-letter 数；
+- `education_operational_component` 的 `last_heartbeat_at`、最后成功/失败和 backlog；
+- `education_dead_letter` 仅保留负载指纹与脱敏错误分类，不得保存业务 payload、凭据或个人数据；
+- Scalar 兼容模式请求成功率、4xx/5xx/timeout、P95/P99 延迟；
 - 练习创建成功/冲突数；
 - 答案保存成功、幂等重放、版本冲突和旧序号拒绝数；
 - 交卷成功、并发冲突和事务失败数；
@@ -97,6 +112,14 @@ yudao:
 
 Scalar 日志只能记录脱敏路径、tenant ID、上游 request ID、状态、耗时和错误分类；不得记录 Authorization、Scalar token、学生答案、正确答案或完整响应体。RuoYi access/error log 中的 trace ID 用于关联入口请求；验收时需保存一条从入口日志到 Scalar request ID 的关联证据。
 
+### Worker、Scanner 与死信处置
+
+1. Worker/Scanner 每次心跳使用固定 `component_key` upsert；部署实例变化写入 `instance_id`。
+2. 心跳状态只能为 `STARTING/UP/DEGRADED/DOWN`，`detail` 必须脱敏且有界。
+3. 重试耗尽后写入 `education_dead_letter`；同一租户、组件、workload 只允许一个 OPEN 记录。
+4. 排障后先把原 OPEN 记录标记为 `REQUEUED` 并填写 `resolved_at`/`resolution_note`，再通过所属业务服务重入队；禁止直接修改业务结果或把原 payload 写入死信表。
+5. 未部署对应 Worker/Scanner 时不得伪造 UP 心跳；能力清单应保持未交付状态。
+
 ## 7. 验证命令
 
 ```bash

tools/education-target-smoke/java-read-readiness.sh

@@ -0,0 +1,25 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+BASE_URL="${BASE_URL:-http://127.0.0.1:48080}"
+TOKEN="${EDUCATION_ADMIN_TOKEN:?EDUCATION_ADMIN_TOKEN is required}"
+TENANT_ID="${EDUCATION_TENANT_ID:?EDUCATION_TENANT_ID is required}"
+
+response="$(curl --fail --silent --show-error \
+  -H "Authorization: Bearer ${TOKEN}" \
+  -H "tenant-id: ${TENANT_ID}" \
+  "${BASE_URL}/admin-api/education/operations/health")"
+
+python3 - "$response" <<'PY'
+import json, sys
+body = json.loads(sys.argv[1])
+assert body.get("code") == 0, body
+health = body["data"]
+deps = {item["key"]: item for item in health["dependencies"]}
+java = deps.get("java-read-postgresql")
+scalar = deps.get("scalar-catalog")
+assert java and java["required"] and java["status"] == "UP", deps
+assert scalar and not scalar["required"] and scalar["status"] == "NOT_SELECTED", deps
+assert health["status"] in ("UP", "DEGRADED"), health
+print("target-only JAVA_READ readiness proof passed")
+PY

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/controller/admin/EducationCapabilityController.java

@@ -68,17 +68,25 @@ public class EducationCapabilityController {
                         List.of("capability-family-disposition", "entitlement-and-export-policy"),
                         List.of(), "FEATURE_FLAG", "CONTRACT_ONLY"),
                 theme("EDU-015", "operational-independence",
-                        catalogDependencies.isEmpty() ? "PARTIAL" : "LEGACY_DEPENDENT",
+                        catalogDependencies.isEmpty() ? "TARGET_READY" : "LEGACY_DEPENDENT",
                         List.of("education", "infra"),
-                        List.of("target-only-runtime-evidence", "worker-scanner-exit-contract"),
-                        catalogDependencies, "PROVIDER_SWITCH", "INTEGRATION_TESTED"));
+                        List.of(),
+                        catalogDependencies, "PROVIDER_SWITCH", "RUNTIME_PROBED",
+                        List.of("/admin-api/education/operations/health")));
     }
 
     private EducationThemeRespVO theme(String key, String name, String status, List<String> owningModules,
             List<String> blockers, List<String> legacyDependencies, String rollbackCategory, String evidenceLevel) {
+        return theme(key, name, status, owningModules, blockers, legacyDependencies, rollbackCategory, evidenceLevel,
+                List.of());
+    }
+
+    private EducationThemeRespVO theme(String key, String name, String status, List<String> owningModules,
+            List<String> blockers, List<String> legacyDependencies, String rollbackCategory, String evidenceLevel,
+            List<String> executableInterfaces) {
         return EducationThemeRespVO.builder()
                 .key(key).name(name).status(status).owningModules(owningModules)
-                .executableInterfaces(List.of()).blockers(blockers).legacyDependencies(legacyDependencies)
+                .executableInterfaces(executableInterfaces).blockers(blockers).legacyDependencies(legacyDependencies)
                 .rollbackCategory(rollbackCategory).evidenceLevel(evidenceLevel).build();
     }
 

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/controller/admin/operations/EducationOperationalHealthController.java

@@ -0,0 +1,57 @@
+package cn.iocoder.yudao.module.education.controller.admin.operations;
+
+import cn.iocoder.yudao.framework.common.pojo.CommonResult;
+import cn.iocoder.yudao.framework.tenant.core.util.TenantUtils;
+import cn.iocoder.yudao.module.education.controller.admin.operations.vo.EducationOperationalHealthRespVO;
+import cn.iocoder.yudao.module.education.dal.mysql.operations.EducationDeadLetterMapper;
+import cn.iocoder.yudao.module.education.dal.mysql.operations.OperationalComponentMapper;
+import cn.iocoder.yudao.module.education.runtime.EducationDependencyStatus;
+import cn.iocoder.yudao.module.education.runtime.EducationRuntimeDependencyRegistry;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.security.access.prepost.PreAuthorize;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+import static cn.iocoder.yudao.framework.common.pojo.CommonResult.success;
+
+@Tag(name = "管理后台 - 教育运行健康")
+@RestController
+@RequestMapping("/education/operations")
+@ConditionalOnProperty(prefix = "yudao.education", name = "enabled", havingValue = "true")
+public class EducationOperationalHealthController {
+
+    private final EducationRuntimeDependencyRegistry dependencyRegistry;
+    private final OperationalComponentMapper componentMapper;
+    private final EducationDeadLetterMapper deadLetterMapper;
+
+    public EducationOperationalHealthController(EducationRuntimeDependencyRegistry dependencyRegistry,
+                                                OperationalComponentMapper componentMapper,
+                                                EducationDeadLetterMapper deadLetterMapper) {
+        this.dependencyRegistry = dependencyRegistry;
+        this.componentMapper = componentMapper;
+        this.deadLetterMapper = deadLetterMapper;
+    }
+
+    @GetMapping("/health")
+    @Operation(summary = "获得教育运行依赖、Worker、Scanner 与死信健康")
+    @PreAuthorize("@ss.hasPermission('education:capability')")
+    public CommonResult<EducationOperationalHealthRespVO> getHealth() {
+        List<EducationDependencyStatus> dependencies = dependencyRegistry.checkAll();
+        return success(TenantUtils.executeIgnore(() -> {
+            var components = componentMapper.selectOperationalComponents().stream()
+                    .map(EducationOperationalHealthRespVO.ComponentHealth::from).toList();
+            long openDeadLetters = deadLetterMapper.countOpenAllTenants();
+            boolean dependenciesUp = dependencies.stream()
+                    .noneMatch(item -> item.isRequired() && "DOWN".equals(item.getStatus()));
+            boolean componentsUp = components.stream().noneMatch(item -> "DOWN".equals(item.getStatus()));
+            String status = dependenciesUp && componentsUp && openDeadLetters == 0 ? "UP" : "DEGRADED";
+            return EducationOperationalHealthRespVO.builder().status(status).dependencies(dependencies)
+                    .components(components).openDeadLetters(openDeadLetters).build();
+        }));
+    }
+}

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/controller/admin/operations/vo/EducationOperationalHealthRespVO.java

@@ -0,0 +1,47 @@
+package cn.iocoder.yudao.module.education.controller.admin.operations.vo;
+
+import cn.iocoder.yudao.module.education.dal.dataobject.operations.OperationalComponentDO;
+import cn.iocoder.yudao.module.education.runtime.EducationDependencyStatus;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.Builder;
+import lombok.Data;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+@Schema(description = "管理后台 - 教育运行依赖与后台组件健康 Response VO")
+@Data
+@Builder
+public class EducationOperationalHealthRespVO {
+
+    private String status;
+    private List<EducationDependencyStatus> dependencies;
+    private List<ComponentHealth> components;
+    private long openDeadLetters;
+
+    @Data
+    @Builder
+    public static class ComponentHealth {
+        private String key;
+        private String type;
+        private String instanceId;
+        private String status;
+        private LocalDateTime lastHeartbeatAt;
+        private LocalDateTime lastSuccessAt;
+        private LocalDateTime lastFailureAt;
+        private String failureCategory;
+        private String detail;
+        private Long pendingCount;
+        private Long retryCount;
+        private Long deadLetterCount;
+
+        public static ComponentHealth from(OperationalComponentDO component) {
+            return ComponentHealth.builder().key(component.getComponentKey()).type(component.getComponentType())
+                    .instanceId(component.getInstanceId()).status(component.getStatus())
+                    .lastHeartbeatAt(component.getLastHeartbeatAt()).lastSuccessAt(component.getLastSuccessAt())
+                    .lastFailureAt(component.getLastFailureAt()).failureCategory(component.getFailureCategory())
+                    .detail(component.getDetail()).pendingCount(component.getPendingCount())
+                    .retryCount(component.getRetryCount()).deadLetterCount(component.getDeadLetterCount()).build();
+        }
+    }
+}

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/operations/EducationDeadLetterDO.java

@@ -0,0 +1,30 @@
+package cn.iocoder.yudao.module.education.dal.dataobject.operations;
+
+import cn.iocoder.yudao.framework.tenant.core.db.TenantBaseDO;
+import com.baomidou.mybatisplus.annotation.KeySequence;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.time.LocalDateTime;
+
+@TableName("education_dead_letter")
+@KeySequence("education_dead_letter_seq")
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class EducationDeadLetterDO extends TenantBaseDO {
+
+    @TableId
+    private Long id;
+    private String componentKey;
+    private String workloadKey;
+    private String payloadFingerprint;
+    private String failureCategory;
+    private Integer attempts;
+    private String status;
+    private LocalDateTime firstFailedAt;
+    private LocalDateTime lastFailedAt;
+    private LocalDateTime resolvedAt;
+    private String resolutionNote;
+}

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/dataobject/operations/OperationalComponentDO.java

@@ -0,0 +1,29 @@
+package cn.iocoder.yudao.module.education.dal.dataobject.operations;
+
+import cn.iocoder.yudao.framework.mybatis.core.dataobject.BaseDO;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+import java.time.LocalDateTime;
+
+@TableName("education_operational_component")
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class OperationalComponentDO extends BaseDO {
+
+    @TableId
+    private String componentKey;
+    private String componentType;
+    private String instanceId;
+    private String status;
+    private LocalDateTime lastHeartbeatAt;
+    private LocalDateTime lastSuccessAt;
+    private LocalDateTime lastFailureAt;
+    private String failureCategory;
+    private String detail;
+    private Long pendingCount;
+    private Long retryCount;
+    private Long deadLetterCount;
+}

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/operations/EducationDeadLetterMapper.java

@@ -0,0 +1,16 @@
+package cn.iocoder.yudao.module.education.dal.mysql.operations;
+
+import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
+import cn.iocoder.yudao.module.education.dal.dataobject.operations.EducationDeadLetterDO;
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Select;
+
+@Mapper
+public interface EducationDeadLetterMapper extends BaseMapperX<EducationDeadLetterDO> {
+
+    @Select("""
+            SELECT count(*) FROM education_dead_letter
+             WHERE status = 'OPEN' AND deleted = false
+            """)
+    long countOpenAllTenants();
+}

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/dal/mysql/operations/OperationalComponentMapper.java

@@ -0,0 +1,38 @@
+package cn.iocoder.yudao.module.education.dal.mysql.operations;
+
+import cn.iocoder.yudao.framework.mybatis.core.mapper.BaseMapperX;
+import cn.iocoder.yudao.module.education.dal.dataobject.operations.OperationalComponentDO;
+import org.apache.ibatis.annotations.Insert;
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Select;
+
+import java.util.List;
+
+@Mapper
+public interface OperationalComponentMapper extends BaseMapperX<OperationalComponentDO> {
+
+    @Insert("""
+            INSERT INTO education_operational_component
+                (component_key, component_type, instance_id, status, last_heartbeat_at,
+                 last_success_at, last_failure_at, failure_category, detail,
+                 pending_count, retry_count, dead_letter_count, update_time)
+            VALUES (#{componentKey}, #{componentType}, #{instanceId}, #{status}, CURRENT_TIMESTAMP,
+                    #{lastSuccessAt}, #{lastFailureAt}, #{failureCategory}, #{detail},
+                    #{pendingCount}, #{retryCount}, #{deadLetterCount}, CURRENT_TIMESTAMP)
+            ON CONFLICT (component_key) DO UPDATE SET
+                component_type = EXCLUDED.component_type, instance_id = EXCLUDED.instance_id,
+                status = EXCLUDED.status, last_heartbeat_at = CURRENT_TIMESTAMP,
+                last_success_at = EXCLUDED.last_success_at, last_failure_at = EXCLUDED.last_failure_at,
+                failure_category = EXCLUDED.failure_category, detail = EXCLUDED.detail,
+                pending_count = EXCLUDED.pending_count, retry_count = EXCLUDED.retry_count,
+                dead_letter_count = EXCLUDED.dead_letter_count, update_time = CURRENT_TIMESTAMP
+            """)
+    int upsertHeartbeat(OperationalComponentDO component);
+
+    @Select("""
+            SELECT * FROM education_operational_component
+             WHERE deleted = false
+             ORDER BY component_type, component_key
+            """)
+    List<OperationalComponentDO> selectOperationalComponents();
+}

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/integration/scalar/config/ScalarAutoConfiguration.java

@@ -8,6 +8,7 @@ import cn.iocoder.yudao.module.education.service.catalog.ScalarCatalogProvider;
 import cn.iocoder.yudao.module.education.service.catalog.UnsupportedModeCatalogProvider;
 import cn.iocoder.yudao.module.education.service.question.UnsupportedModeQuestionCatalogProvider;
 import cn.iocoder.yudao.module.education.service.catalog.provider.JavaCatalogProvider;
+import cn.iocoder.yudao.module.education.runtime.ScalarLegacyDependencyTelemetry;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
@@ -28,8 +29,9 @@ public class ScalarAutoConfiguration {
 
     @Bean
     @ConditionalOnProperty(prefix = "yudao.education", name = "catalog-mode", havingValue = "SCALAR_READ", matchIfMissing = true)
-    public ScalarCatalogProvider scalarCatalogProvider(ScalarProperties scalarProperties) {
-        return new ScalarCatalogProvider(scalarProperties);
+    public ScalarCatalogProvider scalarCatalogProvider(ScalarProperties scalarProperties,
+                                                       ScalarLegacyDependencyTelemetry telemetry) {
+        return new ScalarCatalogProvider(scalarProperties, telemetry);
     }
 
     // JAVA_READ mode is handled by JavaCatalogProvider @Component

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/integration/scalar/config/ScalarProperties.java

@@ -34,6 +34,16 @@ public class ScalarProperties {
      */
     private String token;
 
+    /**
+     * Scalar 兼容依赖负责人。启用 SCALAR_READ 时必须显式配置。
+     */
+    private String owner;
+
+    /**
+     * Scalar 兼容依赖退出日期，ISO-8601 yyyy-MM-dd。启用 SCALAR_READ 时必须显式配置。
+     */
+    private java.time.LocalDate exitDate;
+
     /**
      * 连接超时时间，默认 5 秒。
      */

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/runtime/EducationDependencyStatus.java

@@ -0,0 +1,21 @@
+package cn.iocoder.yudao.module.education.runtime;
+
+import lombok.Builder;
+import lombok.Value;
+
+import java.time.LocalDateTime;
+
+@Value
+@Builder
+public class EducationDependencyStatus {
+    String key;
+    String kind;
+    boolean required;
+    boolean legacy;
+    String owner;
+    LocalDateTime exitAt;
+    String failurePolicy;
+    String status;
+    String detail;
+    LocalDateTime checkedAt;
+}

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/runtime/EducationRuntimeDependency.java

@@ -0,0 +1,6 @@
+package cn.iocoder.yudao.module.education.runtime;
+
+public interface EducationRuntimeDependency {
+
+    EducationDependencyStatus check();
+}

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/runtime/EducationRuntimeDependencyRegistry.java

@@ -0,0 +1,21 @@
+package cn.iocoder.yudao.module.education.runtime;
+
+import org.springframework.stereotype.Component;
+
+import java.util.Comparator;
+import java.util.List;
+
+@Component
+public class EducationRuntimeDependencyRegistry {
+
+    private final List<EducationRuntimeDependency> dependencies;
+
+    public EducationRuntimeDependencyRegistry(List<EducationRuntimeDependency> dependencies) {
+        this.dependencies = dependencies;
+    }
+
+    public List<EducationDependencyStatus> checkAll() {
+        return dependencies.stream().map(EducationRuntimeDependency::check)
+                .sorted(Comparator.comparing(EducationDependencyStatus::getKey)).toList();
+    }
+}

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/runtime/JavaReadPostgreSqlReadiness.java

@@ -0,0 +1,48 @@
+package cn.iocoder.yudao.module.education.runtime;
+
+import cn.iocoder.yudao.module.education.config.EducationProperties;
+import cn.iocoder.yudao.module.education.enums.CatalogProviderMode;
+import org.springframework.stereotype.Component;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.time.LocalDateTime;
+
+@Component
+public class JavaReadPostgreSqlReadiness implements EducationRuntimeDependency {
+
+    private final EducationProperties properties;
+    private final DataSource dataSource;
+
+    public JavaReadPostgreSqlReadiness(EducationProperties properties, DataSource dataSource) {
+        this.properties = properties;
+        this.dataSource = dataSource;
+    }
+
+    @Override
+    public EducationDependencyStatus check() {
+        boolean selected = properties.getCatalogMode() == CatalogProviderMode.JAVA_READ;
+        if (!selected) {
+            return status(false, "NOT_SELECTED", "JAVA_READ is not selected");
+        }
+        try (Connection connection = dataSource.getConnection();
+             var statement = connection.prepareStatement("""
+                     SELECT to_regclass('education_question') IS NOT NULL
+                        AND to_regclass('education_content_node') IS NOT NULL
+                        AND to_regclass('education_question_collection') IS NOT NULL
+                     """); ResultSet result = statement.executeQuery()) {
+            boolean ready = result.next() && result.getBoolean(1);
+            return status(true, ready ? "UP" : "DOWN",
+                    ready ? "PostgreSQL catalog schema reachable" : "required catalog tables are missing");
+        } catch (Exception ex) {
+            return status(true, "DOWN", ex.getClass().getSimpleName());
+        }
+    }
+
+    private EducationDependencyStatus status(boolean required, String status, String detail) {
+        return EducationDependencyStatus.builder().key("java-read-postgresql").kind("POSTGRESQL")
+                .required(required).legacy(false).failurePolicy("FAIL_CLOSED")
+                .status(status).detail(detail).checkedAt(LocalDateTime.now()).build();
+    }
+}

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/runtime/ScalarLegacyDependencyTelemetry.java

@@ -0,0 +1,52 @@
+package cn.iocoder.yudao.module.education.runtime;
+
+import cn.iocoder.yudao.module.education.config.EducationProperties;
+import cn.iocoder.yudao.module.education.enums.CatalogProviderMode;
+import cn.iocoder.yudao.module.education.integration.scalar.config.ScalarProperties;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDateTime;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.LongAdder;
+
+@Component
+public class ScalarLegacyDependencyTelemetry implements EducationRuntimeDependency {
+
+    private final EducationProperties educationProperties;
+    private final ScalarProperties scalarProperties;
+    private final ConcurrentHashMap<String, LongAdder> outcomes = new ConcurrentHashMap<>();
+
+    public ScalarLegacyDependencyTelemetry(EducationProperties educationProperties, ScalarProperties scalarProperties) {
+        this.educationProperties = educationProperties;
+        this.scalarProperties = scalarProperties;
+    }
+
+    public void record(String outcome) {
+        outcomes.computeIfAbsent(outcome, ignored -> new LongAdder()).increment();
+    }
+
+    public long count(String outcome) {
+        LongAdder counter = outcomes.get(outcome);
+        return counter != null ? counter.sum() : 0;
+    }
+
+    @Override
+    public EducationDependencyStatus check() {
+        boolean selected = educationProperties.getCatalogMode() == CatalogProviderMode.SCALAR_READ;
+        boolean configured = scalarProperties.isEnabled() && hasText(scalarProperties.getBaseUrl())
+                && hasText(scalarProperties.getToken()) && hasText(scalarProperties.getOwner())
+                && scalarProperties.getExitDate() != null;
+        String status = !selected ? "NOT_SELECTED" : configured ? "UP" : "DOWN";
+        String detail = "success=" + count("success") + ",failure=" + count("failure");
+        return EducationDependencyStatus.builder().key("scalar-catalog").kind("HTTP")
+                .required(selected).legacy(true).owner(scalarProperties.getOwner())
+                .exitAt(scalarProperties.getExitDate() != null
+                        ? scalarProperties.getExitDate().atStartOfDay() : null)
+                .failurePolicy("FAIL_CLOSED")
+                .status(status).detail(detail).checkedAt(LocalDateTime.now()).build();
+    }
+
+    private static boolean hasText(String value) {
+        return value != null && !value.isBlank();
+    }
+}

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/service/catalog/ScalarCatalogProvider.java

@@ -16,6 +16,7 @@ import cn.iocoder.yudao.module.education.service.question.dto.CatalogQuestionDTO
 import cn.iocoder.yudao.module.education.integration.scalar.dto.ScalarQuestionResponseDto;
 import cn.iocoder.yudao.module.education.integration.scalar.dto.ScalarQuestionOptionDto;
 import cn.iocoder.yudao.module.education.integration.scalar.dto.ScalarPracticeBlueprintResponseDto;
+import cn.iocoder.yudao.module.education.runtime.ScalarLegacyDependencyTelemetry;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.restclient.RestTemplateBuilder;
 import org.springframework.core.ParameterizedTypeReference;
@@ -50,6 +51,7 @@ import static cn.iocoder.yudao.module.education.enums.ErrorCodeConstants.*;
 public class ScalarCatalogProvider implements CatalogProvider, QuestionCatalogProvider {
 
     private final ScalarProperties scalarProperties;
+    private final ScalarLegacyDependencyTelemetry telemetry;
     private final RestTemplate restTemplate;
 
     private static final ParameterizedTypeReference<ScalarListResponse<ScalarCatalogEntityDto>> ENTITY_LIST_TYPE =
@@ -70,7 +72,12 @@ public class ScalarCatalogProvider implements CatalogProvider, QuestionCatalogPr
             new ParameterizedTypeReference<>() {};
 
     public ScalarCatalogProvider(ScalarProperties scalarProperties) {
+        this(scalarProperties, null);
+    }
+
+    public ScalarCatalogProvider(ScalarProperties scalarProperties, ScalarLegacyDependencyTelemetry telemetry) {
         this.scalarProperties = scalarProperties;
+        this.telemetry = telemetry;
         if (scalarProperties.isEnabled()) {
             if (scalarProperties.getBaseUrl() == null || scalarProperties.getBaseUrl().isBlank()) {
                 throw exception(CATALOG_SCALAR_NOT_CONFIGURED);
@@ -226,6 +233,7 @@ public class ScalarCatalogProvider implements CatalogProvider, QuestionCatalogPr
             String requestId = body.getMeta() != null ? body.getMeta().getRequestId() : null;
             log.info("[Scalar] ok path={} tenant={} requestId={} item-type={} elapsed={}ms",
                     sanitizedPath, tenantId, requestId, item.getClass().getSimpleName(), elapsed);
+            recordTelemetry("success");
 
             return item;
         } catch (HttpClientErrorException e) {
@@ -233,12 +241,14 @@ public class ScalarCatalogProvider implements CatalogProvider, QuestionCatalogPr
         } catch (HttpServerErrorException e) {
             return handleServerError(path, sanitizedPath, start, e);
         } catch (ResourceAccessException e) {
+            recordTelemetry("failure");
             long elapsed = Duration.between(start, Instant.now()).toMillis();
             String category = classifyResourceError(e);
             log.error("[Scalar] io-error path={} tenant={} category={} elapsed={}ms",
                     sanitizedPath, tenantId, category, elapsed);
             throw exception(mapResourceCategory(category));
         } catch (RestClientException e) {
+            recordTelemetry("failure");
             long elapsed = Duration.between(start, Instant.now()).toMillis();
             log.error("[Scalar] conversion-error path={} tenant={} type={} elapsed={}ms",
                     sanitizedPath, tenantId, e.getClass().getSimpleName(), elapsed);
@@ -281,6 +291,7 @@ public class ScalarCatalogProvider implements CatalogProvider, QuestionCatalogPr
             String requestId = body.getMeta() != null ? body.getMeta().getRequestId() : null;
             log.info("[Scalar] ok path={} tenant={} requestId={} items={} elapsed={}ms",
                     sanitizedPath, tenantId, requestId, items.size(), elapsed);
+            recordTelemetry("success");
 
             return items;
         } catch (HttpClientErrorException e) {
@@ -288,12 +299,14 @@ public class ScalarCatalogProvider implements CatalogProvider, QuestionCatalogPr
         } catch (HttpServerErrorException e) {
             return handleServerError(path, sanitizedPath, start, e);
         } catch (ResourceAccessException e) {
+            recordTelemetry("failure");
             long elapsed = Duration.between(start, Instant.now()).toMillis();
             String category = classifyResourceError(e);
             log.error("[Scalar] io-error path={} tenant={} category={} elapsed={}ms",
                     sanitizedPath, tenantId, category, elapsed);
             throw exception(mapResourceCategory(category));
         } catch (RestClientException e) {
+            recordTelemetry("failure");
             long elapsed = Duration.between(start, Instant.now()).toMillis();
             log.error("[Scalar] conversion-error path={} tenant={} type={} elapsed={}ms",
                     sanitizedPath, tenantId, e.getClass().getSimpleName(), elapsed);
@@ -310,6 +323,7 @@ public class ScalarCatalogProvider implements CatalogProvider, QuestionCatalogPr
     }
 
     private <T> T handleClientError(String path, String sanitizedPath, Instant start, HttpClientErrorException e) {
+        recordTelemetry("failure");
         long elapsed = Duration.between(start, Instant.now()).toMillis();
         HttpStatusCode statusCode = e.getStatusCode();
         String requestId = extractRequestId(e);
@@ -334,6 +348,7 @@ public class ScalarCatalogProvider implements CatalogProvider, QuestionCatalogPr
     }
 
     private <T> T handleServerError(String path, String sanitizedPath, Instant start, HttpServerErrorException e) {
+        recordTelemetry("failure");
         long elapsed = Duration.between(start, Instant.now()).toMillis();
         String requestId = extractRequestId(e);
         log.error("[Scalar] server-error path={} tenant={} status={} requestId={} elapsed={}ms",
@@ -373,6 +388,12 @@ public class ScalarCatalogProvider implements CatalogProvider, QuestionCatalogPr
         return null;
     }
 
+    private void recordTelemetry(String outcome) {
+        if (telemetry != null) {
+            telemetry.record(outcome);
+        }
+    }
+
     private Long getTenantId() {
         try {
             return TenantContextHolder.getRequiredTenantId();
@@ -674,12 +695,14 @@ public class ScalarCatalogProvider implements CatalogProvider, QuestionCatalogPr
         } catch (HttpServerErrorException e) {
             return handleServerError(path, sanitizedPath, start, e);
         } catch (ResourceAccessException e) {
+            recordTelemetry("failure");
             long elapsed = Duration.between(start, Instant.now()).toMillis();
             String category = classifyResourceError(e);
             log.error("[Scalar] io-error path={} tenant={} category={} elapsed={}ms",
                     sanitizedPath, tenantId, category, elapsed);
             throw exception(mapResourceCategory(category));
         } catch (RestClientException e) {
+            recordTelemetry("failure");
             long elapsed = Duration.between(start, Instant.now()).toMillis();
             log.error("[Scalar] conversion-error path={} tenant={} type={} elapsed={}ms",
                     sanitizedPath, tenantId, e.getClass().getSimpleName(), elapsed);

yudao-module-education/src/main/java/cn/iocoder/yudao/module/education/service/operations/EducationTenantContextPropagation.java

@@ -0,0 +1,24 @@
+package cn.iocoder.yudao.module.education.service.operations;
+
+import cn.iocoder.yudao.framework.tenant.core.util.TenantUtils;
+
+import java.util.concurrent.Callable;
+import java.util.function.Supplier;
+
+public final class EducationTenantContextPropagation {
+
+    private EducationTenantContextPropagation() {
+    }
+
+    public static Runnable wrap(Long tenantId, Runnable task) {
+        return () -> TenantUtils.execute(tenantId, task);
+    }
+
+    public static <T> Callable<T> wrap(Long tenantId, Callable<T> task) {
+        return () -> TenantUtils.execute(tenantId, task);
+    }
+
+    public static <T> Supplier<T> wrapSupplier(Long tenantId, Supplier<T> task) {
+        return () -> TenantUtils.execute(tenantId, task::get);
+    }
+}

yudao-module-education/src/main/resources/db/migration/education/V4170__add_operational_observability.sql

@@ -0,0 +1,72 @@
+-- EDU-015: durable operational heartbeat and dead-letter observability contracts.
+-- These tables contain operational metadata only; dead letters retain fingerprints, never payloads or secrets.
+
+CREATE TABLE education_operational_component (
+    component_key VARCHAR(100) PRIMARY KEY,
+    component_type VARCHAR(16) NOT NULL,
+    instance_id VARCHAR(100) NOT NULL,
+    status VARCHAR(16) NOT NULL,
+    last_heartbeat_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    last_success_at TIMESTAMP,
+    last_failure_at TIMESTAMP,
+    failure_category VARCHAR(64),
+    detail VARCHAR(255),
+    pending_count BIGINT NOT NULL DEFAULT 0,
+    retry_count BIGINT NOT NULL DEFAULT 0,
+    dead_letter_count BIGINT NOT NULL DEFAULT 0,
+    creator VARCHAR(64) DEFAULT '',
+    create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updater VARCHAR(64) DEFAULT '',
+    update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    deleted BOOLEAN NOT NULL DEFAULT false,
+    CONSTRAINT ck_education_operational_component_type
+        CHECK (component_type IN ('WORKER', 'SCANNER')),
+    CONSTRAINT ck_education_operational_component_status
+        CHECK (status IN ('STARTING', 'UP', 'DEGRADED', 'DOWN')),
+    CONSTRAINT ck_education_operational_component_counts
+        CHECK (pending_count >= 0 AND retry_count >= 0 AND dead_letter_count >= 0)
+);
+
+COMMENT ON TABLE education_operational_component IS 'EDU-015 durable worker and scanner heartbeat summary';
+COMMENT ON COLUMN education_operational_component.detail IS 'Sanitized bounded diagnostic; no payloads, credentials, or personal data';
+
+CREATE INDEX idx_education_operational_component_heartbeat
+    ON education_operational_component (component_type, last_heartbeat_at);
+
+CREATE TABLE education_dead_letter (
+    id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
+    tenant_id BIGINT NOT NULL,
+    component_key VARCHAR(100) NOT NULL,
+    workload_key VARCHAR(128) NOT NULL,
+    payload_fingerprint VARCHAR(128) NOT NULL,
+    failure_category VARCHAR(64) NOT NULL,
+    attempts INTEGER NOT NULL,
+    status VARCHAR(16) NOT NULL DEFAULT 'OPEN',
+    first_failed_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    last_failed_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    resolved_at TIMESTAMP,
+    resolution_note VARCHAR(255),
+    creator VARCHAR(64) DEFAULT '',
+    create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updater VARCHAR(64) DEFAULT '',
+    update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    deleted BOOLEAN NOT NULL DEFAULT false,
+    CONSTRAINT fk_education_dead_letter_component
+        FOREIGN KEY (component_key) REFERENCES education_operational_component(component_key),
+    CONSTRAINT ck_education_dead_letter_attempts CHECK (attempts > 0),
+    CONSTRAINT ck_education_dead_letter_status CHECK (status IN ('OPEN', 'REQUEUED', 'RESOLVED')),
+    CONSTRAINT ck_education_dead_letter_resolution CHECK (
+        (status = 'OPEN' AND resolved_at IS NULL) OR
+        (status IN ('REQUEUED', 'RESOLVED') AND resolved_at IS NOT NULL))
+);
+
+COMMENT ON TABLE education_dead_letter IS 'EDU-015 tenant-scoped dead-letter observability without retained business payload';
+COMMENT ON COLUMN education_dead_letter.payload_fingerprint IS 'Stable digest used for correlation and duplicate-safe handling; not the payload';
+
+CREATE UNIQUE INDEX uk_education_dead_letter_open_workload
+    ON education_dead_letter (tenant_id, component_key, workload_key)
+    WHERE status = 'OPEN' AND deleted = false;
+CREATE INDEX idx_education_dead_letter_status_failure
+    ON education_dead_letter (status, last_failed_at, id);
+CREATE INDEX idx_education_dead_letter_tenant_component
+    ON education_dead_letter (tenant_id, component_key, status, id);

yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/runtime/EducationRuntimeDependencyRegistryTest.java

@@ -0,0 +1,27 @@
+package cn.iocoder.yudao.module.education.runtime;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class EducationRuntimeDependencyRegistryTest {
+
+    @Test
+    void shouldReturnDependenciesInStableKeyOrder() {
+        EducationRuntimeDependency second = () -> status("z-database");
+        EducationRuntimeDependency first = () -> status("a-provider");
+
+        var registry = new EducationRuntimeDependencyRegistry(List.of(second, first));
+
+        assertThat(registry.checkAll()).extracting(EducationDependencyStatus::getKey)
+                .containsExactly("a-provider", "z-database");
+    }
+
+    private static EducationDependencyStatus status(String key) {
+        return EducationDependencyStatus.builder().key(key).kind("TEST").required(true).legacy(false)
+                .failurePolicy("FAIL_CLOSED").status("UP").checkedAt(LocalDateTime.now()).build();
+    }
+}

yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/runtime/ScalarLegacyDependencyTelemetryTest.java

@@ -0,0 +1,46 @@
+package cn.iocoder.yudao.module.education.runtime;
+
+import cn.iocoder.yudao.module.education.config.EducationProperties;
+import cn.iocoder.yudao.module.education.enums.CatalogProviderMode;
+import cn.iocoder.yudao.module.education.integration.scalar.config.ScalarProperties;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class ScalarLegacyDependencyTelemetryTest {
+
+    @Test
+    void shouldExposeLegacyFailurePolicyAndCountersWhenSelected() {
+        EducationProperties education = new EducationProperties();
+        education.setCatalogMode(CatalogProviderMode.SCALAR_READ);
+        ScalarProperties scalar = new ScalarProperties();
+        scalar.setEnabled(true);
+        scalar.setBaseUrl("https://scalar.invalid");
+        scalar.setToken("secret");
+        scalar.setOwner("education-platform");
+        scalar.setExitDate(java.time.LocalDate.of(2026, 12, 31));
+        var telemetry = new ScalarLegacyDependencyTelemetry(education, scalar);
+        telemetry.record("success");
+        telemetry.record("failure");
+
+        EducationDependencyStatus status = telemetry.check();
+
+        assertThat(status.isLegacy()).isTrue();
+        assertThat(status.isRequired()).isTrue();
+        assertThat(status.getFailurePolicy()).isEqualTo("FAIL_CLOSED");
+        assertThat(status.getOwner()).isEqualTo("education-platform");
+        assertThat(status.getExitAt()).isEqualTo(java.time.LocalDate.of(2026, 12, 31).atStartOfDay());
+        assertThat(status.getStatus()).isEqualTo("UP");
+        assertThat(status.getDetail()).isEqualTo("success=1,failure=1");
+    }
+
+    @Test
+    void shouldNotRequireScalarInJavaReadMode() {
+        EducationProperties education = new EducationProperties();
+        education.setCatalogMode(CatalogProviderMode.JAVA_READ);
+        var telemetry = new ScalarLegacyDependencyTelemetry(education, new ScalarProperties());
+
+        assertThat(telemetry.check().getStatus()).isEqualTo("NOT_SELECTED");
+        assertThat(telemetry.check().isRequired()).isFalse();
+    }
+}

yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/service/operations/EducationTenantContextPropagationTest.java

@@ -0,0 +1,46 @@
+package cn.iocoder.yudao.module.education.service.operations;
+
+import cn.iocoder.yudao.framework.tenant.core.context.TenantContextHolder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.Executors;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class EducationTenantContextPropagationTest {
+
+    @AfterEach
+    void clear() {
+        TenantContextHolder.clear();
+    }
+
+    @Test
+    void shouldInstallAndRestoreTenantAroundExecutorTask() throws Exception {
+        var executor = Executors.newSingleThreadExecutor();
+        try {
+            assertThat(executor.submit(TenantContextHolder::getTenantId).get()).isNull();
+            TenantContextHolder.setTenantId(7L);
+            Long observed = executor.submit(EducationTenantContextPropagation.wrap(42L,
+                    TenantContextHolder::getRequiredTenantId)).get();
+            Long after = executor.submit(TenantContextHolder::getTenantId).get();
+
+            assertThat(observed).isEqualTo(42L);
+            assertThat(after).isNull();
+            assertThat(TenantContextHolder.getRequiredTenantId()).isEqualTo(7L);
+        } finally {
+            executor.shutdownNow();
+        }
+    }
+
+    @Test
+    void shouldRestoreContextWhenTaskFails() {
+        TenantContextHolder.setTenantId(7L);
+        Runnable wrapped = EducationTenantContextPropagation.wrap(42L, (Runnable) () -> {
+            throw new IllegalStateException("boom");
+        });
+
+        org.junit.jupiter.api.Assertions.assertThrows(IllegalStateException.class, wrapped::run);
+        assertThat(TenantContextHolder.getRequiredTenantId()).isEqualTo(7L);
+    }
+}

yudao-module-education/src/test/java/cn/iocoder/yudao/module/education/test/EducationFlywayMigrationIntegrationTest.java

@@ -119,7 +119,7 @@ class EducationFlywayMigrationIntegrationTest {
 
         assertThat(queryStrings(schema,
                 "SELECT COALESCE(version, 'BASELINE') FROM flyway_schema_history ORDER BY installed_rank"))
-                .containsExactly("4009", "4010", "4020", "4030", "4040", "4050", "4060", "4070", "4080", "4090", "4100", "4110", "4140");
+                .containsExactly("4009", "4010", "4020", "4030", "4040", "4050", "4060", "4070", "4080", "4090", "4100", "4110", "4140", "4170");
         assertThat(queryLong(schema,
                 "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = current_schema() " +
                         "AND table_name = 'education_idempotency'"))
@@ -211,7 +211,7 @@ class EducationFlywayMigrationIntegrationTest {
 
         assertThat(queryStrings(schema,
                 "SELECT version FROM flyway_schema_history WHERE success = TRUE ORDER BY installed_rank"))
-                .containsExactly("4010", "4020", "4030", "4040", "4050", "4060", "4070", "4080", "4090", "4100", "4110", "4140");
+                .containsExactly("4010", "4020", "4030", "4040", "4050", "4060", "4070", "4080", "4090", "4100", "4110", "4140", "4170");
         assertThat(queryStrings(schema,
                 "SELECT table_name FROM information_schema.tables " +
                         "WHERE table_schema = current_schema() AND table_name IN (" +
@@ -260,6 +260,44 @@ class EducationFlywayMigrationIntegrationTest {
                 .isEqualTo(8L);
     }
 
+    @Test
+    void shouldCreateOperationalHeartbeatAndDeadLetterContracts() throws SQLException {
+        String schema = createSchema("operations");
+        configureFlyway(schema, false).load().migrate();
+
+        assertThat(queryStrings(schema, """
+                SELECT table_name FROM information_schema.tables
+                 WHERE table_schema = current_schema()
+                   AND table_name IN ('education_operational_component', 'education_dead_letter')
+                 ORDER BY table_name
+                """)).containsExactly("education_dead_letter", "education_operational_component");
+
+        execute(schema, """
+                INSERT INTO education_operational_component
+                    (component_key, component_type, instance_id, status)
+                VALUES ('asset-scanner', 'SCANNER', 'scanner-1', 'UP');
+                INSERT INTO education_dead_letter
+                    (tenant_id, component_key, workload_key, payload_fingerprint, failure_category, attempts)
+                VALUES (10, 'asset-scanner', 'asset-1', 'sha256:abc', 'MALWARE_SCAN_ERROR', 3);
+                """);
+        assertThatThrownBy(() -> execute(schema, """
+                INSERT INTO education_dead_letter
+                    (tenant_id, component_key, workload_key, payload_fingerprint, failure_category, attempts)
+                VALUES (10, 'asset-scanner', 'asset-1', 'sha256:def', 'MALWARE_SCAN_ERROR', 4);
+                """)).hasMessageContaining("uk_education_dead_letter_open_workload");
+
+        execute(schema, """
+                UPDATE education_dead_letter
+                   SET status = 'REQUEUED', resolved_at = CURRENT_TIMESTAMP
+                 WHERE tenant_id = 10 AND workload_key = 'asset-1';
+                INSERT INTO education_dead_letter
+                    (tenant_id, component_key, workload_key, payload_fingerprint, failure_category, attempts)
+                VALUES (10, 'asset-scanner', 'asset-1', 'sha256:def', 'MALWARE_SCAN_ERROR', 4);
+                """);
+        assertThat(queryLong(schema, "SELECT count(*) FROM education_dead_letter WHERE tenant_id = 10"))
+                .isEqualTo(2L);
+    }
+
     @Test
     void shouldEnforcePublicAndTenantCatalogGraphRules() throws SQLException {
         String schema = createSchema("catalog_graph");