merge store-layer concurrency and layering fixes

# Conflicts:
#	dc3-db/dc3-db-agentic/pom.xml

AGENTIC-AUDIT-REPORT.md

@@ -0,0 +1,125 @@
+# AI 助手（Agentic）体系体检报告
+
+> 审计日期：2026-10-20 · 方法：4 阶段多智能体审计（8 读图 → 7 维度深审 → 每发现 3 镜头对抗验证 → 完备性批评）
+> 规模：220 个 agent · 4,538 次工具调用 · 约 1,100 万 tokens · 81 分钟
+> 结论口径：**62 条确认**（15 high / 33 medium / 14 low），6 条被对抗验证否决，13 条因 API 限流票数不全（均已获 ≥2 票通过，结论可信但置信略降）
+
+---
+
+## 一、总评（直接回答三个问题）
+
+### 1. 健不健全？——**骨架健全，多处未达生产水位**
+
+分层设计、依赖方向、端口-适配器分工全部符合仓库既有模式且干净。但有 15 个 high 级缺陷集中在**韧性**（幻觉工具名/慢消费者杀死整轮会话）、**安全**（SSRF 探测面、API Key 明文落库、提示注入零防御）、**契约**（OpenAI 兼容承诺与实现背离）三块——这些在单用户开发场景无感，多租户生产场景会集中引爆。
+
+### 2. 全不全面？——**主体全面，但有五块"版图外"区域零审计**
+
+12 工具 + 双运行时 + 会话/附件/动作管理 + 前端全套 + 41 测试的主体覆盖完整。但批评员发现五个**边界组成部分**从未被任何维度触及（见第六节），最典型的是 **MCP 外供通道**（平台把 API 以 MCP 工具暴露给外部 AI 客户端）——这是 AI 助手的"第三条暴露面"，完全零审计。
+
+### 3. 是最佳实践吗？——**局部达标，体系不是**
+
+**已达最佳实践水位的**：租户上下文强制注入（不信任模型参数）、AgenticToolResult 统一信封、写操作 10 分钟人审门、前端 SWR 不闪/XSS 消毒/竞态双令牌/千例对话语料、契约测试纪律。
+
+**低于最佳实践的五个体系级差距**（均有 high 级证据）：
+1. 供应商抽象不完整——换 Anthropic 等非 OpenAI 兼容供应商 = 静默砍掉 95% 工具面，且系统提示词仍向模型许诺全部能力
+2. 上下文管理无 token 预算——按条数截窗、字符/4 估算用量、analytics 结果 5000 点无截断直接注入
+3. Agent 循环终止 fail-hard——超 8 轮直接报错杀整轮，无重复调用检测、无末轮去工具强制收尾
+4. 零指标埋点——LLM 延迟/工具成功率/token/队列深度全部不可度量，prometheus 端点空挂
+5. 多副本语义未设计——进程内缓存无 TTL、失效仅本节点、限流按节点而非全局
+
+---
+
+## 二、High 级缺陷（15 条，对抗验证通过）
+
+### 韧性组（agent 循环）
+| # | 缺陷 | 位置 |
+|---|---|---|
+| H1 | **幻觉工具名 `Mono.error` 杀死整轮会话**——未知工具应返回 tool-error 让模型自愈（轮数上限已有兜底） | `OpenAiCompatibleAgenticRuntime.java:456` |
+| H2 | **慢 SSE 消费者被硬性判死**——`requestedFromDownstream()==0` 即终止生成并落库 FAILED，零缓冲容忍 | `OpenAiCompatibleAgenticRuntime.java:271` |
+| H3 | **双通道工具注册造成 Provider 级能力断层**——Spring AI 路径仅 3 类 @Tool（约 10 方法），60+ reactive 工具整体缺席；桥接代码在 L463-466 已有反向先例可对称复用 | `ChatClientConfig.java:106` |
+
+### 安全组
+| # | 缺陷 | 位置 |
+|---|---|---|
+| H4 | **连通性探测成 SSRF 通道**——草稿探测接受任意 baseUrl；providerId+自定义 baseUrl 组合会把存量 API Key 发往攻击者主机。需 https-only + 私网/环回段拒绝（解析后校验防 DNS rebinding）+ 改 baseUrl 必须同时换 key | `AgenticConnectivityCheckServiceImpl.java:160` |
+| H5 | **LLM API Key 明文落库、无 TTL 常驻堆缓存**——静态加密完全缺失；建议信封加密（AES-GCM+主密钥）、缓存存脱敏副本 | `R2dbcModelProviderStore.java:46` |
+| H6 | **提示注入零防御**——附件元数据无定界符直接拼 system prompt，工具结果回流无消毒。需注入防御段 + 随机化定界符 + 已知注入模式过滤 | `ChatClientConfig.java:53` |
+
+### 契约组
+| # | 缺陷 | 位置 |
+|---|---|---|
+| H7 | **非流式响应输出 camelCase 字段名**（finishReason/promptTokens）——背离自称的 OpenAI 兼容契约；请求的 max_tokens 被静默忽略 | `ChatCompletionResponseVO.java:93` |
+| H8 | **conversationId 运行时必填但 OpenAPI 声明可选**——标准 OpenAI 客户端每次请求必失败；文档与实现互相矛盾 | `AgenticChatRequestPreparer.java:178` |
+
+### 可观测组
+| # | 缺陷 | 位置 |
+|---|---|---|
+| H9 | **agentic 模块零指标**——五组必备指标（信号量水位/轮次 Timer/工具 Counter/token 用量/队列深度）全缺 | `AgenticChatServiceImpl.java:173` |
+| H10 | **trace 贯穿断裂**——requestId 从未写入 MDC（日志关联槽恒空），insight RPC 不携带关联 ID，跨服务排障断链 | `RequestIdWebFilter.java:103` |
+
+### 前端组
+| # | 缺陷 | 位置 |
+|---|---|---|
+| H11 | **Composer 回车在 IME 输入中途提交**——中日韩输入法确认候选词的 Enter 被当发送，中文用户无法正常输入 | `AgenticComposer.vue:80` |
+
+### 最佳实践组
+| # | 缺陷 | 位置 |
+|---|---|---|
+| H12 | 同 H3（供应商维度视角） | — |
+| H13 | **上下文无 token 预算**（同总评差距 2） | `AgenticProperties.java:64` |
+| H14 | **工具 schema 手写内联 JSON 且 7 个 analytics 工具的 series/window 是裸 `{"type":"object"}`**——schema 字面量与代码校验常数平行维护必然漂移；QUERY_LATEST 已证明可填全 | `ReactiveAgenticToolRegistry.java:325` |
+| H15 | 同 H1（最佳实践维度视角） | — |
+
+---
+
+## 三、Medium 级主题（33 条，按主题归并）
+
+- **架构**：1180 行注册表 8 组复制粘贴工厂/接口/解析器（一个泛型工厂可删约 500 行）；telescoping 构造器静默丢工具；`tools()` 每轮全量重建且排序被 `Map.copyOf` 丢弃；db 适配器反向注入 service 层执行文件系统副作用（分层倒置）；MessageController 绕过 service 直连 repository 且无分页；ChatClientFactory 四张缓存无 TTL、fallback 客户端永久粘滞
+- **正确性**：软删会话墓碑永久毒化 conversationId（唯一约束冲突抛裸异常）；session_ext 并发读-合-写丢失更新（无锁无 CAS）；阻塞文件操作跑在 Netty 事件循环；TIMESTAMPTZ 绑定 UTC-naive LocalDateTime（时区漂移）；附件 TOCTOU 孤儿文件
+- **安全**：连通性缓存无淘汰且 key 由客户端材料构成（内存 DoS）；租户缺模型配置时静默回退平台全局 Key 且该路径不限流；权限动词错位（批准高危写操作仅需 `action:list`，上传/聊天复用 list/get 动词）；附件无 MIME 白名单无租户配额
+- **契约**：请求 messages 数组声明为完整上下文但只取最后一条 user；三种分页风格并存；AttachmentVO 泄露服务端存储路径；SessionVO 声明永远返回 null 的字段
+- **可观测**：provider 真实 usage 被 includeUsage 请求后丢弃（全靠估算且系统性偏低）；限流半成品（无超时无队列上限，未知名模型完全绕过）；SSE 无心跳无事件名、失败恒 HTTP 200 收尾
+- **前端**：流式热路径 O(n²)（每 delta 全量 markdown 重解析+DOMPurify）；mock 引擎整段输出不模拟逐 token 流；会话列表硬上限 50 无翻页（更早会话 UI 不可达）；停靠模式下拉与工作台栏数据口径分叉
+- **最佳实践**：数值参数 `10.0`/`1e3` 形态抛 NumberFormatException；工具注册表与 provider 格式定义每请求重建
+
+## 四、Low 级（14 条）
+
+5 个注入未用的工具参数（重构残留）；OpenAI 客户端三处复制（运行时每请求新建无连接池复用）；工具异常 message 原样回灌模型（内部拓扑泄露）；provider 路由嵌套不一致；运维日志缺租户归属；fetch 通道 401 处理与 axios 拦截器漂移；失败回合无重试按钮；等——详见附录数据。
+
+---
+
+## 五、做得好的（必须说）
+
+1. **租户隔离是全审计最强项**：每个工具从 ToolContext 强制提取 tenantId（不信任模型参数）、InsightDispatcher 二次校验、DB 层 WHERE 全带 tenant_id、写点值 10 分钟人审门全程租户校验
+2. **统一信封设计**：AgenticToolResult{success,code,message,data,visualizations} 六态 code 枚举，已注册工具的错误被 tracing 层降级为 tool-result 不炸循环（讽刺的是未知工具名反而炸——H1 正是没对齐这个自己的好模式）
+3. **前端工程质量高于社区平均**：SWR 缓存播种+尾部对齐合并（无重载闪烁）、幽灵占位、reduced-motion 双处生效、多宽度适配、DOMPurify 全量消毒、竞态双令牌、1000+ 用例对话语料
+4. **仓库纪律**：禁 @PathVariable 全模块合规（grep 验证）、RFC 9457 problem+json 统一错误、密钥 VO write-only + LogSanitizer 达标、消息索引 FOR UPDATE 正确串行化、action 审批 CAS 竞态安全
+
+---
+
+## 六、完备性缺口（批评员裁定，二轮审计候选）
+
+1. **MCP 外供通道**（`dc3-common-auth` 的 McpManagementController/ReactiveOAuthMcpRuntimeServiceImpl + `openapi-agentic.json` 冻结快照 + dc3-cli mcp.ts）——第三条暴露面，且快照与 7 个 controller 实际契约的漂移**无任何守护**
+2. **ActiveAgenticProfileConfig** 经 spring.factories 以 HIGHEST_PRECEDENCE 无条件激活 agentic profile 到任何宿主（含 center-single）——未审计
+3. **native/AOT 形态**：手写 JSON schema + ObjectMapper + MethodToolCallbackProvider 反射在 native 管线（真实发布形态）下行为从未验证
+4. **权限种子闭环**：全部 initdb/migrations grep 不到任何 agentic 资源码——@perm.can 引用的资源从何而来、默认谁能批准写动作，未定性（verb 错位是笔误还是有意无法判断）
+5. **数据保留与隐私**：dc3_message（全量对话+工具轨迹）与 dc3_attachment 无保留期策略、无租户清理任务、无合规视角
+6. **多副本部署**：SSE 网关粘性、缓存跨节点失效、信号量按节点——多处仅"注释自认退化"
+7. **Insight TSDB 租户隔离未下钻**（ReactiveTsdbStore 是否强制 tenant 分量）——租户隔离是硬规则，此处必须二轮实证
+8. 文档面（iot-dc3-docs 仓库）、可达性/i18n、mock 契约守护、工具层测试盲区（CommandTool/EventTool 恰是最弱校验处且零专测）
+
+## 七、修复路线建议
+
+- **P0（发布前必须）**：H4 SSRF、H5 密钥加密、H1+H2 循环韧性（都是小改动大收益）、H11 IME（一行守卫）
+- **P1（生产水位）**：H3 双通道桥接、H13 token 预算、H7/H8 契约对齐、H9+H10 指标+trace、权限动词、附件白名单+配额、限流补全
+- **P2（工程债）**：注册表泛型化重构（-500 行）、分页统一、缓存 TTL、前端 O(n²) 批量合并、MessageController 分页
+- **二轮审计**：第六节 8 项（MCP 通道优先）
+
+---
+
+## 附：方法论与可信度说明
+
+- 7 维度审查员各自独立读码（非仅依赖读图摘要），每条 finding 带 file:line+代码证据+可落地建议
+- 每条发现经 3 镜头对抗验证（证据核对/影响评估/主动反驳），≥2/3 票保留；13 条因 API 限流仅 2 票（全部为 2 正票，无反对票）
+- 6 条被否决项含一条高价值仲裁：审计员称"信号量许可取消时永久泄漏为最严重缺陷"，验证员否决，人工复核确认**否决正确**（`Flux.usingWhen` 三参重载在 cancel 路径同样执行 release；仅剩 acquire 阻塞中途被取消的极窄边缘）
+- 完整结构化数据（62 条全文+8 张子系统地图+7 份维度总结+批评报告）：会话任务输出 `wegnwa0xu.output`，journal 见 `subagents/workflows/wf_3344a3b1-c53/journal.jsonl`

dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/repository/ReactiveAttachmentStore.java

@@ -25,12 +25,25 @@ import reactor.core.publisher.Mono;
 /** Reactive persistence port for attachment metadata. */
 public interface ReactiveAttachmentStore {
 
-    /** Save the attachment, inserting or updating as needed. */
+    /**
+     * Save the attachment. The insert runs in one transaction with a live-session guard
+     * ({@code SELECT ... FOR UPDATE} on the owning session), so a save racing a concurrent
+     * session delete either lands before the delete or fails with a clear error instead of
+     * leaving a zombie row for a soft-deleted conversation.
+     */
     Mono<AttachmentBO> save(AttachmentBO attachment);
 
     /** Stream attachments matching the request. */
     Flux<AttachmentBO> list(String conversationId, RequestHeader.PrincipalHeader header);
 
+    /**
+     * Stream the stored file paths of the conversation's soft-deleted attachments. Used by the
+     * session delete orchestration to unlink files after the database delete commits; live
+     * attachments are excluded on purpose so a conversation revived after the delete keeps its
+     * newly uploaded files.
+     */
+    Flux<String> listDeletedFilePaths(String conversationId, RequestHeader.PrincipalHeader header);
+
     /** Load the attachments for the given ids. */
     Flux<AttachmentBO> getByIds(Collection<Long> ids, RequestHeader.PrincipalHeader header);
 

dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/repository/ReactiveSessionStore.java

@@ -26,7 +26,15 @@ import reactor.core.publisher.Mono;
 /** Reactive persistence port for agentic conversation sessions. */
 public interface ReactiveSessionStore {
 
-    /** Create or refresh the session and emit the stored state. */
+    /**
+     * Create or refresh the session and emit the stored state.
+     *
+     * <p>Soft-deleted conversation ids are reusable: when the conversation's row is a soft-delete
+     * tombstone, {@code touch} revives that row as a brand-new session (title and extension reset,
+     * creation restarted) instead of failing on the natural-key uniqueness constraint, which must
+     * stay a full constraint because {@code dc3_message}, {@code dc3_attachment}, and
+     * {@code dc3_action} reference it with foreign keys.
+     */
     Mono<SessionBO> touch(String conversationId, SessionExt sessionExt, RequestHeader.PrincipalHeader header);
 
     /** Load the session for the request. */
@@ -40,14 +48,20 @@ public interface ReactiveSessionStore {
             java.util.List<SortSpec> sort,
             RequestHeader.PrincipalHeader header);
 
-    /** Update one session and emit the updated row. */
+    /**
+     * Update one session and emit the updated row. The read-merge-write sequence runs under a
+     * row lock inside one transaction, so concurrent updates merge instead of overwriting each
+     * other. Completes empty when the session does not exist.
+     */
     Mono<SessionBO> update(
             String conversationId, SessionExt sessionExt, String title, RequestHeader.PrincipalHeader header);
 
     /**
-     * Delete the session. Implementations also remove the session's stored attachment files
-     * best-effort after the database delete commits; file cleanup failures must not fail the
-     * delete.
+     * Soft-delete the session and its messages, attachments, and actions in one transaction,
+     * locking the session row first so concurrent attachment uploads serialize against the
+     * delete. Returns the number of deleted session rows (0 when already deleted); stored
+     * attachment file cleanup is the caller's responsibility, driven by
+     * {@link ReactiveAttachmentStore#listDeletedFilePaths}.
      */
     Mono<Long> delete(String conversationId, RequestHeader.PrincipalHeader header);
 }

dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/service/impl/AttachmentFileCleaner.java

@@ -23,33 +23,49 @@ import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.Collection;
-import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
 
 /**
  * Best-effort filesystem cleanup for stored conversation attachments.
  *
- * <p>Owns the attachment storage-root policy so upload path resolution and delete-time
- * file unlinking share one root and one boundary guard and can never drift apart. All
- * operations are warn-only: a missing, undeletable, or out-of-root file must never fail
- * the database side of an attachment or session delete.
+ * <p>Owns the attachment storage-root policy so upload path resolution and delete-time file
+ * unlinking share one root and one boundary guard and can never drift apart. All operations are
+ * warn-only: a missing, undeletable, or out-of-root file must never fail the database side of an
+ * attachment or session delete. The blocking {@link Files} unlink always runs on the injected
+ * {@link Scheduler} (bounded elastic in production), so no caller can execute file deletion on a
+ * Reactor Netty event loop.
  *
  * @author pnoker
  * @since 2026.9.22
  */
 @Slf4j
 @Component
-@RequiredArgsConstructor
 public class AttachmentFileCleaner {
 
     private final AgenticProperties properties;
+    private final Scheduler unlinkScheduler;
 
     private Path storageRoot;
 
+    /** Spring wiring: unlinks run on the shared bounded-elastic scheduler. */
+    @Autowired
+    public AttachmentFileCleaner(AgenticProperties properties) {
+        this(properties, Schedulers.boundedElastic());
+    }
+
+    /** Test and alternate-wiring path: unlinks run on the given scheduler. */
+    AttachmentFileCleaner(AgenticProperties properties, Scheduler unlinkScheduler) {
+        this.properties = properties;
+        this.unlinkScheduler = unlinkScheduler;
+    }
+
     @PostConstruct
     void initializeStorageRoot() {
         storageRoot = storageRoot();
@@ -63,24 +79,26 @@ public class AttachmentFileCleaner {
     /** Unlink one stored attachment file, resolving it and enforcing the storage-root boundary. */
     public Mono<Void> unlink(String storedFilePath) {
         if (StringUtils.isBlank(storedFilePath)) return Mono.empty();
-        return Mono.fromRunnable(() -> {
-            try {
-                Files.deleteIfExists(resolveWithinRoot(storedFilePath));
-            } catch (Exception exception) {
-                log.warn("Attachment file cleanup failed, file={}", storedFilePath, exception);
-            }
-        });
+        return Mono.<Void>fromRunnable(() -> {
+                    try {
+                        Files.deleteIfExists(resolveWithinRoot(storedFilePath));
+                    } catch (Exception exception) {
+                        log.warn("Attachment file cleanup failed, file={}", storedFilePath, exception);
+                    }
+                })
+                .subscribeOn(unlinkScheduler);
     }
 
     /** Unlink one already-resolved attachment file, used for upload rollback of just-created paths. */
     Mono<Void> unlink(Path path) {
-        return Mono.fromRunnable(() -> {
-            try {
-                Files.deleteIfExists(path);
-            } catch (Exception exception) {
-                log.warn("Attachment file cleanup failed, file={}", path, exception);
-            }
-        });
+        return Mono.<Void>fromRunnable(() -> {
+                    try {
+                        Files.deleteIfExists(path);
+                    } catch (Exception exception) {
+                        log.warn("Attachment file cleanup failed, file={}", path, exception);
+                    }
+                })
+                .subscribeOn(unlinkScheduler);
     }
 
     /** Unlink every stored attachment file in the collection, each best-effort. */

dc3-common/dc3-common-agentic/src/main/java/io/github/pnoker/common/agentic/service/impl/SessionServiceImpl.java

@@ -18,6 +18,7 @@ package io.github.pnoker.common.agentic.service.impl;
 
 import io.github.pnoker.common.agentic.entity.bo.SessionBO;
 import io.github.pnoker.common.agentic.entity.model.SessionExt;
+import io.github.pnoker.common.agentic.repository.ReactiveAttachmentStore;
 import io.github.pnoker.common.agentic.repository.ReactiveSessionStore;
 import io.github.pnoker.common.agentic.service.SessionService;
 import io.github.pnoker.common.entity.common.RequestHeader;
@@ -33,6 +34,8 @@ import reactor.core.publisher.Mono;
 public class SessionServiceImpl implements SessionService {
 
     private final ReactiveSessionStore sessionStore;
+    private final ReactiveAttachmentStore attachmentStore;
+    private final AttachmentFileCleaner attachmentFileCleaner;
 
     @Override
     public Mono<SessionBO> touch(String conversationId, RequestHeader.PrincipalHeader header, SessionExt sessionExt) {
@@ -62,6 +65,23 @@ public class SessionServiceImpl implements SessionService {
 
     @Override
     public Mono<Long> delete(String conversationId, RequestHeader.PrincipalHeader header) {
-        return sessionStore.delete(conversationId, header);
+        return sessionStore
+                .delete(conversationId, header)
+                .flatMap(deleted -> deleted > 0
+                        ? unlinkDeletedAttachmentFiles(conversationId, header).thenReturn(deleted)
+                        : Mono.just(deleted));
+    }
+
+    /**
+     * Unlink the stored files of the conversation's soft-deleted attachments after the session
+     * delete commits. Runs strictly after the transaction so file cleanup can never roll the
+     * delete back, and stays best-effort: a missing or undeletable file never fails the committed
+     * delete.
+     */
+    private Mono<Void> unlinkDeletedAttachmentFiles(String conversationId, RequestHeader.PrincipalHeader header) {
+        return attachmentStore
+                .listDeletedFilePaths(conversationId, header)
+                .collectList()
+                .flatMap(attachmentFileCleaner::unlinkAll);
     }
 }

dc3-common/dc3-common-agentic/src/test/java/io/github/pnoker/common/agentic/service/impl/AttachmentFileCleanerTest.java

@@ -0,0 +1,120 @@
+/*
+ * Copyright 2016-present the IoT DC3 original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package io.github.pnoker.common.agentic.service.impl;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.github.pnoker.common.agentic.config.AgenticProperties;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
+
+/**
+ * Unit coverage of the attachment file cleaner's blocking-IO offload: the NIO unlink must run on
+ * the injected scheduler (bounded elastic in production) so no caller can execute file deletion
+ * on a Reactor Netty event loop, while the storage-root boundary guard and best-effort contract
+ * stay intact.
+ */
+class AttachmentFileCleanerTest {
+
+    @TempDir
+    Path storageRoot;
+
+    private final ConcurrentLinkedQueue<String> unlinkThreads = new ConcurrentLinkedQueue<>();
+    private ExecutorServiceHolder holder;
+
+    private AttachmentFileCleaner cleaner;
+
+    /** Keeps a reference to the probe executor so the test can shut it down. */
+    private record ExecutorServiceHolder(java.util.concurrent.ExecutorService service) {}
+
+    @BeforeEach
+    void setUp() {
+        AgenticProperties properties = new AgenticProperties();
+        properties.setAttachmentStoragePath(storageRoot.toString());
+        java.util.concurrent.ExecutorService probe = Executors.newSingleThreadExecutor(runnable -> {
+            Thread thread = new Thread(runnable, "probe-unlink");
+            thread.setDaemon(true);
+            return thread;
+        });
+        holder = new ExecutorServiceHolder(probe);
+        // The probe executor records the thread that actually performs each task, proving the
+        // cleaner offloads instead of running on the subscribing thread.
+        Executor instrumented = command -> probe.execute(() -> {
+            unlinkThreads.add(Thread.currentThread().getName());
+            command.run();
+        });
+        Scheduler scheduler = Schedulers.fromExecutor(instrumented);
+        cleaner = new AttachmentFileCleaner(properties, scheduler);
+        cleaner.initializeStorageRoot();
+    }
+
+    @AfterEach
+    void tearDown() {
+        holder.service().shutdownNow();
+    }
+
+    @Test
+    void unlinkRunsTheBlockingDeletionOnTheSchedulerThread() throws Exception {
+        Path file = Files.writeString(storageRoot.resolve("note.txt"), "payload");
+        String subscribingThread = Thread.currentThread().getName();
+
+        cleaner.unlink(file.toString()).block();
+
+        assertThat(Files.exists(file)).as("the file must be deleted").isFalse();
+        assertThat(unlinkThreads)
+                .as("the unlink must run on the scheduler thread, not on the subscriber")
+                .singleElement()
+                .isNotEqualTo(subscribingThread)
+                .satisfies(thread -> assertThat(thread).startsWith("probe-unlink"));
+    }
+
+    @Test
+    void unlinkAllRunsEveryDeletionOnTheSchedulerThread() throws Exception {
+        Path first = Files.writeString(storageRoot.resolve("a.txt"), "payload");
+        Path second = Files.writeString(storageRoot.resolve("b.txt"), "payload");
+
+        cleaner.unlinkAll(List.of(first.toString(), second.toString(), " ")).block();
+
+        assertThat(Files.exists(first)).isFalse();
+        assertThat(Files.exists(second)).isFalse();
+        assertThat(unlinkThreads).as("each unlink must be offloaded").hasSize(2);
+    }
+
+    @Test
+    void unlinkSkipsBlankPathsAndRefusesPathsOutsideTheStorageRoot() throws Exception {
+        Path outside = Files.createTempDirectory("dc3-cleaner-outside");
+        Path escaped = Files.writeString(outside.resolve("escaped.txt"), "payload");
+
+        cleaner.unlink(" ").block();
+        assertThat(unlinkThreads).as("blank paths must not schedule any work").isEmpty();
+
+        cleaner.unlink(escaped.toString()).block();
+        assertThat(Files.exists(escaped))
+                .as("the boundary guard must refuse out-of-root unlinks")
+                .isTrue();
+    }
+}

dc3-common/dc3-common-agentic/src/test/java/io/github/pnoker/common/agentic/service/impl/SessionServiceImplTest.java

@@ -16,36 +16,61 @@
  */
 package io.github.pnoker.common.agentic.service.impl;
 
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import io.github.pnoker.common.agentic.config.AgenticProperties;
 import io.github.pnoker.common.agentic.entity.bo.SessionBO;
+import io.github.pnoker.common.agentic.repository.ReactiveAttachmentStore;
 import io.github.pnoker.common.agentic.repository.ReactiveSessionStore;
 import io.github.pnoker.common.entity.common.RequestHeader;
 import io.github.pnoker.db.core.page.OffsetPage;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.util.List;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.io.TempDir;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
+/**
+ * Unit coverage of the session service delete orchestration: the service owns the post-commit
+ * attachment file cleanup (the store stays pure SQL), cleanup failures never fail the committed
+ * delete, and an already-deleted session skips cleanup entirely.
+ */
 @ExtendWith(MockitoExtension.class)
 class SessionServiceImplTest {
 
     @Mock
     private ReactiveSessionStore sessionStore;
 
+    @Mock
+    private ReactiveAttachmentStore attachmentStore;
+
     @Mock
     private RequestHeader.PrincipalHeader header;
 
+    @TempDir
+    Path storageRoot;
+
+    private AttachmentFileCleaner fileCleaner;
+
     private SessionServiceImpl service;
 
     @BeforeEach
     void setUp() {
-        service = new SessionServiceImpl(sessionStore);
+        AgenticProperties properties = new AgenticProperties();
+        properties.setAttachmentStoragePath(storageRoot.toString());
+        fileCleaner = new AttachmentFileCleaner(properties);
+        service = new SessionServiceImpl(sessionStore, attachmentStore, fileCleaner);
     }
 
     @Test
@@ -69,4 +94,41 @@ class SessionServiceImplTest {
                 .expectNext(page)
                 .verifyComplete();
     }
+
+    @Test
+    void deleteUnlinksSoftDeletedAttachmentFilesAfterTheStoreDeleteCommits() throws Exception {
+        Path file = Files.writeString(storageRoot.resolve("tenant_1_user_2_conversation_conv-1-a.txt"), "payload");
+        when(sessionStore.delete("conv", header)).thenReturn(Mono.just(1L));
+        when(attachmentStore.listDeletedFilePaths("conv", header)).thenReturn(Flux.just(file.toString(), " "));
+
+        StepVerifier.create(service.delete("conv", header)).expectNext(1L).verifyComplete();
+
+        assertThat(Files.exists(file))
+                .as("the soft-deleted attachment's file must be unlinked")
+                .isFalse();
+        verify(attachmentStore).listDeletedFilePaths("conv", header);
+    }
+
+    @Test
+    void deleteKeepsTheCommittedResultWhenAFileIsOutsideTheStorageRoot() throws Exception {
+        Path outside = Files.createTempDirectory("dc3-session-outside");
+        Path escaped = Files.writeString(outside.resolve("escaped.txt"), "payload");
+        when(sessionStore.delete("conv", header)).thenReturn(Mono.just(1L));
+        when(attachmentStore.listDeletedFilePaths("conv", header)).thenReturn(Flux.just(escaped.toString()));
+
+        StepVerifier.create(service.delete("conv", header)).expectNext(1L).verifyComplete();
+
+        assertThat(Files.exists(escaped))
+                .as("the boundary guard must refuse out-of-root unlinks")
+                .isTrue();
+    }
+
+    @Test
+    void deleteSkipsFileCleanupWhenTheSessionWasAlreadyDeleted() {
+        when(sessionStore.delete("conv", header)).thenReturn(Mono.just(0L));
+
+        StepVerifier.create(service.delete("conv", header)).expectNext(0L).verifyComplete();
+
+        verify(attachmentStore, never()).listDeletedFilePaths(any(), any());
+    }
 }

dc3-db/dc3-db-agentic/pom.xml

@@ -45,17 +45,12 @@
             <artifactId>dc3-common-agentic</artifactId>
         </dependency>
 
-        <!-- Store-level unit tests (API-key encryption wiring) -->
+        <!-- Store-level unit tests (key encryption wiring + concurrency scripts) -->
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-test</artifactId>
             <scope>test</scope>
         </dependency>
-        <dependency>
-            <groupId>org.junit.jupiter</groupId>
-            <artifactId>junit-jupiter</artifactId>
-            <scope>test</scope>
-        </dependency>
         <dependency>
             <groupId>io.projectreactor</groupId>
             <artifactId>reactor-test</artifactId>

dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcActionStore.java

@@ -38,7 +38,14 @@ import reactor.core.publisher.Mono;
 import tools.jackson.core.type.TypeReference;
 import tools.jackson.databind.ObjectMapper;
 
-/** Explicit SQL adapter for agentic actions. */
+/**
+ * Explicit SQL adapter for agentic actions.
+ *
+ * <p>Timestamps are stored in {@code TIMESTAMPTZ} columns and bound as {@link Instant} through
+ * {@link R2dbcDialect#bindInstant}, so encoding never depends on the database session timezone.
+ * The {@link ActionBO} wall-clock fields stay UTC-naive; reads normalize through
+ * {@link DatabaseInstant#toLocalDateTimeUtc}.
+ */
 @Repository
 @ConditionalOnClass({DatabaseClient.class, PageTransaction.class})
 @RequiredArgsConstructor
@@ -69,6 +76,7 @@ public class R2dbcActionStore implements ReactiveActionStore {
         if (action.getId() == null) {
             action.setId(UuidV7.nextLong());
         }
+        Instant now = Instant.now();
         String sql = "INSERT INTO " + TABLE
                 + " (id, action_id, conversation_id, action_type, title, description, payload, status, expire_time, "
                 + "tenant_id, user_id, remark, creator_id, creator_name, create_time, operator_id, operator_name, operate_time, deleted) "
@@ -91,14 +99,15 @@ public class R2dbcActionStore implements ReactiveActionStore {
                 .bind("remark", valueOrEmpty(action.getRemark()))
                 .bind("creator_id", valueOrZero(action.getCreatorId(), action.getUserId()))
                 .bind("creator_name", valueOrEmpty(action.getCreatorName()))
-                .bind("create_time", utc(action.getCreateTime(), Instant.now()))
+                .bind("create_time", dialect.bindInstant(instant(action.getCreateTime(), now)))
                 .bind("operator_id", valueOrZero(action.getOperatorId(), action.getUserId()))
                 .bind("operator_name", valueOrEmpty(action.getOperatorName()))
-                .bind("operate_time", utc(action.getOperateTime(), Instant.now()));
+                .bind("operate_time", dialect.bindInstant(instant(action.getOperateTime(), now)));
         if (action.getExpireTime() == null) {
-            statement = statement.bindNull("expire_time", LocalDateTime.class);
+            statement = statement.bindNull("expire_time", Instant.class);
         } else {
-            statement = statement.bind("expire_time", action.getExpireTime());
+            statement = statement.bind(
+                    "expire_time", dialect.bindInstant(action.getExpireTime().toInstant(ZoneOffset.UTC)));
         }
         return statement
                 .fetch()
@@ -133,7 +142,7 @@ public class R2dbcActionStore implements ReactiveActionStore {
             return Mono.just(OffsetPage.of(java.util.List.of(), offset, limit, 0));
         }
         new PageRequest(offset, limit);
-        LocalDateTime current = utc(now, Instant.now());
+        Instant current = instant(now, Instant.now());
         String predicate = " FROM " + TABLE
                 + " WHERE conversation_id = :conversation_id AND tenant_id = :tenant_id AND user_id = :user_id"
                 + " AND status = :status AND deleted = 0"
@@ -145,7 +154,7 @@ public class R2dbcActionStore implements ReactiveActionStore {
                 .bind("tenant_id", header.getTenantId())
                 .bind("user_id", header.getUserId())
                 .bind("status", AgenticActionStatusEnum.PENDING.getIndex())
-                .bind("now", current)
+                .bind("now", dialect.bindInstant(current))
                 .bind("limit", limit)
                 .bind("offset", offset);
         DatabaseClient.GenericExecuteSpec countStatement = databaseClient
@@ -154,7 +163,7 @@ public class R2dbcActionStore implements ReactiveActionStore {
                 .bind("tenant_id", header.getTenantId())
                 .bind("user_id", header.getUserId())
                 .bind("status", AgenticActionStatusEnum.PENDING.getIndex())
-                .bind("now", current);
+                .bind("now", dialect.bindInstant(current));
         Mono<Long> total = countStatement
                 .map((row, metadata) -> {
                     Number value = row.get("total", Number.class);
@@ -174,7 +183,7 @@ public class R2dbcActionStore implements ReactiveActionStore {
         if (nextStatus == null || nextStatus == AgenticActionStatusEnum.PENDING) {
             return Mono.error(new IllegalArgumentException("nextStatus must be a terminal action status"));
         }
-        LocalDateTime current = utc(now, Instant.now());
+        Instant current = instant(now, Instant.now());
         return databaseClient
                 .sql("UPDATE " + TABLE + " SET status = :next_status, operator_id = :operator_id, "
                         + "operator_name = :operator_name, operate_time = :operate_time"
@@ -184,12 +193,12 @@ public class R2dbcActionStore implements ReactiveActionStore {
                 .bind("next_status", nextStatus.getIndex())
                 .bind("operator_id", header.getUserId())
                 .bind("operator_name", valueOrEmpty(header.getUserName()))
-                .bind("operate_time", current)
+                .bind("operate_time", dialect.bindInstant(current))
                 .bind("action_id", actionId)
                 .bind("tenant_id", header.getTenantId())
                 .bind("user_id", header.getUserId())
                 .bind("pending_status", AgenticActionStatusEnum.PENDING.getIndex())
-                .bind("now", current)
+                .bind("now", dialect.bindInstant(current))
                 .fetch()
                 .rowsUpdated()
                 .flatMap(rows -> rows == 1 ? get(actionId, header) : Mono.empty());
@@ -215,7 +224,7 @@ public class R2dbcActionStore implements ReactiveActionStore {
                 .bind("remark", valueOrEmpty(remark))
                 .bind("operator_id", header.getUserId())
                 .bind("operator_name", valueOrEmpty(header.getUserName()))
-                .bind("operate_time", utc(now, Instant.now()))
+                .bind("operate_time", dialect.bindInstant(instant(now, Instant.now())))
                 .bind("action_id", actionId)
                 .bind("tenant_id", header.getTenantId())
                 .bind("user_id", header.getUserId())
@@ -292,11 +301,12 @@ public class R2dbcActionStore implements ReactiveActionStore {
         return value == null ? "" : value;
     }
 
-    private LocalDateTime utc(LocalDateTime value, Instant fallback) {
-        return value == null ? utc(fallback, Instant.now()) : value;
+    /** Interpret a BO wall-clock value as UTC and fall back to the given instant when absent. */
+    private Instant instant(LocalDateTime value, Instant fallback) {
+        return value == null ? fallback : value.toInstant(ZoneOffset.UTC);
     }
 
-    private LocalDateTime utc(Instant value, Instant fallback) {
-        return LocalDateTime.ofInstant(value == null ? fallback : value, ZoneOffset.UTC);
+    private Instant instant(Instant value, Instant fallback) {
+        return value == null ? fallback : value;
     }
 }

dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcAttachmentStore.java

@@ -19,10 +19,9 @@ package io.github.pnoker.db.agentic;
 import io.github.pnoker.common.agentic.entity.bo.AttachmentBO;
 import io.github.pnoker.common.agentic.repository.ReactiveAttachmentStore;
 import io.github.pnoker.common.entity.common.RequestHeader;
+import io.github.pnoker.db.core.dialect.R2dbcDialect;
 import io.github.pnoker.db.core.time.DatabaseInstant;
 import java.time.Instant;
-import java.time.LocalDateTime;
-import java.time.ZoneOffset;
 import java.util.Collection;
 import java.util.Objects;
 import java.util.stream.Collectors;
@@ -34,7 +33,14 @@ import org.springframework.transaction.reactive.TransactionalOperator;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-/** Explicit SQL adapter for attachment metadata. */
+/**
+ * Explicit SQL adapter for attachment metadata.
+ *
+ * <p>Timestamps are stored in {@code TIMESTAMPTZ} columns and bound as {@link Instant} through
+ * {@link R2dbcDialect#bindInstant}, so encoding never depends on the database session timezone.
+ * The {@link AttachmentBO} wall-clock fields stay UTC-naive; reads normalize through
+ * {@link DatabaseInstant#toLocalDateTimeUtc}.
+ */
 @Repository
 @ConditionalOnClass({DatabaseClient.class, TransactionalOperator.class})
 @RequiredArgsConstructor
@@ -47,18 +53,34 @@ public class R2dbcAttachmentStore implements ReactiveAttachmentStore {
 
     private final DatabaseClient databaseClient;
     private final TransactionalOperator transactionalOperator;
+    private final R2dbcDialect dialect;
 
     @Override
     public Mono<AttachmentBO> save(AttachmentBO attachment) {
         Objects.requireNonNull(attachment, "attachment must not be null");
+        Instant now = Instant.now();
         String sql = "INSERT INTO " + TABLE
                 + " (conversation_id, file_name, content_type, size, file_path, tenant_id, user_id,"
                 + " creator_id, creator_name, operator_id, operator_name, create_time, operate_time, deleted)"
                 + " VALUES (:conversation_id, :file_name, :content_type, :size, :file_path, :tenant_id, :user_id,"
                 + " :creator_id, :creator_name, :operator_id, :operator_name, :create_time, :operate_time, 0)";
-        LocalDateTime now = utcNow();
+        // Guard on the live session row under the same lock the session delete takes: a save
+        // racing a delete either lands before the delete soft-deletes the children or fails
+        // here, never surviving as a live row of a soft-deleted conversation.
+        Mono<Void> sessionGuard = databaseClient
+                .sql("SELECT id FROM dc3_agentic.dc3_session"
+                        + " WHERE conversation_id = :conversation_id AND tenant_id = :tenant_id"
+                        + " AND user_id = :user_id AND deleted = 0 FOR UPDATE")
+                .bind("conversation_id", attachment.getConversationId())
+                .bind("tenant_id", attachment.getTenantId())
+                .bind("user_id", attachment.getUserId())
+                .fetch()
+                .all()
+                .switchIfEmpty(Mono.error(
+                        new IllegalStateException("session must exist and must be live before saving an attachment")))
+                .then();
         return transactionalOperator
-                .transactional(databaseClient
+                .transactional(sessionGuard.then(databaseClient
                         .sql(sql)
                         .bind("conversation_id", attachment.getConversationId())
                         .bind("file_name", attachment.getFileName())
@@ -71,14 +93,14 @@ public class R2dbcAttachmentStore implements ReactiveAttachmentStore {
                         .bind("creator_name", attachment.getCreatorName())
                         .bind("operator_id", attachment.getOperatorId())
                         .bind("operator_name", attachment.getOperatorName())
-                        .bind("create_time", now)
-                        .bind("operate_time", now)
+                        .bind("create_time", dialect.bindInstant(now))
+                        .bind("operate_time", dialect.bindInstant(now))
                         .fetch()
                         .rowsUpdated()
                         .flatMap(rows -> rows == 1
                                 ? getByPath(attachment.getFilePath(), attachment.getTenantId(), attachment.getUserId())
                                 : Mono.error(
-                                        new IllegalStateException("attachment insert affected " + rows + " rows"))))
+                                        new IllegalStateException("attachment insert affected " + rows + " rows")))))
                 .switchIfEmpty(Mono.error(new IllegalStateException("attachment insert returned no row")));
     }
 
@@ -97,6 +119,21 @@ public class R2dbcAttachmentStore implements ReactiveAttachmentStore {
                 .all();
     }
 
+    @Override
+    public Flux<String> listDeletedFilePaths(String conversationId, RequestHeader.PrincipalHeader header) {
+        if (header == null) return Flux.error(new IllegalArgumentException("header must not be null"));
+        if (conversationId == null || conversationId.isBlank()) return Flux.empty();
+        return databaseClient
+                .sql("SELECT file_path FROM " + TABLE
+                        + " WHERE conversation_id = :conversation_id AND tenant_id = :tenant_id"
+                        + " AND user_id = :user_id AND deleted = 1")
+                .bind("conversation_id", conversationId)
+                .bind("tenant_id", header.getTenantId())
+                .bind("user_id", header.getUserId())
+                .map((row, metadata) -> row.get("file_path", String.class))
+                .all();
+    }
+
     @Override
     public Flux<AttachmentBO> getByIds(Collection<Long> ids, RequestHeader.PrincipalHeader header) {
         if (header == null) return Flux.error(new IllegalArgumentException("header must not be null"));
@@ -139,7 +176,7 @@ public class R2dbcAttachmentStore implements ReactiveAttachmentStore {
                         + " operate_time = :operate_time WHERE id = :id AND tenant_id = :tenant_id"
                         + " AND user_id = :user_id AND deleted = 0")
                 .bind("operator_id", header.getUserId())
-                .bind("operate_time", utcNow())
+                .bind("operate_time", dialect.bindInstant(Instant.now()))
                 .bind("id", id)
                 .bind("tenant_id", header.getTenantId())
                 .bind("user_id", header.getUserId())
@@ -179,8 +216,4 @@ public class R2dbcAttachmentStore implements ReactiveAttachmentStore {
         attachment.setOperateTime(DatabaseInstant.toLocalDateTimeUtc(row.get("operate_time")));
         return attachment;
     }
-
-    private LocalDateTime utcNow() {
-        return LocalDateTime.ofInstant(Instant.now(), ZoneOffset.UTC);
-    }
 }

dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcMessageStore.java

@@ -24,8 +24,6 @@ import io.github.pnoker.common.enums.AgenticMessageStatusEnum;
 import io.github.pnoker.db.core.dialect.R2dbcDialect;
 import io.github.pnoker.db.core.time.DatabaseInstant;
 import java.time.Instant;
-import java.time.LocalDateTime;
-import java.time.ZoneOffset;
 import lombok.RequiredArgsConstructor;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.r2dbc.core.DatabaseClient;
@@ -35,7 +33,14 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import tools.jackson.databind.ObjectMapper;
 
-/** Explicit SQL adapter for the agentic message store. */
+/**
+ * Explicit SQL adapter for the agentic message store.
+ *
+ * <p>Timestamps are stored in {@code TIMESTAMPTZ} columns and bound as {@link Instant} through
+ * {@link R2dbcDialect#bindInstant}, so encoding never depends on the database session timezone.
+ * The {@link MessageBO} wall-clock fields stay UTC-naive; reads normalize through
+ * {@link DatabaseInstant#toLocalDateTimeUtc}.
+ */
 @Repository
 @ConditionalOnClass({DatabaseClient.class, TransactionalOperator.class})
 @RequiredArgsConstructor
@@ -131,8 +136,8 @@ public class R2dbcMessageStore implements ReactiveMessageStore {
                 .bind("creator_name", header.getUserName() == null ? "" : header.getUserName())
                 .bind("operator_id", header.getUserId())
                 .bind("operator_name", header.getUserName() == null ? "" : header.getUserName())
-                .bind("create_time", LocalDateTime.ofInstant(now, ZoneOffset.UTC))
-                .bind("operate_time", LocalDateTime.ofInstant(now, ZoneOffset.UTC))
+                .bind("create_time", dialect.bindInstant(now))
+                .bind("operate_time", dialect.bindInstant(now))
                 .fetch()
                 .rowsUpdated()
                 .flatMap(rows -> rows == 1
@@ -202,7 +207,7 @@ public class R2dbcMessageStore implements ReactiveMessageStore {
                 .sql("UPDATE " + TABLE + " SET deleted = 1, operate_time = :operate_time"
                         + " WHERE conversation_id = :conversation_id AND tenant_id = :tenant_id"
                         + " AND user_id = :user_id AND deleted = 0")
-                .bind("operate_time", LocalDateTime.ofInstant(Instant.now(), ZoneOffset.UTC))
+                .bind("operate_time", dialect.bindInstant(Instant.now()))
                 .bind("conversation_id", conversationId)
                 .bind("tenant_id", header.getTenantId())
                 .bind("user_id", header.getUserId())

dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcModelConfigStore.java

@@ -24,7 +24,9 @@ import io.github.pnoker.common.enums.ConnectivityStatusEnum;
 import io.github.pnoker.common.enums.DefaultFlagEnum;
 import io.github.pnoker.common.enums.EnableFlagEnum;
 import io.github.pnoker.common.utils.UuidV7;
+import io.github.pnoker.db.core.dialect.R2dbcDialect;
 import io.github.pnoker.db.core.time.DatabaseInstant;
+import java.time.Instant;
 import java.time.LocalDateTime;
 import java.time.ZoneOffset;
 import lombok.RequiredArgsConstructor;
@@ -35,7 +37,14 @@ import org.springframework.transaction.reactive.TransactionalOperator;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-/** Explicit SQL adapter for agentic model configurations. */
+/**
+ * Explicit SQL adapter for agentic model configurations.
+ *
+ * <p>Timestamps are stored in {@code TIMESTAMPTZ} columns and bound as {@link Instant} through
+ * {@link R2dbcDialect#bindInstant}, so encoding never depends on the database session timezone.
+ * The {@link ModelConfigBO} wall-clock fields stay UTC-naive; reads normalize through
+ * {@link DatabaseInstant#toLocalDateTimeUtc}.
+ */
 @Repository
 @ConditionalOnClass({DatabaseClient.class, TransactionalOperator.class})
 @RequiredArgsConstructor
@@ -50,6 +59,7 @@ public class R2dbcModelConfigStore implements ReactiveModelConfigStore {
 
     private final DatabaseClient databaseClient;
     private final TransactionalOperator transactionalOperator;
+    private final R2dbcDialect dialect;
 
     @Override
     public Flux<ModelConfigBO> list(RequestHeader.PrincipalHeader header, boolean enabledOnly) {
@@ -142,10 +152,10 @@ public class R2dbcModelConfigStore implements ReactiveModelConfigStore {
                 .bind("remark", value.getRemark())
                 .bind("creator_id", value.getCreatorId())
                 .bind("creator_name", value.getCreatorName())
-                .bind("create_time", utc(value.getCreateTime()))
+                .bind("create_time", dialect.bindInstant(instant(value.getCreateTime())))
                 .bind("operator_id", value.getOperatorId())
                 .bind("operator_name", value.getOperatorName())
-                .bind("operate_time", utc(value.getOperateTime()))
+                .bind("operate_time", dialect.bindInstant(instant(value.getOperateTime())))
                 .fetch()
                 .rowsUpdated()
                 .flatMap(rows -> rows == 1
@@ -185,7 +195,7 @@ public class R2dbcModelConfigStore implements ReactiveModelConfigStore {
                 .bind("remark", value.getRemark())
                 .bind("operator_id", value.getOperatorId())
                 .bind("operator_name", value.getOperatorName())
-                .bind("operate_time", utc(value.getOperateTime()))
+                .bind("operate_time", dialect.bindInstant(instant(value.getOperateTime())))
                 .bind("id", value.getId())
                 .bind("tenant_id", value.getTenantId())
                 .fetch()
@@ -202,7 +212,7 @@ public class R2dbcModelConfigStore implements ReactiveModelConfigStore {
                 .sql("UPDATE " + TABLE + " SET deleted = 1, operator_id = :operator_id,"
                         + " operate_time = :operate_time WHERE id = :id AND tenant_id = :tenant_id AND deleted = 0")
                 .bind("operator_id", header.getUserId())
-                .bind("operate_time", utcNow())
+                .bind("operate_time", dialect.bindInstant(Instant.now()))
                 .bind("id", id)
                 .bind("tenant_id", header.getTenantId())
                 .fetch()
@@ -228,8 +238,9 @@ public class R2dbcModelConfigStore implements ReactiveModelConfigStore {
                 ? spec.bindNull("last_check_status", Byte.class)
                 : spec.bind("last_check_status", profile.status().getIndex());
         spec = profile.time() == null
-                ? spec.bindNull("last_check_time", LocalDateTime.class)
-                : spec.bind("last_check_time", profile.time());
+                ? spec.bindNull("last_check_time", Instant.class)
+                : spec.bind(
+                        "last_check_time", dialect.bindInstant(profile.time().toInstant(ZoneOffset.UTC)));
         spec = profile.latencyMs() == null
                 ? spec.bindNull("last_check_latency_ms", Integer.class)
                 : spec.bind("last_check_latency_ms", profile.latencyMs());
@@ -257,7 +268,7 @@ public class R2dbcModelConfigStore implements ReactiveModelConfigStore {
         return databaseClient
                 .sql("UPDATE " + TABLE + " SET default_flag = 0, operate_time = :operate_time"
                         + " WHERE tenant_id = :tenant_id AND default_flag = 1 AND deleted = 0 AND id <> :id")
-                .bind("operate_time", utcNow())
+                .bind("operate_time", dialect.bindInstant(Instant.now()))
                 .bind("tenant_id", tenantId)
                 .bind("id", exceptId)
                 .fetch()
@@ -332,12 +343,9 @@ public class R2dbcModelConfigStore implements ReactiveModelConfigStore {
                 : value.getEnableFlag().getIndex();
     }
 
-    private LocalDateTime utc(LocalDateTime value) {
-        return value == null ? utcNow() : value;
-    }
-
-    private LocalDateTime utcNow() {
-        return LocalDateTime.now(ZoneOffset.UTC);
+    /** Interpret a BO wall-clock value as UTC, defaulting to now when absent. */
+    private Instant instant(LocalDateTime value) {
+        return value == null ? Instant.now() : value.toInstant(ZoneOffset.UTC);
     }
 
     private LocalDateTime time(Object value) {

dc3-db/dc3-db-agentic/src/main/java/io/github/pnoker/db/agentic/R2dbcSessionStore.java

@@ -19,7 +19,6 @@ package io.github.pnoker.db.agentic;
 import io.github.pnoker.common.agentic.entity.bo.SessionBO;
 import io.github.pnoker.common.agentic.entity.model.SessionExt;
 import io.github.pnoker.common.agentic.repository.ReactiveSessionStore;
-import io.github.pnoker.common.agentic.service.impl.AttachmentFileCleaner;
 import io.github.pnoker.common.constant.service.AgenticConstant;
 import io.github.pnoker.common.entity.common.RequestHeader;
 import io.github.pnoker.db.core.dialect.R2dbcDialect;
@@ -28,22 +27,23 @@ import io.github.pnoker.db.core.page.SortSpec;
 import io.github.pnoker.db.core.time.DatabaseInstant;
 import io.github.pnoker.db.core.transaction.PageTransaction;
 import java.time.Instant;
-import java.time.LocalDateTime;
-import java.time.ZoneOffset;
-import java.util.List;
 import lombok.RequiredArgsConstructor;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.r2dbc.core.DatabaseClient;
 import org.springframework.stereotype.Repository;
 import org.springframework.transaction.reactive.TransactionalOperator;
-import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.util.function.Tuple2;
-import reactor.util.function.Tuples;
 import tools.jackson.databind.ObjectMapper;
 
-/** Explicit SQL adapter for agentic sessions. */
+/**
+ * Explicit SQL adapter for agentic sessions.
+ *
+ * <p>Timestamps are stored in {@code TIMESTAMPTZ} columns and bound as {@link Instant} through
+ * {@link R2dbcDialect#bindInstant}, so encoding never depends on the database session timezone.
+ * The {@link SessionBO} wall-clock fields stay UTC-naive; reads normalize through
+ * {@link DatabaseInstant#toLocalDateTimeUtc}.
+ */
 @Repository
 @ConditionalOnClass({DatabaseClient.class, TransactionalOperator.class, PageTransaction.class})
 @RequiredArgsConstructor
@@ -58,22 +58,39 @@ public class R2dbcSessionStore implements ReactiveSessionStore {
     private final PageTransaction pageTransaction;
     private final ObjectMapper objectMapper;
     private final R2dbcDialect dialect;
-    private final AttachmentFileCleaner attachmentFileCleaner;
 
     @Override
     public Mono<SessionBO> touch(String conversationId, SessionExt sessionExt, RequestHeader.PrincipalHeader header) {
+        return touchAttempt(conversationId, sessionExt, header, true);
+    }
+
+    /**
+     * One touch attempt: lock the conversation's row regardless of tombstone state, then update,
+     * revive, or insert. A unique-key violation means a concurrent touch inserted the same
+     * conversation first, so the caller retries the whole decision once instead of surfacing the
+     * raw constraint error.
+     */
+    private Mono<SessionBO> touchAttempt(
+            String conversationId,
+            SessionExt sessionExt,
+            RequestHeader.PrincipalHeader header,
+            boolean retryOnConflict) {
         if (conversationId == null || conversationId.isBlank()) {
             return Mono.error(new IllegalArgumentException("conversationId must not be blank"));
         }
         if (header == null) return Mono.error(new IllegalArgumentException("header must not be null"));
-        Mono<SessionBO> transaction = lock(conversationId, header)
-                .flatMap(existing -> updateExisting(conversationId, sessionExt, existing, header))
-                .switchIfEmpty(insert(conversationId, serialize(sessionExt), header));
+        Mono<SessionBO> transaction = lockAny(conversationId, header)
+                .flatMap(locked -> locked.deleted()
+                        ? revive(conversationId, sessionExt, locked.session(), header)
+                        : updateRow(conversationId, sessionExt, null, locked.session(), header))
+                .switchIfEmpty(Mono.defer(() -> insert(conversationId, serialize(sessionExt), header)));
         return transactionalOperator
                 .transactional(transaction)
                 .onErrorResume(
                         DataIntegrityViolationException.class,
-                        ignored -> get(conversationId, header).switchIfEmpty(Mono.error(ignored)));
+                        ignored -> retryOnConflict
+                                ? touchAttempt(conversationId, sessionExt, header, false)
+                                : Mono.error(ignored));
     }
 
     @Override
@@ -167,25 +184,11 @@ public class R2dbcSessionStore implements ReactiveSessionStore {
             return Mono.error(new IllegalArgumentException("conversationId must not be blank"));
         }
         if (header == null) return Mono.error(new IllegalArgumentException("header must not be null"));
-        return get(conversationId, header).flatMap(existing -> {
-            String nextTitle = title == null || title.isBlank() ? existing.getTitle() : title.trim();
-            String nextExt = serialize(merge(existing.getSessionExt(), sessionExt));
-            return databaseClient
-                    .sql("UPDATE " + TABLE + " SET title = :title, session_ext = "
-                            + dialect.jsonWriteExpression(":session_ext") + ","
-                            + " operator_id = :operator_id, operate_time = :operate_time WHERE id = :id"
-                            + " AND tenant_id = :tenant_id AND user_id = :user_id AND deleted = 0")
-                    .bind("title", nextTitle)
-                    .bind("session_ext", nextExt)
-                    .bind("operator_id", header.getUserId())
-                    .bind("operate_time", utcNow())
-                    .bind("id", existing.getId())
-                    .bind("tenant_id", header.getTenantId())
-                    .bind("user_id", header.getUserId())
-                    .fetch()
-                    .rowsUpdated()
-                    .flatMap(rows -> rows == 1 ? get(conversationId, header) : Mono.empty());
-        });
+        // The row lock serializes concurrent updates and the merge reads the locked row, so
+        // concurrent preference changes merge instead of the last writer silently winning.
+        Mono<SessionBO> transaction = lock(conversationId, header)
+                .flatMap(existing -> updateRow(conversationId, sessionExt, title, existing, header));
+        return transactionalOperator.transactional(transaction);
     }
 
     @Override
@@ -194,62 +197,44 @@ public class R2dbcSessionStore implements ReactiveSessionStore {
             return Mono.error(new IllegalArgumentException("conversationId must not be blank"));
         }
         if (header == null) return Mono.error(new IllegalArgumentException("header must not be null"));
-        Mono<Long> deleteMessages = databaseClient
-                .sql("UPDATE dc3_agentic.dc3_message SET deleted = 1,"
-                        + " operate_time = :operate_time WHERE conversation_id = :conversation_id"
-                        + " AND tenant_id = :tenant_id AND user_id = :user_id AND deleted = 0")
-                .bind("operate_time", utcNow())
+        // Lock the live session row first: attachment saves take the same lock before inserting,
+        // so a concurrent upload either lands before this delete (and is soft-deleted with the
+        // rest) or fails its guard once the session is gone. Children then the session row.
+        Mono<Long> transaction = lockSessionId(conversationId, header)
+                .flatMap(sessionId -> softDeleteChildren("dc3_agentic.dc3_message", conversationId, header)
+                        .then(softDeleteChildren("dc3_agentic.dc3_attachment", conversationId, header))
+                        .then(softDeleteChildren("dc3_agentic.dc3_action", conversationId, header))
+                        .then(deleteSession(conversationId, header)))
+                .defaultIfEmpty(0L);
+        return transactionalOperator.transactional(transaction);
+    }
+
+    /** Lock the live session row and emit its id, or empty when the session is already deleted. */
+    private Mono<Long> lockSessionId(String conversationId, RequestHeader.PrincipalHeader header) {
+        return databaseClient
+                .sql("SELECT id FROM " + TABLE
+                        + " WHERE conversation_id = :conversation_id AND tenant_id = :tenant_id"
+                        + " AND user_id = :user_id AND deleted = 0 FOR UPDATE")
                 .bind("conversation_id", conversationId)
                 .bind("tenant_id", header.getTenantId())
                 .bind("user_id", header.getUserId())
-                .fetch()
-                .rowsUpdated()
-                .map(Long::valueOf);
-        Mono<Long> deleteAttachments = softDeleteChildren("dc3_agentic.dc3_attachment", conversationId, header);
-        Mono<Long> deleteActions = softDeleteChildren("dc3_agentic.dc3_action", conversationId, header);
-        Mono<Long> deleteSession = databaseClient
+                .map((row, metadata) -> row.get("id", Long.class))
+                .one();
+    }
+
+    private Mono<Long> deleteSession(String conversationId, RequestHeader.PrincipalHeader header) {
+        return databaseClient
                 .sql("UPDATE " + TABLE + " SET deleted = 1,"
                         + " operator_id = :operator_id, operate_time = :operate_time WHERE conversation_id = :conversation_id"
                         + " AND tenant_id = :tenant_id AND user_id = :user_id AND deleted = 0")
                 .bind("operator_id", header.getUserId())
-                .bind("operate_time", utcNow())
+                .bind("operate_time", dialect.bindInstant(Instant.now()))
                 .bind("conversation_id", conversationId)
                 .bind("tenant_id", header.getTenantId())
                 .bind("user_id", header.getUserId())
                 .fetch()
                 .rowsUpdated()
                 .map(Long::valueOf);
-        return transactionalOperator
-                .transactional(listAttachmentFilePaths(conversationId, header)
-                        .collectList()
-                        .flatMap(filePaths -> deleteMessages
-                                .then(deleteAttachments)
-                                .then(deleteActions)
-                                .then(deleteSession)
-                                .map(deleted -> Tuples.of(deleted, filePaths))))
-                .flatMap(deletedSession -> unlinkAttachmentFiles(deletedSession).thenReturn(deletedSession.getT1()));
-    }
-
-    /** Read attachment file locations while the rows are still live, so the files can be unlinked after commit. */
-    private Flux<String> listAttachmentFilePaths(String conversationId, RequestHeader.PrincipalHeader header) {
-        return databaseClient
-                .sql("SELECT file_path FROM dc3_agentic.dc3_attachment"
-                        + " WHERE conversation_id = :conversation_id AND tenant_id = :tenant_id"
-                        + " AND user_id = :user_id AND deleted = 0")
-                .bind("conversation_id", conversationId)
-                .bind("tenant_id", header.getTenantId())
-                .bind("user_id", header.getUserId())
-                .map((row, metadata) -> row.get("file_path", String.class))
-                .all();
-    }
-
-    /**
-     * Unlink the conversation's stored attachment files after the session delete commits. Runs
-     * strictly after the transaction so file cleanup can never roll the delete back, and is
-     * warn-only so a missing or undeletable file never fails the committed delete.
-     */
-    private Mono<Void> unlinkAttachmentFiles(Tuple2<Long, List<String>> deletedSession) {
-        return attachmentFileCleaner.unlinkAll(deletedSession.getT2());
     }
 
     private Mono<Long> softDeleteChildren(String table, String conversationId, RequestHeader.PrincipalHeader header) {
@@ -257,7 +242,7 @@ public class R2dbcSessionStore implements ReactiveSessionStore {
                 .sql("UPDATE " + table + " SET deleted = 1, operate_time = :operate_time"
                         + " WHERE conversation_id = :conversation_id AND tenant_id = :tenant_id"
                         + " AND user_id = :user_id AND deleted = 0")
-                .bind("operate_time", utcNow())
+                .bind("operate_time", dialect.bindInstant(Instant.now()))
                 .bind("conversation_id", conversationId)
                 .bind("tenant_id", header.getTenantId())
                 .bind("user_id", header.getUserId())
@@ -266,6 +251,7 @@ public class R2dbcSessionStore implements ReactiveSessionStore {
                 .map(Long::valueOf);
     }
 
+    /** Lock the live session row for update; empty when the conversation has no live row. */
     private Mono<SessionBO> lock(String conversationId, RequestHeader.PrincipalHeader header) {
         return databaseClient
                 .sql("SELECT " + COLUMNS + " FROM " + TABLE
@@ -278,16 +264,75 @@ public class R2dbcSessionStore implements ReactiveSessionStore {
                 .one();
     }
 
-    private Mono<SessionBO> updateExisting(
-            String conversationId, SessionExt sessionExt, SessionBO existing, RequestHeader.PrincipalHeader header) {
-        String nextJson = serialize(merge(existing.getSessionExt(), sessionExt));
+    /**
+     * Lock the conversation's row regardless of soft-delete state. The natural key is unique
+     * across live and deleted rows (it is the foreign-key target for the child tables), so the
+     * tombstone row is the row that must be locked and revived to reuse the conversation id.
+     */
+    private Mono<LockedSession> lockAny(String conversationId, RequestHeader.PrincipalHeader header) {
         return databaseClient
-                .sql("UPDATE " + TABLE + " SET session_ext = " + dialect.jsonWriteExpression(":session_ext")
-                        + ", operator_id = :operator_id," + " operate_time = :operate_time WHERE id = :id")
-                .bind("session_ext", nextJson)
+                .sql("SELECT " + COLUMNS + " FROM " + TABLE
+                        + " WHERE conversation_id = :conversation_id AND tenant_id = :tenant_id"
+                        + " AND user_id = :user_id LIMIT 1 FOR UPDATE")
+                .bind("conversation_id", conversationId)
+                .bind("tenant_id", header.getTenantId())
+                .bind("user_id", header.getUserId())
+                .map((row, metadata) -> new LockedSession(
+                        map(row, metadata), row.get("deleted", Number.class).intValue() != 0))
+                .one();
+    }
+
+    /**
+     * Revive a soft-deleted session row as a brand-new conversation: the tombstone's title,
+     * extension, creator, and creation time are reset, because the deleted conversation's
+     * messages and attachments stay soft-deleted and must never reappear.
+     */
+    private Mono<SessionBO> revive(
+            String conversationId, SessionExt sessionExt, SessionBO tombstone, RequestHeader.PrincipalHeader header) {
+        return databaseClient
+                .sql("UPDATE " + TABLE + " SET deleted = 0, title = :title, session_ext = "
+                        + dialect.jsonWriteExpression(":session_ext") + ", creator_id = :creator_id,"
+                        + " creator_name = :creator_name, create_time = :create_time, operator_id = :operator_id,"
+                        + " operate_time = :operate_time WHERE id = :id AND tenant_id = :tenant_id"
+                        + " AND user_id = :user_id")
+                .bind("title", AgenticConstant.Session.DEFAULT_TITLE)
+                .bind("session_ext", serialize(sessionExt))
+                .bind("creator_id", header.getUserId())
+                .bind("creator_name", header.getUserName() == null ? "" : header.getUserName())
+                .bind("create_time", dialect.bindInstant(Instant.now()))
                 .bind("operator_id", header.getUserId())
-                .bind("operate_time", utcNow())
+                .bind("operate_time", dialect.bindInstant(Instant.now()))
+                .bind("id", tombstone.getId())
+                .bind("tenant_id", header.getTenantId())
+                .bind("user_id", header.getUserId())
+                .fetch()
+                .rowsUpdated()
+                .flatMap(rows -> rows == 1
+                        ? get(conversationId, header)
+                        : Mono.error(new IllegalStateException("session revive affected " + rows + " rows")));
+    }
+
+    /** Merge the incoming extension onto the locked row and persist it with the optional title. */
+    private Mono<SessionBO> updateRow(
+            String conversationId,
+            SessionExt sessionExt,
+            String title,
+            SessionBO existing,
+            RequestHeader.PrincipalHeader header) {
+        String nextTitle = title == null || title.isBlank() ? existing.getTitle() : title.trim();
+        String nextExt = serialize(merge(existing.getSessionExt(), sessionExt));
+        return databaseClient
+                .sql("UPDATE " + TABLE + " SET title = :title, session_ext = "
+                        + dialect.jsonWriteExpression(":session_ext") + ","
+                        + " operator_id = :operator_id, operate_time = :operate_time WHERE id = :id"
+                        + " AND tenant_id = :tenant_id AND user_id = :user_id AND deleted = 0")
+                .bind("title", nextTitle)
+                .bind("session_ext", nextExt)
+                .bind("operator_id", header.getUserId())
+                .bind("operate_time", dialect.bindInstant(Instant.now()))
                 .bind("id", existing.getId())
+                .bind("tenant_id", header.getTenantId())
+                .bind("user_id", header.getUserId())
                 .fetch()
                 .rowsUpdated()
                 .flatMap(rows -> rows == 1
@@ -308,8 +353,8 @@ public class R2dbcSessionStore implements ReactiveSessionStore {
                 .bind("user_id", header.getUserId())
                 .bind("creator_id", header.getUserId())
                 .bind("operator_id", header.getUserId())
-                .bind("create_time", utcNow())
-                .bind("operate_time", utcNow())
+                .bind("create_time", dialect.bindInstant(Instant.now()))
+                .bind("operate_time", dialect.bindInstant(Instant.now()))
                 .fetch()
                 .rowsUpdated()
                 .flatMap(rows -> rows == 1
@@ -376,7 +421,6 @@ public class R2dbcSessionStore implements ReactiveSessionStore {
         return merged;
     }
 
-    private LocalDateTime utcNow() {
-        return LocalDateTime.ofInstant(Instant.now(), ZoneOffset.UTC);
-    }
+    /** A session row read under lock together with its soft-delete tombstone flag. */
+    private record LockedSession(SessionBO session, boolean deleted) {}
 }

dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/AgenticStoreLayeringTest.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2016-present the IoT DC3 original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package io.github.pnoker.db.agentic;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.stream.Stream;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Layering guard: the database adapters must stay pure SQL ports. They implement the
+ * {@code Reactive*Store} contracts of {@code dc3-common-agentic} and must never import or
+ * orchestrate service-layer implementation classes (business rules, filesystem cleanup), so the
+ * dependency direction never reverses.
+ */
+class AgenticStoreLayeringTest {
+
+    @Test
+    void storeAdaptersNeverImportServiceImplementations() throws IOException {
+        List<Path> offenders;
+        try (Stream<Path> sources = Files.walk(Path.of("src", "main", "java"))) {
+            offenders = sources.filter(path -> path.toString().endsWith(".java"))
+                    .filter(path -> containsServiceImport(path))
+                    .toList();
+        }
+        assertThat(offenders)
+                .as("db adapters must not reference io.github.pnoker.common.agentic.service.impl")
+                .isEmpty();
+    }
+
+    private boolean containsServiceImport(Path path) {
+        try {
+            return Files.readString(path).contains("io.github.pnoker.common.agentic.service.impl");
+        } catch (IOException exception) {
+            throw new IllegalStateException("cannot read " + path, exception);
+        }
+    }
+}

dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcActionStoreTest.java

@@ -0,0 +1,118 @@
+/*
+ * Copyright 2016-present the IoT DC3 original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package io.github.pnoker.db.agentic;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.github.pnoker.common.agentic.entity.bo.ActionBO;
+import io.github.pnoker.common.entity.common.RequestHeader;
+import io.github.pnoker.common.enums.AgenticActionStatusEnum;
+import io.github.pnoker.db.core.dialect.StandardR2dbcDialect;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import reactor.test.StepVerifier;
+import tools.jackson.databind.ObjectMapper;
+
+/** Regression coverage for timezone-safe timestamp binds across the action store statements. */
+class R2dbcActionStoreTest {
+
+    private static final long TENANT_ID = 7L;
+    private static final long USER_ID = 42L;
+    private static final String CONVERSATION_ID = "conv-1";
+
+    private final ScriptedAgenticDatabase database = new ScriptedAgenticDatabase();
+    private final R2dbcActionStore store = new R2dbcActionStore(
+            database,
+            ScriptedAgenticDatabase.passThroughPageTransaction(),
+            new ObjectMapper(),
+            new StandardR2dbcDialect("postgres", "public.dc3_schema_fingerprint"));
+
+    private RequestHeader.PrincipalHeader header;
+
+    @BeforeEach
+    void setUp() {
+        header = new RequestHeader.PrincipalHeader();
+        header.setTenantId(TENANT_ID);
+        header.setPrincipalId(USER_ID);
+        header.setPrincipalName("operator");
+    }
+
+    @Test
+    void createBindsEveryTimestampAsAnInstant() {
+        ActionBO action = action(LocalDateTime.now(ZoneOffset.UTC).plusMinutes(10));
+
+        StepVerifier.create(store.create(action)).expectNextCount(1).verifyComplete();
+
+        ScriptedAgenticDatabase.Execution insert = database.firstMatching("INSERT INTO dc3_agentic.dc3_action");
+        assertThat(insert.binds().get("expire_time"))
+                .as("TIMESTAMPTZ binds must carry an explicit instant, not a zone-less wall clock")
+                .isInstanceOf(Instant.class);
+        assertThat(insert.binds().get("create_time")).isInstanceOf(Instant.class);
+        assertThat(insert.binds().get("operate_time")).isInstanceOf(Instant.class);
+    }
+
+    @Test
+    void createBindsANullExpireTimeWithTheInstantType() {
+        StepVerifier.create(store.create(action(null))).expectNextCount(1).verifyComplete();
+
+        ScriptedAgenticDatabase.Execution insert = database.firstMatching("INSERT INTO dc3_agentic.dc3_action");
+        assertThat(insert.nulls()).containsEntry("expire_time", Instant.class);
+        assertThat(insert.binds()).doesNotContainKey("expire_time");
+    }
+
+    @Test
+    void listPendingBindsTheExpiryWindowAsInstants() {
+        database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Live", "{}", false);
+
+        StepVerifier.create(store.listPending(0, 10, CONVERSATION_ID, header, Instant.now()))
+                .expectNextCount(1)
+                .verifyComplete();
+
+        ScriptedAgenticDatabase.Execution query = database.firstMatching("FROM dc3_agentic.dc3_action");
+        assertThat(query.binds().get("now")).isInstanceOf(Instant.class);
+    }
+
+    @Test
+    void claimPendingBindsOperateTimeAsAnInstant() {
+        ActionBO created = action(null);
+        StepVerifier.create(store.create(created)).expectNextCount(1).verifyComplete();
+
+        StepVerifier.create(store.claimPending("action-1", header, AgenticActionStatusEnum.CONFIRMED, Instant.now()))
+                .expectNextCount(1)
+                .verifyComplete();
+
+        ScriptedAgenticDatabase.Execution update = database.firstMatching("UPDATE dc3_agentic.dc3_action");
+        assertThat(update.binds().get("operate_time")).isInstanceOf(Instant.class);
+        assertThat(update.binds().get("now")).isInstanceOf(Instant.class);
+    }
+
+    private ActionBO action(LocalDateTime expireTime) {
+        ActionBO action = new ActionBO();
+        action.setActionId("action-1");
+        action.setConversationId(CONVERSATION_ID);
+        action.setActionType("write");
+        action.setTitle("Write value");
+        action.setDescription("Writes a device point value");
+        action.setTenantId(TENANT_ID);
+        action.setUserId(USER_ID);
+        action.setExpireTime(expireTime);
+        return action;
+    }
+}

dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcAttachmentStoreTest.java

@@ -0,0 +1,184 @@
+/*
+ * Copyright 2016-present the IoT DC3 original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package io.github.pnoker.db.agentic;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.github.pnoker.common.agentic.entity.bo.AttachmentBO;
+import io.github.pnoker.common.entity.common.RequestHeader;
+import io.github.pnoker.db.core.dialect.StandardR2dbcDialect;
+import java.time.Instant;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import reactor.test.StepVerifier;
+
+/**
+ * Regression coverage for the attachment save guard: the insert runs in one transaction behind a
+ * live-session row lock, so a save racing a concurrent session delete either lands before the
+ * delete or fails with a clear error, never surviving as a zombie row of a soft-deleted
+ * conversation. Timestamps bind as {@link Instant} so TIMESTAMPTZ encoding cannot drift.
+ */
+class R2dbcAttachmentStoreTest {
+
+    private static final long TENANT_ID = 7L;
+    private static final long USER_ID = 42L;
+    private static final String CONVERSATION_ID = "conv-1";
+
+    private final ScriptedAgenticDatabase database = new ScriptedAgenticDatabase();
+    private final R2dbcAttachmentStore store = new R2dbcAttachmentStore(
+            database,
+            ScriptedAgenticDatabase.passThroughOperator(),
+            new StandardR2dbcDialect("postgres", "public.dc3_schema_fingerprint"));
+
+    private RequestHeader.PrincipalHeader header;
+
+    @BeforeEach
+    void setUp() {
+        header = new RequestHeader.PrincipalHeader();
+        header.setTenantId(TENANT_ID);
+        header.setPrincipalId(USER_ID);
+        header.setPrincipalName("operator");
+    }
+
+    @Test
+    void saveInsertsBehindTheLiveSessionGuardWhenTheSessionIsLive() {
+        database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Live", "{}", false);
+
+        StepVerifier.create(store.save(attachment("/store/note.txt")))
+                .assertNext(saved -> {
+                    assertThat(saved.getConversationId()).isEqualTo(CONVERSATION_ID);
+                    assertThat(saved.getFilePath()).isEqualTo("/store/note.txt");
+                })
+                .verifyComplete();
+
+        int guardIndex = database.indexOfFirst("SELECT id FROM dc3_agentic.dc3_session");
+        int insertIndex = database.indexOfFirst("INSERT INTO dc3_agentic.dc3_attachment");
+        assertThat(guardIndex)
+                .as("the live-session guard must run before the insert in the same transaction")
+                .isGreaterThanOrEqualTo(0)
+                .isLessThan(insertIndex);
+    }
+
+    @Test
+    void saveFailsClosedWhenTheSessionIsASoftDeletedTombstone() {
+        database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Gone", "{}", true);
+
+        StepVerifier.create(store.save(attachment("/store/note.txt")))
+                .expectErrorSatisfies(error -> {
+                    assertThat(error).isInstanceOf(IllegalStateException.class);
+                    assertThat(error.getMessage()).contains("session must exist");
+                })
+                .verify();
+
+        assertThat(database.countMatching("INSERT INTO dc3_agentic.dc3_attachment"))
+                .as("a tombstoned session must never accept a new attachment row")
+                .isZero();
+    }
+
+    @Test
+    void saveFailsClosedWhenTheSessionDoesNotExist() {
+        StepVerifier.create(store.save(attachment("/store/note.txt")))
+                .expectErrorSatisfies(error -> assertThat(error).isInstanceOf(IllegalStateException.class))
+                .verify();
+
+        assertThat(database.countMatching("INSERT INTO dc3_agentic.dc3_attachment"))
+                .isZero();
+    }
+
+    @Test
+    void saveLosesTheRaceAgainstAConcurrentSessionDelete() throws Exception {
+        database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Doomed", "{}", false);
+        // Park the delete's first child soft-delete so the upload's guard subscribes while the
+        // delete transaction holds the session row lock.
+        ScriptedAgenticDatabase.Hold deleteHold = database.holdNext("UPDATE dc3_agentic.dc3_message");
+
+        R2dbcSessionStore sessionStore = sessionStore();
+        CompletableFuture<Long> delete =
+                sessionStore.delete(CONVERSATION_ID, header).toFuture();
+        deleteHold.awaitRouted();
+
+        CompletableFuture<AttachmentBO> upload =
+                store.save(attachment("/store/late.txt")).toFuture();
+        assertThat(upload.isDone())
+                .as("the upload guard must wait on the session row lock held by the delete")
+                .isFalse();
+
+        deleteHold.release();
+        assertThat(delete.get(5, TimeUnit.SECONDS)).isEqualTo(1L);
+        assertThat(upload)
+                .as("the upload must fail once the delete committed, instead of surviving it")
+                .isCompletedExceptionally();
+
+        assertThat(database.countMatching("INSERT INTO dc3_agentic.dc3_attachment"))
+                .as("no attachment row may outlive the session delete")
+                .isZero();
+        assertThat(database.session(TENANT_ID, USER_ID, CONVERSATION_ID).deleted)
+                .isOne();
+    }
+
+    @Test
+    void listDeletedFilePathsStreamsOnlySoftDeletedRows() {
+        database.seedAttachment(TENANT_ID, USER_ID, CONVERSATION_ID, "/store/gone.txt", true);
+        database.seedAttachment(TENANT_ID, USER_ID, CONVERSATION_ID, "/store/live.txt", false);
+
+        StepVerifier.create(store.listDeletedFilePaths(CONVERSATION_ID, header))
+                .expectNext("/store/gone.txt")
+                .verifyComplete();
+    }
+
+    @Test
+    void saveBindsTimestampsAsInstantsSoEncodingIsTimezoneIndependent() {
+        database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Live", "{}", false);
+
+        StepVerifier.create(store.save(attachment("/store/note.txt")))
+                .expectNextCount(1)
+                .verifyComplete();
+
+        ScriptedAgenticDatabase.Execution insert = database.firstMatching("INSERT INTO dc3_agentic.dc3_attachment");
+        assertThat(insert.binds().get("create_time"))
+                .as("TIMESTAMPTZ binds must carry an explicit instant, not a zone-less wall clock")
+                .isInstanceOf(Instant.class);
+        assertThat(insert.binds().get("operate_time")).isInstanceOf(Instant.class);
+    }
+
+    private R2dbcSessionStore sessionStore() {
+        return new R2dbcSessionStore(
+                database,
+                ScriptedAgenticDatabase.passThroughOperator(),
+                org.mockito.Mockito.mock(io.github.pnoker.db.core.transaction.PageTransaction.class),
+                new tools.jackson.databind.ObjectMapper(),
+                new StandardR2dbcDialect("postgres", "public.dc3_schema_fingerprint"));
+    }
+
+    private AttachmentBO attachment(String path) {
+        AttachmentBO attachment = new AttachmentBO();
+        attachment.setConversationId(CONVERSATION_ID);
+        attachment.setFileName("note.txt");
+        attachment.setContentType("text/plain");
+        attachment.setSize(7L);
+        attachment.setFilePath(path);
+        attachment.setTenantId(TENANT_ID);
+        attachment.setUserId(USER_ID);
+        attachment.setCreatorId(USER_ID);
+        attachment.setCreatorName("operator");
+        attachment.setOperatorId(USER_ID);
+        attachment.setOperatorName("operator");
+        return attachment;
+    }
+}

dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcMessageStoreTest.java

@@ -0,0 +1,100 @@
+/*
+ * Copyright 2016-present the IoT DC3 original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package io.github.pnoker.db.agentic;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.github.pnoker.common.agentic.entity.model.AgenticMessageContent;
+import io.github.pnoker.common.entity.common.RequestHeader;
+import io.github.pnoker.common.enums.AgenticMessageStatusEnum;
+import io.github.pnoker.db.core.dialect.StandardR2dbcDialect;
+import java.time.Instant;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import reactor.test.StepVerifier;
+import tools.jackson.databind.ObjectMapper;
+
+/** Regression coverage for the message store save protocol and timezone-safe timestamp binds. */
+class R2dbcMessageStoreTest {
+
+    private static final long TENANT_ID = 7L;
+    private static final long USER_ID = 42L;
+    private static final String CONVERSATION_ID = "conv-1";
+
+    private final ScriptedAgenticDatabase database = new ScriptedAgenticDatabase();
+    private final R2dbcMessageStore store = new R2dbcMessageStore(
+            database,
+            ScriptedAgenticDatabase.passThroughOperator(),
+            new ObjectMapper(),
+            new StandardR2dbcDialect("postgres", "public.dc3_schema_fingerprint"));
+
+    private RequestHeader.PrincipalHeader header;
+
+    @BeforeEach
+    void setUp() {
+        header = new RequestHeader.PrincipalHeader();
+        header.setTenantId(TENANT_ID);
+        header.setPrincipalId(USER_ID);
+        header.setPrincipalName("operator");
+    }
+
+    @Test
+    void saveRunsTheSessionGuardBeforeInsertingAndBindsInstants() {
+        database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Live", "{}", false);
+
+        StepVerifier.create(store.save(
+                        CONVERSATION_ID,
+                        "user",
+                        AgenticMessageContent.ofText("hello"),
+                        "m1",
+                        AgenticMessageStatusEnum.COMPLETED,
+                        header))
+                .assertNext(message -> {
+                    assertThat(message.getConversationId()).isEqualTo(CONVERSATION_ID);
+                    assertThat(message.getMessageIndex()).isZero();
+                })
+                .verifyComplete();
+
+        int guardIndex = database.indexOfFirst("SELECT id FROM dc3_agentic.dc3_session");
+        int insertIndex = database.indexOfFirst("INSERT INTO dc3_agentic.dc3_message");
+        assertThat(guardIndex).isGreaterThanOrEqualTo(0).isLessThan(insertIndex);
+
+        ScriptedAgenticDatabase.Execution insert = database.firstMatching("INSERT INTO dc3_agentic.dc3_message");
+        assertThat(insert.binds().get("create_time"))
+                .as("TIMESTAMPTZ binds must carry an explicit instant, not a zone-less wall clock")
+                .isInstanceOf(Instant.class);
+        assertThat(insert.binds().get("operate_time")).isInstanceOf(Instant.class);
+    }
+
+    @Test
+    void saveFailsClosedWhenTheSessionIsMissingOrDeleted() {
+        database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Gone", "{}", true);
+
+        StepVerifier.create(store.save(
+                        CONVERSATION_ID,
+                        "user",
+                        AgenticMessageContent.ofText("hello"),
+                        "m1",
+                        AgenticMessageStatusEnum.COMPLETED,
+                        header))
+                .expectError(IllegalStateException.class)
+                .verify();
+
+        assertThat(database.countMatching("INSERT INTO dc3_agentic.dc3_message"))
+                .isZero();
+    }
+}

dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcModelConfigStoreTest.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2016-present the IoT DC3 original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package io.github.pnoker.db.agentic;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.github.pnoker.common.agentic.entity.model.ConnectivityCheckProfile;
+import io.github.pnoker.common.entity.common.RequestHeader;
+import io.github.pnoker.common.enums.ConnectivityStatusEnum;
+import io.github.pnoker.db.core.dialect.StandardR2dbcDialect;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import reactor.test.StepVerifier;
+
+/** Regression coverage for timezone-safe timestamp binds across the model config statements. */
+class R2dbcModelConfigStoreTest {
+
+    private final ScriptedAgenticDatabase database = new ScriptedAgenticDatabase();
+    private final R2dbcModelConfigStore store = new R2dbcModelConfigStore(
+            database,
+            ScriptedAgenticDatabase.passThroughOperator(),
+            new StandardR2dbcDialect("postgres", "public.dc3_schema_fingerprint"));
+
+    private RequestHeader.PrincipalHeader header;
+
+    @BeforeEach
+    void setUp() {
+        header = new RequestHeader.PrincipalHeader();
+        header.setTenantId(7L);
+        header.setPrincipalId(42L);
+        header.setPrincipalName("operator");
+    }
+
+    @Test
+    void updateCheckProfileBindsTheCheckTimeAsAnInstant() {
+        ConnectivityCheckProfile profile = new ConnectivityCheckProfile(
+                ConnectivityStatusEnum.PASS, LocalDateTime.now(ZoneOffset.UTC), 120, null, null, null);
+
+        StepVerifier.create(store.updateCheckProfile(1L, profile, header))
+                .expectNext(true)
+                .verifyComplete();
+
+        ScriptedAgenticDatabase.Execution update = database.firstMatching("last_check_time = :last_check_time");
+        assertThat(update.binds().get("last_check_time"))
+                .as("TIMESTAMPTZ binds must carry an explicit instant, not a zone-less wall clock")
+                .isInstanceOf(Instant.class);
+    }
+
+    @Test
+    void deleteBindsOperateTimeAsAnInstant() {
+        StepVerifier.create(store.delete(1L, header)).expectNext(true).verifyComplete();
+
+        ScriptedAgenticDatabase.Execution delete = database.firstMatching("SET deleted = 1");
+        assertThat(delete.binds().get("operate_time")).isInstanceOf(Instant.class);
+    }
+}

dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/R2dbcSessionStoreTest.java

@@ -0,0 +1,270 @@
+/*
+ * Copyright 2016-present the IoT DC3 original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package io.github.pnoker.db.agentic;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import io.github.pnoker.common.agentic.entity.bo.SessionBO;
+import io.github.pnoker.common.agentic.entity.model.SessionExt;
+import io.github.pnoker.common.entity.common.RequestHeader;
+import io.github.pnoker.db.core.dialect.StandardR2dbcDialect;
+import io.github.pnoker.db.core.transaction.PageTransaction;
+import java.time.Instant;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import reactor.test.StepVerifier;
+import tools.jackson.databind.ObjectMapper;
+
+/**
+ * Regression coverage for the session store concurrency contract: soft-deleted conversation ids
+ * revive instead of failing on the natural-key constraint, concurrent updates merge under the
+ * row lock instead of losing writes, the delete locks the session row before touching children,
+ * and every timestamp bind is an {@link Instant} so TIMESTAMPTZ encoding cannot drift with the
+ * database session timezone.
+ */
+class R2dbcSessionStoreTest {
+
+    private static final long TENANT_ID = 7L;
+    private static final long USER_ID = 42L;
+    private static final String CONVERSATION_ID = "conv-1";
+
+    private final ScriptedAgenticDatabase database = new ScriptedAgenticDatabase();
+    private final ObjectMapper objectMapper = new ObjectMapper();
+    private final R2dbcSessionStore store = new R2dbcSessionStore(
+            database,
+            ScriptedAgenticDatabase.passThroughOperator(),
+            Mockito.mock(PageTransaction.class),
+            objectMapper,
+            new StandardR2dbcDialect("postgres", "public.dc3_schema_fingerprint"));
+
+    private RequestHeader.PrincipalHeader header;
+
+    @BeforeEach
+    void setUp() {
+        header = new RequestHeader.PrincipalHeader();
+        header.setTenantId(TENANT_ID);
+        header.setPrincipalId(USER_ID);
+        header.setPrincipalName("operator");
+    }
+
+    @Test
+    void touchRevivesASoftDeletedConversationAsAFreshSession() {
+        database.seedSession(
+                TENANT_ID, USER_ID, CONVERSATION_ID, "Old Title", "{\"model\":\"dead\",\"temperature\":0.1}", true);
+        SessionExt incoming = new SessionExt();
+        incoming.setModel("fresh-model");
+
+        StepVerifier.create(store.touch(CONVERSATION_ID, incoming, header))
+                .assertNext(session -> {
+                    assertThat(session.getTitle()).isEqualTo("New Conversation");
+                    assertThat(session.getSessionExt().getModel()).isEqualTo("fresh-model");
+                    assertThat(session.getSessionExt().getTemperature())
+                            .as("the tombstone's dead preferences must not leak into the revived session")
+                            .isNull();
+                })
+                .verifyComplete();
+
+        ScriptedAgenticDatabase.SessionRow row = database.session(TENANT_ID, USER_ID, CONVERSATION_ID);
+        assertThat(row.deleted).as("the tombstone row must be live again").isZero();
+        assertThat(row.title).isEqualTo("New Conversation");
+        assertThat((Instant) row.createTime)
+                .as("revive restarts the conversation lifecycle")
+                .isAfter(Instant.parse("2026-01-02T00:00:00Z"));
+        assertThat(database.countMatching("INSERT INTO dc3_agentic.dc3_session"))
+                .as("revive must not attempt an insert against the unique natural key")
+                .isZero();
+    }
+
+    @Test
+    void touchMergesIntoTheLiveSessionUnderLock() {
+        database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Keep", "{\"temperature\":0.5}", false);
+        SessionExt incoming = new SessionExt();
+        incoming.setModel("m1");
+
+        StepVerifier.create(store.touch(CONVERSATION_ID, incoming, header))
+                .assertNext(session -> {
+                    assertThat(session.getSessionExt().getModel()).isEqualTo("m1");
+                    assertThat(session.getSessionExt().getTemperature()).isEqualTo(0.5);
+                    assertThat(session.getTitle()).isEqualTo("Keep");
+                })
+                .verifyComplete();
+
+        assertThat(database.firstMatching("LIMIT 1 FOR UPDATE").sql())
+                .as("the decision must read the row under FOR UPDATE without a deleted filter")
+                .doesNotContain("deleted = 0");
+    }
+
+    @Test
+    void touchRecoversWhenAConcurrentInsertWinsTheConversation() {
+        database.scriptNextSessionInsertViolation(() ->
+                database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Winner", "{\"temperature\":0.7}", false));
+        SessionExt incoming = new SessionExt();
+        incoming.setModel("mine");
+
+        StepVerifier.create(store.touch(CONVERSATION_ID, incoming, header))
+                .assertNext(session -> {
+                    assertThat(session.getSessionExt().getModel()).isEqualTo("mine");
+                    assertThat(session.getSessionExt().getTemperature())
+                            .as("the retry merges onto the concurrent winner's state")
+                            .isEqualTo(0.7);
+                })
+                .verifyComplete();
+
+        assertThat(database.session(TENANT_ID, USER_ID, CONVERSATION_ID).deleted)
+                .isZero();
+    }
+
+    @Test
+    void concurrentUpdatesMergeInsteadOfLosingTheFirstWrite() throws Exception {
+        database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Base", "{\"temperature\":0.5}", false);
+        // Park the first updater's session UPDATE mid-flight so the second updater subscribes
+        // while the first is still inside its locked read-merge-write window.
+        ScriptedAgenticDatabase.Hold firstUpdateHold = database.holdNext("UPDATE dc3_agentic.dc3_session");
+
+        SessionExt temperatureChange = new SessionExt();
+        temperatureChange.setTemperature(0.9);
+        CompletableFuture<SessionBO> first = store.update(CONVERSATION_ID, temperatureChange, "First Title", header)
+                .toFuture();
+        firstUpdateHold.awaitRouted();
+
+        SessionExt modelChange = new SessionExt();
+        modelChange.setModel("m2");
+        CompletableFuture<SessionBO> second =
+                store.update(CONVERSATION_ID, modelChange, null, header).toFuture();
+        assertThat(second.isDone())
+                .as("the second updater must wait on the row lock held by the first")
+                .isFalse();
+
+        firstUpdateHold.release();
+        SessionBO firstResult = first.get(5, TimeUnit.SECONDS);
+        SessionBO secondResult = second.get(5, TimeUnit.SECONDS);
+
+        assertThat(firstResult.getSessionExt().getTemperature()).isEqualTo(0.9);
+        assertThat(secondResult.getSessionExt().getModel()).isEqualTo("m2");
+        assertThat(secondResult.getSessionExt().getTemperature())
+                .as("the second update must merge over the first update's committed state, not lose it")
+                .isEqualTo(0.9);
+        assertThat(secondResult.getTitle())
+                .as("a blank title keeps the first updater's rename")
+                .isEqualTo("First Title");
+    }
+
+    @Test
+    void updateRunsTheLockMergeWriteSequenceInsideOneTransaction() {
+        database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Base", "{}", false);
+        StepVerifier.create(store.update(CONVERSATION_ID, null, "Renamed", header))
+                .assertNext(session -> assertThat(session.getTitle()).isEqualTo("Renamed"))
+                .verifyComplete();
+
+        assertThat(database.indexOfFirst("deleted = 0 LIMIT 1 FOR UPDATE"))
+                .as("the update path must open with a locking read of the live row")
+                .isZero();
+        assertThat(database.firstMatching("UPDATE dc3_agentic.dc3_session").binds())
+                .containsEntry("title", "Renamed");
+    }
+
+    @Test
+    void updateCompletesEmptyWhenTheSessionDoesNotExist() {
+        StepVerifier.create(store.update("missing", null, "Renamed", header)).verifyComplete();
+        assertThat(database.countMatching("UPDATE dc3_agentic.dc3_session")).isZero();
+    }
+
+    @Test
+    void deleteLocksTheLiveSessionBeforeSoftDeletingChildren() {
+        database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Doomed", "{}", false);
+        database.seedAttachment(TENANT_ID, USER_ID, CONVERSATION_ID, "/store/a.txt", false);
+        database.seedAttachment(TENANT_ID, USER_ID, CONVERSATION_ID, "/store/b.txt", false);
+
+        StepVerifier.create(store.delete(CONVERSATION_ID, header))
+                .expectNext(1L)
+                .verifyComplete();
+
+        assertThat(database.indexOfFirst("SELECT id FROM dc3_agentic.dc3_session"))
+                .as("the delete must lock the session row first so uploads serialize against it")
+                .isZero();
+        assertThat(database.session(TENANT_ID, USER_ID, CONVERSATION_ID).deleted)
+                .isOne();
+        assertThat(database.attachments())
+                .as("the delete must soft-delete the conversation's attachments")
+                .allSatisfy(row -> assertThat(row.deleted).isOne());
+        List<ScriptedAgenticDatabase.Execution> statements = database.executions();
+        assertThat(statements.get(statements.size() - 1).sql())
+                .as("the session row must be soft-deleted after the children")
+                .contains("UPDATE dc3_agentic.dc3_session SET deleted = 1");
+    }
+
+    @Test
+    void deleteIsANoOpWhenTheSessionIsAlreadyDeleted() {
+        database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Gone", "{}", true);
+
+        StepVerifier.create(store.delete(CONVERSATION_ID, header))
+                .expectNext(0L)
+                .verifyComplete();
+
+        assertThat(database.countMatching("UPDATE dc3_agentic.dc3_session")).isZero();
+        assertThat(database.countMatching("UPDATE dc3_agentic.dc3_message")).isZero();
+    }
+
+    @Test
+    void touchBindsTimestampsAsInstantsSoEncodingIsTimezoneIndependent() {
+        StepVerifier.create(store.touch(CONVERSATION_ID, null, header))
+                .expectNextCount(1)
+                .verifyComplete();
+
+        ScriptedAgenticDatabase.Execution insert = database.firstMatching("INSERT INTO dc3_agentic.dc3_session");
+        assertThat(insert.binds().get("create_time"))
+                .as("TIMESTAMPTZ binds must carry an explicit instant, not a zone-less wall clock")
+                .isInstanceOf(Instant.class);
+        assertThat(insert.binds().get("operate_time")).isInstanceOf(Instant.class);
+    }
+
+    @Test
+    void reviveAndDeleteBindsTimestampsAsInstants() {
+        database.seedSession(TENANT_ID, USER_ID, CONVERSATION_ID, "Old", "{}", true);
+        StepVerifier.create(store.touch(CONVERSATION_ID, null, header))
+                .expectNextCount(1)
+                .verifyComplete();
+        assertThat(database.firstMatching("SET deleted = 0").binds().get("create_time"))
+                .isInstanceOf(Instant.class);
+
+        StepVerifier.create(store.delete(CONVERSATION_ID, header))
+                .expectNext(1L)
+                .verifyComplete();
+        assertThat(database.firstMatching("UPDATE dc3_agentic.dc3_session SET deleted = 1")
+                        .binds()
+                        .get("operate_time"))
+                .isInstanceOf(Instant.class);
+    }
+
+    @Test
+    void getNormalizesNonUtcDriverTimesToUtcWallClock() {
+        database.seedSessionWithOffsetTimes(TENANT_ID, USER_ID, CONVERSATION_ID);
+
+        StepVerifier.create(store.get(CONVERSATION_ID, header))
+                .assertNext(session -> {
+                    assertThat(session.getCreateTime())
+                            .as("an offset driver decode must normalize to the UTC wall clock")
+                            .isEqualTo(java.time.LocalDateTime.parse("2026-03-01T04:00"));
+                    assertThat(session.getOperateTime()).isEqualTo(java.time.LocalDateTime.parse("2026-03-01T04:00"));
+                })
+                .verifyComplete();
+    }
+}

dc3-db/dc3-db-agentic/src/test/java/io/github/pnoker/db/agentic/ScriptedAgenticDatabase.java

@@ -0,0 +1,975 @@
+/*
+ * Copyright 2016-present the IoT DC3 original author or authors.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+package io.github.pnoker.db.agentic;
+
+import io.r2dbc.spi.ConnectionFactory;
+import io.r2dbc.spi.Row;
+import io.r2dbc.spi.RowMetadata;
+import java.time.Instant;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Supplier;
+import org.mockito.ArgumentMatchers;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.r2dbc.core.DatabaseClient;
+import org.springframework.r2dbc.core.FetchSpec;
+import org.springframework.r2dbc.core.RowsFetchSpec;
+import org.springframework.transaction.reactive.TransactionalOperator;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * In-memory stand-in for the agentic PostgreSQL schema, purpose-built for the store unit tests.
+ *
+ * <p>Routes the stores' explicit SQL by statement shape, keeps mutable session, attachment, and
+ * message rows, records every executed statement with its bind values, and models READ COMMITTED
+ * row locking: {@code SELECT ... FOR UPDATE} serializes per conversation (FIFO waiters that
+ * re-read row state once granted), a session-row mutation releases the lock, and a natural-key
+ * insert colliding with an existing row raises {@link DataIntegrityViolationException}. Tests can
+ * park one statement mid-flight with {@link #holdNext(String)} to force real interleavings.
+ */
+final class ScriptedAgenticDatabase implements DatabaseClient {
+
+    private final Map<String, SessionRow> sessions = new LinkedHashMap<>();
+    private final List<AttachmentRow> attachments = new ArrayList<>();
+    private final List<MessageRow> messages = new ArrayList<>();
+    private final List<ActionRow> actions = new ArrayList<>();
+    private final Map<String, LockBox> locks = new LinkedHashMap<>();
+    private final List<Execution> executions = new ArrayList<>();
+    private final ArrayDeque<Hold> holds = new ArrayDeque<>();
+    private final AtomicLong sessionIds = new AtomicLong(100);
+    private final AtomicLong attachmentIds = new AtomicLong(200);
+    private final AtomicLong messageIds = new AtomicLong(300);
+    private final AtomicLong actionIds = new AtomicLong(400);
+    private Runnable sessionInsertViolationScript;
+
+    /** One recorded statement with its bind values and typed null binds. */
+    record Execution(String sql, Map<String, Object> binds, Map<String, Class<?>> nulls) {
+
+        boolean contains(String marker) {
+            return sql.contains(marker);
+        }
+    }
+
+    /** A mutable soft-deletable {@code dc3_session} row. */
+    static final class SessionRow {
+        Long id;
+        String conversationId;
+        String title;
+        String sessionExtJson;
+        Long tenantId;
+        Long userId;
+        Long creatorId;
+        String creatorName;
+        Object createTime;
+        Long operatorId;
+        String operatorName;
+        Object operateTime;
+        int deleted;
+    }
+
+    /** A mutable soft-deletable {@code dc3_attachment} row. */
+    static final class AttachmentRow {
+        Long id;
+        String conversationId;
+        String fileName;
+        String contentType;
+        Long size;
+        String filePath;
+        Long tenantId;
+        Long userId;
+        Long creatorId;
+        String creatorName;
+        Instant createTime;
+        Long operatorId;
+        String operatorName;
+        Instant operateTime;
+        int deleted;
+    }
+
+    /** A mutable soft-deletable {@code dc3_message} row. */
+    static final class MessageRow {
+        Long id;
+        String conversationId;
+        String role;
+        String content;
+        String model;
+        Long messageIndex;
+        Byte status;
+        Long tenantId;
+        Long userId;
+        Instant createTime;
+        int deleted;
+    }
+
+    /** A mutable soft-deletable {@code dc3_action} row. */
+    static final class ActionRow {
+        Long id;
+        String actionId;
+        String conversationId;
+        String actionType;
+        String title;
+        String description;
+        String payload;
+        Byte status;
+        Instant expireTime;
+        Long tenantId;
+        Long userId;
+        Instant createTime;
+        int deleted;
+    }
+
+    /** Park the next statement whose SQL contains the marker until {@link Hold#release()}. */
+    Hold holdNext(String marker) {
+        Hold hold = new Hold(marker);
+        synchronized (holds) {
+            holds.add(hold);
+        }
+        return hold;
+    }
+
+    /** Fail the next session insert with a unique-key violation, running the script first. */
+    void scriptNextSessionInsertViolation(Runnable script) {
+        sessionInsertViolationScript = script;
+    }
+
+    List<Execution> executions() {
+        synchronized (executions) {
+            return List.copyOf(executions);
+        }
+    }
+
+    Execution firstMatching(String marker) {
+        return executions().stream()
+                .filter(execution -> execution.contains(marker))
+                .findFirst()
+                .orElseThrow(() -> new AssertionError("no statement containing: " + marker));
+    }
+
+    int countMatching(String marker) {
+        return (int) executions().stream()
+                .filter(execution -> execution.contains(marker))
+                .count();
+    }
+
+    int indexOfFirst(String marker) {
+        List<Execution> all = executions();
+        for (int index = 0; index < all.size(); index++) {
+            if (all.get(index).contains(marker)) return index;
+        }
+        return -1;
+    }
+
+    void seedSession(
+            long tenantId, long userId, String conversationId, String title, String sessionExtJson, boolean deleted) {
+        SessionRow row = new SessionRow();
+        row.id = sessionIds.incrementAndGet();
+        row.tenantId = tenantId;
+        row.userId = userId;
+        row.conversationId = conversationId;
+        row.title = title;
+        row.sessionExtJson = sessionExtJson;
+        row.creatorId = userId;
+        row.creatorName = "seeder";
+        row.operatorId = userId;
+        row.operatorName = "seeder";
+        row.createTime = Instant.parse("2026-01-01T00:00:00Z");
+        row.operateTime = row.createTime;
+        row.deleted = deleted ? 1 : 0;
+        sessions.put(key(tenantId, userId, conversationId), row);
+    }
+
+    /** Seed a session whose decoded times carry a non-UTC offset, like a driver in a non-UTC session timezone. */
+    void seedSessionWithOffsetTimes(long tenantId, long userId, String conversationId) {
+        seedSession(tenantId, userId, conversationId, "Offset", "{}", false);
+        SessionRow row = sessions.get(key(tenantId, userId, conversationId));
+        row.createTime = java.time.OffsetDateTime.parse("2026-03-01T12:00:00+08:00");
+        row.operateTime = java.time.OffsetDateTime.parse("2026-03-01T12:00:00+08:00");
+    }
+
+    void seedAttachment(long tenantId, long userId, String conversationId, String filePath, boolean deleted) {
+        AttachmentRow row = new AttachmentRow();
+        row.id = attachmentIds.incrementAndGet();
+        row.tenantId = tenantId;
+        row.userId = userId;
+        row.conversationId = conversationId;
+        row.fileName = "seed.txt";
+        row.contentType = "text/plain";
+        row.size = 1L;
+        row.filePath = filePath;
+        row.creatorId = userId;
+        row.createTime = Instant.parse("2026-01-01T00:00:00Z");
+        row.deleted = deleted ? 1 : 0;
+        attachments.add(row);
+    }
+
+    SessionRow session(long tenantId, long userId, String conversationId) {
+        return sessions.get(key(tenantId, userId, conversationId));
+    }
+
+    List<AttachmentRow> attachments() {
+        return attachments;
+    }
+
+    List<MessageRow> messages() {
+        return messages;
+    }
+
+    /** A transactional operator mock that passes publishers through unchanged. */
+    static TransactionalOperator passThroughOperator() {
+        TransactionalOperator operator = org.mockito.Mockito.mock(TransactionalOperator.class);
+        org.mockito.Mockito.when(operator.transactional(ArgumentMatchers.<Mono<Object>>any()))
+                .thenAnswer(invocation -> invocation.getArgument(0));
+        org.mockito.Mockito.when(operator.transactional(ArgumentMatchers.<Flux<Object>>any()))
+                .thenAnswer(invocation -> invocation.getArgument(0));
+        return operator;
+    }
+
+    /** A page transaction mock that passes publishers through unchanged. */
+    static io.github.pnoker.db.core.transaction.PageTransaction passThroughPageTransaction() {
+        io.github.pnoker.db.core.transaction.PageTransaction transaction =
+                org.mockito.Mockito.mock(io.github.pnoker.db.core.transaction.PageTransaction.class);
+        org.mockito.Mockito.when(transaction.transactional(ArgumentMatchers.<Mono<Object>>any()))
+                .thenAnswer(invocation -> invocation.getArgument(0));
+        return transaction;
+    }
+
+    @Override
+    public GenericExecuteSpec sql(String sql) {
+        return new ScriptedSpec(sql);
+    }
+
+    @Override
+    public GenericExecuteSpec sql(Supplier<String> sqlSupplier) {
+        return new ScriptedSpec(sqlSupplier.get());
+    }
+
+    @Override
+    public ConnectionFactory getConnectionFactory() {
+        throw new UnsupportedOperationException("not used by the store unit tests");
+    }
+
+    @Override
+    public <T> Mono<T> inConnection(java.util.function.Function<io.r2dbc.spi.Connection, Mono<T>> connectionFunction) {
+        throw new UnsupportedOperationException("not used by the store unit tests");
+    }
+
+    @Override
+    public <T> Flux<T> inConnectionMany(
+            java.util.function.Function<io.r2dbc.spi.Connection, Flux<T>> connectionFunction) {
+        throw new UnsupportedOperationException("not used by the store unit tests");
+    }
+
+    /** Parked-statement handle; the routed latch proves the statement reached the database. */
+    static final class Hold {
+        private final String marker;
+        private final CompletableFuture<Void> gate = new CompletableFuture<>();
+        private final CountDownLatch routed = new CountDownLatch(1);
+
+        private Hold(String marker) {
+            this.marker = marker;
+        }
+
+        void awaitRouted() throws InterruptedException {
+            if (!routed.await(5, java.util.concurrent.TimeUnit.SECONDS)) {
+                throw new AssertionError("held statement never routed: " + marker);
+            }
+        }
+
+        void release() {
+            gate.complete(null);
+        }
+    }
+
+    private static String key(Long tenantId, Long userId, String conversationId) {
+        return tenantId + "/" + userId + "/" + conversationId;
+    }
+
+    private static String key(Map<String, Object> binds) {
+        return key((Long) binds.get("tenant_id"), (Long) binds.get("user_id"), (String) binds.get("conversation_id"));
+    }
+
+    private static final class LockBox {
+        private boolean held;
+        private final ArrayDeque<CompletableFuture<Void>> waiters = new ArrayDeque<>();
+
+        synchronized boolean tryAcquire() {
+            if (held) return false;
+            held = true;
+            return true;
+        }
+
+        synchronized CompletableFuture<Void> enqueue() {
+            CompletableFuture<Void> granted = new CompletableFuture<>();
+            waiters.add(granted);
+            return granted;
+        }
+
+        synchronized void release() {
+            CompletableFuture<Void> next = waiters.poll();
+            if (next == null) {
+                held = false;
+            } else {
+                next.complete(null);
+            }
+        }
+    }
+
+    private Mono<Void> acquire(String lockKey) {
+        return Mono.defer(() -> {
+            LockBox box = locks.computeIfAbsent(lockKey, ignored -> new LockBox());
+            if (box.tryAcquire()) return Mono.empty();
+            return Mono.fromFuture(box.enqueue()).then();
+        });
+    }
+
+    private void release(String lockKey) {
+        LockBox box = locks.get(lockKey);
+        if (box != null) box.release();
+    }
+
+    private record Outcome(List<Map<String, Object>> rows, long rowsUpdated) {
+
+        static final Outcome EMPTY = new Outcome(List.of(), 0);
+    }
+
+    private final class ScriptedSpec implements GenericExecuteSpec {
+
+        private final String sql;
+        private final Map<String, Object> binds = new LinkedHashMap<>();
+        private final Map<String, Class<?>> nulls = new LinkedHashMap<>();
+
+        private ScriptedSpec(String sql) {
+            this.sql = sql;
+        }
+
+        @Override
+        public GenericExecuteSpec bind(int index, Object value) {
+            throw new UnsupportedOperationException("positional binds are not used");
+        }
+
+        @Override
+        public GenericExecuteSpec bindNull(int index, Class<?> type) {
+            throw new UnsupportedOperationException("positional binds are not used");
+        }
+
+        @Override
+        public GenericExecuteSpec bind(String name, Object value) {
+            binds.put(name, value);
+            return this;
+        }
+
+        @Override
+        public GenericExecuteSpec bindNull(String name, Class<?> type) {
+            nulls.put(name, type);
+            return this;
+        }
+
+        @Override
+        public GenericExecuteSpec bindValues(List<?> values) {
+            throw new UnsupportedOperationException("not used");
+        }
+
+        @Override
+        public GenericExecuteSpec bindValues(Map<String, ?> values) {
+            binds.putAll(values);
+            return this;
+        }
+
+        @Override
+        public GenericExecuteSpec bindProperties(Object source) {
+            throw new UnsupportedOperationException("not used");
+        }
+
+        @Override
+        public GenericExecuteSpec filter(org.springframework.r2dbc.core.StatementFilterFunction filter) {
+            throw new UnsupportedOperationException("not used");
+        }
+
+        @Override
+        public GenericExecuteSpec filter(
+                java.util.function.Function<? super io.r2dbc.spi.Statement, ? extends io.r2dbc.spi.Statement> filter) {
+            throw new UnsupportedOperationException("not used");
+        }
+
+        @Override
+        public <R> RowsFetchSpec<R> map(java.util.function.Function<? super io.r2dbc.spi.Readable, R> mapping) {
+            return mapped((row, ignored) -> mapping.apply(row));
+        }
+
+        @Override
+        public <R> RowsFetchSpec<R> map(java.util.function.BiFunction<Row, RowMetadata, R> mapping) {
+            return mapped(mapping);
+        }
+
+        @Override
+        public <R> RowsFetchSpec<R> mapValue(Class<R> type) {
+            throw new UnsupportedOperationException("not used");
+        }
+
+        @Override
+        public <R> RowsFetchSpec<R> mapProperties(Class<R> type) {
+            throw new UnsupportedOperationException("not used");
+        }
+
+        @Override
+        public <R> Flux<R> flatMap(
+                java.util.function.Function<io.r2dbc.spi.Result, org.reactivestreams.Publisher<R>> mapping) {
+            throw new UnsupportedOperationException("not used");
+        }
+
+        @Override
+        public FetchSpec<Map<String, Object>> fetch() {
+            return new ScriptedFetchSpec(this);
+        }
+
+        @Override
+        public Mono<Void> then() {
+            return resolveOutcome().then();
+        }
+
+        /** Resolve this statement: record it, honor any hold, then route and apply effects. */
+        Mono<Outcome> resolveOutcome() {
+            return Mono.defer(() -> {
+                        Hold hold = takeHold(sql);
+                        Mono<Outcome> body = Mono.defer(() -> route(sql, binds));
+                        if (hold == null) {
+                            return body;
+                        }
+                        return Mono.fromFuture(hold.gate).then(body);
+                    })
+                    .doOnSubscribe(subscription -> record(sql, binds, nulls));
+        }
+
+        private <R> RowsFetchSpec<R> mapped(java.util.function.BiFunction<Row, RowMetadata, R> mapping) {
+            return new RowsFetchSpec<>() {
+                @Override
+                public Mono<R> one() {
+                    return resolveOutcome()
+                            .flatMap(outcome -> outcome.rows().isEmpty()
+                                    ? Mono.<R>empty()
+                                    : Mono.just(mapping.apply(
+                                            new FakeRow(outcome.rows().get(0)), (RowMetadata) null)));
+                }
+
+                @Override
+                public Mono<R> first() {
+                    return one();
+                }
+
+                @Override
+                public Flux<R> all() {
+                    return resolveOutcome()
+                            .flatMapMany(outcome -> Flux.fromIterable(outcome.rows().stream()
+                                    .map(row -> mapping.apply(new FakeRow(row), null))
+                                    .toList()));
+                }
+            };
+        }
+
+        private Hold takeHold(String statement) {
+            synchronized (holds) {
+                for (Hold hold : holds) {
+                    if (statement.contains(hold.marker)) {
+                        holds.remove(hold);
+                        hold.routed.countDown();
+                        return hold;
+                    }
+                }
+            }
+            return null;
+        }
+
+        private void record(String statement, Map<String, Object> values, Map<String, Class<?>> typedNulls) {
+            synchronized (executions) {
+                executions.add(new Execution(statement, Map.copyOf(values), Map.copyOf(typedNulls)));
+            }
+        }
+    }
+
+    private final class ScriptedFetchSpec implements FetchSpec<Map<String, Object>> {
+
+        private final ScriptedSpec spec;
+
+        private ScriptedFetchSpec(ScriptedSpec spec) {
+            this.spec = spec;
+        }
+
+        @Override
+        public Mono<Map<String, Object>> one() {
+            return resolveRows().flatMap(rows -> rows.isEmpty() ? Mono.empty() : Mono.just(rows.get(0)));
+        }
+
+        @Override
+        public Mono<Map<String, Object>> first() {
+            return one();
+        }
+
+        @Override
+        public Flux<Map<String, Object>> all() {
+            return resolveRows().flatMapMany(Flux::fromIterable);
+        }
+
+        @Override
+        public Mono<Long> rowsUpdated() {
+            return spec.resolveOutcome().map(Outcome::rowsUpdated);
+        }
+
+        private Mono<List<Map<String, Object>>> resolveRows() {
+            return spec.resolveOutcome().map(Outcome::rows);
+        }
+    }
+
+    private Mono<Outcome> route(String sql, Map<String, Object> binds) {
+        if (sql.startsWith("SELECT id FROM dc3_agentic.dc3_session")) {
+            return lockSessionId(binds);
+        }
+        if (sql.startsWith("INSERT INTO dc3_agentic.dc3_session")) {
+            return insertSession(binds);
+        }
+        if (sql.startsWith("UPDATE dc3_agentic.dc3_session")) {
+            return updateSession(sql, binds);
+        }
+        if (sql.startsWith("SELECT id, conversation_id, title, session_ext")) {
+            return selectSession(sql, binds);
+        }
+        if (sql.startsWith("INSERT INTO dc3_agentic.dc3_attachment")) {
+            return insertAttachment(binds);
+        }
+        if (sql.startsWith("SELECT file_path FROM dc3_agentic.dc3_attachment")) {
+            return listDeletedFilePaths(binds);
+        }
+        if (sql.startsWith("SELECT id, conversation_id, file_name")) {
+            return selectAttachment(binds);
+        }
+        if (sql.startsWith("UPDATE dc3_agentic.dc3_attachment SET deleted = 1")) {
+            return softDeleteAttachments(binds);
+        }
+        if (sql.startsWith("UPDATE dc3_agentic.dc3_message SET deleted = 1")) {
+            return softDeleteMessages(binds);
+        }
+        if (sql.startsWith("UPDATE dc3_agentic.dc3_action SET deleted = 1")) {
+            return Mono.just(new Outcome(List.of(), 0));
+        }
+        if (sql.startsWith("INSERT INTO dc3_agentic.dc3_action")) {
+            return insertAction(binds);
+        }
+        if (sql.startsWith("UPDATE dc3_agentic.dc3_action")) {
+            return updateAction(binds);
+        }
+        if (sql.startsWith("INSERT INTO dc3_agentic.dc3_message")) {
+            return insertMessage(binds);
+        }
+        if (sql.startsWith("SELECT COALESCE(MAX(message_index)")) {
+            return nextMessageIndex(binds);
+        }
+        if (sql.startsWith("SELECT id, action_id, conversation_id")) {
+            return selectAction(binds);
+        }
+        if (sql.startsWith("SELECT COUNT(*) AS total")) {
+            return Mono.just(new Outcome(List.of(Map.of("total", 0L)), 0));
+        }
+        if (sql.startsWith("SELECT id, conversation_id, role, content")) {
+            return selectMessage(binds);
+        }
+        // Statements outside the tested protocol (model config updates) still execute and record
+        // their binds; they simply report one affected row and no data.
+        return Mono.just(new Outcome(List.of(), 1));
+    }
+
+    private Mono<Outcome> lockSessionId(Map<String, Object> binds) {
+        // FOR UPDATE on no matching row locks nothing, so an absent session must not queue.
+        if (liveSession(binds) == null) return Mono.just(Outcome.EMPTY);
+        String lockKey = key(binds);
+        return acquire(lockKey).then(Mono.fromCallable(() -> {
+            SessionRow row = liveSession(binds);
+            return row == null ? Outcome.EMPTY : new Outcome(List.of(Map.of("id", row.id)), 0);
+        }));
+    }
+
+    private Mono<Outcome> insertSession(Map<String, Object> binds) {
+        Runnable script = sessionInsertViolationScript;
+        if (script != null) {
+            sessionInsertViolationScript = null;
+            script.run();
+            if (sessions.containsKey(key(binds))) {
+                return Mono.error(new DataIntegrityViolationException("uq_session_tenant_user_conversation"));
+            }
+        }
+        if (sessions.containsKey(key(binds))) {
+            return Mono.error(new DataIntegrityViolationException("uq_session_tenant_user_conversation"));
+        }
+        SessionRow row = new SessionRow();
+        row.id = sessionIds.incrementAndGet();
+        row.tenantId = (Long) binds.get("tenant_id");
+        row.userId = (Long) binds.get("user_id");
+        row.conversationId = (String) binds.get("conversation_id");
+        row.title = (String) binds.get("title");
+        row.sessionExtJson = (String) binds.get("session_ext");
+        row.creatorId = (Long) binds.get("creator_id");
+        row.operatorId = (Long) binds.get("operator_id");
+        row.createTime = (Instant) binds.get("create_time");
+        row.operateTime = (Instant) binds.get("operate_time");
+        row.deleted = 0;
+        sessions.put(key(row.tenantId, row.userId, row.conversationId), row);
+        release(key(binds));
+        return Mono.just(new Outcome(List.of(), 1));
+    }
+
+    private Mono<Outcome> updateSession(String sql, Map<String, Object> binds) {
+        if (sql.contains("SET deleted = 1")) {
+            SessionRow row = liveSession(binds);
+            if (row == null) return Mono.just(Outcome.EMPTY);
+            row.deleted = 1;
+            row.operatorId = (Long) binds.get("operator_id");
+            row.operateTime = (Instant) binds.get("operate_time");
+            release(key(row.tenantId, row.userId, row.conversationId));
+            return Mono.just(new Outcome(List.of(), 1));
+        }
+        SessionRow row = sessionById((Long) binds.get("id"));
+        // The plain update's WHERE clause guards on deleted = 0; the revive's SET clause clears
+        // the tombstone flag, so distinguish by the statement tail, not by a bare substring.
+        if (row == null || (sql.endsWith("deleted = 0") && row.deleted != 0)) {
+            return Mono.just(Outcome.EMPTY);
+        }
+        if (sql.contains("SET deleted = 0")) {
+            row.deleted = 0;
+            row.title = (String) binds.get("title");
+            row.sessionExtJson = (String) binds.get("session_ext");
+            row.creatorId = (Long) binds.get("creator_id");
+            row.creatorName = (String) binds.get("creator_name");
+            row.createTime = (Instant) binds.get("create_time");
+            row.operatorId = (Long) binds.get("operator_id");
+            row.operateTime = (Instant) binds.get("operate_time");
+        } else {
+            row.title = (String) binds.get("title");
+            row.sessionExtJson = (String) binds.get("session_ext");
+            row.operatorId = (Long) binds.get("operator_id");
+            row.operateTime = (Instant) binds.get("operate_time");
+        }
+        release(key(row.tenantId, row.userId, row.conversationId));
+        return Mono.just(new Outcome(List.of(), 1));
+    }
+
+    private Mono<Outcome> selectSession(String sql, Map<String, Object> binds) {
+        boolean liveOnly = sql.contains("AND deleted = 0");
+        if (sql.contains("FOR UPDATE")) {
+            // FOR UPDATE on no matching row locks nothing, so an absent row must not queue.
+            if ((liveOnly ? liveSession(binds) : anySession(binds)) == null) {
+                return Mono.just(Outcome.EMPTY);
+            }
+            String lockKey = key(binds);
+            return acquire(lockKey).then(Mono.fromCallable(() -> {
+                SessionRow row = liveOnly ? liveSession(binds) : anySession(binds);
+                return row == null ? Outcome.EMPTY : new Outcome(List.of(sessionMap(row)), 0);
+            }));
+        }
+        SessionRow row = liveSession(binds);
+        return Mono.just(row == null ? Outcome.EMPTY : new Outcome(List.of(sessionMap(row)), 0));
+    }
+
+    private Mono<Outcome> insertAttachment(Map<String, Object> binds) {
+        AttachmentRow row = new AttachmentRow();
+        row.id = attachmentIds.incrementAndGet();
+        row.tenantId = (Long) binds.get("tenant_id");
+        row.userId = (Long) binds.get("user_id");
+        row.conversationId = (String) binds.get("conversation_id");
+        row.fileName = (String) binds.get("file_name");
+        row.contentType = (String) binds.get("content_type");
+        row.size = (Long) binds.get("size");
+        row.filePath = (String) binds.get("file_path");
+        row.creatorId = (Long) binds.get("creator_id");
+        row.creatorName = (String) binds.get("creator_name");
+        row.createTime = (Instant) binds.get("create_time");
+        row.operatorId = (Long) binds.get("operator_id");
+        row.operateTime = (Instant) binds.get("operate_time");
+        row.deleted = 0;
+        attachments.add(row);
+        release(key(binds));
+        return Mono.just(new Outcome(List.of(), 1));
+    }
+
+    private Mono<Outcome> listDeletedFilePaths(Map<String, Object> binds) {
+        List<Map<String, Object>> rows = attachments.stream()
+                .filter(row -> matchesConversation(row, binds) && row.deleted == 1)
+                .map(row -> Map.<String, Object>of("file_path", row.filePath))
+                .toList();
+        return Mono.just(new Outcome(rows, 0));
+    }
+
+    private Mono<Outcome> selectAttachment(Map<String, Object> binds) {
+        if (binds.containsKey("file_path")) {
+            AttachmentRow found = attachments.stream()
+                    .filter(row -> row.filePath.equals(binds.get("file_path"))
+                            && row.tenantId.equals(binds.get("tenant_id"))
+                            && row.userId.equals(binds.get("user_id"))
+                            && row.deleted == 0)
+                    .findFirst()
+                    .orElse(null);
+            return Mono.just(found == null ? Outcome.EMPTY : new Outcome(List.of(attachmentMap(found)), 0));
+        }
+        List<Map<String, Object>> rows = attachments.stream()
+                .filter(row -> matchesConversation(row, binds) && row.deleted == 0)
+                .map(this::attachmentMap)
+                .toList();
+        return Mono.just(new Outcome(rows, 0));
+    }
+
+    private Mono<Outcome> softDeleteAttachments(Map<String, Object> binds) {
+        long affected = attachments.stream()
+                .filter(row -> matchesConversation(row, binds) && row.deleted == 0)
+                .peek(row -> row.deleted = 1)
+                .count();
+        return Mono.just(new Outcome(List.of(), affected));
+    }
+
+    private Mono<Outcome> softDeleteMessages(Map<String, Object> binds) {
+        long affected = messages.stream()
+                .filter(row -> matchesConversation(row, binds) && row.deleted == 0)
+                .peek(row -> row.deleted = 1)
+                .count();
+        return Mono.just(new Outcome(List.of(), affected));
+    }
+
+    private Mono<Outcome> insertMessage(Map<String, Object> binds) {
+        MessageRow row = new MessageRow();
+        row.id = messageIds.incrementAndGet();
+        row.tenantId = (Long) binds.get("tenant_id");
+        row.userId = (Long) binds.get("user_id");
+        row.conversationId = (String) binds.get("conversation_id");
+        row.role = (String) binds.get("role");
+        row.content = (String) binds.get("content");
+        row.model = (String) binds.get("model");
+        row.messageIndex = (Long) binds.get("message_index");
+        row.status = (Byte) binds.get("status");
+        row.createTime = (Instant) binds.get("create_time");
+        row.deleted = 0;
+        messages.add(row);
+        release(key(binds));
+        return Mono.just(new Outcome(List.of(), 1));
+    }
+
+    private Mono<Outcome> nextMessageIndex(Map<String, Object> binds) {
+        long next = messages.stream()
+                        .filter(row -> matchesConversation(row, binds) && row.deleted == 0)
+                        .mapToLong(row -> row.messageIndex)
+                        .max()
+                        .orElse(-1L)
+                + 1;
+        return Mono.just(new Outcome(List.of(Map.of("next_index", next)), 0));
+    }
+
+    private Mono<Outcome> selectMessage(Map<String, Object> binds) {
+        MessageRow found = messages.stream()
+                .filter(row -> matchesConversation(row, binds)
+                        && row.deleted == 0
+                        && row.messageIndex.equals(binds.get("message_index")))
+                .findFirst()
+                .orElse(null);
+        if (found == null) return Mono.just(Outcome.EMPTY);
+        Map<String, Object> row = new LinkedHashMap<>();
+        row.put("id", found.id);
+        row.put("conversation_id", found.conversationId);
+        row.put("role", found.role);
+        row.put("content", found.content);
+        row.put("model", found.model);
+        row.put("message_index", found.messageIndex);
+        row.put("status", found.status);
+        row.put("tenant_id", found.tenantId);
+        row.put("user_id", found.userId);
+        row.put("remark", "");
+        row.put("creator_id", found.tenantId);
+        row.put("creator_name", "");
+        row.put("create_time", found.createTime);
+        row.put("operator_id", found.tenantId);
+        row.put("operator_name", "");
+        row.put("operate_time", found.createTime);
+        return Mono.just(new Outcome(List.of(row), 0));
+    }
+
+    private SessionRow liveSession(Map<String, Object> binds) {
+        SessionRow row = anySession(binds);
+        return row != null && row.deleted == 0 ? row : null;
+    }
+
+    private Mono<Outcome> insertAction(Map<String, Object> binds) {
+        ActionRow row = new ActionRow();
+        row.id = actionIds.incrementAndGet();
+        row.actionId = (String) binds.get("action_id");
+        row.conversationId = (String) binds.get("conversation_id");
+        row.actionType = (String) binds.get("action_type");
+        row.title = (String) binds.get("title");
+        row.description = (String) binds.get("description");
+        row.payload = (String) binds.get("payload");
+        row.status = (Byte) binds.get("status");
+        row.tenantId = (Long) binds.get("tenant_id");
+        row.userId = (Long) binds.get("user_id");
+        row.createTime = (Instant) binds.get("create_time");
+        row.expireTime = (Instant) binds.get("expire_time");
+        row.deleted = 0;
+        actions.add(row);
+        return Mono.just(new Outcome(List.of(), 1));
+    }
+
+    private Mono<Outcome> updateAction(Map<String, Object> binds) {
+        ActionRow row = actions.stream()
+                .filter(candidate -> candidate.actionId.equals(binds.get("action_id"))
+                        && candidate.tenantId.equals(binds.get("tenant_id"))
+                        && candidate.userId.equals(binds.get("user_id"))
+                        && candidate.deleted == 0)
+                .findFirst()
+                .orElse(null);
+        if (row == null) return Mono.just(Outcome.EMPTY);
+        if (binds.containsKey("status")) row.status = (Byte) binds.get("status");
+        return Mono.just(new Outcome(List.of(), 1));
+    }
+
+    private Mono<Outcome> selectAction(Map<String, Object> binds) {
+        ActionRow found = actions.stream()
+                .filter(candidate -> candidate.actionId.equals(binds.get("action_id"))
+                        && candidate.tenantId.equals(binds.get("tenant_id"))
+                        && candidate.userId.equals(binds.get("user_id"))
+                        && candidate.deleted == 0)
+                .findFirst()
+                .orElse(null);
+        if (found == null) return Mono.just(Outcome.EMPTY);
+        Map<String, Object> row = new LinkedHashMap<>();
+        row.put("id", found.id);
+        row.put("action_id", found.actionId);
+        row.put("conversation_id", found.conversationId);
+        row.put("action_type", found.actionType);
+        row.put("title", found.title);
+        row.put("description", found.description);
+        row.put("payload", found.payload);
+        row.put("status", found.status);
+        row.put("expire_time", found.expireTime);
+        row.put("tenant_id", found.tenantId);
+        row.put("user_id", found.userId);
+        row.put("remark", "");
+        row.put("creator_id", found.tenantId);
+        row.put("creator_name", "");
+        row.put("create_time", found.createTime);
+        row.put("operator_id", found.tenantId);
+        row.put("operator_name", "");
+        row.put("operate_time", found.createTime);
+        return Mono.just(new Outcome(List.of(row), 0));
+    }
+
+    private SessionRow anySession(Map<String, Object> binds) {
+        return sessions.get(key(binds));
+    }
+
+    private SessionRow sessionById(Long id) {
+        return sessions.values().stream()
+                .filter(row -> row.id.equals(id))
+                .findFirst()
+                .orElse(null);
+    }
+
+    private boolean matchesConversation(AttachmentRow row, Map<String, Object> binds) {
+        return row.conversationId.equals(binds.get("conversation_id"))
+                && row.tenantId.equals(binds.get("tenant_id"))
+                && row.userId.equals(binds.get("user_id"));
+    }
+
+    private boolean matchesConversation(MessageRow row, Map<String, Object> binds) {
+        return row.conversationId.equals(binds.get("conversation_id"))
+                && row.tenantId.equals(binds.get("tenant_id"))
+                && row.userId.equals(binds.get("user_id"));
+    }
+
+    private Map<String, Object> sessionMap(SessionRow row) {
+        Map<String, Object> values = new LinkedHashMap<>();
+        values.put("id", row.id);
+        values.put("conversation_id", row.conversationId);
+        values.put("title", row.title);
+        values.put("session_ext", row.sessionExtJson);
+        values.put("tenant_id", row.tenantId);
+        values.put("user_id", row.userId);
+        values.put("remark", "");
+        values.put("creator_id", row.creatorId);
+        values.put("creator_name", row.creatorName);
+        values.put("create_time", row.createTime);
+        values.put("operator_id", row.operatorId);
+        values.put("operator_name", row.operatorName);
+        values.put("operate_time", row.operateTime);
+        values.put("deleted", row.deleted);
+        return values;
+    }
+
+    private Map<String, Object> attachmentMap(AttachmentRow row) {
+        Map<String, Object> values = new LinkedHashMap<>();
+        values.put("id", row.id);
+        values.put("conversation_id", row.conversationId);
+        values.put("file_name", row.fileName);
+        values.put("content_type", row.contentType);
+        values.put("size", row.size);
+        values.put("file_path", row.filePath);
+        values.put("tenant_id", row.tenantId);
+        values.put("user_id", row.userId);
+        values.put("remark", "");
+        values.put("creator_id", row.creatorId);
+        values.put("creator_name", row.creatorName);
+        values.put("create_time", row.createTime);
+        values.put("operator_id", row.operatorId);
+        values.put("operator_name", row.operatorName);
+        values.put("operate_time", row.operateTime);
+        values.put("deleted", row.deleted);
+        return values;
+    }
+
+    /** Minimal map-backed {@link Row} with the numeric coercions the store mappers rely on. */
+    private static final class FakeRow implements Row {
+
+        private final Map<String, Object> values;
+
+        private FakeRow(Map<String, Object> values) {
+            this.values = values;
+        }
+
+        @Override
+        public <T> T get(String name, Class<T> type) {
+            Object value = values.get(name);
+            if (value == null) return null;
+            if (type.isInstance(value)) return type.cast(value);
+            if (value instanceof Number number) {
+                if (type == Long.class) return type.cast(number.longValue());
+                if (type == Integer.class) return type.cast(number.intValue());
+                if (type == Double.class) return type.cast(number.doubleValue());
+                if (type == Number.class) return type.cast(number);
+            }
+            if (type == String.class) return type.cast(value.toString());
+            throw new IllegalStateException("unsupported column type: " + name);
+        }
+
+        @Override
+        public <T> T get(int index, Class<T> type) {
+            throw new UnsupportedOperationException("not used");
+        }
+
+        @Override
+        public RowMetadata getMetadata() {
+            throw new UnsupportedOperationException("not used");
+        }
+    }
+}

dc3/bin/check_r2dbc_migration.py

@@ -115,6 +115,11 @@ def is_expected_startup_block(path: Path, category: str, match: re.Match[str]) -
     # bounded elastic scheduler so neither bridge ever runs on the event loop.
     if path.name == "SystemTool.java" and match.group(0).startswith(".block"):
         return True
+    # Attachment file cleanup deletes files with blocking NIO; the cleaner owns the
+    # bounded-elastic offload for every unlink so callers (session delete, upload
+    # rollback) can never run file deletion on an event loop.
+    if path.name == "AttachmentFileCleaner.java" and match.group(0).startswith("Schedulers.boundedElastic"):
+        return True
     return (
         path.name == "OpenAiCompatibleAgenticRuntime.java"
         and match.group(0).startswith("Schedulers.boundedElastic")