feat: deliver N07 message recall

2 months ago · 277b3c0409
28 changed files with 2656 additions and 50 deletions
--- a/docs/backend-development-roadmap.md
+++ b/docs/backend-development-roadmap.md
@ -8,9 +8,10 @@
 |---|---|---|
 | 核心招聘与账户能力 | 已有代码与历史验收记录；本轮未重跑验收 | 按清单C01–C08做一期集成回归 |
 | V39–V42 | 已核对提交55e3975、2698d06、5cc2c86、96b3113；不再标为待Git检查点 | 保留回归，不重复开发企业回复/发现/邀约 |
-| V43 | 独立代码复验与真实 MySQL/Docker 验收通过，待 Git 检查点 | N01 已关闭版本内小样本补集泄露；30/300/3000 查询预算通过，EXPLAIN 扩容风险保留观察 |
-| 评分规则解释与合理性 | N02 独立代码复验与真实 MySQL/Docker 黑盒验收通过，待 Git 检查点 | 保持分数/权重/排名不变；下一步仅做精确暂存与提交审查 |
-| 其余六个功能组 | 技能词云、入职清单/公司知识、薪资计算器/参考、诚信积分/激励、消息撤回、简历查重 | N03–N08全部一期必做，细节设计后编码 |
+| V43 | 独立代码复验、真实 MySQL/Docker 验收及 Git 检查点已完成（cf39737） | N01 已关闭版本内小样本补集泄露；30/300/3000 查询预算通过，EXPLAIN 扩容风险保留观察 |
+| 评分规则解释与合理性 | N02 独立代码复验、真实 MySQL/Docker 验收及 Git 检查点已完成（dbd27e6） | 保持分数、权重、排名和企业评分契约不变 |
+| 消息撤回与会话一致性 | N07 定向整改完成并通过自动化与干净构建，待独立代码复验与真实 MySQL/Docker 验收 | 两分钟窗口、发送者权限、正文覆盖、幂等并发、DIRECT/邀约隐私均已锁定 |
+| 其余五个功能组 | 技能词云、入职清单/公司知识、薪资计算器/参考、诚信积分/激励、简历查重 | N03–N06、N08仍属一期，细节设计后编码 |
 | 上线保障 | 现有资产待逐项核验 | G01–G04，一期交付门槛 |
 | 后续候选池 | AI简历、AI模拟面试/预测、SHAP高级解释等 | 以清单第6节为准，移出一期 |

--- a/docs/frontend-integration.md
+++ b/docs/frontend-integration.md
@ -936,6 +936,12 @@ POST /admin/review-replies/{replyId}/decision

 V43 仅供管理员后台使用，前端不得向学生或企业展示。当前状态为版本内补集定向修复完成、待独立复验；总览和申诉使用最终 SQL 聚合，并统一执行独立候选人数与补集抑制。`drift` 和 `ranking-stability` 在当前没有不可变评分历史与排名快照时返回 `availability=UNAVAILABLE`，页面应显示“暂无可比历史数据”。

+## 27. N07 消息撤回与会话一致性
+
+消息发送者可在发送后两分钟内调用 `POST /conversations/{conversationId}/messages/{messageId}/recall` 撤回本人消息。请求不接收用户 ID；服务端从登录身份、会话成员关系和消息发送者校验归属。窗口边界采用 `capturedNow <= createdAt + 2 minutes`，恰好两分钟仍允许，超过窗口返回 409。非成员、非原发送者、管理员和跨会话访问均按既有 403/2002 语义处理。
+
+成功撤回后，消息状态为 `RECALLED`，正文统一显示“消息已撤回”，并保留原时间线位置；响应、消息列表、发送响应和审计均不再返回原正文。重复撤回幂等返回当前撤回状态，不重复写审计。撤回消息不计入未读数，`markRead` 不为撤回消息新增已读记录。`DIRECT` 会话保持既有 `senderUserId` 语义，`TALENT_INVITATION` 会话继续只返回 `CURRENT_USER`/`OTHER_PARTICIPANT` 相对身份。完整接口契约见 [messaging-api.md](messaging-api.md)。
+
 ## 26. N02 候选评分规则解释与合理性

 学生评分卡和单维详情在保持原有路径不变的前提下增加固定解释字段。综合卡新增 `explanationVersion=candidate-scoring-explanation-v1`、`coverage`（实际参与 overallScore 的维度及排除原因）和 `comparabilityNotice`；单维详情新增中性 `displayName`、`whatItMeasures`、`interpretation`、`limitations`、`evidenceLevel`、`ruleContributions`、`improvementActions`、`explanationVersion`、`validity`（仅 ABILITY）和 `calculatedAt`。
--- a/docs/message-recall-design.md
+++ b/docs/message-recall-design.md
@ -0,0 +1,368 @@
+# N07 消息撤回与会话一致性：详细设计
+
+状态：N07 定向整改完成并通过自动化与干净构建，待独立代码复验与真实 MySQL/Docker 验收。
+
+文档类型：解释与技术参考。目标读者为产品负责人、后端开发、前端联调和独立验收人员。
+
+## 1. 目标
+
+N07 为现有站内会话增加有限、可审计、并发安全的消息撤回能力。发送者可在发送后两分钟内撤回自己的消息；会话成员继续看到一条撤回占位记录，但不能再通过列表、详情、缓存或审计取得原正文。
+
+本版同时统一消息响应的白名单映射，保证 `DIRECT` 会话保持既有身份字段语义，`TALENT_INVITATION` 会话继续使用相对身份标识，不重新暴露学生内部 ID。
+
+## 2. 当前实现事实
+
+- 消息表由 `V9__messaging.sql` 创建，并由 N07 的 `V41__message_recall.sql` 增加 `status`、`recalled_at` 和 `version`。
+- `Message` 实体使用字符串状态、撤回时间和 JPA `@Version`；撤回通过条件更新保证并发安全。
+- `MessagingService` 的既有 Map 响应和 N07 撤回响应均委托给统一的 `MessageViewMapper`，不直接返回实体。
+- `DIRECT` 消息当前返回 `senderUserId`；这是既有兼容语义。
+- `TALENT_INVITATION` 消息返回 `senderScope=CURRENT_USER/OTHER_PARTICIPANT`，不返回另一方用户 ID。
+- 当前没有附件、引用消息、消息编辑、WebSocket 推送或消息撤回。
+- 未读数当前基于最近 500 条消息与 `message_reads` 计算。
+- 最新主线迁移为 V40；并行工作存在迁移号冲突风险。
+
+## 3. 范围
+
+### 3.1 本版包含
+
+- 发送者撤回本人发送的消息；
+- `DIRECT` 与 `TALENT_INVITATION` 两种会话；
+- 两分钟撤回窗口；
+- 重复撤回幂等；
+- 数据库条件更新与乐观锁；
+- 固定撤回占位文本；
+- 撤回消息不计入未读数；
+- 白名单消息响应 DTO；
+- 一条脱敏安全审计；
+- 自动化、OpenAPI、联调文档和真实 MySQL 黑盒验收。
+
+### 3.2 本版不包含
+
+- 消息编辑、删除、批量撤回；
+- 管理员代撤回；
+- 撤回恢复；
+- 附件和引用消息；
+- WebSocket、SSE 或 RabbitMQ 撤回通知；
+- 离线推送、短信或邮件；
+- 保存或公开原正文历史；
+- 内容审核、AI 摘要或聊天风控扩展；
+- 改变 DIRECT 会话创建的 MATCHED 守卫；
+- 改变 V42 邀约接受后建会话的语义。
+
+## 4. 产品规则
+
+### 4.1 时间窗口
+
+撤回截止时间为：
+
+```text
+recallDeadline = message.createdAt + 2 minutes
+```
+
+当 `capturedNow <= recallDeadline` 时允许撤回；当 `capturedNow > recallDeadline` 时返回冲突。一次请求只捕获一次 `capturedNow`，所有判断和 `recalledAt` 均使用该时间。
+
+### 4.2 权限
+
+- 请求者必须已登录；
+- 请求者必须是会话成员；
+- 消息必须属于路径中的会话；
+- 只有 `senderUserId` 对应的原发送者可以撤回；
+- 其他成员、ADMIN 以及不属于该会话的用户均返回既有 `403/2002`；
+- 不通过不同错误暴露某条消息是否属于其他会话。
+
+### 4.3 状态机
+
+```text
+SENT -> RECALLED
+```
+
+- `SENT` 可以在窗口内转为 `RECALLED`；
+- `RECALLED` 再次撤回返回同一当前结果，HTTP 200，不重复更新、不重复审计；
+- 窗口外的 `SENT` 返回 409；
+- 没有其他状态迁移。
+
+## 5. 数据库方案
+
+编码前从实际主线确定下一可用 Flyway 版本，文件说明建议为：
+
+```text
+V<next>__message_recall.sql
+```
+
+不得预先假定 V41 未被其他分支占用，也不得修改 V1–V40。
+
+在 `messages` 增加：
+
+| 字段 | 类型 | 约束 | 说明 |
+|---|---|---|---|
+| status | VARCHAR(20) | NOT NULL DEFAULT 'SENT' | `SENT` 或 `RECALLED` |
+| recalled_at | DATETIME | NULL | 实际撤回时间 |
+| version | BIGINT | NOT NULL DEFAULT 0 | JPA `@Version` |
+
+约束建议：
+
+```sql
+CHECK (status IN ('SENT','RECALLED'))
+```
+
+现有 `conversation_id, created_at` 索引继续用于分页。本版不新增撤回历史表，不保存原正文副本，不新增包含正文的审计表。
+
+迁移必须同时通过 H2 `MODE=MySQL` 与真实 MySQL 8.x，并保持 `ddl-auto=validate`。
+
+## 6. 正文与数据保留
+
+撤回成功时，将 `content` 覆盖为固定内部占位值：
+
+```text
+消息已撤回
+```
+
+响应层也只返回这一固定文本。原正文不得写入：
+
+- audit_logs；
+- 应用日志；
+-异常消息；
+- history 表；
+- RabbitMQ；
+-缓存或导出结果。
+
+数据库备份和存储层面的物理恢复策略属于 G02，不在 N07 内建立正文恢复接口。
+
+## 7. HTTP 接口
+
+新增一个接口：
+
+```http
+POST /conversations/{conversationId}/messages/{messageId}/recall
+```
+
+无请求体。身份仅从 JWT/SecurityContext 获取。
+
+成功响应使用白名单 `MessageResponse`：
+
+```json
+{
+  "code": 200,
+  "data": {
+    "id": 81,
+    "conversationId": 12,
+    "senderUserId": 7,
+    "content": "消息已撤回",
+    "status": "RECALLED",
+    "recalledAt": "2026-09-10T10:01:30",
+    "createdAt": "2026-09-10T10:00:00",
+    "isRead": false
+  }
+}
+```
+
+对于 `TALENT_INVITATION`，响应使用 `senderScope`，不得返回 `senderUserId`。
+
+### 7.1 错误语义
+
+| 场景 | HTTP / 业务码 |
+|---|---|
+| 未登录 | 401 / 既有未认证码 |
+| 非成员、非发送者、跨会话探测、ADMIN | 403 / 2002 |
+| 会话不存在 | 404 / 既有会话不存在码 |
+| 当前会话内消息不存在 | 404 / 新增消息不存在码或经评审复用现有码 |
+| 超过撤回窗口 | 409 / 新增撤回窗口错误码 |
+| 并发状态冲突 | 409 / 新增并发冲突错误码 |
+
+ErrorCode 属于高冲突文件。编码前应先核对现有码值，并只新增确有必要的错误码。
+
+## 8. 响应 DTO 与兼容
+
+新增白名单 DTO，建议：
+
+- `MessageResponse`；
+- 如分页结构尚无类型，可新增 `MessagePageResponse`。
+
+允许字段：
+
+- id；
+- conversationId；
+- senderUserId，仅 DIRECT；
+- senderScope，仅 TALENT_INVITATION；
+- content；
+- status；
+- recalledAt；
+- createdAt；
+- isRead。
+
+禁止字段：
+
+- 对方在 TALENT_INVITATION 中的 userId；
+- 手机号、邮箱、姓名、简历；
+- 邀约 candidateKey 或摘要；
+- 原撤回正文；
+- IP、User-Agent、Token、密码；
+- 内部审计内容。
+
+现有发送与列表接口兼容新增 `status` 和可空 `recalledAt`。DIRECT 的 `senderUserId` 不删除；TALENT_INVITATION 继续只提供相对身份。
+
+## 9. 服务与事务
+
+建议新增：
+
+```java
+MessageResponse recallMessage(
+    Long currentUserId,
+    Long conversationId,
+    Long messageId,
+    LocalDateTime capturedNow,
+    AuditContext auditContext)
+```
+
+事务步骤：
+
+1. 查询会话并校验成员；
+2. 查询消息，校验其属于当前会话；
+3. 校验请求者为发送者；
+4. 若已撤回，返回当前白名单响应；
+5. 校验两分钟窗口；
+6. 使用 `id + conversationId + senderUserId + status=SENT + version` 条件更新；
+7. 更新 `status=RECALLED`、固定占位正文、`recalledAt=capturedNow`；
+8. 条件更新成功后写一条审计；
+9. 返回重新读取或由确定字段构造的响应。
+
+只有成功发生 `SENT -> RECALLED` 的请求写审计。幂等重放和失败请求不重复写成功审计。
+
+## 10. 并发与幂等
+
+- `Message.version` 使用 JPA `@Version`；
+- Repository 提供条件更新，不能只依赖“先查后改”；
+- 两个并发撤回请求只能一个执行状态变更并写审计；
+- 另一请求重新读取后若状态已为 `RECALLED`，按幂等成功返回；
+- 撤回与列表读取允许并发，读取结果只能是完整 `SENT` 或完整 `RECALLED`，不能返回已撤回状态配原正文；
+- SQL异常和锁异常不得转为空成功。
+
+## 11. 未读与列表
+
+- `RECALLED` 消息保留在消息分页中，维持时间线位置；
+- 撤回占位消息的 `isRead` 可以保留既有记录语义；
+- `RECALLED` 消息不计入 unreadCount；
+- 不删除已有 `message_reads`；
+- `markRead` 不需要为撤回消息新增 read 记录；
+- 列表、发送响应、撤回响应必须共用同一消息映射器。
+
+现有 `countUnread` 与 `markRead` 存在最多读取 500 条的历史限制。N07 只修正撤回消息口径，不借机重写整个未读架构；容量治理记录到 G03。
+
+## 12. 审计
+
+成功状态变更写：
+
+```text
+MESSAGE_RECALLED
+```
+
+details 白名单：
+
+```json
+{
+  "messageId": 81,
+  "conversationId": 12,
+  "conversationType": "DIRECT",
+  "status": "RECALLED",
+  "requestId": "..."
+}
+```
+
+禁止写入正文、对方用户 ID、手机号、邮箱、简历、IP/UA 明文、Token、密码或邀约标识。
+
+## 13. 性能与查询
+
+- 消息分页不得因撤回字段产生逐消息额外查询；
+- 会话类型只读取一次，不能逐消息查询 Conversation；
+- 已读状态继续批量查询，禁止逐消息 `existsBy...`；
+- unreadCount 应通过数据库计数或固定批量查询实现，排除 `RECALLED`；
+- 查询次数不得随消息数量线性增长；
+- 为真实 MySQL EXPLAIN 保留消息分页和未读计数 SQL。
+
+## 14. OpenAPI 影响
+
+当前基线为 172 paths / 210 operations。若仅新增一个撤回接口，预计为：
+
+```text
+173 paths / 211 operations
+```
+
+实际计数以静态 OpenAPI、运行时 `/api/v1/v3/api-docs` 和校验脚本为准。
+
+OpenAPI 必须断言：
+
+- 仅新增一个 N07 path/operation；
+- 撤回接口 200 响应有非空 schema；
+- DIRECT 与邀约消息隐私契约明确；
+- 响应不包含原正文历史或用户隐私字段；
+- 既有 6 个消息 operations 保持存在。
+
+## 15. 测试与验收
+
+### 15.1 单元与服务测试
+
+- 发送后 0 秒、1:59、恰好 2:00 可撤回；
+- 2:00:00 后最小时间单位不可撤回；
+- 非发送者、非成员、ADMIN 拒绝；
+- 消息和路径会话不一致返回 403，避免枚举；
+- 重复撤回幂等；
+- 并发撤回仅一次更新和一次审计；
+- 固定 Clock，全请求只捕获一次；
+- 审计白名单不含正文。
+
+### 15.2 HTTP 与隐私测试
+
+- DIRECT 发送、列表和撤回保持 senderUserId；
+- TALENT_INVITATION 仅返回 senderScope；
+- 已撤回消息所有接口只返回占位内容；
+- 完整 JSON 不含原正文、对方 ID、联系方式或凭据；
+- 未登录 401、越权 403/2002。
+
+### 15.3 持久化与 MySQL
+
+- 迁移在 H2 MySQL 模式和 MySQL 8.x 执行；
+- `CHECK`、默认状态和 `@Version` 正确；
+- 撤回后数据库正文为固定占位；
+- 并发条件更新正确；
+- `ddl-auto=validate` 通过；
+- 消息分页和未读 SQL EXPLAIN 无发布阻断。
+
+### 15.4 回归
+
+- DIRECT MATCHED 守卫不变；
+- V33 屏蔽发送规则不变；
+- V42 邀约接受、会话隐私和唯一会话不变；
+- 消息发送、列表、已读原有功能通过；
+- 不写 candidate_dimension_scores；
+- 不发送新增 RabbitMQ 消息。
+
+## 16. 风险
+
+- 并行分支可能占用迁移号，编码前必须重新确认；
+- Map 响应迁移为 DTO 时容易意外删除旧字段，必须做兼容 JSON 对照；
+- 撤回与列表并发可能暴露状态和正文不一致，必须原子更新并统一映射；
+- 当前未读实现有 500 条上限，不能将其误称为完整历史未读统计；
+- 数据库备份可能保留历史物理页，本功能只保证业务接口、当前数据库值、日志和审计不暴露原正文。
+
+## 17. 已确认决策
+
+| 决策 | 结论 |
+|---|---|
+| D01 | 撤回窗口为发送后 2 分钟 |
+| D02 | `capturedNow <= deadline` 允许撤回 |
+| D03 | DIRECT 与 TALENT_INVITATION 均支持 |
+| D04 | 仅原发送者可撤回 |
+| D05 | 使用 conversationId 与 messageId 的 POST 撤回接口 |
+| D06 | 重复撤回幂等返回 200 |
+| D07 | 保留固定撤回占位消息 |
+| D08 | 数据库正文覆盖为固定占位，不保存原正文历史 |
+| D09 | 撤回消息不计入 unreadCount，不删除已有 read 记录 |
+| D10 | messages 增加 status、recalled_at、version，无历史表 |
+| D11 | @Version 与条件更新共同保证并发 |
+| D12 | 仅成功转换写 MESSAGE_RECALLED 脱敏审计 |
+| D13 | 一期不新增实时推送或 RabbitMQ 事件 |
+| D14 | 附件与引用消息不在本版范围 |
+| D15 | 使用白名单 DTO，保持两种会话的既有身份语义 |
+| D16 | 401/403/404/409 按认证、归属、存在性和状态区分 |
+| D17 | 预计新增 1 path / 1 operation |
+| D18 | Flyway 版本在编码前按实际主线分配 |
--- a/docs/message-recall-implementation-instruction.md
+++ b/docs/message-recall-implementation-instruction.md
@ -0,0 +1,293 @@
+# N07 消息撤回：实施指令
+
+状态：N07 定向整改完成并通过自动化与干净构建，待独立代码复验与真实 MySQL/Docker 验收。
+
+文档类型：操作指南。用于指导实现者在不扩大范围的前提下完成 N07。
+
+## 1. 开始前
+
+项目根目录：
+
+```text
+C:\Users\15642\Desktop\zhilai
+```
+
+先完整阅读：
+
+- `docs/message-recall-design.md`；
+- `docs/frontend-integration.md`；
+- `docs/enterprise-talent-invitation-api.md`；
+- `docs/report-block-api.md`；
+- `docs/phase1-scope-and-acceptance.md`。
+
+重点代码：
+
+- `messaging/Message.java`；
+- `messaging/MessageRepository.java`；
+- `messaging/MessagingService.java`；
+- `messaging/MessagingController.java`；
+- `messaging/Conversation.java`；
+- `messaging/ConversationMemberRepository.java`；
+- `talentinvitation/`；
+- `safety/BlockRelationRepository.java`；
+- `audit/SecurityAuditService.java`；
+- `config/SecurityConfig.java`；
+- `common/exception/ErrorCode.java`；
+- `common/exception/GlobalExceptionHandler.java`。
+
+先执行：
+
+```powershell
+git status --short
+git diff --cached --name-only
+git log -5 --oneline
+git diff --check
+```
+
+如果暂存区非空，停止并报告。保护所有历史未跟踪文件，不得 reset、restore、checkout、clean 或 `git add -A`。
+
+## 2. 迁移号分配
+
+编码前列出迁移：
+
+```powershell
+Get-ChildItem mydreamjob-server/src/main/resources/db/migration -File |
+  Sort-Object Name |
+  Select-Object -ExpandProperty Name
+```
+
+同时检查其他待合并分支是否已经占用下一版本。只有负责人确认后才创建：
+
+```text
+V<next>__message_recall.sql
+```
+
+禁止修改 V1–V40，禁止擅自假定 V41 可用。
+
+迁移只允许为 `messages` 增加：
+
+- `status VARCHAR(20) NOT NULL DEFAULT 'SENT'`；
+- `recalled_at DATETIME NULL`；
+- `version BIGINT NOT NULL DEFAULT 0`；
+- `status` 的受控 CHECK。
+
+不新增历史表，不保存正文副本。
+
+## 3. 实体与 Repository
+
+修改 `Message`：
+
+- 新增字符串状态或受控枚举；
+- 新增 `recalledAt`；
+- 新增 `@Version Long version`；
+- 发送创建状态固定为 `SENT`。
+
+Repository 增加条件更新，条件至少包括：
+
+- messageId；
+- conversationId；
+- senderUserId；
+- status=`SENT`；
+- expectedVersion。
+
+更新内容：
+
+- status=`RECALLED`；
+- content=`消息已撤回`；
+- recalledAt=capturedNow；
+- version 原子递增。
+
+不得只用 `findById -> set -> save` 作为并发保证。
+
+消息分页和未读查询要排除撤回消息的未读贡献，并采用批量查询，禁止逐消息查 Conversation 或 read 状态。
+
+## 4. DTO 与映射
+
+在 `messaging/dto/` 新增白名单 DTO：
+
+- `MessageResponse`；
+- 必要时新增 `MessagePageResponse`。
+
+把发送、列表、撤回响应收敛到同一个映射器。
+
+DIRECT 兼容保留 `senderUserId`；TALENT_INVITATION 只返回 `senderScope`。撤回状态下任何路径都只能返回固定占位正文。
+
+不得直接返回 Message Entity，不得继续在 Controller 拼 Map。
+
+## 5. 服务实现
+
+为 `MessagingService` 注入 `Clock`，生产默认系统时区。撤回入口只捕获一次 `capturedNow`。
+
+实现顺序：
+
+1. requireConversation；
+2. assertMember；
+3. 查询消息并防止跨会话枚举；
+4. 校验发送者；
+5. 已撤回则幂等返回；
+6. 校验 `capturedNow <= createdAt.plusMinutes(2)`；
+7. 执行条件更新；
+8. 失败时重新读取并区分幂等完成与真实冲突；
+9. 成功后写一次脱敏审计；
+10. 返回白名单响应。
+
+不得把原正文写入日志、异常或审计。
+
+## 6. Controller 与接口
+
+新增：
+
+```http
+POST /conversations/{conversationId}/messages/{messageId}/recall
+```
+
+无 requestBody，不接收 userId。身份从 SecurityContext/JWT subject 获取。
+
+Controller 只做参数接收、身份解析和响应包装，不包含撤回规则。
+
+## 7. ErrorCode
+
+先搜索现有消息错误码。只在没有可准确复用的代码时新增：
+
+- MESSAGE_NOT_FOUND；
+- MESSAGE_RECALL_WINDOW_EXPIRED；
+- MESSAGE_RECALL_CONCURRENT_UPDATE。
+
+保持：
+
+- 未登录 401；
+- 非成员、非发送者和跨会话探测 403/2002；
+- 超时和并发冲突 409。
+
+不得改变已有错误码含义。ErrorCode 和 GlobalExceptionHandler 修改前应与集成负责人确认。
+
+## 8. 审计
+
+成功状态转换写一条 `MESSAGE_RECALLED`。
+
+details 只允许：
+
+- messageId；
+- conversationId；
+- conversationType；
+- status；
+- requestId。
+
+幂等重放、超时、越权和并发失败不写成功审计。禁止记录正文、对方 ID、联系方式、IP/UA 明文或凭据。
+
+## 9. 必须新增的测试
+
+建议测试类：
+
+- `MessageRecallServiceTest`；
+- `MessageRecallControllerTest`；
+- `MessageRecallPersistenceIntegrationTest`；
+- `MessageRecallPrivacyIntegrationTest`。
+
+必须覆盖：
+
+1. 窗口内撤回；
+2. 1:59 与恰好 2:00 成功；
+3. 超过 2:00 的最小时间单位失败；
+4. 重复撤回 HTTP 200 且不重复审计；
+5. 非发送者、非成员、ADMIN、匿名；
+6. messageId 属于另一 conversationId 时 403/2002；
+7. 两个并发请求只有一次条件更新和审计；
+8. 撤回后数据库正文为固定占位；
+9. 列表、撤回响应均不返回原正文；
+10. DIRECT 保留 senderUserId；
+11. TALENT_INVITATION 不返回 senderUserId、createdBy 或学生 ID；
+12. 撤回消息不计 unreadCount；
+13. 已有 message_reads 不删除；
+14. markRead 不为撤回消息产生新记录；
+15. 消息分页无逐消息查询；
+16. 审计 details 白名单；
+17. V33 屏蔽发送回归；
+18. V42 邀约会话隐私回归；
+19. DIRECT MATCHED 守卫回归；
+20. `candidate_dimension_scores` 不变。
+
+完整 JSON 禁止出现：
+
+```text
+originalContent
+previousContent
+studentUserId
+candidateId
+phone
+email
+resumeUrl
+candidateKey
+candidateKeyDigest
+ipAddress
+userAgent
+password
+token
+secret
+```
+
+## 10. OpenAPI 与文档
+
+更新：
+
+- `docs/messaging-api.md`，不存在则新增；
+- `docs/frontend-integration.md`；
+- `docs/backend-development-roadmap.md`；
+- `docs/phase1-scope-and-acceptance.md`；
+- `docs/openapi-rc.json`；
+- `docs/verify_openapi.py`；
+- 本设计和实施指令状态。
+
+目标预计为 173 paths / 211 operations，以实际静态和运行时结果为准。
+
+校验脚本必须递归检查消息响应 schema，不得只检查计数。
+
+## 11. 自动化验证
+
+依次执行：
+
+```powershell
+cd C:\Users\15642\Desktop\zhilai\mydreamjob-server
+mvn -Dtest="*MessageRecall*Test,*Messaging*Test,*TalentInvitation*Test" test
+mvn test
+mvn clean verify
+python ..\docs\verify_openapi.py
+git diff --check
+```
+
+记录真实 Tests run、Failures、Errors、Skipped 和 OpenAPI 计数。禁止删除测试或修改期望语义掩盖回归。
+
+## 12. 真实 MySQL/Docker 验收
+
+自动化完成后由负责人授权执行：
+
+- Flyway 新迁移 success=1；
+- `ddl-auto=validate` 通过；
+- 窗口边界与并发撤回；
+- 重复撤回只一条审计；
+- 数据库正文不可恢复为业务明文；
+- 权限矩阵；
+- DIRECT 与邀约完整 JSON 脱敏；
+- unreadCount 与 read 记录；
+- 消息分页和未读 SQL EXPLAIN；
+- RabbitMQ 无新增消息；
+- 运行时 OpenAPI 与静态文件一致。
+
+未执行真库验收不得标记冻结。
+
+## 13. Git 边界
+
+本轮实现期间：
+
+- 不操作 Docker，除非负责人明确要求黑盒；
+- 不暂存验收脚本、日志、请求体或 `.env`；
+- 不执行 `git add -A`；
+- 不提交、不推送，直到独立验收完成；
+- 不删除历史未跟踪文件；
+- 不进入 N08。
+
+完成代码和自动化后，状态只能写为：
+
+```text
+N07 定向整改完成并通过自动化与干净构建，待独立代码复验与真实 MySQL/Docker 验收。
+```
--- a/docs/messaging-api.md
+++ b/docs/messaging-api.md
@ -0,0 +1,33 @@
+# 站内消息接口
+
+## 消息撤回
+
+`POST /conversations/{conversationId}/messages/{messageId}/recall`
+
+当前登录用户从 JWT subject 解析。请求无 body，只有会话成员中的原发送者可在消息创建后两分钟内撤回；截止瞬间仍允许撤回。会话不存在返回 404/8305，跨会话、非成员或非发送者返回 403/2002，窗口结束返回 409/8307，并发状态冲突返回 409/8308。
+
+成功响应为白名单 `MessageResponse`：
+
+```json
+{
+  "code": 200,
+  "data": {
+    "id": 81,
+    "conversationId": 12,
+    "senderUserId": 7,
+    "content": "消息已撤回",
+    "status": "RECALLED",
+    "recalledAt": "2026-09-10T10:02:00",
+    "createdAt": "2026-09-10T10:00:00",
+    "isRead": false
+  }
+}
+```
+
+撤回会在同一事务中将数据库正文覆盖为固定占位文本 `消息已撤回`，并将状态从 `SENT` 更新为 `RECALLED`。重复请求幂等返回当前撤回结果，不重复写成功审计。原正文不写入审计、异常、缓存、消息队列或历史表。
+
+`TALENT_INVITATION` 会话继续使用 `senderScope=CURRENT_USER` 或 `OTHER_PARTICIPANT`，不返回对方的 `senderUserId`；`DIRECT` 会话保留既有 `senderUserId` 兼容字段。撤回消息仍保留在时间线中，但不计入未读数，也不会因标记已读新增 `message_reads`。
+
+## 既有消息接口兼容
+
+原有会话列表、会话详情、发送消息、消息列表和标记已读接口路径不变。消息列表和发送响应增加 `status`，撤回消息的 `content` 统一为占位文本，其他既有字段语义保持不变。
--- a/docs/n07-independent-mysql-acceptance-20260910.md
+++ b/docs/n07-independent-mysql-acceptance-20260910.md
@ -0,0 +1,87 @@
+# N07 消息撤回与会话一致性：独立代码复验与真实 MySQL/Docker 验收
+
+日期：2026-09-10
+分支：`feat/message-recall`
+结论：**PASS（真实 MySQL/Docker 黑盒已完成；可进入精确暂存检查点）**
+
+## 1. Git 基线与边界
+
+- 当前分支：`feat/message-recall`。
+- HEAD：`dbd27e6 feat: enhance candidate scoring explanations`。
+- 验收开始及结束时暂存区均为空；既有 N07 修改、历史脚本、日志、JSON、SQL、xlsx 和 `.env` 均保留。
+- `git diff --check`：通过；迁移目录没有 V1–V40 修改。
+- N07 唯一新增迁移：`V41__message_recall.sql`。
+- 未执行 `git add -A`、commit、push、reset、restore、checkout、clean；未删除或移动历史文件。
+
+## 2. 自动化独立复验
+
+| 命令 | Tests run | Failures | Errors | Skipped | 结果 |
+|---|---:|---:|---:|---:|---|
+| N07 定向测试 | 30 | 0 | 0 | 0 | BUILD SUCCESS |
+| 消息与邀约关联回归 | 80 | 0 | 0 | 0 | BUILD SUCCESS |
+| `mvn clean verify` | 1601 | 0 | 0 | 0 | BUILD SUCCESS |
+
+`python docs\\verify_openapi.py` 通过，静态 OpenAPI 为 173 paths / 211 operations。
+
+## 3. 代码审查
+
+- 状态机：PASS。V41 CHECK 仅允许 SENT/RECALLED；条件更新包含 messageId、conversationId、senderUserId、SENT、expectedVersion；正文固定占位、recalledAt 非空、version 仅递增一次；重复撤回幂等且不重复成功审计。
+- 权限与归属：PASS。真实 HTTP 中原发送者成功，另一成员/非成员/ADMIN/跨会话为 403/2002，匿名为 401，会话/消息不存在分别为 404/8305、404/8306，超时为 409/8307。
+- 并发与事务：PASS。H2 独立事务测试以及真实 HTTP 两个并发请求均无 500；同消息最终 version=1、恰好一条成功审计；不同消息并发各自成功。
+- 依赖注入：PASS。生产目录不存在 `messageViewMapper == null`、`objectMapper == null`、`new MessageViewMapper()`、`new ObjectMapper()` 兜底。
+- 审计隐私：PASS。真实 MySQL `MESSAGE_RECALLED` 行 `ip_address`/`user_agent` 均为 NULL，details 为合法 JSON 且字段严格为 `messageId,conversationId,conversationType,status,requestId`。
+- DTO/查询结构：PASS。撤回及分页路径使用白名单 DTO；会话与成员固定查询、消息分页固定查询、message_reads 一次批量查询，无逐消息 exists/find。
+- 副作用：PASS。目标消息仅发生 SENT→RECALLED、占位正文、recalledAt、version+1；message_reads、conversation_members、candidate_dimension_scores、notifications、风险表内容和 version 均未改变。
+
+## 4. Docker、MySQL、Flyway 与启动
+
+- Docker Engine：Server 29.6.2；MySQL 8.4.11、Redis 7.4、RabbitMQ 3.13 容器均 healthy。未重建、删除容器或 volume。
+- 真实 Flyway 查询：最新三条为 V41 `message recall` success=1、V40 `enterprise talent invitations` success=1、V39 `enterprise talent discovery` success=1；无失败迁移、无 N07 之外迁移。
+- `SHOW CREATE TABLE messages`：`status` 默认 SENT、`recalled_at` 可空、`version` 默认 0，CHECK 仅允许 SENT/RECALLED，既有外键与 `idx_messages_conversation_created` 保留；无消息历史表。
+- 使用 `mvn clean verify` 生成的 JAR、dev 配置并保持 `ddl-auto=validate` 启动，日志出现 `Started MyDreamjobApplication`，无 Flyway/schema-validation 错误。
+- `GET /api/v1/health`：HTTP 200，status=UP。
+- 本轮启动的应用 PID 4412（后续边界复验重启为 PID 21168）均在验收结束后正常停止；8080 已恢复空闲。Docker 未停止。
+
+## 5. 真实 HTTP 主链路与权限
+
+- DIRECT：真实发送、列表、撤回、重复撤回全部 200；撤回后 status=RECALLED、正文为固定占位、recalledAt 非空、version 0→1，重复请求不再增加 version 或审计；senderUserId 按既有语义保留。
+- TALENT_INVITATION：真实发送、撤回、列表全部 200；仅返回 senderScope，不返回 senderUserId、学生 ID或 candidateKey；会话详情和会话列表中的 TALENT_INVITATION 项不返回 createdBy；撤回后正文为固定占位。
+- 权限配对结果：原发送者 `200/200`；另一成员 `403/2002`；非成员 `403/2002`；ADMIN `403/2002`；匿名 `401/401`；跨会话 `403/2002`；不存在会话 `404/8305`；不存在消息 `404/8306`；超时 `409/8307`。
+- 递归 JSON 脱敏：DIRECT 撤回/列表、TALENT_INVITATION 撤回/列表及会话详情中的 TALENT_INVITATION 项均未发现原正文、senderUserId（TALENT）、studentUserId、candidateKey、联系方式、IP/UA、凭据等禁止字段。
+
+## 6. 并发、时间与 message_reads
+
+- 真实 MySQL HTTP 并发：同一消息两请求均 200，最终 version=1、单条审计；两条不同消息并发均 200、各 version=1、各一条审计。
+- 时间：H2 正式测试覆盖 0 秒、1:59、恰好 2:00 成功及超过 2:00 最小单位失败；真实 MySQL HTTP 119 秒返回 200/200、120 秒返回 409/8307（数据库状态分别为 RECALLED/SENT），创建后立即撤回也成功。边界表达式保持 `capturedNow <= createdAt.plusMinutes(2)`。
+- 已有 `message_reads` 行在撤回后仍存在（精确查询=1）；撤回消息不计入 unreadCount（精确查询=0），消息仍保留时间线，未新增第二条 read 行。
+
+## 7. 查询预算与 EXPLAIN
+
+- 使用真实 MySQL `performance_schema.events_statements_summary_by_digest` 测量正式 HTTP 列表路径，夹具规模 1、20、50 条消息：每次均为 `messages=2、message_reads=1、conversations=1、conversation_members=1`，查询数不随消息量线性增长。
+- 真实 EXPLAIN FORMAT=JSON：消息分页使用 `idx_messages_conversation_created`、access=ref、无 filesort/temporary；message_reads 使用 reader 索引、成员校验使用 `uk_conv_members`；条件撤回使用 PRIMARY、access=range；未读查询使用消息索引和 `uk_message_reads`。未新增索引。
+- 说明：该 EXPLAIN 为当前验收数据的代表性计划；更大生产数据量的容量/延迟压测仍需后续授权，不以固定查询数替代性能结论。
+
+## 8. RabbitMQ 与副作用
+
+RabbitMQ `ping` 成功；`queue.sms.verification`、`queue.notification`、`queue.audit`、`queue.notification.dlq` 均真实存在，ready=0、unacked=0。N07 撤回未产生新增消息、通知、短信或邮件。
+
+目标消息撤回前后稳定摘要证明：除目标消息允许的四项状态变化和一条白名单 `MESSAGE_RECALLED` 审计外，message_reads、会话、成员、评分快照、通知、风险治理表内容与 version 不变。
+
+## 9. OpenAPI
+
+- 静态：173 paths / 211 operations，`verify_openapi.py` 通过。
+- 运行时：`GET /api/v1/v3/api-docs` 返回 173 paths / 211 operations，与静态计数一致；撤回无 requestBody，200 响应引用 `ApiResponseMessageResponse`，分页响应引用 `ApiResponseMessagePageResponse`。
+- 未新增编辑、删除、批量撤回、恢复接口；TALENT_INVITATION 的条件性 senderScope 脱敏由真实 JSON 验证。
+
+## 10. PASS/FAIL/SKIP 与剩余风险
+
+- PASS：自动化、V41 迁移与 schema validation、启动/健康、权限、DIRECT/TALENT_INVITATION 脱敏、并发、审计隐私、message_reads/unread、只读副作用、查询预算、EXPLAIN、RabbitMQ、静态/运行时 OpenAPI。
+- FAIL：0。
+- SKIP：0 个核心验收项。生产级更大规模容量压测和持续运行监控不属于本轮黑盒范围，需独立性能验收，但不构成当前 N07 功能阻断。
+- 不允许把本报告扩展为对 V1–V40 其他业务的全面生产保证；仅确认 N07 变更边界。
+
+## 11. 最终判定
+
+**PASS：N07 独立代码复验和真实 MySQL/Docker 黑盒全部通过，可以进入精确暂存检查点。**
+
+本报告不包含数据库密码、JWT、测试账号密码、手机号、邮箱或原消息正文。验收数据仅使用本轮唯一前缀并已按精确 ID 清理；历史验收文件未删除或移动。
--- a/docs/openapi-rc.json
+++ b/docs/openapi-rc.json
@ -2348,7 +2348,7 @@
            "content": {
              "*/*": {
                "schema": {
-                  "$ref": "#/components/schemas/ApiResponseMapStringObject"
+                  "$ref": "#/components/schemas/ApiResponseMessagePageResponse"
                }
              }
            }
@ -2389,6 +2389,60 @@
        }
      }
    },
+    "/conversations/{conversationId}/messages/{messageId}/recall": {
+      "post": {
+        "tags": [
+          "消息"
+        ],
+        "summary": "撤回消息",
+        "description": "仅原发送者可在发送后两分钟内撤回消息；重复撤回幂等返回当前状态",
+        "operationId": "recallMessage",
+        "parameters": [
+          {
+            "name": "conversationId",
+            "in": "path",
+            "required": true,
+            "schema": {
+              "type": "integer",
+              "format": "int64"
+            }
+          },
+          {
+            "name": "messageId",
+            "in": "path",
+            "required": true,
+            "schema": {
+              "type": "integer",
+              "format": "int64"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "description": "OK",
+            "content": {
+              "*/*": {
+                "schema": {
+                  "$ref": "#/components/schemas/ApiResponseMessageResponse"
+                }
+              }
+            }
+          },
+          "401": {
+            "description": "未登录"
+          },
+          "403": {
+            "description": "非会话成员或非原发送者"
+          },
+          "404": {
+            "description": "会话或消息不存在"
+          },
+          "409": {
+            "description": "撤回窗口已过期或并发冲突"
+          }
+        }
+      }
+    },
    "/health": {
      "get": {
        "tags": [
@ -15423,6 +15477,120 @@
          }
        }
      },
+      "MessageResponse": {
+        "type": "object",
+        "description": "脱敏消息响应；TALENT_INVITATION 使用 senderScope 而非 senderUserId",
+        "properties": {
+          "id": {
+            "type": "integer",
+            "format": "int64"
+          },
+          "conversationId": {
+            "type": "integer",
+            "format": "int64"
+          },
+          "senderUserId": {
+            "type": "integer",
+            "format": "int64"
+          },
+          "senderScope": {
+            "type": "string"
+          },
+          "content": {
+            "type": "string"
+          },
+          "status": {
+            "type": "string",
+            "enum": [
+              "SENT",
+              "RECALLED"
+            ]
+          },
+          "recalledAt": {
+            "type": "string",
+            "format": "date-time"
+          },
+          "createdAt": {
+            "type": "string",
+            "format": "date-time"
+          },
+          "isRead": {
+            "type": "boolean"
+          }
+        }
+      },
+      "MessagePageResponse": {
+        "type": "object",
+        "description": "脱敏消息分页响应",
+        "properties": {
+          "items": {
+            "type": "array",
+            "items": {
+              "$ref": "#/components/schemas/MessageResponse"
+            }
+          },
+          "total": {
+            "type": "integer",
+            "format": "int64"
+          },
+          "page": {
+            "type": "integer",
+            "format": "int32"
+          },
+          "size": {
+            "type": "integer",
+            "format": "int32"
+          },
+          "totalPages": {
+            "type": "integer",
+            "format": "int32"
+          }
+        }
+      },
+      "ApiResponseMessagePageResponse": {
+        "type": "object",
+        "properties": {
+          "code": {
+            "type": "integer",
+            "format": "int32"
+          },
+          "message": {
+            "type": "string"
+          },
+          "data": {
+            "$ref": "#/components/schemas/MessagePageResponse"
+          },
+          "requestId": {
+            "type": "string"
+          },
+          "timestamp": {
+            "type": "string",
+            "format": "date-time"
+          }
+        }
+      },
+      "ApiResponseMessageResponse": {
+        "type": "object",
+        "properties": {
+          "code": {
+            "type": "integer",
+            "format": "int32"
+          },
+          "message": {
+            "type": "string"
+          },
+          "data": {
+            "$ref": "#/components/schemas/MessageResponse"
+          },
+          "requestId": {
+            "type": "string"
+          },
+          "timestamp": {
+            "type": "string",
+            "format": "date-time"
+          }
+        }
+      },
      "ApiResponsePageResponseDataExportTaskResponse": {
        "type": "object",
        "properties": {
--- a/docs/phase1-scope-and-acceptance.md
+++ b/docs/phase1-scope-and-acceptance.md
@ -36,13 +36,13 @@

 | 编号/功能 | 当前状态与一期需要什么 | 代码在哪里 | 如何验收 | 主责 / 复核 |
 |---|---|---|---|---|
-| N01 V43过程公平性与稳定性监测 | 独立代码复验与真实 MySQL/Docker 验收通过，待 Git 检查点；历史不足返回UNAVAILABLE | 现有 scoringmonitoring/；docs/v43-candidate-scoring-fairness-design.md | 版本内补集、申诉集合、权限、只读性及30/300/3000查询预算通过；EXPLAIN扩容风险纳入容量观察 | 张顺 / 小刘 |
-| N02 评分规则解释与合理性 | 独立代码复验与真实 MySQL/Docker 黑盒验收通过，待 Git 检查点 | candidatescore/CandidateScoringCalculator.java、CandidateScoringEngineService.java、candidatescore/explanation/；docs/candidate-scoring-explanation-design.md、docs/candidate-scoring-rule-evidence.md | 未知规则安全失败；真实 H2 时间边界、四入口单次 capturedNow、企业完整 JSON 脱敏、运行时 OpenAPI 与真库回归均已通过 | 张顺 / 小刘 |
+| N01 V43过程公平性与稳定性监测 | 独立代码复验、真实 MySQL/Docker 验收及 Git 检查点已完成（cf39737）；历史不足返回UNAVAILABLE | 现有 scoringmonitoring/；docs/v43-candidate-scoring-fairness-design.md | 版本内补集、申诉集合、权限、只读性及30/300/3000查询预算通过；EXPLAIN扩容风险纳入容量观察 | 张顺 / 小刘 |
+| N02 评分规则解释与合理性 | 独立代码复验、真实 MySQL/Docker 验收及 Git 检查点已完成（dbd27e6） | candidatescore/CandidateScoringCalculator.java、CandidateScoringEngineService.java、candidatescore/explanation/；docs/candidate-scoring-explanation-design.md、docs/candidate-scoring-rule-evidence.md | 未知规则安全失败；真实 H2 时间边界、四入口单次 capturedNow、企业完整 JSON 脱敏、运行时 OpenAPI 与真库回归均已通过 | 张顺 / 小刘 |
 | N03 行业技能词云 | 未确认有独立实现；用有授权、可追溯的数据生成技能频次，支持空数据与更新时间；不是装饰性随机词云 | 现有 job/、recommendation/ 可复用数据入口；拟新增 skillinsight/ | 同义词归一/去重/计数用已知样本对账；行业筛选与来源日期准确；无数据不编造；批量性能与页面联调 | 小刘 / 张顺 |
 | N04 入职清单与公司知识介绍 | 未确认有独立实现；清单模板、个人完成进度、公司可公开介绍；公司原有介绍与新增入职知识明确区分 | 现有 company/、offer/、careerprogress/；拟新增 onboarding/ | 本人进度隔离、企业内容归属、模板变更行为、重复更新；未认证内容不得冒充平台核验；页面完整走通 | 小刘 / 张顺 |
 | N05 薪资计算器与薪资参考 | 未确认有独立实现；可解释的计算明细与有来源的薪资参考；明确币种、周期、地区、规则生效期和估算性质 | 现有 job/、offer/ 可提供薪资字段；拟新增 salary/ | 经确认的规则用独立算例验证舍入与边界；来源/版本/更新时间；缺地区或参考样本时明确不可计算/不足；不伪造市场数据 | 小刘 / 张顺 |
 | N06 诚信积分与激励 | 独立积分能力未完成，现有INTEGRITY不是积分账户；先确定合格行为、积分流水、撤销和展示语义，不以简单行为推断人格 | 现有 candidatescore/CandidateScoringCalculator.java#computeIntegrity、audit/；拟新增 incentive/ | 积分事件幂等、并发不重复加分、撤销可追溯、余额与流水一致、刷行为不无限获益；与候选评分默认隔离，接入必须单独评审 | 小刘 / 张顺 |
-| N07 消息撤回 | 未确认有独立实现；发送者在批准窗口内撤回，列表/详情/重试响应同步显示撤回态；保留必要审计 | 现有 messaging/MessagingService.java、talentinvitation/；新增位置由设计确定 | 非本人不可撤回；时间边界、并发发送/读取/撤回、幂等；撤回后正文不从缓存或历史响应泄漏；DIRECT与邀约会话回归 | 张顺 / 小刘 |
+| N07 消息撤回 | N07 定向整改完成并通过自动化与干净构建，待独立代码复验与真实 MySQL/Docker 验收；发送者在两分钟窗口内撤回，列表/响应统一显示撤回态并覆盖原正文 | 现有 messaging/MessagingService.java、talentinvitation/；docs/message-recall-design.md、docs/message-recall-implementation-instruction.md、docs/messaging-api.md | 非本人不可撤回；时间边界、并发读取/撤回、幂等；撤回后正文不从当前数据库、响应、日志或审计泄漏；DIRECT与邀约会话回归 | 张顺 / 小刘 |
 | N08 简历查重 | 未确认有独立实现；先界定同人重复上传与跨文档相似检测，给出依据和局限；不得直接判抄袭或自动扣分 | 现有 resume/ResumeVersionService.java、profile/storage/；拟新增 resumededup/ | 重复文件/改名/模板相似/不同格式/解析失败样例；阈值误报验证；不得借查重返回他人简历、姓名或联系方式；上传性能与租户隔离 | 张顺 / 小刘 |

 ### 编码前必须补齐的产品决策
--- a/docs/verify_openapi.py
+++ b/docs/verify_openapi.py
@ -6,7 +6,7 @@ p = PROJECT_ROOT / 'docs' / 'openapi-rc.json'
 d = json.load(open(p, encoding='utf-8'))

 print('paths:', len(d['paths']))
-assert len(d['paths']) == 172, f"Expected 172 paths, got {len(d['paths'])}"
+assert len(d['paths']) == 173, f"Expected 173 paths, got {len(d['paths'])}"

 paths = d['paths']

@ -559,10 +559,42 @@ assert all(path in paths for path in candidate_scoring_paths), 'Candidate scorin
 assert sum(len(paths[path]) for path in candidate_scoring_paths) == 4, 'Candidate scoring operation count changed'
 print('  N02 explanation schemas validated; enterprise scoring schema unchanged; 4 scoring operations preserved')

-# 统计 operation 总数并校验 V43 计数 210
+# N07 消息撤回：仅新增一个 POST，消息响应为白名单 DTO。
+n07_path = '/conversations/{conversationId}/messages/{messageId}/recall'
+assert n07_path in paths, 'N07 recall path missing'
+assert list(paths[n07_path].keys()) == ['post'], 'N07 must expose exactly one POST operation'
+n07_operation = paths[n07_path]['post']
+assert n07_operation.get('parameters') and len(n07_operation['parameters']) == 2, 'N07 path parameters missing'
+assert n07_operation.get('responses', {}).get('200', {}).get('content', {}).get('*/*', {}).get('schema', {}).get('$ref', '').endswith('ApiResponseMessageResponse'), 'N07 200 schema missing'
+assert 'MessageResponse' in d['components']['schemas'], 'MessageResponse schema missing'
+assert 'ApiResponseMessageResponse' in d['components']['schemas'], 'ApiResponseMessageResponse schema missing'
+assert 'MessagePageResponse' in d['components']['schemas'], 'MessagePageResponse schema missing'
+assert 'ApiResponseMessagePageResponse' in d['components']['schemas'], 'ApiResponseMessagePageResponse schema missing'
+# Inspect the raw array item reference here.  ``resolve_schema`` intentionally
+# expands local refs and therefore removes the ``$ref`` node; using the
+# expanded object for this assertion made a valid MessageResponse array look
+# invalid even though the static contract was correct.
+message_page_schema = resolve_schema(d['components']['schemas']['MessagePageResponse'])
+message_page_items = d['components']['schemas']['MessagePageResponse'].get('properties', {}).get('items', {})
+assert message_page_items.get('type') == 'array', 'MessagePageResponse items missing'
+assert message_page_items.get('items', {}).get('$ref', '').endswith('MessageResponse'), 'MessagePageResponse item schema missing'
+assert paths['/conversations/{id}/messages']['get']['responses']['200']['content']['*/*']['schema'].get('$ref', '').endswith('ApiResponseMessagePageResponse'), 'message list schema missing'
+message_props = d['components']['schemas']['MessageResponse'].get('properties', {})
+for field in ('id', 'conversationId', 'content', 'status', 'createdAt'):
+    assert field in message_props, f'MessageResponse missing {field}'
+assert message_props['status'].get('enum') == ['SENT', 'RECALLED'], 'MessageResponse status enum mismatch'
+message_text = json.dumps(resolve_schema(d['components']['schemas']['MessageResponse']), ensure_ascii=False)
+for forbidden in ('originalContent', 'previousContent', 'studentUserId', 'candidateId', 'phone', 'email',
+                  'resumeUrl', 'candidateKey', 'candidateKeyDigest', 'password', 'token', 'secret'):
+    assert forbidden not in message_text, f'MessageResponse exposes forbidden field {forbidden}'
+assert '/conversations' in paths and '/conversations/messages' in paths
+assert '/conversations/{id}/messages' in paths and '/conversations/{id}/read' in paths
+print('  N07 recall path/schema validated; existing messaging operations preserved')
+
+# 统计 operation 总数并校验 V43 + N07 计数 211
 operations = sum(len(methods) for methods in d['paths'].values())
 print('operations:', operations)
-assert operations == 210, f"Expected 210 operations, got {operations}"
+assert operations == 211, f"Expected 211 operations, got {operations}"
 print("  Offer advance endpoints all present")
 print("  26 operations all present")
 print("  Account deletion endpoints present (3 operations)")
--- a/mydreamjob-server/src/main/java/com/mydreamjob/common/exception/ErrorCode.java
+++ b/mydreamjob-server/src/main/java/com/mydreamjob/common/exception/ErrorCode.java
@ -80,8 +80,11 @@ public enum ErrorCode {
    MATCH_SCORES_LOCKED(8302, "match already succeeded, scores are locked"),
    MATCH_REQUIRED_FOR_CONVERSATION(8303, "direct conversation between company and candidate requires a matched application"),
    APPLICATION_NOT_MATCHED(8304, "application is not matched"),
-    // 站内信 8305
+    // 站内信 83xx
    CONVERSATION_NOT_FOUND(8305, "conversation not found"),
+    MESSAGE_NOT_FOUND(8306, "message not found"),
+    MESSAGE_RECALL_WINDOW_EXPIRED(8307, "message recall window expired"),
+    MESSAGE_RECALL_CONCURRENT_UPDATE(8308, "message recall concurrent update"),

    // 评价 91xx
    DUPLICATE_REVIEW(9101, "duplicate review"),
--- a/mydreamjob-server/src/main/java/com/mydreamjob/common/exception/GlobalExceptionHandler.java
+++ b/mydreamjob-server/src/main/java/com/mydreamjob/common/exception/GlobalExceptionHandler.java
@ -75,6 +75,7 @@ public class GlobalExceptionHandler {
                || errorCode == ErrorCode.REVIEW_NOT_FOUND.getCode()
                || errorCode == ErrorCode.FEEDBACK_NOT_FOUND.getCode()
                || errorCode == ErrorCode.CONVERSATION_NOT_FOUND.getCode()
+                || errorCode == ErrorCode.MESSAGE_NOT_FOUND.getCode()
                || errorCode == ErrorCode.APPEAL_NOT_FOUND.getCode()
                || errorCode == ErrorCode.DATA_EXPORT_TASK_NOT_FOUND.getCode()
                || errorCode == ErrorCode.REVIEW_REPLY_NOT_FOUND.getCode()
@ -111,6 +112,8 @@ public class GlobalExceptionHandler {
                || errorCode == ErrorCode.REVIEW_REPLY_INVALID_STATE.getCode()
                || errorCode == ErrorCode.REVIEW_REPLY_CONCURRENT_UPDATE.getCode()
                || errorCode == ErrorCode.REVIEW_REPLY_IDEMPOTENCY_CONFLICT.getCode()
+                || errorCode == ErrorCode.MESSAGE_RECALL_WINDOW_EXPIRED.getCode()
+                || errorCode == ErrorCode.MESSAGE_RECALL_CONCURRENT_UPDATE.getCode()
                || errorCode == ErrorCode.TALENT_INVITATION_CANDIDATE_UNAVAILABLE.getCode()
                || errorCode == ErrorCode.TALENT_INVITATION_ALREADY_ACTIVE.getCode()
                || errorCode == ErrorCode.TALENT_INVITATION_INVALID_STATE.getCode()
--- a/mydreamjob-server/src/main/java/com/mydreamjob/messaging/Message.java
+++ b/mydreamjob-server/src/main/java/com/mydreamjob/messaging/Message.java
@ -32,6 +32,18 @@ public class Message {
    @Column(nullable = false, columnDefinition = "text")
    private String content;

+    @Column(nullable = false, length = 20)
+    @Builder.Default
+    private String status = "SENT";
+
+    @Column(name = "recalled_at")
+    private LocalDateTime recalledAt;
+
+    @Version
+    @Column(nullable = false)
+    @Builder.Default
+    private Long version = 0L;
+
    @CreatedDate
    @Column(name = "created_at", nullable = false, updatable = false)
    private LocalDateTime createdAt;
--- a/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageRepository.java
+++ b/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageRepository.java
@ -3,10 +3,45 @@ package com.mydreamjob.messaging;
 import org.springframework.data.domain.Page;
 import org.springframework.data.domain.Pageable;
 import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Lock;
+import org.springframework.data.jpa.repository.Modifying;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
 import org.springframework.stereotype.Repository;
+import jakarta.persistence.LockModeType;
+
+import java.time.LocalDateTime;

@Repository
 public interface MessageRepository extends JpaRepository<Message, Long> {

    Page<Message> findByConversationIdOrderByCreatedAtDesc(Long conversationId, Pageable pageable);
+
+    /**
+     * 条件更新未命中后的当前读，用于区分并发请求已完成撤回与真实冲突。
+     * PESSIMISTIC_WRITE 强制读取提交后的最新行，避免 MySQL REPEATABLE READ 使用旧快照。
+     */
+    @Lock(LockModeType.PESSIMISTIC_WRITE)
+    @Query("select m from Message m where m.id = :messageId")
+    java.util.Optional<Message> findCurrentByIdForRecall(@Param("messageId") Long messageId);
+
+    @Modifying(clearAutomatically = true, flushAutomatically = true)
+    @Query("""
+            update Message m
+               set m.status = 'RECALLED',
+                   m.content = :placeholder,
+                   m.recalledAt = :recalledAt,
+                   m.version = m.version + 1
+             where m.id = :messageId
+               and m.conversationId = :conversationId
+               and m.senderUserId = :senderUserId
+               and m.status = 'SENT'
+               and m.version = :expectedVersion
+            """)
+    int recallIfCurrent(@Param("messageId") Long messageId,
+                        @Param("conversationId") Long conversationId,
+                        @Param("senderUserId") Long senderUserId,
+                        @Param("expectedVersion") Long expectedVersion,
+                        @Param("placeholder") String placeholder,
+                        @Param("recalledAt") LocalDateTime recalledAt);
 }
--- a/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageViewMapper.java
+++ b/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessageViewMapper.java
@ -0,0 +1,32 @@
+package com.mydreamjob.messaging;
+
+import com.mydreamjob.messaging.dto.MessageResponse;
+import org.springframework.stereotype.Component;
+
+import java.util.Objects;
+
+/** 集中处理 DIRECT 与 TALENT_INVITATION 消息的白名单映射。 */
+@Component
+public class MessageViewMapper {
+
+    public static final String RECALLED_PLACEHOLDER = "消息已撤回";
+
+    public MessageResponse toResponse(Message message, Long currentUserId, boolean talentInvitation) {
+        if (message == null) return null;
+        boolean recalled = "RECALLED".equals(message.getStatus());
+        MessageResponse.MessageResponseBuilder builder = MessageResponse.builder()
+                .id(message.getId())
+                .conversationId(message.getConversationId())
+                .content(recalled ? RECALLED_PLACEHOLDER : message.getContent())
+                .status(message.getStatus() == null ? "SENT" : message.getStatus())
+                .recalledAt(message.getRecalledAt())
+                .createdAt(message.getCreatedAt());
+        if (talentInvitation) {
+            builder.senderScope(Objects.equals(message.getSenderUserId(), currentUserId)
+                    ? "CURRENT_USER" : "OTHER_PARTICIPANT");
+        } else {
+            builder.senderUserId(message.getSenderUserId());
+        }
+        return builder.build();
+    }
+}
--- a/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessagingController.java
+++ b/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessagingController.java
@ -1,20 +1,21 @@
 package com.mydreamjob.messaging;

 import com.mydreamjob.common.api.ApiResponse;
+import com.mydreamjob.audit.AuditContext;
 import com.mydreamjob.messaging.dto.CreateConversationRequest;
+import com.mydreamjob.messaging.dto.MessagePageResponse;
+import com.mydreamjob.messaging.dto.MessageResponse;
 import com.mydreamjob.messaging.dto.SendMessageRequest;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.tags.Tag;
 import jakarta.servlet.http.HttpServletRequest;
 import jakarta.validation.Valid;
 import lombok.RequiredArgsConstructor;
-import org.springframework.data.domain.Page;
 import org.springframework.http.ResponseEntity;
 import org.springframework.security.core.Authentication;
 import org.springframework.security.core.context.SecurityContextHolder;
 import org.springframework.web.bind.annotation.*;

-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;

@ -60,18 +61,14 @@ public class MessagingController {

    @Operation(summary = "消息列表", description = "分页查询会话消息，倒序；仅成员可查看")
    @GetMapping("/{id}/messages")
-    public ResponseEntity<ApiResponse<Map<String, Object>>> listMessages(
+    public ResponseEntity<ApiResponse<MessagePageResponse>> listMessages(
            @PathVariable Long id,
            @RequestParam(defaultValue = "0") int page,
            @RequestParam(defaultValue = "20") int size,
            HttpServletRequest request) {
        Long userId = resolveUserId(request);
-        Page<Map<String, Object>> result = messagingService.listMessages(id, userId, page, size);
-        Map<String, Object> data = new HashMap<>();
-        data.put("items", result.getContent()); data.put("total", result.getTotalElements());
-        data.put("page", result.getNumber()); data.put("size", result.getSize());
-        data.put("totalPages", result.getTotalPages());
-        return ResponseEntity.ok(ApiResponse.ok(data, getRequestId(request)));
+        MessagePageResponse result = messagingService.listMessages(id, userId, page, size);
+        return ResponseEntity.ok(ApiResponse.ok(result, getRequestId(request)));
    }

    @Operation(summary = "标记已读", description = "将当前会话所有消息标记为已读")
@ -83,6 +80,18 @@ public class MessagingController {
        return ResponseEntity.ok(ApiResponse.ok(Map.of("message", "marked as read"), getRequestId(request)));
    }

+    @Operation(summary = "撤回消息", description = "仅原发送者可在发送后两分钟内撤回消息；重复撤回幂等返回当前状态")
+    @PostMapping("/{conversationId}/messages/{messageId}/recall")
+    public ResponseEntity<ApiResponse<MessageResponse>> recallMessage(
+            @PathVariable Long conversationId,
+            @PathVariable Long messageId,
+            HttpServletRequest request) {
+        Long userId = resolveUserId(request);
+        MessageResponse response = messagingService.recallMessage(
+                userId, conversationId, messageId, AuditContext.from(request));
+        return ResponseEntity.ok(ApiResponse.ok(response, getRequestId(request)));
+    }
+
    private Long resolveUserId(HttpServletRequest request) {
        Authentication auth = SecurityContextHolder.getContext().getAuthentication();
        if (auth == null || !auth.isAuthenticated())
--- a/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessagingService.java
+++ b/mydreamjob-server/src/main/java/com/mydreamjob/messaging/MessagingService.java
@ -1,7 +1,10 @@
 package com.mydreamjob.messaging;

+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.mydreamjob.auth.User;
 import com.mydreamjob.auth.UserRepository;
+import com.mydreamjob.audit.AuditService;
 import com.mydreamjob.audit.AuditContext;
 import com.mydreamjob.audit.SecurityAuditService;
 import com.mydreamjob.bidirectionalmatch.ApplicationMatchRecordRepository;
@ -10,6 +13,8 @@ import com.mydreamjob.common.exception.ErrorCode;
 import com.mydreamjob.company.Company;
 import com.mydreamjob.company.CompanyRepository;
 import com.mydreamjob.messaging.dto.CreateConversationRequest;
+import com.mydreamjob.messaging.dto.MessagePageResponse;
+import com.mydreamjob.messaging.dto.MessageResponse;
 import com.mydreamjob.messaging.dto.SendMessageRequest;
 import com.mydreamjob.safety.BlockRelationRepository;
 import lombok.RequiredArgsConstructor;
@ -22,6 +27,8 @@ import org.springframework.web.context.request.RequestContextHolder;
 import org.springframework.web.context.request.ServletRequestAttributes;

 import java.util.*;
+import java.time.Clock;
+import java.time.LocalDateTime;

@Slf4j
@Service
@ -37,9 +44,16 @@ public class MessagingService {
    private final ApplicationMatchRecordRepository applicationMatchRecordRepository;
    private final BlockRelationRepository blockRelationRepository;
    private final SecurityAuditService securityAuditService;
+    private final AuditService auditService;
+    private final Clock clock;
+    private final MessageViewMapper messageViewMapper;
+    private final ObjectMapper objectMapper;

    private static final int MAX_PAGE_SIZE = 50;
    private static final int MAX_CONTENT_LENGTH = 2000;
+    private static final String RECALLED_STATUS = "RECALLED";
+    private static final String SENT_STATUS = "SENT";
+    private static final int RECALL_WINDOW_MINUTES = 2;

    private void assertMember(Long conversationId, Long userId) {
        if (!memberRepository.existsByConversationIdAndUserId(conversationId, userId))
@ -177,16 +191,29 @@ public class MessagingService {
        return toView(msg, userId);
    }

-    public Page<Map<String, Object>> listMessages(Long conversationId, Long userId, int page, int size) {
+    public MessagePageResponse listMessages(Long conversationId, Long userId, int page, int size) {
        Conversation conversation = requireConversation(conversationId);
        assertMember(conversationId, userId);
        if (page < 0 || size < 1 || size > MAX_PAGE_SIZE) throw new BusinessException(ErrorCode.BAD_REQUEST);
-        return messageRepository.findByConversationIdOrderByCreatedAtDesc(conversationId, PageRequest.of(page, size))
-                .map(m -> {
-                    Map<String, Object> v = toView(m, userId, "TALENT_INVITATION".equals(conversation.getType()));
-                    v.put("isRead", readRepository.existsByMessageIdAndReaderUserId(m.getId(), userId));
-                    return v;
-                });
+        Page<Message> result = messageRepository.findByConversationIdOrderByCreatedAtDesc(
+                conversationId, PageRequest.of(page, size));
+        List<Long> messageIds = result.getContent().stream().map(Message::getId).toList();
+        Set<Long> readIds = messageIds.isEmpty() ? Set.of() : readRepository
+                .findByMessageIdInAndReaderUserId(messageIds, userId).stream()
+                .map(MessageRead::getMessageId).collect(java.util.stream.Collectors.toSet());
+        boolean talentInvitation = "TALENT_INVITATION".equals(conversation.getType());
+        List<MessageResponse> items = result.getContent().stream().map(m -> {
+            MessageResponse response = messageViewMapper.toResponse(m, userId, talentInvitation);
+            response.setIsRead(readIds.contains(m.getId()));
+            return response;
+        }).toList();
+        return MessagePageResponse.builder()
+                .items(items)
+                .total(result.getTotalElements())
+                .page(result.getNumber())
+                .size(result.getSize())
+                .totalPages(result.getTotalPages())
+                .build();
    }

    @Transactional
@ -195,7 +222,10 @@ public class MessagingService {
        assertMember(conversationId, userId);
        List<Message> msgs = messageRepository
                .findByConversationIdOrderByCreatedAtDesc(conversationId, PageRequest.of(0, 500)).getContent();
-        List<Long> msgIds = msgs.stream().map(Message::getId).toList();
+        List<Long> msgIds = msgs.stream()
+                .filter(m -> !RECALLED_STATUS.equals(m.getStatus()))
+                .map(Message::getId).toList();
+        if (msgIds.isEmpty()) return;
        List<MessageRead> existing = readRepository.findByMessageIdInAndReaderUserId(msgIds, userId);
        Set<Long> existingIds = new HashSet<>();
        for (MessageRead r : existing) existingIds.add(r.getMessageId());
@ -209,10 +239,104 @@ public class MessagingService {
        List<Message> msgs = messageRepository
                .findByConversationIdOrderByCreatedAtDesc(convId, PageRequest.of(0, 500)).getContent();
        if (msgs.isEmpty()) return 0;
-        List<Long> msgIds = msgs.stream().map(Message::getId).toList();
+        List<Long> msgIds = msgs.stream()
+                .filter(m -> !RECALLED_STATUS.equals(m.getStatus()))
+                .map(Message::getId).toList();
+        if (msgIds.isEmpty()) return 0;
        return msgIds.size() - readRepository.findByMessageIdInAndReaderUserId(msgIds, userId).size();
    }

+    /**
+     * 在事务内执行消息撤回。调用者身份、会话成员关系和原发送者关系均由服务端校验。
+     * capturedNow 由入口统一捕获，便于边界测试，也避免同一请求混用多个时间。
+     */
+    @Transactional
+    public MessageResponse recallMessage(Long currentUserId, Long conversationId, Long messageId,
+                                         AuditContext auditContext, LocalDateTime capturedNow) {
+        Conversation conversation = requireConversation(conversationId);
+        assertMember(conversationId, currentUserId);
+        User currentUser = userRepository.findById(currentUserId).orElse(null);
+        if (currentUser != null && hasRole(currentUser, User.Role.ADMIN)) {
+            throw new BusinessException(ErrorCode.FORBIDDEN);
+        }
+        Message message = messageRepository.findById(messageId)
+                .orElseThrow(() -> new BusinessException(ErrorCode.MESSAGE_NOT_FOUND));
+        if (!Objects.equals(message.getConversationId(), conversationId)) {
+            throw new BusinessException(ErrorCode.FORBIDDEN);
+        }
+        if (!Objects.equals(message.getSenderUserId(), currentUserId)) {
+            throw new BusinessException(ErrorCode.FORBIDDEN);
+        }
+
+        String status = message.getStatus() == null ? SENT_STATUS : message.getStatus();
+        if (RECALLED_STATUS.equals(status)) {
+            return toMessageResponse(message, currentUserId, conversation);
+        }
+        LocalDateTime createdAt = message.getCreatedAt();
+        if (createdAt == null || capturedNow == null
+                || capturedNow.isAfter(createdAt.plusMinutes(RECALL_WINDOW_MINUTES))) {
+            throw new BusinessException(ErrorCode.MESSAGE_RECALL_WINDOW_EXPIRED);
+        }
+        Long expectedVersion = message.getVersion() == null ? 0L : message.getVersion();
+        int updated = messageRepository.recallIfCurrent(messageId, conversationId, currentUserId,
+                expectedVersion, MessageViewMapper.RECALLED_PLACEHOLDER, capturedNow);
+        if (updated != 1) {
+            Message latest = messageRepository.findCurrentByIdForRecall(messageId).orElse(null);
+            if (latest != null && RECALLED_STATUS.equals(latest.getStatus())) {
+                return toMessageResponse(latest, currentUserId, conversation);
+            }
+            throw new BusinessException(ErrorCode.MESSAGE_RECALL_CONCURRENT_UPDATE);
+        }
+
+        auditService.log(currentUserId, "MESSAGE_RECALLED", "MESSAGE", messageId,
+                recallAuditDetails(messageId, conversationId, conversation.getType(), auditContext),
+                recallAuditContext(auditContext));
+        Message recalled = messageRepository.findById(messageId).orElse(message);
+        return toMessageResponse(recalled, currentUserId, conversation);
+    }
+
+    /** 生产入口使用 Spring Clock 捕获一次当前时间。 */
+    @Transactional
+    public MessageResponse recallMessage(Long currentUserId, Long conversationId, Long messageId,
+                                         AuditContext auditContext) {
+        Clock effectiveClock = clock == null ? Clock.systemDefaultZone() : clock;
+        return recallMessage(currentUserId, conversationId, messageId, auditContext,
+                LocalDateTime.now(effectiveClock));
+    }
+
+    private MessageResponse toMessageResponse(Message message, Long currentUserId, Conversation conversation) {
+        MessageResponse response = messageViewMapper.toResponse(message, currentUserId,
+                conversation != null && "TALENT_INVITATION".equals(conversation.getType()));
+        if (response != null) {
+            response.setIsRead(readRepository.existsByMessageIdAndReaderUserId(message.getId(), currentUserId));
+        }
+        return response;
+    }
+
+    private AuditContext recallAuditContext(AuditContext source) {
+        return AuditContext.builder()
+                .requestId(source == null ? null : source.getRequestId())
+                .build();
+    }
+
+    private String recallAuditDetails(Long messageId, Long conversationId, String conversationType,
+                                      AuditContext context) {
+        String safeConversationType = "DIRECT".equals(conversationType)
+                || "TALENT_INVITATION".equals(conversationType)
+                ? conversationType : "UNKNOWN";
+        Map<String, Object> details = new LinkedHashMap<>();
+        details.put("messageId", messageId);
+        details.put("conversationId", conversationId);
+        details.put("conversationType", safeConversationType);
+        details.put("status", "RECALLED");
+        details.put("requestId", context == null ? null : context.getRequestId());
+        try {
+            return objectMapper.writeValueAsString(details);
+        } catch (JsonProcessingException ex) {
+            throw new IllegalStateException("message recall audit serialization failed");
+        }
+    }
+
    public Map<String, Object> toView(Conversation c) {
        Map<String, Object> v = new HashMap<>();
        v.put("id", c.getId()); v.put("title", c.getTitle()); v.put("type", c.getType());
@ -231,10 +355,7 @@ public class MessagingService {
    }

    public Map<String, Object> toView(Message m) {
-        Map<String, Object> v = new HashMap<>();
-        v.put("id", m.getId()); v.put("conversationId", m.getConversationId());
-        v.put("senderUserId", m.getSenderUserId()); v.put("content", m.getContent()); v.put("createdAt", m.getCreatedAt());
-        return v;
+        return toLegacyView(messageViewMapper.toResponse(m, null, false));
    }

    /** V42 消息隐私映射：TALENT_INVITATION 仅返回相对发送者身份。 */
@ -244,11 +365,21 @@ public class MessagingService {
    }

    private Map<String, Object> toView(Message m, Long currentUserId, boolean talentInvitation) {
-        if (!talentInvitation) return toView(m);
+        return toLegacyView(messageViewMapper.toResponse(m, currentUserId, talentInvitation));
+    }
+
+    private Map<String, Object> toLegacyView(MessageResponse response) {
        Map<String, Object> v = new HashMap<>();
-        v.put("id", m.getId()); v.put("conversationId", m.getConversationId());
-        v.put("senderScope", Objects.equals(m.getSenderUserId(), currentUserId) ? "CURRENT_USER" : "OTHER_PARTICIPANT");
-        v.put("content", m.getContent()); v.put("createdAt", m.getCreatedAt());
+        if (response == null) return v;
+        v.put("id", response.getId());
+        v.put("conversationId", response.getConversationId());
+        if (response.getSenderUserId() != null) v.put("senderUserId", response.getSenderUserId());
+        if (response.getSenderScope() != null) v.put("senderScope", response.getSenderScope());
+        v.put("content", response.getContent());
+        v.put("status", response.getStatus());
+        if (response.getRecalledAt() != null) v.put("recalledAt", response.getRecalledAt());
+        v.put("createdAt", response.getCreatedAt());
+        if (response.getIsRead() != null) v.put("isRead", response.getIsRead());
        return v;
    }

--- a/mydreamjob-server/src/main/java/com/mydreamjob/messaging/dto/MessagePageResponse.java
+++ b/mydreamjob-server/src/main/java/com/mydreamjob/messaging/dto/MessagePageResponse.java
@ -0,0 +1,24 @@
+package com.mydreamjob.messaging.dto;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+/** 消息分页白名单响应，保持既有分页字段兼容。 */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@Schema(name = "MessagePageResponse", description = "脱敏消息分页响应")
+public class MessagePageResponse {
+
+    private List<MessageResponse> items;
+    private long total;
+    private int page;
+    private int size;
+    private int totalPages;
+}
--- a/mydreamjob-server/src/main/java/com/mydreamjob/messaging/dto/MessageResponse.java
+++ b/mydreamjob-server/src/main/java/com/mydreamjob/messaging/dto/MessageResponse.java
@ -0,0 +1,31 @@
+package com.mydreamjob.messaging.dto;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.time.LocalDateTime;
+
+/** 消息白名单响应；不直接暴露 Message 实体。 */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@JsonInclude(JsonInclude.Include.NON_NULL)
+@Schema(name = "MessageResponse", description = "脱敏消息响应")
+public class MessageResponse {
+
+    private Long id;
+    private Long conversationId;
+    /** 仅 DIRECT 会话返回；TALENT_INVITATION 使用 senderScope。 */
+    private Long senderUserId;
+    private String senderScope;
+    private String content;
+    private String status;
+    private LocalDateTime recalledAt;
+    private LocalDateTime createdAt;
+    private Boolean isRead;
+}
--- a/mydreamjob-server/src/main/resources/db/migration/V41__message_recall.sql
+++ b/mydreamjob-server/src/main/resources/db/migration/V41__message_recall.sql
@ -0,0 +1,6 @@
+ALTER TABLE messages ADD COLUMN status VARCHAR(20) NOT NULL DEFAULT 'SENT';
+ALTER TABLE messages ADD COLUMN recalled_at DATETIME NULL;
+ALTER TABLE messages ADD COLUMN version BIGINT NOT NULL DEFAULT 0;
+
+ALTER TABLE messages
+    ADD CONSTRAINT chk_messages_status CHECK (status IN ('SENT', 'RECALLED'));
--- a/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessageListQueryBudgetIntegrationTest.java
+++ b/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessageListQueryBudgetIntegrationTest.java
@ -0,0 +1,215 @@
+package com.mydreamjob.messaging;
+
+import com.mydreamjob.auth.User;
+import com.mydreamjob.auth.UserRepository;
+import com.mydreamjob.messaging.dto.MessagePageResponse;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.context.TestConfiguration;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Import;
+import org.springframework.context.annotation.Primary;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.jdbc.datasource.AbstractDataSource;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.test.context.ActiveProfiles;
+
+import javax.sql.DataSource;
+import java.lang.reflect.Proxy;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * 真实 MessagingService + Repository 的消息列表查询预算证据。
+ * 只统计夹具准备完成后 listMessages 的 JDBC 查询，不以 Mockito verify 替代 SQL 计数。
+ */
+@SpringBootTest
+@ActiveProfiles("integration-test")
+@Import(MessageListQueryBudgetIntegrationTest.CountingDataSourceConfiguration.class)
+class MessageListQueryBudgetIntegrationTest {
+
+    @Autowired private MessagingService messagingService;
+    @Autowired private UserRepository userRepository;
+    @Autowired private ConversationRepository conversationRepository;
+    @Autowired private ConversationMemberRepository memberRepository;
+    @Autowired private MessageRepository messageRepository;
+    @Autowired private MessageReadRepository readRepository;
+    @Autowired private JdbcTemplate jdbcTemplate;
+    @Autowired private CountingDataSource countingDataSource;
+    @MockBean private RedisTemplate<String, Object> redisTemplate;
+
+    private final List<Long> userIds = new ArrayList<>();
+    private final List<Long> conversationIds = new ArrayList<>();
+
+    @AfterEach
+    void cleanup() {
+        if (!conversationIds.isEmpty()) {
+            String marks = placeholders(conversationIds.size());
+            jdbcTemplate.update("DELETE FROM message_reads WHERE message_id IN "
+                    + "(SELECT id FROM messages WHERE conversation_id IN (" + marks + "))",
+                    conversationIds.toArray());
+            jdbcTemplate.update("DELETE FROM messages WHERE conversation_id IN (" + marks + ")",
+                    conversationIds.toArray());
+            jdbcTemplate.update("DELETE FROM conversation_members WHERE conversation_id IN (" + marks + ")",
+                    conversationIds.toArray());
+            jdbcTemplate.update("DELETE FROM conversations WHERE id IN (" + marks + ")",
+                    conversationIds.toArray());
+        }
+        if (!userIds.isEmpty()) {
+            String marks = placeholders(userIds.size());
+            jdbcTemplate.update("DELETE FROM user_roles WHERE user_id IN (" + marks + ")", userIds.toArray());
+            jdbcTemplate.update("DELETE FROM users WHERE id IN (" + marks + ")", userIds.toArray());
+        }
+        userIds.clear();
+        conversationIds.clear();
+    }
+
+    @Test
+    void listMessagesUsesFixedBatchQueriesForOneTwentyAndFiftyMessages() {
+        List<Integer> measuredStatements = new ArrayList<>();
+        for (int count : List.of(1, 20, 50)) {
+            long sender = user("budget-sender-" + count);
+            long receiver = user("budget-receiver-" + count);
+            long conversation = conversation(sender, receiver, count % 2 == 0 ? "TALENT_INVITATION" : "DIRECT");
+            LocalDateTime created = LocalDateTime.of(2026, 9, 10, 10, 0);
+            List<Message> messages = new ArrayList<>();
+            for (int i = 0; i < count; i++) {
+                boolean recalled = i % 7 == 0;
+                messages.add(Message.builder()
+                        .conversationId(conversation)
+                        .senderUserId(sender)
+                        .content(recalled ? MessageViewMapper.RECALLED_PLACEHOLDER : "budget-body-" + i)
+                        .status(recalled ? "RECALLED" : "SENT")
+                        .version(recalled ? 1L : 0L)
+                        .recalledAt(recalled ? created.plusSeconds(i) : null)
+                        .createdAt(created.plusSeconds(i))
+                        .build());
+            }
+            messageRepository.saveAllAndFlush(messages);
+
+            countingDataSource.reset();
+            MessagePageResponse response = messagingService.listMessages(conversation, receiver, 0, 50);
+            int statements = countingDataSource.messageStatements();
+            measuredStatements.add(statements);
+            assertTrue(statements <= 5,
+                    "message list must stay within the fixed conversation/member/page/read budget (count=" + count + ")");
+            assertEquals(count, response.getItems().size());
+            assertEquals(count, response.getTotal());
+            assertTrue(response.getItems().stream().allMatch(item -> item.getConversationId().equals(conversation)));
+            assertFalse(response.getItems().stream().anyMatch(item -> item.getContent() != null
+                    && item.getContent().startsWith("budget-body-")
+                    && item.getStatus().equals("RECALLED")),
+                    "recalled messages must expose only the placeholder");
+        }
+        assertTrue(measuredStatements.stream().max(Integer::compareTo).orElse(0)
+                        - measuredStatements.stream().min(Integer::compareTo).orElse(0) <= 1,
+                "pagination count query optimization may add one statement, but must not scale with item count: "
+                        + measuredStatements);
+    }
+
+    private long user(String nickname) {
+        User saved = userRepository.saveAndFlush(User.builder()
+                .phone("139" + Math.abs(System.nanoTime() % 1_000_000_000L))
+                .nickname(nickname).role(User.Role.STUDENT).roles(Set.of(User.Role.STUDENT))
+                .active(true).verified(true).build());
+        userIds.add(saved.getId());
+        return saved.getId();
+    }
+
+    private long conversation(long sender, long receiver, String type) {
+        Conversation saved = conversationRepository.saveAndFlush(Conversation.builder()
+                .type(type).createdBy(sender).title("N07 query budget").build());
+        conversationIds.add(saved.getId());
+        memberRepository.save(ConversationMember.builder().conversationId(saved.getId()).userId(sender).build());
+        memberRepository.saveAndFlush(ConversationMember.builder().conversationId(saved.getId()).userId(receiver).build());
+        return saved.getId();
+    }
+
+    private static String placeholders(int size) {
+        return String.join(",", java.util.Collections.nCopies(size, "?"));
+    }
+
+    @TestConfiguration
+    static class CountingDataSourceConfiguration {
+        @Bean
+        @Primary
+        CountingDataSource dataSource(DataSourceProperties properties) {
+            return new CountingDataSource(properties.initializeDataSourceBuilder().build());
+        }
+    }
+
+    static final class CountingDataSource extends AbstractDataSource {
+        private final DataSource delegate;
+        private final AtomicInteger messageStatements = new AtomicInteger();
+
+        CountingDataSource(DataSource delegate) {
+            this.delegate = delegate;
+        }
+
+        void reset() {
+            messageStatements.set(0);
+        }
+
+        int messageStatements() {
+            return messageStatements.get();
+        }
+
+        @Override
+        public Connection getConnection() throws java.sql.SQLException {
+            return wrap(delegate.getConnection());
+        }
+
+        @Override
+        public Connection getConnection(String username, String password) throws java.sql.SQLException {
+            return wrap(delegate.getConnection(username, password));
+        }
+
+        private Connection wrap(Connection connection) {
+            return (Connection) Proxy.newProxyInstance(Connection.class.getClassLoader(),
+                    new Class<?>[]{Connection.class}, (proxy, method, args) -> {
+                        Object value = method.invoke(connection, args);
+                        if (value instanceof Statement statement) {
+                            String sql = args != null && args.length > 0 && args[0] instanceof String s ? s : "";
+                            return wrapStatement(statement, sql);
+                        }
+                        return value;
+                    });
+        }
+
+        private Object wrapStatement(Statement statement, String preparedSql) {
+            Class<?> iface = statement instanceof java.sql.PreparedStatement
+                    ? java.sql.PreparedStatement.class : Statement.class;
+            return Proxy.newProxyInstance(iface.getClassLoader(), new Class<?>[]{iface},
+                    (proxy, method, args) -> {
+                        String sql = preparedSql;
+                        if (sql.isBlank() && args != null && args.length > 0 && args[0] instanceof String s) sql = s;
+                        if (method.getName().startsWith("execute") && isMessageQuery(sql)) {
+                            messageStatements.incrementAndGet();
+                        }
+                        return method.invoke(statement, args);
+                    });
+        }
+
+        private boolean isMessageQuery(String sql) {
+            String normalized = sql.toLowerCase(java.util.Locale.ROOT);
+            return normalized.contains(" from conversations")
+                    || normalized.contains(" from conversation_members")
+                    || normalized.contains(" from messages")
+                    || normalized.contains(" from message_reads")
+                    || normalized.contains(" from message_reads");
+        }
+    }
+}
--- a/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessageRecallConcurrencyIntegrationTest.java
+++ b/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessageRecallConcurrencyIntegrationTest.java
@ -0,0 +1,174 @@
+package com.mydreamjob.messaging;
+
+import com.mydreamjob.audit.AuditLog;
+import com.mydreamjob.audit.AuditLogRepository;
+import com.mydreamjob.auth.User;
+import com.mydreamjob.auth.UserRepository;
+import com.mydreamjob.common.exception.BusinessException;
+import com.mydreamjob.common.exception.ErrorCode;
+import com.mydreamjob.messaging.dto.MessageResponse;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.support.TransactionTemplate;
+
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/** 真实事务竞争测试：不 Mock 撤回服务、仓库或审计。 */
+@SpringBootTest
+@ActiveProfiles("integration-test")
+class MessageRecallConcurrencyIntegrationTest {
+
+    @Autowired private MessagingService messagingService;
+    @Autowired private UserRepository userRepository;
+    @Autowired private ConversationRepository conversationRepository;
+    @Autowired private ConversationMemberRepository memberRepository;
+    @Autowired private MessageRepository messageRepository;
+    @Autowired private AuditLogRepository auditLogRepository;
+    @Autowired private JdbcTemplate jdbcTemplate;
+    @Autowired private PlatformTransactionManager transactionManager;
+    @MockBean private RedisTemplate<String, Object> redisTemplate;
+
+    private final List<Long> messages = new ArrayList<>();
+    private Long conversation;
+    private Long sender;
+    private Long receiver;
+
+    @AfterEach
+    void cleanup() {
+        if (!messages.isEmpty()) {
+            String marks = String.join(",", messages.stream().map(id -> "?").toList());
+            jdbcTemplate.update("DELETE FROM audit_logs WHERE action='MESSAGE_RECALLED' AND entity_id IN (" + marks + ")",
+                    messages.toArray());
+            jdbcTemplate.update("DELETE FROM message_reads WHERE message_id IN (" + marks + ")", messages.toArray());
+            jdbcTemplate.update("DELETE FROM messages WHERE id IN (" + marks + ")", messages.toArray());
+        }
+        if (conversation != null) {
+            memberRepository.deleteAll(memberRepository.findByConversationId(conversation));
+            conversationRepository.deleteById(conversation);
+        }
+        if (sender != null) userRepository.deleteById(sender);
+        if (receiver != null) userRepository.deleteById(receiver);
+        messages.clear();
+        conversation = sender = receiver = null;
+    }
+
+    private void fixture(int messageCount) {
+        sender = userRepository.save(user("concurrent-sender")).getId();
+        receiver = userRepository.save(user("concurrent-receiver")).getId();
+        conversation = conversationRepository.save(Conversation.builder().type("DIRECT").createdBy(sender).build()).getId();
+        memberRepository.save(ConversationMember.builder().conversationId(conversation).userId(sender).build());
+        memberRepository.save(ConversationMember.builder().conversationId(conversation).userId(receiver).build());
+        LocalDateTime created = LocalDateTime.of(2026, 9, 10, 10, 0);
+        for (int i = 0; i < messageCount; i++) {
+            Message saved = messageRepository.saveAndFlush(Message.builder()
+                    .conversationId(conversation).senderUserId(sender)
+                    .content("N07_CONCURRENT_ORIGINAL_" + i).status("SENT").version(0L)
+                    .createdAt(created).build());
+            jdbcTemplate.update("UPDATE messages SET created_at=? WHERE id=?", created, saved.getId());
+            messages.add(saved.getId());
+        }
+    }
+
+    private User user(String name) {
+        return User.builder().phone("139" + Math.abs(System.nanoTime() % 1_000_000_000L))
+                .nickname(name).role(User.Role.STUDENT).roles(Set.of(User.Role.STUDENT))
+                .active(true).verified(true).build();
+    }
+
+    private List<Attempt> runConcurrently(List<Long> messageIds) throws Exception {
+        ExecutorService pool = Executors.newFixedThreadPool(messageIds.size());
+        CountDownLatch ready = new CountDownLatch(messageIds.size());
+        CountDownLatch start = new CountDownLatch(1);
+        TransactionTemplate tx = new TransactionTemplate(transactionManager);
+        try {
+            List<Future<Attempt>> futures = messageIds.stream().map(id -> pool.submit(() -> {
+                ready.countDown();
+                assertTrue(start.await(5, TimeUnit.SECONDS));
+                try {
+                    MessageResponse response = tx.execute(status -> messagingService.recallMessage(
+                            sender, conversation, id, null, LocalDateTime.of(2026, 9, 10, 10, 1)));
+                    return new Attempt(response, null);
+                } catch (Throwable failure) {
+                    return new Attempt(null, failure);
+                }
+            })).toList();
+            assertTrue(ready.await(5, TimeUnit.SECONDS));
+            start.countDown();
+            List<Attempt> attempts = new ArrayList<>();
+            for (Future<Attempt> future : futures) attempts.add(future.get(10, TimeUnit.SECONDS));
+            return attempts;
+        } finally {
+            pool.shutdownNow();
+        }
+    }
+
+    @Test
+    void concurrentRecallOnlyTransitionsAndAuditsOnce() throws Exception {
+        fixture(1);
+        List<Attempt> attempts = runConcurrently(List.of(messages.get(0), messages.get(0)));
+
+        assertEquals(2, attempts.size());
+        for (Attempt attempt : attempts) {
+            if (attempt.failure() != null) {
+                Integer code = findBusinessCode(attempt.failure());
+                assertEquals(ErrorCode.MESSAGE_RECALL_CONCURRENT_UPDATE.getCode(), code,
+                        "并发失败只能是既定 8308，而不能泄漏 500");
+            } else {
+                assertEquals("RECALLED", attempt.response().getStatus());
+            }
+        }
+        Message persisted = messageRepository.findById(messages.get(0)).orElseThrow();
+        assertEquals("RECALLED", persisted.getStatus());
+        assertEquals(MessageViewMapper.RECALLED_PLACEHOLDER, persisted.getContent());
+        assertNotNull(persisted.getRecalledAt());
+        assertEquals(1L, persisted.getVersion());
+        assertEquals(1, auditLogRepository.findAll().stream()
+                .filter(a -> "MESSAGE_RECALLED".equals(a.getAction()) && messages.get(0).equals(a.getEntityId())).count());
+    }
+
+    @Test
+    void concurrentRecallOfDifferentMessagesDoesNotBlockOrCrossUpdate() throws Exception {
+        fixture(2);
+        List<Attempt> attempts = runConcurrently(List.of(messages.get(0), messages.get(1)));
+
+        assertEquals(2, attempts.size());
+        assertTrue(attempts.stream().allMatch(a -> a.failure() == null),
+                "不同消息的并发撤回不得产生异常");
+        for (Long messageId : messages) {
+            Message persisted = messageRepository.findById(messageId).orElseThrow();
+            assertEquals("RECALLED", persisted.getStatus());
+            assertEquals(MessageViewMapper.RECALLED_PLACEHOLDER, persisted.getContent());
+            assertEquals(1L, persisted.getVersion());
+        }
+        assertEquals(2, auditLogRepository.findAll().stream()
+                .filter(a -> "MESSAGE_RECALLED".equals(a.getAction()) && messages.contains(a.getEntityId())).count());
+    }
+
+    private Integer findBusinessCode(Throwable failure) {
+        Throwable current = failure;
+        while (current != null) {
+            if (current instanceof BusinessException business) return business.getErrorCodeInt();
+            current = current.getCause();
+        }
+        return null;
+    }
+
+    private record Attempt(MessageResponse response, Throwable failure) { }
+}
--- a/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessageRecallControllerTest.java
+++ b/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessageRecallControllerTest.java
@ -0,0 +1,91 @@
+package com.mydreamjob.messaging;
+
+import com.mydreamjob.common.exception.BusinessException;
+import com.mydreamjob.common.exception.ErrorCode;
+import com.mydreamjob.messaging.dto.MessageResponse;
+import com.mydreamjob.auth.JwtTokenProvider;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.BeforeEach;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.core.ValueOperations;
+import org.springframework.http.MediaType;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.web.servlet.MockMvc;
+
+import java.time.LocalDateTime;
+import java.util.Map;
+
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.*;
+
+@SpringBootTest
+@AutoConfigureMockMvc
+@ActiveProfiles("integration-test")
+class MessageRecallControllerTest {
+
+    @Autowired private MockMvc mockMvc;
+    @Autowired private JwtTokenProvider jwtTokenProvider;
+    @MockBean private MessagingService messagingService;
+    @MockBean private RedisTemplate<String, Object> redisTemplate;
+    @Autowired private ObjectMapper objectMapper;
+
+    @BeforeEach
+    @SuppressWarnings("unchecked")
+    void setUpRedis() {
+        ValueOperations<String, Object> ops = mock(ValueOperations.class);
+        when(redisTemplate.opsForValue()).thenReturn(ops);
+        when(redisTemplate.hasKey(any())).thenReturn(false);
+    }
+
+    private String token(Long userId) {
+        return "Bearer " + jwtTokenProvider.generateAccessToken(userId, "1", Map.of("roles", "STUDENT"));
+    }
+
+    @Test
+    void recallReturnsWhitelistedMessageResponse() throws Exception {
+        MessageResponse response = MessageResponse.builder()
+                .id(81L).conversationId(12L).senderUserId(7L)
+                .content(MessageViewMapper.RECALLED_PLACEHOLDER).status("RECALLED")
+                .recalledAt(LocalDateTime.of(2026, 9, 10, 10, 2))
+                .createdAt(LocalDateTime.of(2026, 9, 10, 10, 0)).isRead(false).build();
+        when(messagingService.recallMessage(eq(7L), eq(12L), eq(81L), any())).thenReturn(response);
+
+        mockMvc.perform(post("/conversations/12/messages/81/recall")
+                        .header("Authorization", token(7L))
+                        .header("X-Request-Id", "recall-1"))
+                .andExpect(status().isOk())
+                .andExpect(jsonPath("$.code").value(200))
+                .andExpect(jsonPath("$.data.content").value(MessageViewMapper.RECALLED_PLACEHOLDER))
+                .andExpect(jsonPath("$.data.status").value("RECALLED"))
+                .andExpect(jsonPath("$.data.senderUserId").value(7));
+        verify(messagingService).recallMessage(eq(7L), eq(12L), eq(81L), any());
+    }
+
+    @Test
+    void recallWithoutTokenReturns401() throws Exception {
+        mockMvc.perform(post("/conversations/12/messages/81/recall"))
+                .andExpect(status().isUnauthorized());
+        verifyNoInteractions(messagingService);
+    }
+
+    @Test
+    void recallWindowErrorMapsTo409WithoutInternalContent() throws Exception {
+        when(messagingService.recallMessage(eq(7L), eq(12L), eq(81L), any()))
+                .thenThrow(new BusinessException(ErrorCode.MESSAGE_RECALL_WINDOW_EXPIRED));
+
+        String body = mockMvc.perform(post("/conversations/12/messages/81/recall")
+                        .header("Authorization", token(7L)))
+                .andExpect(status().isConflict())
+                .andExpect(jsonPath("$.code").value(ErrorCode.MESSAGE_RECALL_WINDOW_EXPIRED.getCode()))
+                .andReturn().getResponse().getContentAsString();
+        org.junit.jupiter.api.Assertions.assertFalse(body.contains("original"));
+        org.junit.jupiter.api.Assertions.assertFalse(body.contains("password"));
+    }
+}
--- a/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessageRecallHttpIntegrationTest.java
+++ b/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessageRecallHttpIntegrationTest.java
@ -0,0 +1,272 @@
+package com.mydreamjob.messaging;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.mydreamjob.audit.AuditLogRepository;
+import com.mydreamjob.auth.JwtTokenProvider;
+import com.mydreamjob.auth.User;
+import com.mydreamjob.auth.UserRepository;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.http.MediaType;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.web.servlet.MockMvc;
+
+import java.time.Clock;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.when;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.*;
+
+/** 真实 Service、Repository 和 Security 链路的消息撤回 HTTP 脱敏验证。 */
+@SpringBootTest
+@AutoConfigureMockMvc
+@ActiveProfiles("integration-test")
+class MessageRecallHttpIntegrationTest {
+
+    @Autowired private MockMvc mockMvc;
+    @Autowired private JwtTokenProvider jwtTokenProvider;
+    @Autowired private UserRepository userRepository;
+    @Autowired private ConversationRepository conversationRepository;
+    @Autowired private ConversationMemberRepository memberRepository;
+    @Autowired private MessageRepository messageRepository;
+    @Autowired private AuditLogRepository auditLogRepository;
+    @Autowired private JdbcTemplate jdbcTemplate;
+    @Autowired private ObjectMapper objectMapper;
+    @MockBean private RedisTemplate<String, Object> redisTemplate;
+    @MockBean private Clock clock;
+
+    private static final ZoneId BUSINESS_ZONE = ZoneId.of("Asia/Shanghai");
+    private static final LocalDateTime CAPTURED_NOW = LocalDateTime.of(2026, 9, 10, 10, 1);
+    private static final LocalDateTime FIXTURE_CREATED_AT = LocalDateTime.of(2026, 9, 10, 10, 0);
+
+    private static final AtomicLong PHONE = new AtomicLong(7400);
+    private final List<Long> messageIds = new ArrayList<>();
+    private final List<Long> conversationIds = new ArrayList<>();
+    private final List<Long> userIds = new ArrayList<>();
+
+    @org.junit.jupiter.api.BeforeEach
+    void fixApplicationClock() {
+        when(clock.instant()).thenReturn(CAPTURED_NOW.atZone(BUSINESS_ZONE).toInstant());
+        when(clock.getZone()).thenReturn(BUSINESS_ZONE);
+    }
+
+    @AfterEach
+    void cleanup() {
+        if (!messageIds.isEmpty()) {
+            String marks = String.join(",", messageIds.stream().map(id -> "?").toList());
+            jdbcTemplate.update("DELETE FROM audit_logs WHERE action='MESSAGE_RECALLED' AND entity_id IN (" + marks + ")",
+                    messageIds.toArray());
+            jdbcTemplate.update("DELETE FROM message_reads WHERE message_id IN (" + marks + ")", messageIds.toArray());
+            jdbcTemplate.update("DELETE FROM messages WHERE id IN (" + marks + ")", messageIds.toArray());
+        }
+        if (!conversationIds.isEmpty()) {
+            String marks = String.join(",", conversationIds.stream().map(id -> "?").toList());
+            jdbcTemplate.update("DELETE FROM conversation_members WHERE conversation_id IN (" + marks + ")",
+                    conversationIds.toArray());
+            jdbcTemplate.update("DELETE FROM conversations WHERE id IN (" + marks + ")", conversationIds.toArray());
+        }
+        if (!userIds.isEmpty()) {
+            String marks = String.join(",", userIds.stream().map(id -> "?").toList());
+            jdbcTemplate.update("DELETE FROM user_roles WHERE user_id IN (" + marks + ")", userIds.toArray());
+            jdbcTemplate.update("DELETE FROM users WHERE id IN (" + marks + ")", userIds.toArray());
+        }
+        messageIds.clear();
+        conversationIds.clear();
+        userIds.clear();
+    }
+
+    private Long user(String label, User.Role role) {
+        User saved = userRepository.saveAndFlush(User.builder()
+                .phone("139" + String.format("%08d", PHONE.incrementAndGet()))
+                .nickname(label).role(role).roles(Set.of(role))
+                .active(true).verified(true).build());
+        userIds.add(saved.getId());
+        return saved.getId();
+    }
+
+    private Long conversation(Long sender, Long receiver, String type) {
+        Long id = conversationRepository.saveAndFlush(Conversation.builder()
+                .type(type).createdBy(sender).title("N07 HTTP test").build()).getId();
+        conversationIds.add(id);
+        memberRepository.save(ConversationMember.builder().conversationId(id).userId(sender).build());
+        memberRepository.save(ConversationMember.builder().conversationId(id).userId(receiver).build());
+        return id;
+    }
+
+    private Long message(Long conversation, Long sender, String body) {
+        LocalDateTime created = FIXTURE_CREATED_AT;
+        Message saved = messageRepository.saveAndFlush(Message.builder()
+                .conversationId(conversation).senderUserId(sender).content(body)
+                .status("SENT").version(0L).createdAt(created).build());
+        jdbcTemplate.update("UPDATE messages SET created_at=? WHERE id=?", created, saved.getId());
+        messageIds.add(saved.getId());
+        return saved.getId();
+    }
+
+    private String token(Long userId, User.Role role) {
+        return "Bearer " + jwtTokenProvider.generateAccessToken(userId, "1",
+                java.util.Map.of("roles", role.name()));
+    }
+
+    @Test
+    void directRecallAndListUsePlaceholderAndKeepExistingSenderId() throws Exception {
+        Long sender = user("direct-sender", User.Role.STUDENT);
+        Long receiver = user("direct-receiver", User.Role.STUDENT);
+        Long conversation = conversation(sender, receiver, "DIRECT");
+        Long message = message(conversation, sender, "N07_PRIVATE_ORIGINAL_BODY");
+
+        String recalled = mockMvc.perform(post("/conversations/{conversationId}/messages/{messageId}/recall",
+                        conversation, message)
+                        .header("Authorization", token(sender, User.Role.STUDENT))
+                        .header("X-Request-Id", "n07-http-direct"))
+                .andExpect(status().isOk())
+                .andExpect(jsonPath("$.data.senderUserId").value(sender))
+                .andExpect(jsonPath("$.data.content").value(MessageViewMapper.RECALLED_PLACEHOLDER))
+                .andExpect(jsonPath("$.data.status").value("RECALLED"))
+                .andExpect(jsonPath("$.data.recalledAt").exists())
+                .andReturn().getResponse().getContentAsString();
+        assertSafe(recalled);
+
+        String listed = mockMvc.perform(get("/conversations/{conversationId}/messages", conversation)
+                        .header("Authorization", token(receiver, User.Role.STUDENT)))
+                .andExpect(status().isOk())
+                .andExpect(jsonPath("$.data.items[0].content").value(MessageViewMapper.RECALLED_PLACEHOLDER))
+                .andExpect(jsonPath("$.data.items[0].status").value("RECALLED"))
+                .andReturn().getResponse().getContentAsString();
+        assertSafe(listed);
+    }
+
+    @Test
+    void talentInvitationRecallUsesRelativeIdentityAcrossResponses() throws Exception {
+        Long sender = user("talent-sender", User.Role.COMPANY);
+        Long receiver = user("talent-receiver", User.Role.STUDENT);
+        Long conversation = conversation(sender, receiver, "TALENT_INVITATION");
+        Long message = message(conversation, sender, "N07_PRIVATE_ORIGINAL_BODY");
+
+        String recalled = mockMvc.perform(post("/conversations/{conversationId}/messages/{messageId}/recall",
+                        conversation, message)
+                        .header("Authorization", token(sender, User.Role.COMPANY)))
+                .andExpect(status().isOk())
+                .andExpect(jsonPath("$.data.senderScope").value("CURRENT_USER"))
+                .andExpect(jsonPath("$.data.senderUserId").doesNotExist())
+                .andExpect(jsonPath("$.data.content").value(MessageViewMapper.RECALLED_PLACEHOLDER))
+                .andReturn().getResponse().getContentAsString();
+        assertSafe(recalled);
+
+        String listed = mockMvc.perform(get("/conversations/{conversationId}/messages", conversation)
+                        .header("Authorization", token(receiver, User.Role.STUDENT)))
+                .andExpect(status().isOk())
+                .andExpect(jsonPath("$.data.items[0].senderScope").value("OTHER_PARTICIPANT"))
+                .andExpect(jsonPath("$.data.items[0].senderUserId").doesNotExist())
+                .andReturn().getResponse().getContentAsString();
+        assertSafe(listed);
+
+        String conversationView = mockMvc.perform(get("/conversations/{conversationId}", conversation)
+                        .header("Authorization", token(receiver, User.Role.STUDENT)))
+                .andExpect(status().isOk())
+                .andExpect(jsonPath("$.data.createdBy").doesNotExist())
+                .andReturn().getResponse().getContentAsString();
+        assertSafe(conversationView);
+    }
+
+    @Test
+    void realHttpPermissionAndErrorContractIsPreserved() throws Exception {
+        Long sender = user("permission-sender", User.Role.STUDENT);
+        Long receiver = user("permission-receiver", User.Role.STUDENT);
+        Long admin = user("permission-admin", User.Role.ADMIN);
+        Long conversation = conversation(sender, receiver, "DIRECT");
+        memberRepository.saveAndFlush(ConversationMember.builder().conversationId(conversation).userId(admin).build());
+        Long message = message(conversation, sender, "N07_PRIVATE_ORIGINAL_BODY");
+
+        mockMvc.perform(post("/conversations/{conversationId}/messages/{messageId}/recall", conversation, message)
+                        .header("Authorization", token(receiver, User.Role.STUDENT)))
+                .andExpect(status().isForbidden()).andExpect(jsonPath("$.code").value(2002));
+        mockMvc.perform(post("/conversations/{conversationId}/messages/{messageId}/recall", conversation, message)
+                        .header("Authorization", token(admin, User.Role.ADMIN)))
+                .andExpect(status().isForbidden()).andExpect(jsonPath("$.code").value(2002));
+        mockMvc.perform(post("/conversations/{conversationId}/messages/{messageId}/recall", conversation, message))
+                .andExpect(status().isUnauthorized());
+        mockMvc.perform(post("/conversations/{conversationId}/messages/{messageId}/recall", conversation, 999999L)
+                        .header("Authorization", token(sender, User.Role.STUDENT)))
+                .andExpect(status().isNotFound()).andExpect(jsonPath("$.code").value(8306));
+    }
+
+    @Test
+    void realHttpCoversNonMemberCrossConversationMissingConversationAndExpiredWindow() throws Exception {
+        Long sender = user("matrix-sender", User.Role.STUDENT);
+        Long receiver = user("matrix-receiver", User.Role.STUDENT);
+        Long nonMember = user("matrix-non-member", User.Role.STUDENT);
+        Long primaryConversation = conversation(sender, receiver, "DIRECT");
+        Long otherConversation = conversation(sender, receiver, "DIRECT");
+        Long message = message(primaryConversation, sender, "N07_PRIVATE_ORIGINAL_BODY");
+
+        String nonMemberBody = mockMvc.perform(post(
+                        "/conversations/{conversationId}/messages/{messageId}/recall",
+                        primaryConversation, message)
+                        .header("Authorization", token(nonMember, User.Role.STUDENT)))
+                .andExpect(status().isForbidden())
+                .andExpect(jsonPath("$.code").value(2002))
+                .andReturn().getResponse().getContentAsString();
+        assertSafe(nonMemberBody);
+
+        String crossConversationBody = mockMvc.perform(post(
+                        "/conversations/{conversationId}/messages/{messageId}/recall",
+                        otherConversation, message)
+                        .header("Authorization", token(sender, User.Role.STUDENT)))
+                .andExpect(status().isForbidden())
+                .andExpect(jsonPath("$.code").value(2002))
+                .andReturn().getResponse().getContentAsString();
+        assertSafe(crossConversationBody);
+
+        String missingConversationBody = mockMvc.perform(post(
+                        "/conversations/{conversationId}/messages/{messageId}/recall",
+                        999999L, message)
+                        .header("Authorization", token(sender, User.Role.STUDENT)))
+                .andExpect(status().isNotFound())
+                .andExpect(jsonPath("$.code").value(8305))
+                .andReturn().getResponse().getContentAsString();
+        assertSafe(missingConversationBody);
+
+        LocalDateTime expiredNow = FIXTURE_CREATED_AT.plusMinutes(2).plusNanos(1);
+        when(clock.instant()).thenReturn(expiredNow.atZone(BUSINESS_ZONE).toInstant());
+        String expiredBody = mockMvc.perform(post(
+                        "/conversations/{conversationId}/messages/{messageId}/recall",
+                        primaryConversation, message)
+                        .header("Authorization", token(sender, User.Role.STUDENT)))
+                .andExpect(status().isConflict())
+                .andExpect(jsonPath("$.code").value(8307))
+                .andReturn().getResponse().getContentAsString();
+        assertSafe(expiredBody);
+
+        Message unchanged = messageRepository.findById(message).orElseThrow();
+        assertEquals("SENT", unchanged.getStatus());
+        assertEquals("N07_PRIVATE_ORIGINAL_BODY", unchanged.getContent());
+        assertEquals(0L, unchanged.getVersion());
+        assertEquals(0, auditLogRepository.findAll().stream()
+                .filter(a -> "MESSAGE_RECALLED".equals(a.getAction()) && message.equals(a.getEntityId()))
+                .count());
+    }
+
+    private void assertSafe(String json) {
+        assertFalse(json.contains("N07_PRIVATE_ORIGINAL_BODY"));
+        for (String forbidden : List.of("originalContent", "previousContent", "studentUserId", "candidateId",
+                "phone", "email", "resumeUrl", "candidateKey", "candidateKeyDigest", "ipAddress", "userAgent",
+                "password", "token", "secret")) {
+            assertFalse(json.contains(forbidden), "response contains forbidden field: " + forbidden);
+        }
+    }
+}
--- a/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessageRecallPersistenceIntegrationTest.java
+++ b/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessageRecallPersistenceIntegrationTest.java
@ -0,0 +1,355 @@
+package com.mydreamjob.messaging;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.mydreamjob.audit.AuditContext;
+import com.mydreamjob.audit.AuditLog;
+import com.mydreamjob.audit.AuditLogRepository;
+import com.mydreamjob.auth.User;
+import com.mydreamjob.auth.UserRepository;
+import com.mydreamjob.common.exception.BusinessException;
+import com.mydreamjob.common.exception.ErrorCode;
+import com.mydreamjob.messaging.dto.MessageResponse;
+import com.mydreamjob.safety.BlockRelationRepository;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import jakarta.persistence.EntityManager;
+import jakarta.persistence.EntityManagerFactory;
+import jakarta.persistence.OptimisticLockException;
+import org.springframework.dao.DataAccessException;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.domain.Page;
+import org.springframework.test.context.ActiveProfiles;
+
+import java.time.LocalDateTime;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+@SpringBootTest
+@ActiveProfiles("integration-test")
+class MessageRecallPersistenceIntegrationTest {
+
+    @Autowired private MessagingService messagingService;
+    @Autowired private UserRepository userRepository;
+    @Autowired private ConversationRepository conversationRepository;
+    @Autowired private ConversationMemberRepository memberRepository;
+    @Autowired private MessageRepository messageRepository;
+    @Autowired private MessageReadRepository readRepository;
+    @Autowired private AuditLogRepository auditLogRepository;
+    @Autowired private BlockRelationRepository blockRelationRepository;
+    @Autowired private JdbcTemplate jdbcTemplate;
+    @Autowired private EntityManager entityManager;
+    @Autowired private EntityManagerFactory entityManagerFactory;
+    @Autowired private ObjectMapper objectMapper;
+    @MockBean private RedisTemplate<String, Object> redisTemplate;
+
+    private Long sender;
+    private Long receiver;
+    private Long conversation;
+    private Long message;
+    private Long candidateScoreStudent;
+
+    @AfterEach
+    void cleanup() {
+        if (candidateScoreStudent != null) {
+            jdbcTemplate.update("DELETE FROM candidate_dimension_scores WHERE student_user_id=?", candidateScoreStudent);
+            candidateScoreStudent = null;
+        }
+        if (message != null) readRepository.deleteAll();
+        messageRepository.deleteAll();
+        memberRepository.deleteAll();
+        blockRelationRepository.deleteAll();
+        conversationRepository.deleteAll();
+        auditLogRepository.deleteAll();
+        if (sender != null || receiver != null) {
+            if (sender != null) {
+                jdbcTemplate.update("DELETE FROM scoring_risk_actions WHERE operator_user_id=?", sender);
+                jdbcTemplate.update("DELETE FROM scoring_risk_events WHERE subject_user_id=?", sender);
+                jdbcTemplate.update("DELETE FROM notifications WHERE user_id=?", sender);
+            }
+            if (receiver != null) {
+                jdbcTemplate.update("DELETE FROM scoring_risk_actions WHERE operator_user_id=?", receiver);
+                jdbcTemplate.update("DELETE FROM scoring_risk_events WHERE subject_user_id=?", receiver);
+                jdbcTemplate.update("DELETE FROM notifications WHERE user_id=?", receiver);
+            }
+        }
+        userRepository.deleteAll();
+    }
+
+    private void fixture(LocalDateTime createdAt) {
+        sender = userRepository.save(user("sender")).getId();
+        receiver = userRepository.save(user("receiver")).getId();
+        conversation = conversationRepository.save(Conversation.builder().type("DIRECT").createdBy(sender).build()).getId();
+        memberRepository.save(ConversationMember.builder().conversationId(conversation).userId(sender).build());
+        memberRepository.save(ConversationMember.builder().conversationId(conversation).userId(receiver).build());
+        Message saved = messageRepository.saveAndFlush(Message.builder().conversationId(conversation).senderUserId(sender)
+                .content("secret-body").status("SENT").version(0L).createdAt(createdAt).build());
+        message = saved.getId();
+        // Auditing may refresh @CreatedDate on update; force the deterministic boundary in the fixture only.
+        jdbcTemplate.update("UPDATE messages SET created_at=? WHERE id=?", createdAt, message);
+        entityManager.clear();
+    }
+
+    private User user(String name) {
+        return User.builder().phone("139" + Math.abs(System.nanoTime() % 1_000_000_000L))
+                .nickname(name).role(User.Role.STUDENT).roles(Set.of(User.Role.STUDENT))
+                .active(true).verified(true).build();
+    }
+
+    @Test
+    void migrationFieldsAndConditionalUpdatePersistRecall() {
+        LocalDateTime created = LocalDateTime.of(2026, 9, 10, 10, 0);
+        LocalDateTime recalledAt = created.plusMinutes(2);
+        fixture(created);
+
+        MessageResponse response = messagingService.recallMessage(sender, conversation, message, null, recalledAt);
+        Message persisted = messageRepository.findById(message).orElseThrow();
+
+        assertEquals("RECALLED", persisted.getStatus());
+        assertEquals(MessageViewMapper.RECALLED_PLACEHOLDER, persisted.getContent());
+        assertEquals(recalledAt, persisted.getRecalledAt());
+        assertEquals(1L, persisted.getVersion());
+        assertEquals(MessageViewMapper.RECALLED_PLACEHOLDER, response.getContent());
+        assertEquals(1, auditLogRepository.findAll().stream()
+                .filter(a -> "MESSAGE_RECALLED".equals(a.getAction()) && message.equals(a.getEntityId())).count());
+        assertFalse(auditLogRepository.findAll().stream()
+                .filter(a -> "MESSAGE_RECALLED".equals(a.getAction()))
+                .anyMatch(a -> a.getDetails() != null && a.getDetails().contains("secret-body")));
+    }
+
+    @Test
+    void messageRecallAuditStoresRequestIdButNotIpOrUserAgent() throws Exception {
+        LocalDateTime created = LocalDateTime.of(2026, 9, 10, 10, 0);
+        fixture(created);
+        AuditContext context = AuditContext.builder()
+                .requestId("recall-audit-1")
+                .ipAddress("127.0.0.66")
+                .userAgent("N07-secret-agent")
+                .build();
+
+        messagingService.recallMessage(sender, conversation, message, context, created.plusMinutes(1));
+
+        AuditLog audit = auditLogRepository.findAll().stream()
+                .filter(a -> "MESSAGE_RECALLED".equals(a.getAction()) && message.equals(a.getEntityId()))
+                .findFirst().orElseThrow();
+        assertEquals("recall-audit-1", audit.getRequestId());
+        assertNull(audit.getIpAddress());
+        assertNull(audit.getUserAgent());
+        JsonNode details = objectMapper.readTree(audit.getDetails());
+        java.util.Set<String> fields = new java.util.HashSet<>();
+        details.fieldNames().forEachRemaining(fields::add);
+        assertEquals(java.util.Set.of("messageId", "conversationId", "conversationType", "status", "requestId"), fields);
+        assertEquals(message.longValue(), details.get("messageId").longValue());
+        assertEquals(conversation.longValue(), details.get("conversationId").longValue());
+        assertEquals("DIRECT", details.get("conversationType").textValue());
+        assertEquals("RECALLED", details.get("status").textValue());
+        assertEquals("recall-audit-1", details.get("requestId").textValue());
+        assertFalse(audit.getDetails().contains("secret-body"));
+        assertFalse(audit.getDetails().contains("127.0.0.66"));
+        assertFalse(audit.getDetails().contains("N07-secret-agent"));
+    }
+
+    @Test
+    void v41DefaultsAndStatusCheckAreEnforced() {
+        fixture(LocalDateTime.of(2026, 9, 10, 10, 0));
+        jdbcTemplate.update(
+                "INSERT INTO messages(conversation_id, sender_user_id, content, created_at) VALUES (?, ?, ?, ?)",
+                conversation, sender, "legacy-body", LocalDateTime.of(2026, 9, 10, 10, 0));
+        Long legacyMessage = jdbcTemplate.queryForObject("SELECT MAX(id) FROM messages WHERE conversation_id=?",
+                Long.class, conversation);
+        java.util.Map<String, Object> row = jdbcTemplate.queryForMap(
+                "SELECT status, recalled_at, version FROM messages WHERE id=?", legacyMessage);
+        assertEquals("SENT", row.get("STATUS"));
+        assertNull(row.get("RECALLED_AT"));
+        assertEquals(0L, ((Number) row.get("VERSION")).longValue());
+
+        Long finalLegacyMessage = legacyMessage;
+        assertThrows(DataAccessException.class, () -> jdbcTemplate.update(
+                "UPDATE messages SET status='DELETED' WHERE id=?", finalLegacyMessage));
+    }
+
+    @Test
+    void optimisticVersionConflictCannotSilentlyOverwrite() {
+        fixture(LocalDateTime.of(2026, 9, 10, 10, 0));
+        EntityManager first = entityManagerFactory.createEntityManager();
+        EntityManager second = entityManagerFactory.createEntityManager();
+        jakarta.persistence.EntityTransaction tx1 = first.getTransaction();
+        jakarta.persistence.EntityTransaction tx2 = second.getTransaction();
+        try {
+            Message firstMessage = first.find(Message.class, message);
+            Message secondMessage = second.find(Message.class, message);
+            tx1.begin();
+            firstMessage.setContent("first-update");
+            tx1.commit();
+
+            tx2.begin();
+            secondMessage.setContent("stale-update");
+            RuntimeException conflict = assertThrows(RuntimeException.class, tx2::commit);
+            assertTrue(hasCause(conflict, OptimisticLockException.class),
+                    "stale update must be reported as an optimistic lock conflict");
+            if (tx2.isActive()) tx2.rollback();
+            assertEquals("first-update", jdbcTemplate.queryForObject(
+                    "SELECT content FROM messages WHERE id=?", String.class, message));
+        } finally {
+            if (tx1.isActive()) tx1.rollback();
+            if (tx2.isActive()) tx2.rollback();
+            first.close();
+            second.close();
+        }
+    }
+
+    @Test
+    @Transactional
+    void conditionalRecallRejectsWrongOwnerConversationVersionAndStatus() {
+        fixture(LocalDateTime.of(2026, 9, 10, 10, 0));
+        LocalDateTime recalledAt = LocalDateTime.of(2026, 9, 10, 10, 1);
+        assertEquals(0, messageRepository.recallIfCurrent(message, conversation, 999L, 0L,
+                MessageViewMapper.RECALLED_PLACEHOLDER, recalledAt));
+        assertEquals(0, messageRepository.recallIfCurrent(message, 999L, sender, 0L,
+                MessageViewMapper.RECALLED_PLACEHOLDER, recalledAt));
+        assertEquals(0, messageRepository.recallIfCurrent(message, conversation, sender, 1L,
+                MessageViewMapper.RECALLED_PLACEHOLDER, recalledAt));
+        assertEquals(1, messageRepository.recallIfCurrent(message, conversation, sender, 0L,
+                MessageViewMapper.RECALLED_PLACEHOLDER, recalledAt));
+        assertEquals(0, messageRepository.recallIfCurrent(message, conversation, sender, 0L,
+                MessageViewMapper.RECALLED_PLACEHOLDER, recalledAt));
+    }
+
+    private boolean hasCause(Throwable error, Class<? extends Throwable> type) {
+        Throwable current = error;
+        while (current != null) {
+            if (type.isInstance(current)) return true;
+            current = current.getCause();
+        }
+        return false;
+    }
+
+    @Test
+    void existingMessageReadRowsRemainAfterRecall() {
+        LocalDateTime created = LocalDateTime.of(2026, 9, 10, 10, 0);
+        fixture(created);
+        readRepository.saveAndFlush(MessageRead.builder().messageId(message).readerUserId(receiver).build());
+
+        messagingService.recallMessage(sender, conversation, message, null, created.plusMinutes(1));
+
+        assertEquals(1, readRepository.findByMessageIdInAndReaderUserId(java.util.List.of(message), receiver).size());
+        messagingService.markRead(conversation, receiver);
+        assertEquals(1, readRepository.findByMessageIdInAndReaderUserId(java.util.List.of(message), receiver).size());
+        assertEquals(1, messagingService.listMessages(conversation, receiver, 0, 20)
+                .getItems().stream().filter(item -> Boolean.TRUE.equals(item.getIsRead())).count());
+    }
+
+    @Test
+    void expiredRecallDoesNotMutateDatabase() {
+        LocalDateTime created = LocalDateTime.of(2026, 9, 10, 10, 0);
+        fixture(created);
+        assertEquals(created, messageRepository.findById(message).orElseThrow().getCreatedAt());
+        BusinessException ex = assertThrows(BusinessException.class,
+                () -> messagingService.recallMessage(sender, conversation, message, null, created.plusMinutes(2).plusNanos(1)));
+        assertEquals(ErrorCode.MESSAGE_RECALL_WINDOW_EXPIRED.getCode(), ex.getErrorCodeInt());
+        Message persisted = messageRepository.findById(message).orElseThrow();
+        assertEquals("SENT", persisted.getStatus());
+        assertEquals("secret-body", persisted.getContent());
+        assertEquals(0L, persisted.getVersion());
+        assertEquals(0, auditLogRepository.findAll().stream()
+                .filter(a -> "MESSAGE_RECALLED".equals(a.getAction()) && message.equals(a.getEntityId())).count());
+    }
+
+    @Test
+    void recalledMessagesAreNotUnreadOrMarkedRead() {
+        LocalDateTime created = LocalDateTime.of(2026, 9, 10, 10, 0);
+        fixture(created);
+        messagingService.recallMessage(sender, conversation, message, null, created.plusMinutes(1));
+        var messages = messagingService.listMessages(conversation, receiver, 0, 20);
+        assertEquals(1, messages.getTotal());
+        assertEquals(MessageViewMapper.RECALLED_PLACEHOLDER, messages.getItems().get(0).getContent());
+        assertEquals("RECALLED", messages.getItems().get(0).getStatus());
+        messagingService.markRead(conversation, receiver);
+        assertEquals(0, readRepository.findByMessageIdInAndReaderUserId(java.util.List.of(message), receiver).size());
+    }
+
+    @Test
+    void messageRecallDoesNotModifyCandidateDimensionScores() {
+        LocalDateTime created = LocalDateTime.of(2026, 9, 10, 10, 0);
+        fixture(created);
+        candidateScoreStudent = receiver;
+        jdbcTemplate.update("INSERT INTO candidate_dimension_scores "
+                        + "(student_user_id, dimension, period, score, status, weight, rules_version, calculated_at) "
+                        + "VALUES (?, 'ACTIVITY', '2026-09', 73.25, 'OK', 12.50, 'candidate-scoring-v0.2', ?)",
+                candidateScoreStudent, created);
+        java.util.Map<String, Object> before = jdbcTemplate.queryForMap(
+                "SELECT id, student_user_id, dimension, score, status, weight, rules_version, calculated_at, period, version "
+                        + "FROM candidate_dimension_scores WHERE student_user_id=? AND dimension='ACTIVITY'",
+                candidateScoreStudent);
+
+        messagingService.recallMessage(sender, conversation, message, null, created.plusMinutes(1));
+
+        java.util.Map<String, Object> after = jdbcTemplate.queryForMap(
+                "SELECT id, student_user_id, dimension, score, status, weight, rules_version, calculated_at, period, version "
+                        + "FROM candidate_dimension_scores WHERE student_user_id=? AND dimension='ACTIVITY'",
+                candidateScoreStudent);
+        assertEquals(before, after, "message recall must not modify candidate_dimension_scores");
+    }
+
+    @Test
+    void messageRecallDoesNotModifyNotificationsOrRiskGovernanceData() throws Exception {
+        LocalDateTime created = LocalDateTime.of(2026, 9, 10, 10, 0);
+        fixture(created);
+        jdbcTemplate.update("INSERT INTO notifications(user_id, type, title, content, is_read, created_at) "
+                        + "VALUES (?, 'SYSTEM', 'N07 baseline', 'unchanged', 0, ?)",
+                receiver, created);
+        Long ruleId = jdbcTemplate.queryForObject(
+                "SELECT id FROM scoring_rule_configs ORDER BY id LIMIT 1", Long.class);
+        jdbcTemplate.update("INSERT INTO scoring_risk_events(rule_config_id, subject_user_id, source_audit_log_id, "
+                        + "source_action, severity, window_started_at, window_ended_at, occurrence_count, status, "
+                        + "detected_at, version, created_at, updated_at) "
+                        + "VALUES (?, ?, 987654321, 'REPORT_RATE_LIMITED', 'LOW', ?, ?, 1, 'OPEN', ?, 0, ?, ?)",
+                ruleId, sender, created.minusMinutes(1), created, created, created, created);
+        Long riskEventId = jdbcTemplate.queryForObject(
+                "SELECT id FROM scoring_risk_events WHERE subject_user_id=?", Long.class, sender);
+        jdbcTemplate.update("INSERT INTO scoring_risk_actions(risk_event_id, operator_user_id, action, reason_code, created_at) "
+                        + "VALUES (?, ?, 'ACKNOWLEDGE', 'MONITORING', ?)",
+                riskEventId, sender, created);
+
+        java.util.List<java.util.Map<String, Object>> notificationsBefore = stableRows("notifications");
+        java.util.List<java.util.Map<String, Object>> riskEventsBefore = stableRows("scoring_risk_events");
+        java.util.List<java.util.Map<String, Object>> riskActionsBefore = stableRows("scoring_risk_actions");
+        java.util.List<java.util.Map<String, Object>> riskRulesBefore = stableRows("scoring_rule_configs");
+        java.util.List<java.util.Map<String, Object>> checkpointsBefore = stableRows("scoring_risk_scan_checkpoints");
+        long auditsBefore = auditLogRepository.count();
+
+        messagingService.recallMessage(sender, conversation, message,
+                AuditContext.builder().requestId("n07-side-effect-check").build(), created.plusMinutes(1));
+
+        assertEquals(notificationsBefore, stableRows("notifications"));
+        assertEquals(riskEventsBefore, stableRows("scoring_risk_events"));
+        assertEquals(riskActionsBefore, stableRows("scoring_risk_actions"));
+        assertEquals(riskRulesBefore, stableRows("scoring_rule_configs"));
+        assertEquals(checkpointsBefore, stableRows("scoring_risk_scan_checkpoints"));
+        assertEquals(auditsBefore + 1, auditLogRepository.count());
+
+        AuditLog recallAudit = auditLogRepository.findAll().stream()
+                .filter(a -> "MESSAGE_RECALLED".equals(a.getAction()) && message.equals(a.getEntityId()))
+                .findFirst().orElseThrow();
+        JsonNode details = objectMapper.readTree(recallAudit.getDetails());
+        java.util.Set<String> fields = new java.util.HashSet<>();
+        details.fieldNames().forEachRemaining(fields::add);
+        assertEquals(java.util.Set.of("messageId", "conversationId", "conversationType", "status", "requestId"), fields);
+        assertNull(recallAudit.getIpAddress());
+        assertNull(recallAudit.getUserAgent());
+    }
+
+    private java.util.List<java.util.Map<String, Object>> stableRows(String table) {
+        java.util.Set<String> allowed = java.util.Set.of(
+                "notifications", "scoring_risk_events", "scoring_risk_actions",
+                "scoring_rule_configs", "scoring_risk_scan_checkpoints");
+        if (!allowed.contains(table)) throw new IllegalArgumentException("unsupported snapshot table");
+        return jdbcTemplate.queryForList("SELECT * FROM " + table + " ORDER BY id");
+    }
+}
--- a/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessageRecallServiceTest.java
+++ b/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessageRecallServiceTest.java
@ -0,0 +1,221 @@
+package com.mydreamjob.messaging;
+
+import com.mydreamjob.audit.AuditContext;
+import com.mydreamjob.audit.AuditService;
+import com.mydreamjob.auth.UserRepository;
+import com.mydreamjob.auth.User;
+import com.mydreamjob.bidirectionalmatch.ApplicationMatchRecordRepository;
+import com.mydreamjob.common.exception.BusinessException;
+import com.mydreamjob.common.exception.ErrorCode;
+import com.mydreamjob.company.CompanyRepository;
+import com.mydreamjob.messaging.dto.MessageResponse;
+import com.mydreamjob.safety.BlockRelationRepository;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.time.Clock;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.util.List;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+@ExtendWith(MockitoExtension.class)
+class MessageRecallServiceTest {
+
+    @Mock private ConversationRepository conversationRepository;
+    @Mock private ConversationMemberRepository memberRepository;
+    @Mock private MessageRepository messageRepository;
+    @Mock private MessageReadRepository readRepository;
+    @Mock private UserRepository userRepository;
+    @Mock private CompanyRepository companyRepository;
+    @Mock private ApplicationMatchRecordRepository applicationMatchRecordRepository;
+    @Mock private BlockRelationRepository blockRelationRepository;
+    @Mock private com.mydreamjob.audit.SecurityAuditService securityAuditService;
+    @Mock private AuditService auditService;
+    @Mock private Clock clock;
+    @org.mockito.Spy private MessageViewMapper messageViewMapper = new MessageViewMapper();
+    @org.mockito.Spy private com.fasterxml.jackson.databind.ObjectMapper objectMapper =
+            new com.fasterxml.jackson.databind.ObjectMapper();
+    @InjectMocks private MessagingService messagingService;
+
+    private static final Long USER = 7L;
+    private static final Long OTHER = 8L;
+    private static final Long CONVERSATION = 12L;
+    private static final Long MESSAGE = 81L;
+    private static final LocalDateTime CREATED = LocalDateTime.of(2026, 9, 10, 10, 0);
+    private static final LocalDateTime AT_DEADLINE = CREATED.plusMinutes(2);
+    private static final AuditContext CONTEXT = AuditContext.builder().requestId("req-1").build();
+
+    @BeforeEach
+    void setUp() {
+        when(conversationRepository.findById(CONVERSATION))
+                .thenReturn(Optional.of(Conversation.builder().id(CONVERSATION).type("DIRECT").build()));
+        lenient().when(memberRepository.existsByConversationIdAndUserId(CONVERSATION, USER)).thenReturn(true);
+        lenient().when(memberRepository.existsByConversationIdAndUserId(CONVERSATION, OTHER)).thenReturn(true);
+        lenient().when(readRepository.existsByMessageIdAndReaderUserId(MESSAGE, USER)).thenReturn(false);
+    }
+
+    private Message sentMessage() {
+        return Message.builder().id(MESSAGE).conversationId(CONVERSATION).senderUserId(USER)
+                .content("private original").status("SENT").version(0L).createdAt(CREATED).build();
+    }
+
+    @Test
+    void recallWithinWindow_overwritesContentAndAuditsOnce() {
+        Message message = sentMessage();
+        when(messageRepository.findById(MESSAGE)).thenReturn(Optional.of(message));
+        when(messageRepository.recallIfCurrent(eq(MESSAGE), eq(CONVERSATION), eq(USER), eq(0L),
+                eq(MessageViewMapper.RECALLED_PLACEHOLDER), eq(AT_DEADLINE))).thenAnswer(invocation -> {
+            message.setStatus("RECALLED");
+            message.setContent(MessageViewMapper.RECALLED_PLACEHOLDER);
+            message.setRecalledAt(AT_DEADLINE);
+            message.setVersion(1L);
+            return 1;
+        });
+
+        MessageResponse response = messagingService.recallMessage(USER, CONVERSATION, MESSAGE,
+                CONTEXT, AT_DEADLINE);
+
+        assertEquals("RECALLED", response.getStatus());
+        assertEquals(MessageViewMapper.RECALLED_PLACEHOLDER, response.getContent());
+        assertEquals(AT_DEADLINE, response.getRecalledAt());
+        verify(messageRepository).recallIfCurrent(anyLong(), anyLong(), anyLong(), anyLong(), anyString(), eq(AT_DEADLINE));
+        verify(auditService).log(eq(USER), eq("MESSAGE_RECALLED"), eq("MESSAGE"), eq(MESSAGE),
+                contains("\"conversationId\":12"), eq(CONTEXT));
+    }
+
+    @Test
+    void exactTwoMinutesIsStillAllowed() {
+        Message message = sentMessage();
+        when(messageRepository.findById(MESSAGE)).thenReturn(Optional.of(message));
+        when(messageRepository.recallIfCurrent(anyLong(), anyLong(), anyLong(), anyLong(), anyString(), eq(AT_DEADLINE)))
+                .thenReturn(1);
+        assertDoesNotThrow(() -> messagingService.recallMessage(USER, CONVERSATION, MESSAGE, CONTEXT, AT_DEADLINE));
+    }
+
+    @Test
+    void recallAtOneMinuteFiftyNineSecondsSucceeds() {
+        Message message = sentMessage();
+        LocalDateTime atOneMinuteFiftyNineSeconds = CREATED.plusMinutes(1).plusSeconds(59);
+        when(messageRepository.findById(MESSAGE)).thenReturn(Optional.of(message));
+        when(messageRepository.recallIfCurrent(anyLong(), anyLong(), anyLong(), anyLong(), anyString(),
+                eq(atOneMinuteFiftyNineSeconds))).thenAnswer(invocation -> {
+            message.setStatus("RECALLED");
+            message.setContent(MessageViewMapper.RECALLED_PLACEHOLDER);
+            message.setRecalledAt(atOneMinuteFiftyNineSeconds);
+            message.setVersion(1L);
+            return 1;
+        });
+
+        MessageResponse response = messagingService.recallMessage(USER, CONVERSATION, MESSAGE,
+                CONTEXT, atOneMinuteFiftyNineSeconds);
+
+        assertEquals("RECALLED", response.getStatus());
+        verify(messageRepository).recallIfCurrent(anyLong(), anyLong(), anyLong(), anyLong(), anyString(),
+                eq(atOneMinuteFiftyNineSeconds));
+    }
+
+    @Test
+    void afterTwoMinutesReturnsConflictWithoutMutationOrAudit() {
+        when(messageRepository.findById(MESSAGE)).thenReturn(Optional.of(sentMessage()));
+        BusinessException ex = assertThrows(BusinessException.class,
+                () -> messagingService.recallMessage(USER, CONVERSATION, MESSAGE, CONTEXT, AT_DEADLINE.plusNanos(1)));
+        assertEquals(ErrorCode.MESSAGE_RECALL_WINDOW_EXPIRED.getCode(), ex.getErrorCodeInt());
+        verify(messageRepository, never()).recallIfCurrent(anyLong(), anyLong(), anyLong(), anyLong(), anyString(), any());
+        verifyNoInteractions(auditService);
+    }
+
+    @Test
+    void nonSenderAndNonMemberAreForbidden() {
+        when(messageRepository.findById(MESSAGE)).thenReturn(Optional.of(sentMessage()));
+        BusinessException sender = assertThrows(BusinessException.class,
+                () -> messagingService.recallMessage(OTHER, CONVERSATION, MESSAGE, CONTEXT, CREATED));
+        assertEquals(ErrorCode.FORBIDDEN.getCode(), sender.getErrorCodeInt());
+
+        when(memberRepository.existsByConversationIdAndUserId(CONVERSATION, USER)).thenReturn(false);
+        BusinessException member = assertThrows(BusinessException.class,
+                () -> messagingService.recallMessage(USER, CONVERSATION, MESSAGE, CONTEXT, CREATED));
+        assertEquals(ErrorCode.FORBIDDEN.getCode(), member.getErrorCodeInt());
+    }
+
+    @Test
+    void adminCannotRecallEvenWhenMemberAndSenderChecksWouldOtherwisePass() {
+        User admin = User.builder().id(USER).role(User.Role.ADMIN).roles(java.util.Set.of(User.Role.ADMIN)).build();
+        when(userRepository.findById(USER)).thenReturn(Optional.of(admin));
+
+        BusinessException ex = assertThrows(BusinessException.class,
+                () -> messagingService.recallMessage(USER, CONVERSATION, MESSAGE, CONTEXT, CREATED));
+
+        assertEquals(ErrorCode.FORBIDDEN.getCode(), ex.getErrorCodeInt());
+        verify(messageRepository, never()).findById(anyLong());
+        verifyNoInteractions(auditService);
+    }
+
+    @Test
+    void crossConversationProbeIsForbidden() {
+        when(messageRepository.findById(MESSAGE)).thenReturn(Optional.of(
+                Message.builder().id(MESSAGE).conversationId(999L).senderUserId(USER).content("x")
+                        .status("SENT").createdAt(CREATED).version(0L).build()));
+        BusinessException ex = assertThrows(BusinessException.class,
+                () -> messagingService.recallMessage(USER, CONVERSATION, MESSAGE, CONTEXT, CREATED));
+        assertEquals(ErrorCode.FORBIDDEN.getCode(), ex.getErrorCodeInt());
+    }
+
+    @Test
+    void repeatedRecallIsIdempotentAndDoesNotAuditAgain() {
+        Message message = sentMessage();
+        message.setStatus("RECALLED");
+        message.setContent(MessageViewMapper.RECALLED_PLACEHOLDER);
+        message.setRecalledAt(CREATED.plusMinutes(1));
+        when(messageRepository.findById(MESSAGE)).thenReturn(Optional.of(message));
+
+        MessageResponse response = messagingService.recallMessage(USER, CONVERSATION, MESSAGE, CONTEXT, AT_DEADLINE);
+
+        assertEquals("RECALLED", response.getStatus());
+        assertEquals(MessageViewMapper.RECALLED_PLACEHOLDER, response.getContent());
+        verify(messageRepository, never()).recallIfCurrent(anyLong(), anyLong(), anyLong(), anyLong(), anyString(), any());
+        verifyNoInteractions(auditService);
+    }
+
+    @Test
+    void concurrentUpdateReturningRecalledRowIsIdempotent() {
+        Message message = sentMessage();
+        Message latest = sentMessage();
+        latest.setStatus("RECALLED");
+        latest.setContent(MessageViewMapper.RECALLED_PLACEHOLDER);
+        latest.setRecalledAt(AT_DEADLINE);
+        latest.setVersion(1L);
+        when(messageRepository.findById(MESSAGE)).thenReturn(Optional.of(message));
+        when(messageRepository.findCurrentByIdForRecall(MESSAGE)).thenReturn(Optional.of(latest));
+        when(messageRepository.recallIfCurrent(anyLong(), anyLong(), anyLong(), anyLong(), anyString(), any()))
+                .thenReturn(0);
+
+        MessageResponse response = messagingService.recallMessage(USER, CONVERSATION, MESSAGE, CONTEXT, CREATED);
+
+        assertEquals("RECALLED", response.getStatus());
+        verifyNoInteractions(auditService);
+    }
+
+    @Test
+    void talentInvitationResponseUsesRelativeSenderScope() {
+        Conversation invitation = Conversation.builder().id(CONVERSATION).type("TALENT_INVITATION").build();
+        when(conversationRepository.findById(CONVERSATION)).thenReturn(Optional.of(invitation));
+        Message message = sentMessage();
+        message.setStatus("RECALLED");
+        when(messageRepository.findById(MESSAGE)).thenReturn(Optional.of(message));
+
+        MessageResponse response = messagingService.recallMessage(USER, CONVERSATION, MESSAGE, CONTEXT, AT_DEADLINE);
+
+        assertNull(response.getSenderUserId());
+        assertEquals("CURRENT_USER", response.getSenderScope());
+    }
+}
--- a/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessagingBlockIntegrationTest.java
+++ b/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessagingBlockIntegrationTest.java
@ -148,9 +148,9 @@ class MessagingBlockIntegrationTest {
        Assertions.assertTrue(convs.stream().anyMatch(c ->
                ((Number) c.get("id")).longValue() == convId), "被屏蔽后仍应能看到既有会话");

-        Page<Map<String, Object>> msgs = messagingService.listMessages(convId, b, 0, 20);
-        Assertions.assertEquals(1, msgs.getTotalElements());
-        Assertions.assertEquals("历史消息", msgs.getContent().get(0).get("content"));
+        var msgs = messagingService.listMessages(convId, b, 0, 20);
+        Assertions.assertEquals(1, msgs.getTotal());
+        Assertions.assertEquals("历史消息", msgs.getItems().get(0).getContent());

        Map<String, Object> single = messagingService.getConversation(convId, b);
        Assertions.assertEquals(convId, ((Number) single.get("id")).longValue());
@ -165,15 +165,15 @@ class MessagingBlockIntegrationTest {
        Assertions.assertThrows(BusinessException.class, () -> send(a, convId, "屏蔽期间的消息"));

        // 失败消息未落库
-        Page<Map<String, Object>> before = messagingService.listMessages(convId, a, 0, 20);
-        Assertions.assertEquals(0, before.getTotalElements());
+        var before = messagingService.listMessages(convId, a, 0, 20);
+        Assertions.assertEquals(0, before.getTotal());

        // 即使是已存在的会话，解除屏蔽后需手动重新发送；历史不会自动补发
        unblock(a, b);
        send(a, convId, "解除屏蔽后发送");
-        Page<Map<String, Object>> after = messagingService.listMessages(convId, a, 0, 20);
-        Assertions.assertEquals(1, after.getTotalElements());
-        Assertions.assertEquals("解除屏蔽后发送", after.getContent().get(0).get("content"));
+        var after = messagingService.listMessages(convId, a, 0, 20);
+        Assertions.assertEquals(1, after.getTotal());
+        Assertions.assertEquals("解除屏蔽后发送", after.getItems().get(0).getContent());
    }

    // 未屏蔽时不受影响（回归）
@ -215,4 +215,4 @@ class MessagingBlockIntegrationTest {
                .count();
        Assertions.assertTrue(cnt >= 1, "屏蔽建会话后 BLOCKED_SEND_PROHIBITED 审计应落库，实际=" + cnt);
    }
-}
+}
--- a/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessagingControllerTest.java
+++ b/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessagingControllerTest.java
@ -3,6 +3,7 @@ package com.mydreamjob.messaging;
 import com.mydreamjob.common.exception.BusinessException;
 import com.mydreamjob.common.exception.ErrorCode;
 import com.mydreamjob.messaging.dto.CreateConversationRequest;
+import com.mydreamjob.messaging.dto.MessagePageResponse;
 import com.mydreamjob.messaging.dto.SendMessageRequest;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.junit.jupiter.api.BeforeEach;
@ -82,7 +83,7 @@ class MessagingControllerTest {

    @Test void listMessages_shouldReturn200() throws Exception {
        when(messagingService.listMessages(eq(1L), eq(100L), eq(0), eq(20)))
-                .thenReturn(org.springframework.data.domain.Page.empty());
+                .thenReturn(MessagePageResponse.builder().items(List.of()).total(0).page(0).size(20).totalPages(0).build());
        mockMvc.perform(get("/conversations/1/messages").header("Authorization", token()))
                .andExpect(status().isOk()).andExpect(jsonPath("$.code").value(200));
    }
--- a/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessagingServiceTest.java
+++ b/mydreamjob-server/src/test/java/com/mydreamjob/messaging/MessagingServiceTest.java
@ -40,6 +40,9 @@ class MessagingServiceTest {
    @Mock private ApplicationMatchRecordRepository applicationMatchRecordRepository;
    @Mock private BlockRelationRepository blockRelationRepository;
    @Mock private SecurityAuditService securityAuditService;
+    @org.mockito.Spy private MessageViewMapper messageViewMapper = new MessageViewMapper();
+    @org.mockito.Spy private com.fasterxml.jackson.databind.ObjectMapper objectMapper =
+            new com.fasterxml.jackson.databind.ObjectMapper();
    @InjectMocks private MessagingService messagingService;

    private static final Long USER_A = 100L;
@ -255,11 +258,11 @@ class MessagingServiceTest {
        Message msg = Message.builder().id(1L).conversationId(CONV_ID).senderUserId(USER_A).content("Hi").build();
        when(messageRepository.findByConversationIdOrderByCreatedAtDesc(eq(CONV_ID), any()))
                .thenReturn(new PageImpl<>(List.of(msg)));
-        when(readRepository.existsByMessageIdAndReaderUserId(1L, USER_A)).thenReturn(false);
+        when(readRepository.findByMessageIdInAndReaderUserId(List.of(1L), USER_A)).thenReturn(List.of());

        var r = messagingService.listMessages(CONV_ID, USER_A, 0, 20);
-        assertEquals(1, r.getTotalElements());
-        assertEquals(false, r.getContent().get(0).get("isRead"));
+        assertEquals(1, r.getTotal());
+        assertEquals(false, r.getItems().get(0).getIsRead());
    }

    @Test void listMessages_asNonMember_shouldReturn403() {