docs(harness): annotate tool domain classes and add tool chain learning notes

- Annotate 43 tool domain classes (contract/projection/boundary/store/adapter/mysql)
- Add tool registration and execution chain learning note
- Add tool call chain runtime journey note (model decision to observation)
This commit is contained in:
zhuyongxin
2026-08-04 18:37:17 +08:00
parent 7844bcea40
commit ff0752a16c
45 changed files with 1021 additions and 9 deletions
@@ -15,13 +15,20 @@ import com.superbiz.agent.harness.tool.mysql.MysqlSqlValidator;
import java.util.Objects;
/** Validates and runs the logical MySQL Tool through the canonical boundary. */
/**
* MySQL 逻辑工具的接线员:反序列化请求 → SQL 沙箱校验(fail-closed)→
* 把只读执行器(executor)和投影器(projector)组装进 ToolBoundary 统一门禁。
* 安全/参数异常映射为 INVALID_REQUEST(不泄露内部细节)。
*/
public final class MysqlToolAdapter {
private final ToolBoundary boundary;
private final ObjectMapper objectMapper;
/** SQL 沙箱:白名单表列 + fail-closed 策略。 */
private final MysqlSqlValidator validator;
/** 只读执行器(JDBC 只读连接 + 超时 + 行数 + 取消)。 */
private final MysqlReadOnlyExecutor executor;
/** 投影器:raw 行 → 有界脱敏契约。 */
private final MysqlResultProjector projector;
public MysqlToolAdapter(ToolBoundary boundary, ObjectMapper objectMapper,
@@ -34,12 +41,19 @@ public final class MysqlToolAdapter {
this.projector = Objects.requireNonNull(projector, "projector must not be null");
}
/**
* 执行入口:解析请求 → SQL 沙箱校验(生成执行计划)→ 组装 executor/projector
* 交给 ToolBoundary。任何安全/参数异常统一映射 INVALID_REQUEST。
*/
public ToolBoundaryResult execute(RunContext context, ToolCallRequestEnvelope envelope) {
try {
MysqlToolRequest request = objectMapper.readValue(envelope.requestJson(), MysqlToolRequest.class);
// 沙箱校验:表列白名单 + fail-closed 策略 → 规范化执行计划
MysqlQueryPlan plan = validator.validate(request);
return boundary.execute(context, envelope,
// executor:只读执行器返回 raw 行 JSON
ignored -> objectMapper.writeValueAsString(executor.execute(plan, context)),
// projector:有界脱敏投影(用该数据源的限制)
raw -> projector.project(request, envelope.toolCallId(), raw, plan.dataSource().limits()));
} catch (MysqlSecurityException | IllegalArgumentException e) {
return ToolBoundaryResult.error(envelope == null ? null : envelope.toolCallId(),
@@ -17,9 +17,14 @@ import java.time.Instant;
import java.time.format.DateTimeFormatter;
import java.util.Objects;
/** Bridges logical query-log requests and the existing Mock tool through ToolBoundary. */
/**
* 逻辑日志请求与既有 Mock 工具的接线员:
* 反序列化请求 → 校验 topic/query/lookback → 构造查询范围 scope →
* 把 legacy executor 和投影器组装进 ToolBoundary 统一门禁执行。
*/
public final class QueryLogsToolAdapter {
/** legacy backend 执行端口(region + 旧主题 + 关键词 + 条数)。 */
@FunctionalInterface
public interface LegacyExecutor {
String execute(String region, String legacyTopic, String query, Integer limit) throws Exception;
@@ -58,25 +63,33 @@ public final class QueryLogsToolAdapter {
this.legacyLimit = legacyLimit;
}
/**
* 执行入口:解析请求 → 校验 → 构造 scope → 组装 executor/projector 交给 ToolBoundary。
* 业务参数非法返回 INVALID_REQUEST(不抛异常打断 ReAct)。
*/
public ToolBoundaryResult execute(RunContext context, ToolCallRequestEnvelope envelope) {
try {
QueryLogsRequest request = objectMapper.readValue(envelope.requestJson(), QueryLogsRequest.class);
if (request.topic() == null || request.query() == null || request.query().isBlank()) {
return ToolBoundaryResult.error(envelope.toolCallId(), ToolBoundaryErrorCode.INVALID_REQUEST);
}
// 回看窗口:缺省 30 分钟,上限 24 小时
int lookback = request.lookbackMinutes() == null
? DEFAULT_LOOKBACK_MINUTES : request.lookbackMinutes();
if (lookback <= 0 || lookback > 24 * 60) {
return ToolBoundaryResult.error(envelope.toolCallId(), ToolBoundaryErrorCode.INVALID_REQUEST);
}
Instant end = clock.instant();
// 实际查询范围(审计/公开 scope 用)
LogQueryScope scope = new LogQueryScope(
request.topic(), request.query(),
DateTimeFormatter.ISO_INSTANT.format(end.minus(Duration.ofMinutes(lookback))),
DateTimeFormatter.ISO_INSTANT.format(end));
String legacyTopic = legacyTopic(request.topic());
return boundary.execute(context, envelope,
// executor:调用 legacy 日志 backend
ignored -> legacyExecutor.execute(region, legacyTopic, request.query(), legacyLimit),
// projector:投影成冻结契约(含 scope)
raw -> projector.project(request, envelope.toolCallId(), scope, raw));
} catch (Exception e) {
return ToolBoundaryResult.error(envelope == null ? null : envelope.toolCallId(),
@@ -84,6 +97,7 @@ public final class QueryLogsToolAdapter {
}
}
/** 逻辑主题 → legacy 日志主题名映射。 */
private static String legacyTopic(LogTopic topic) {
return switch (topic) {
case APPLICATION -> "application-logs";
@@ -4,6 +4,10 @@ import com.superbiz.agent.harness.contract.EvidenceStatus;
import java.util.Objects;
/**
* Projector 的产出:有界 agent_result 文本 + 客观 evidence status。
* ToolBoundary 只接受 FOUND/NO_EVIDENCE(ERROR 走错误路径,不产生投影结果)。
*/
public record ProjectedToolResult(String agentResult, EvidenceStatus evidenceStatus) {
public ProjectedToolResult {
@@ -11,6 +15,7 @@ public record ProjectedToolResult(String agentResult, EvidenceStatus evidenceSta
throw new IllegalArgumentException("agentResult must not be blank");
}
Objects.requireNonNull(evidenceStatus, "evidenceStatus must not be null");
// 投影结果只能是合法证据语义(空不空),错误状态不从这里出
if (evidenceStatus != EvidenceStatus.EVIDENCE_FOUND
&& evidenceStatus != EvidenceStatus.NO_EVIDENCE) {
throw new IllegalArgumentException("projected result must be evidence or no-evidence");
@@ -2,11 +2,22 @@ package com.superbiz.agent.harness.tool.boundary;
import com.fasterxml.jackson.annotation.JsonProperty;
/**
* 一次 Tool 执行的内部信封:同时证明 Run、调用 ID、工具、参数、授权意图和只读意图。
* 由 Adapter 在调 ToolBoundary 前构造(HarnessEvidenceTools 的 bridge 固定 authorized/readOnly=true)。
* 与模型侧 progress Envelope(previous_observation + input)不同:这是边界内部信封。
*/
public record ToolCallRequestEnvelope(
/** 所属 Run。 */
@JsonProperty("run_id") String runId,
/** 本次调用 ID(canonical key 的一部分)。 */
@JsonProperty("tool_call_id") String toolCallId,
/** 工具名。 */
@JsonProperty("tool_name") String toolName,
/** 业务请求 JSON(纯业务参数,无协议字段)。 */
@JsonProperty("request") String requestJson,
/** 是否授权(bridge 恒为 true;策略拒绝走 UNAUTHORIZED)。 */
@JsonProperty("authorized") boolean authorized,
/** 是否只读意图(诊断 Tool 必须只读,否则 NOT_READ_ONLY)。 */
@JsonProperty("read_only") boolean readOnly) {
}
@@ -1,6 +1,11 @@
package com.superbiz.agent.harness.tool.boundary;
/**
* 具体 backend 的 raw 执行函数端口(函数式):输入业务请求 JSON,输出原始响应文本。
* ToolBoundary 不依赖具体 backend,只认这个端口——Adapter 把各自 backend 接进来。
*/
@FunctionalInterface
public interface ToolExecutor {
/** 执行 backend,返回原始响应(非 null);失败抛异常由边界映射错误码。 */
String execute(String requestJson) throws Exception;
}
@@ -1,6 +1,12 @@
package com.superbiz.agent.harness.tool.boundary;
/**
* raw → 有界 agent 契约的投影端口(函数式):输入原始响应,输出投影结果
* (有界 agent_result + 客观 evidence status)。每个 Tool 一个实现
* (Rag/QueryLogs/Mysql ResultProjector),ToolBoundary 通过它解耦投影逻辑。
*/
@FunctionalInterface
public interface ToolResultProjector {
/** 投影 raw:必须返回非 null 的有界结果;失败抛异常由边界映射 PROJECTION_ERROR。 */
ProjectedToolResult project(String rawResponse) throws Exception;
}
@@ -1,21 +1,33 @@
package com.superbiz.agent.harness.tool.contract;
/**
* 三个证据 Tool 的「名字 + 模型可见描述」的单一事实源(冻结契约)。
*
* <p>所有地方(拦截器判重、Normalizer 分派、Projector 分派、注册表)都引用这里的
* 常量而不是字符串字面量——避免工具名拼错导致跨层漂移。
*/
public final class AgentToolContracts {
/** 知识库检索工具名。 */
public static final String LOOKUP_KNOWLEDGE = "lookup_knowledge";
/** 日志查询工具名。 */
public static final String QUERY_LOGS = "query_logs";
/** MySQL 只读查询工具名。 */
public static final String QUERY_MYSQL = "query_mysql";
/** 模型可见的 RAG 工具描述:稳定背景知识,不用于实时日志/指标。 */
public static final String LOOKUP_KNOWLEDGE_DESCRIPTION =
"查询内部知识库中的文档、接口说明、错误码和排障手册。"
+ "适用于稳定背景知识,不用于查询实时日志、指标或数据库状态。"
+ "输入 query:需要查询的问题或关键词。";
/** 模型可见的日志工具描述:应用错误/慢查询/系统事件,不用于指标或表。 */
public static final String QUERY_LOGS_DESCRIPTION =
"查询指定逻辑日志主题在时间窗口内与目标相关的日志证据。"
+ "适用于应用错误、慢查询和系统事件,不用于查询指标或数据库表。"
+ "输入 topic、query、lookback_minutes。";
/** 模型可见的 MySQL 工具描述:授权数据源的参数化只读 SELECT,禁止发现表结构/写操作。 */
public static final String QUERY_MYSQL_DESCRIPTION =
"在授权的逻辑数据源上执行参数化只读查询,获取业务数据库事实。"
+ "只用于已知库表字段的 SELECT,不用于发现表结构或执行写操作。"
@@ -2,6 +2,9 @@ package com.superbiz.agent.harness.tool.contract;
import com.fasterxml.jackson.annotation.JsonProperty;
/**
* 单条日志事件(冻结契约):时间/级别/服务/消息四元组。
*/
public record LogEvent(
@JsonProperty("timestamp") String timestamp,
@JsonProperty("level") String level,
@@ -2,11 +2,17 @@ package com.superbiz.agent.harness.tool.contract;
import com.fasterxml.jackson.annotation.JsonProperty;
/**
* 日志模式聚合(冻结契约):把相似事件压缩成一条「模式」,
* 供模型快速了解事件全貌而不用读每条原始事件。
*/
public record LogPattern(
/** 该模式出现次数。 */
@JsonProperty("count") long count,
@JsonProperty("first_seen") String firstSeen,
@JsonProperty("last_seen") String lastSeen,
@JsonProperty("level") String level,
@JsonProperty("service") String service,
/** 示例事件(有界)。 */
@JsonProperty("example") String example) {
}
@@ -2,6 +2,10 @@ package com.superbiz.agent.harness.tool.contract;
import com.fasterxml.jackson.annotation.JsonProperty;
/**
* 实际日志查询范围(冻结契约):反映「这次到底查了什么」,
* 供审计、ProgressProjector 的公开 scope、以及重复检测参考。
*/
public record LogQueryScope(
@JsonProperty("topic") LogTopic topic,
@JsonProperty("query") String query,
@@ -1,5 +1,9 @@
package com.superbiz.agent.harness.tool.contract;
/**
* 日志来源类型(冻结契约)。当前只有 MOCK(演示/评估环境),
* 后续可扩展 ES/ClickHouse 等真实来源。
*/
public enum LogSourceKind {
MOCK
}
@@ -1,7 +1,13 @@
package com.superbiz.agent.harness.tool.contract;
/**
* 逻辑日志主题(冻结契约):模型只能在这三个主题内查询,不能自由指定任意来源。
*/
public enum LogTopic {
/** 应用错误/业务日志。 */
APPLICATION,
/** 数据库慢查询。 */
DATABASE_SLOW_QUERY,
/** 系统事件。 */
SYSTEM_EVENTS
}
@@ -3,7 +3,13 @@ package com.superbiz.agent.harness.tool.contract;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.superbiz.agent.harness.progress.PreviousObservation;
/**
* 模型发起的 MySQL 只读查询 Tool 调用 Envelope(Agent-facing 冻结契约):
* 协议字段 previous_observation + 业务输入 input。
*/
public record MysqlToolCall(
/** 对上一轮观察的评价(首次调用可为 null)。 */
@JsonProperty("previous_observation") PreviousObservation previousObservation,
/** 业务输入:data_source + sql + params。 */
@JsonProperty("input") MysqlToolRequest input) {
}
@@ -4,9 +4,16 @@ import com.fasterxml.jackson.annotation.JsonProperty;
import java.util.List;
/**
* MySQL 只读查询业务输入(冻结契约):授权逻辑数据源 + 参数化 SQL + 绑定参数。
* 判重指纹 = {data_source, sql, params}。
*/
public record MysqlToolRequest(
/** 授权数据源名(不是任意 JDBC URL)。 */
@JsonProperty("data_source") String dataSource,
/** 参数化 SQL(只允许 SELECT,沙箱校验)。 */
@JsonProperty("sql") String sql,
/** 绑定参数(防注入)。 */
@JsonProperty("params") List<Object> params) {
public MysqlToolRequest {
@@ -6,12 +6,21 @@ import com.superbiz.agent.harness.contract.EvidenceStatus;
import java.util.List;
import java.util.Map;
/**
* MySQL 查询结果(冻结契约):Projector 投影后的有界结果。
* 列名 + 行数据(嵌套不可变),供模型看事实、Harness 看 returned_count/truncated。
*/
public record MysqlToolResult(
/** 证据语义:rows 空不空(客观判定)。 */
@JsonProperty("evidence_status") EvidenceStatus evidenceStatus,
@JsonProperty("tool_call_id") String toolCallId,
/** 列名列表(有界)。 */
@JsonProperty("columns") List<String> columns,
/** 行数据(有界、截断过,每行不可变 Map)。 */
@JsonProperty("rows") List<Map<String, Object>> rows,
/** 返回的行数。 */
@JsonProperty("returned_count") int returnedCount,
/** 是否因预算截断。 */
@JsonProperty("truncated") boolean truncated) {
public MysqlToolResult {
@@ -2,8 +2,15 @@ package com.superbiz.agent.harness.tool.contract;
import com.fasterxml.jackson.annotation.JsonProperty;
/**
* 日志查询业务输入(冻结契约):逻辑主题 + 关键词 + 回看窗口。
* 判重指纹 = {topic, query, lookback_minutes(缺省 30)}。
*/
public record QueryLogsRequest(
/** 逻辑日志主题(APPLICATION / DATABASE_SLOW_QUERY / SYSTEM_EVENTS)。 */
@JsonProperty("topic") LogTopic topic,
/** 查询关键词。 */
@JsonProperty("query") String query,
/** 回看分钟数;null 时 Normalizer 用默认 30。 */
@JsonProperty("lookback_minutes") Integer lookbackMinutes) {
}
@@ -3,7 +3,13 @@ package com.superbiz.agent.harness.tool.contract;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.superbiz.agent.harness.progress.PreviousObservation;
/**
* 模型发起的日志查询 Tool 调用 Envelope(Agent-facing 冻结契约):
* 协议字段 previous_observation + 业务输入 input。
*/
public record QueryLogsToolCall(
/** 对上一轮观察的评价(首次调用可为 null)。 */
@JsonProperty("previous_observation") PreviousObservation previousObservation,
/** 业务输入:topic + query + lookback_minutes。 */
@JsonProperty("input") QueryLogsRequest input) {
}
@@ -5,15 +5,28 @@ import com.superbiz.agent.harness.contract.EvidenceStatus;
import java.util.List;
/**
* 日志查询结果(冻结契约):Projector 投影后的有界结果。
* 包含聚合 pattern(压缩)与原始 event(有界、截断过),
* 供模型看事件、Harness 看 match_count/truncated。
*/
public record QueryLogsToolResult(
/** 证据语义:events 空不空(客观判定)。 */
@JsonProperty("evidence_status") EvidenceStatus evidenceStatus,
@JsonProperty("tool_call_id") String toolCallId,
/** 日志来源类型(当前 MOCK)。 */
@JsonProperty("source_kind") LogSourceKind sourceKind,
/** 实际查询范围(topic/query/时间窗)。 */
@JsonProperty("scope") LogQueryScope scope,
/** 匹配总数(可能大于 returned_count)。 */
@JsonProperty("match_count") long matchCount,
/** 实际返回的事件条数。 */
@JsonProperty("returned_count") int returnedCount,
/** 压缩后的模式聚合(有界)。 */
@JsonProperty("patterns") List<LogPattern> patterns,
/** 事件明细(有界、截断过)。 */
@JsonProperty("events") List<LogEvent> events,
/** 是否因预算截断。 */
@JsonProperty("truncated") boolean truncated) {
public QueryLogsToolResult {
@@ -1,7 +1,15 @@
package com.superbiz.agent.harness.tool.contract;
/**
* RAG 检索相关度(冻结契约):Projector 客观计算,供 Harness/Release 参考。
* 注意:REFERENCE(一般相关)不能自动映射为信息 NO_GAIN——可能仍排除一个假设,
* 需要模型结合诊断上下文判断。
*/
public enum RagRelevanceLevel {
/** 精确匹配。 */
PRECISE,
/** 高度相关。 */
HIGHLY_RELEVANT,
/** 一般相关(参考级)。 */
REFERENCE
}
@@ -3,7 +3,14 @@ package com.superbiz.agent.harness.tool.contract;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.superbiz.agent.harness.progress.PreviousObservation;
/**
* 模型发起的 RAG 工具调用 Envelope(Agent-facing 冻结契约):
* 协议字段 previous_observation + 业务输入 input。
* 拦截器解析后剥离 previous_observation,只把 input 传给业务执行。
*/
public record RagToolCall(
/** 对上一轮观察的评价(首次调用可为 null)。 */
@JsonProperty("previous_observation") PreviousObservation previousObservation,
/** 业务输入:检索 query。 */
@JsonProperty("input") RagToolRequest input) {
}
@@ -2,6 +2,10 @@ package com.superbiz.agent.harness.tool.contract;
import com.fasterxml.jackson.annotation.JsonProperty;
/**
* RAG 业务输入(冻结契约):单个检索查询关键词。
*/
public record RagToolRequest(
/** 检索 query(判重指纹的一部分)。 */
@JsonProperty("query") String query) {
}
@@ -6,20 +6,31 @@ import com.superbiz.agent.harness.contract.EvidenceStatus;
import java.util.List;
/**
* RAG 工具结果(冻结契约):Projector 投影后的有界结果。
* 是模型看到的 observation 与 canonical 记录的 agent_result 的统一形态。
*/
public record RagToolResult(
/** 证据语义:证据数组空不空(客观判定)。 */
@JsonProperty("evidence_status") EvidenceStatus evidenceStatus,
@JsonProperty("tool_call_id") String toolCallId,
/** 原始检索 query。 */
@JsonProperty("query") String query,
/** 投影后的证据列表(有界、截断过)。 */
@JsonProperty("evidence") List<RagEvidence> evidence,
/** 返回的证据条数。 */
@JsonProperty("returned_count") int returnedCount,
/** 检索相关度(仅 EVIDENCE_FOUND 时有值;NO_EVIDENCE 时为 null 不序列化)。 */
@JsonProperty("relevance_level") @JsonInclude(JsonInclude.Include.NON_NULL)
RagRelevanceLevel relevanceLevel,
/** 是否因预算截断。 */
@JsonProperty("truncated") boolean truncated) {
public RagToolResult {
evidence = ToolContractCollections.immutable(evidence);
}
/** 便捷构造:无相关度(NO_EVIDENCE / 内部使用)。 */
public RagToolResult(EvidenceStatus evidenceStatus,
String toolCallId,
String query,
@@ -5,15 +5,21 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
/**
* 契约对象的不可变集合工具(包私有):所有 Result 的列表字段统一用这里保证不可变,
* 防止投影层/消费方意外修改冻结契约。
*/
final class ToolContractCollections {
private ToolContractCollections() {
}
/** 列表不可变拷贝(null → 空列表)。 */
static <T> List<T> immutable(List<T> values) {
return values == null ? List.of() : List.copyOf(values);
}
/** 行集合不可变拷贝:每行 Map 也做深拷贝(嵌套不可变)。 */
static List<Map<String, Object>> immutableRows(List<Map<String, Object>> rows) {
if (rows == null) {
return List.of();
@@ -23,6 +29,7 @@ final class ToolContractCollections {
.toList();
}
/** 单行不可变拷贝(null → 空 Map)。 */
private static Map<String, Object> immutableRow(Map<String, Object> row) {
if (row == null) {
return Map.of();
@@ -20,11 +20,15 @@ import java.util.Map;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicReference;
/** JDBC implementation with read-only, timeout, row and cancellation controls. */
/**
* JDBC 只读执行器:连接强制只读 + 查询超时 + 行数上限 + Run 取消联动。
* 是 MysqlReadOnlyExecutor 的唯一实现——沙箱的「执行侧」防线。
*/
public final class JdbcMysqlReadOnlyExecutor implements MysqlReadOnlyExecutor {
private static final Logger log = LoggerFactory.getLogger(JdbcMysqlReadOnlyExecutor.class);
/** 逻辑数据源 id → 真实 DataSource 映射(配置时注入)。 */
private final Map<String, DataSource> dataSources;
private final Clock clock;
@@ -41,13 +45,16 @@ public final class JdbcMysqlReadOnlyExecutor implements MysqlReadOnlyExecutor {
}
MysqlToolLimits limits = plan.dataSource().limits();
try (Connection connection = dataSource.getConnection()) {
// 强制只读连接(双保险:Validator 语义层 + JDBC 连接层)
connection.setReadOnly(true);
try (PreparedStatement statement = connection.prepareStatement(
plan.normalizedSql(), ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)) {
statement.setQueryTimeout(limits.queryTimeoutSeconds());
// 多取一行用于检测截断
statement.setMaxRows(limits.maxRows() + 1);
bind(statement, plan.params());
// 注册取消回调:Run 取消时同步 cancel 正在执行的语句
AtomicReference<Statement> statementRef = new AtomicReference<>(statement);
context.cancellation().onCancel(ignored -> cancel(statementRef.get()));
checkRun(context);
@@ -61,6 +68,7 @@ public final class JdbcMysqlReadOnlyExecutor implements MysqlReadOnlyExecutor {
java.util.ArrayList<Map<String, Object>> rows = new java.util.ArrayList<>();
boolean truncated = false;
while (resultSet.next()) {
// 每行前检查 Run 终态/取消/超时
checkRun(context);
if (rows.size() >= limits.maxRows()) {
truncated = true;
@@ -74,6 +82,7 @@ public final class JdbcMysqlReadOnlyExecutor implements MysqlReadOnlyExecutor {
truncated |= cell.truncated();
}
rows.add(row);
// 结果字节预算:超限移除最后一行并标记截断
if (estimatedBytes(rows) > limits.maxResultBytes()) {
rows.remove(rows.size() - 1);
truncated = true;
@@ -86,6 +95,7 @@ public final class JdbcMysqlReadOnlyExecutor implements MysqlReadOnlyExecutor {
}
}
} catch (MysqlSecurityException e) {
// 安全异常原样穿出(Adapter 映射稳定错误码)
throw e;
} catch (SQLException e) {
log.debug("MySQL read-only execution failed: sqlState={}", e.getSQLState());
@@ -93,18 +103,21 @@ public final class JdbcMysqlReadOnlyExecutor implements MysqlReadOnlyExecutor {
}
}
/** 绑定参数(PreparedStatement 参数化,防注入)。 */
private static void bind(PreparedStatement statement, List<Object> params) throws SQLException {
for (int i = 0; i < params.size(); i++) {
statement.setObject(i + 1, params.get(i));
}
}
/** 执行前/每行检查:Run 已取消或过 deadline 则中止(与 core 终态联动)。 */
private void checkRun(RunContext context) throws SQLException {
if (context.cancellation().isCancelled() || !clock.instant().isBefore(context.deadline())) {
throw new SQLException("run cancelled or deadline exceeded");
}
}
/** 取消正在执行的语句(Run 取消回调)。 */
private static void cancel(Statement statement) {
if (statement == null) {
return;
@@ -116,6 +129,9 @@ public final class JdbcMysqlReadOnlyExecutor implements MysqlReadOnlyExecutor {
}
}
/**
* 单元格 JSON 安全化:数字/布尔原样;byte[] 转 Base64;字符串截断到 maxCellChars。
*/
private static CellValue jsonSafe(Object value, int maxCellChars) {
if (value == null || value instanceof Number || value instanceof Boolean) {
return new CellValue(value, false);
@@ -131,6 +147,7 @@ public final class JdbcMysqlReadOnlyExecutor implements MysqlReadOnlyExecutor {
: new CellValue(text.substring(0, maxCellChars), true);
}
/** 行集预估字节数(结果预算用)。 */
private static int estimatedBytes(List<Map<String, Object>> rows) {
return rows.toString().getBytes(StandardCharsets.UTF_8).length;
}
@@ -7,11 +7,18 @@ import java.util.Objects;
import java.util.Set;
import java.util.TreeSet;
/** Logical datasource metadata and exact schema/table/column authorization. */
/**
* 逻辑数据源元数据 + 精确的 schema/表/列授权白名单:
* 模型只能查询白名单内的表与列——这是 MySQL 只读沙箱的「访问边界」。
*/
public record MysqlDataSourceDefinition(
/** 逻辑数据源 id(模型用这个,不暴露真实 JDBC)。 */
String id,
/** 默认 schema(必须出现在白名单里)。 */
String defaultSchema,
/** 授权白名单:schema → table → 允许的列集合。 */
Map<String, Map<String, Set<String>>> allowedSchemas,
/** 该数据源的查询限制。 */
MysqlToolLimits limits) {
public MysqlDataSourceDefinition {
@@ -19,6 +26,7 @@ public record MysqlDataSourceDefinition(
requireText(defaultSchema, "defaultSchema");
Objects.requireNonNull(allowedSchemas, "allowedSchemas must not be null");
Objects.requireNonNull(limits, "limits must not be null");
// 深拷贝白名单(列集合 TreeSet 排序保证确定性),防止外部修改
Map<String, Map<String, Set<String>>> schemas = new LinkedHashMap<>();
allowedSchemas.forEach((schema, tables) -> {
requireText(schema, "schema");
@@ -35,10 +43,12 @@ public record MysqlDataSourceDefinition(
}
}
/** 表是否被授权。 */
public boolean allowsTable(String schema, String table) {
return allowedSchemas.containsKey(schema) && allowedSchemas.get(schema).containsKey(table);
}
/** 列是否被授权。 */
public boolean allowsColumn(String schema, String table, String column) {
return allowsTable(schema, table) && allowedSchemas.get(schema).get(table).contains(column);
}
@@ -4,10 +4,18 @@ import com.superbiz.agent.harness.tool.contract.MysqlToolRequest;
import java.util.List;
/**
* 一次 MySQL 查询的执行计划:请求 + 解析出的逻辑数据源 + 规范化 SQL + 绑定参数。
* 由 MysqlToolAdapter 在 SQL 校验后构造,交给只读执行器执行。
*/
public record MysqlQueryPlan(
/** 模型原始请求(data_source/sql/params)。 */
MysqlToolRequest request,
/** 解析出的授权数据源定义(含 schema/表/列白名单与限制)。 */
MysqlDataSourceDefinition dataSource,
/** 校验并规范化后的 SQL(单条、无尾分号、禁注释等)。 */
String normalizedSql,
/** 绑定参数(防注入)。 */
List<Object> params) {
public MysqlQueryPlan {
@@ -5,7 +5,10 @@ import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.Map;
/** Harness-only raw query result; never returned directly to an Agent. */
/**
* Harness-only raw 查询结果(不直接给 Agent):列 + 行 + 是否截断。
* 之后由 MysqlResultProjector 投影成冻结的 MysqlToolResult 才对外。
*/
public record MysqlRawResult(
List<String> columns,
List<Map<String, Object>> rows,
@@ -2,7 +2,12 @@ package com.superbiz.agent.harness.tool.mysql;
import com.superbiz.agent.harness.core.RunContext;
/**
* MySQL 只读执行端口(函数式):输入已校验的查询计划 + Run 上下文,输出 raw 结果。
* JdbcMysqlReadOnlyExecutor 是唯一实现(只读连接 + 超时 + 行数 + 取消控制)。
*/
@FunctionalInterface
public interface MysqlReadOnlyExecutor {
/** 执行只读查询,返回 Harness-only raw 结果;失败抛异常(安全/超时/SQL)。 */
MysqlRawResult execute(MysqlQueryPlan plan, RunContext context) throws Exception;
}
@@ -14,7 +14,16 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
/** Projects raw JDBC rows into the bounded Agent-facing MySQL contract. */
/**
* 把 raw JDBC 行投影成有界、脱敏的 Agent 可见 MySQL 契约。
*
* <p>关键职责:
* <ul>
* <li>脱敏:列名含 password/token/secret/api_key 等敏感 token 时单元格置为 [REDACTED];</li>
* <li>有界:行数(maxRows)+ 单元格字符(maxCellChars)+ 总字节(maxResultBytes)三重截断;</li>
* <li>客观证据语义:rows 空不空 → NO_EVIDENCE / EVIDENCE_FOUND。</li>
* </ul>
*/
public final class MysqlResultProjector {
private static final List<String> SENSITIVE_TOKENS = List.of(
@@ -37,6 +46,12 @@ public final class MysqlResultProjector {
return project(request, toolCallId, rawResponse, limits);
}
/**
* 主入口:raw JDBC 行 JSON → 冻结的 MysqlToolResult。
*
* <p>流程:校验 columns/rows 结构 → 列名去重 → 逐行逐列投影(敏感列脱敏 + 字符截断)
* → 行数/字节截断 → 判定 evidence status → fitBudget 总字节兜底。
*/
public ProjectedToolResult project(MysqlToolRequest request, String toolCallId,
String rawResponse, MysqlToolLimits projectionLimits) throws Exception {
if (request == null || toolCallId == null || toolCallId.isBlank()) {
@@ -49,6 +64,7 @@ public final class MysqlResultProjector {
}
List<String> columns = new ArrayList<>();
java.util.LinkedHashSet<String> uniqueColumns = new java.util.LinkedHashSet<>();
// 列名必须非空且唯一(避免歧义投影)
root.path("columns").forEach(node -> {
String column = node.asText();
if (column.isBlank() || !uniqueColumns.add(column)) {
@@ -59,6 +75,7 @@ public final class MysqlResultProjector {
List<Map<String, Object>> rows = new ArrayList<>();
boolean truncated = root.path("truncated").asBoolean(false);
for (JsonNode rowNode : root.path("rows")) {
// 行数上限:超出置 truncated 并停止
if (rows.size() >= projectionLimits.maxRows()) {
truncated = true;
break;
@@ -71,12 +88,14 @@ public final class MysqlResultProjector {
truncated |= cell.truncated();
}
rows.add(row);
// 结果字节预算:超限移除最后一行并标记截断
if (utf8Bytes(rows.toString()) > projectionLimits.maxResultBytes()) {
rows.remove(rows.size() - 1);
truncated = true;
break;
}
}
// 客观证据语义:行空不空
MysqlToolResult result = new MysqlToolResult(
rows.isEmpty() ? EvidenceStatus.NO_EVIDENCE : EvidenceStatus.EVIDENCE_FOUND,
toolCallId, columns, rows, rows.size(), truncated);
@@ -84,6 +103,10 @@ public final class MysqlResultProjector {
return new ProjectedToolResult(objectMapper.writeValueAsString(result), result.evidenceStatus());
}
/**
* 总字节兜底:超过 maxResultBytes 时逐行裁掉尾部;裁空则诚实降级为 NO_EVIDENCE;
* 仍超限则抛异常(fail closed)。
*/
private MysqlToolResult fitBudget(MysqlToolResult result, int maxResultBytes) throws Exception {
MysqlToolResult current = result;
while (utf8Bytes(objectMapper.writeValueAsString(current)) > maxResultBytes
@@ -100,11 +123,16 @@ public final class MysqlResultProjector {
return current;
}
/**
* 单单元格投影:null 原样;敏感列 → [REDACTED](含脱敏标记);
* 数字/布尔原样;字符串截断到 maxCellChars。
*/
private CellProjection projectCell(String column, JsonNode value, int maxCellChars) {
if (value == null || value.isNull()) {
return new CellProjection(null, false);
}
if (isSensitive(column)) {
// 敏感列(password/token/secret 等):绝不把真实值给 Agent
return new CellProjection("[REDACTED]", true);
}
if (value.isNumber()) {
@@ -119,6 +147,7 @@ public final class MysqlResultProjector {
: new CellProjection(text.substring(0, maxCellChars), true);
}
/** 列名是否含敏感 token(password/passwd/token/secret/api_key/apikey/credential)。 */
private static boolean isSensitive(String column) {
String normalized = column == null ? "" : column.toLowerCase(Locale.ROOT);
return SENSITIVE_TOKENS.stream().anyMatch(normalized::contains);
@@ -1,5 +1,9 @@
package com.superbiz.agent.harness.tool.mysql;
/**
* MySQL 沙箱安全异常:触发只读红线/未授权访问/危险 SQL 时抛出,
* 由 Adapter 映射为稳定的错误码(而非泄露内部细节)。
*/
public final class MysqlSecurityException extends RuntimeException {
public MysqlSecurityException(String message) {
@@ -41,7 +41,18 @@ import java.util.Map;
import java.util.Objects;
import java.util.Set;
/** Fail-closed SQL policy for the Agent-facing MySQL Tool. */
/**
* MySQL 工具的 fail-closed SQL 策略(沙箱的「语义层」防线):
* 用 JSqlParser 解析 AST,逐一拒绝所有不安全形态。
*
* <p>禁止:非 SELECT / 多条语句 / WITH / 子查询 / 通配符投影(*)/
* 窗口函数 / CASE / EXISTS / 分层查询 / 字面量(必须参数化)/ 未授权表/列 /
* 锁读 / 复杂子句(OFFSET/FETCH/TOP 等)。
*
* <p>允许:白名单内表与列的 INNER/LEFT JOIN、聚合函数
* (COUNT/SUM/AVG/MIN/MAX)、占位符参数——且占位符数量必须与 params 匹配。
* 任何解析/校验异常统一转 MysqlSecurityException(fail closed,不泄露细节)。
*/
public final class MysqlSqlValidator {
private static final Set<String> ALLOWED_FUNCTIONS = Set.of("COUNT", "SUM", "AVG", "MIN", "MAX");
@@ -53,11 +64,15 @@ public final class MysqlSqlValidator {
this.dataSources = Map.copyOf(dataSources);
}
/**
* 校验并生成执行计划。全流程 fail-closed:任何一步不满足直接抛 MysqlSecurityException。
*/
public MysqlQueryPlan validate(MysqlToolRequest request) {
if (request == null || request.dataSource() == null || request.dataSource().isBlank()
|| request.sql() == null || request.sql().isBlank()) {
throw new MysqlSecurityException("data_source and sql are required");
}
// 数据源必须存在于授权映射
MysqlDataSourceDefinition dataSource = dataSources.get(request.dataSource());
if (dataSource == null) {
throw new MysqlSecurityException("unknown logical data source");
@@ -66,6 +81,7 @@ public final class MysqlSqlValidator {
throw new MysqlSecurityException("SQL exceeds policy length");
}
try {
// 必须恰好一条语句,且是 SELECT
Statements statements = CCJSqlParserUtil.parseStatements(request.sql());
if (statements.getStatements() == null || statements.getStatements().size() != 1) {
throw new MysqlSecurityException("exactly one SQL statement is required");
@@ -78,6 +94,7 @@ public final class MysqlSqlValidator {
throw new MysqlSecurityException("WITH is not allowed");
}
SelectBody body = select.getSelectBody();
// 只允许普通 PlainSelect(无集合操作/值语句)
if (!(body instanceof PlainSelect plainSelect)
|| body instanceof SetOperationList
|| body instanceof ValuesStatement) {
@@ -97,10 +114,12 @@ public final class MysqlSqlValidator {
throw new MysqlSecurityException("unsupported SELECT clause");
}
// 表注册:FROM 必须是白名单内的实体表(禁子查询/非表来源),重复别名拒绝
Map<String, TableRef> tables = new LinkedHashMap<>();
registerTable(plainSelect.getFromItem(), dataSource, tables);
List<Join> joins = plainSelect.getJoins() == null ? List.of() : plainSelect.getJoins();
for (Join join : joins) {
// 只允许 INNER/LEFT JOIN(禁 CROSS/RIGHT/FULL/OUTER)
if (join.isCross() || join.isRight() || join.isFull() || join.isOuter()
|| (!join.isInner() && !join.isLeft())) {
throw new MysqlSecurityException("only INNER/LEFT JOIN is allowed");
@@ -117,6 +136,7 @@ public final class MysqlSqlValidator {
}
}
// 投影必须显式(禁 * / t.*),所有表达式逐节点校验
if (plainSelect.getSelectItems() == null || plainSelect.getSelectItems().isEmpty()) {
throw new MysqlSecurityException("projection must be explicit");
}
@@ -129,6 +149,7 @@ public final class MysqlSqlValidator {
}
validateExpression(expressionItem.getExpression(), tables, dataSource);
}
// WHERE/HAVING/GROUP BY/ORDER BY 全表达式校验
validateExpression(plainSelect.getWhere(), tables, dataSource);
validateExpression(plainSelect.getHaving(), tables, dataSource);
if (plainSelect.getGroupBy() != null) {
@@ -140,6 +161,7 @@ public final class MysqlSqlValidator {
plainSelect.getOrderByElements().forEach(order ->
validateExpression(order.getExpression(), tables, dataSource));
}
// 占位符数量必须与 params 匹配(防止参数错位/少传)
int placeholders = countPlaceholders(plainSelect);
int provided = request.params() == null ? 0 : request.params().size();
if (placeholders != provided) {
@@ -148,12 +170,15 @@ public final class MysqlSqlValidator {
return new MysqlQueryPlan(request, dataSource, statement.toString(),
request.params() == null ? List.of() : request.params());
} catch (MysqlSecurityException e) {
// 业务规则违规:原样穿出(已分类)
throw e;
} catch (Exception e) {
// 解析/其他异常:统一 fail closed(不泄露内部细节)
throw new MysqlSecurityException("SQL cannot be safely validated", e);
}
}
/** 注册 FROM 表:必须是白名单内实体表,别名唯一。 */
private void registerTable(FromItem item, MysqlDataSourceDefinition dataSource,
Map<String, TableRef> tables) {
if (!(item instanceof Table table)) {
@@ -174,6 +199,7 @@ public final class MysqlSqlValidator {
}
}
/** 表达式逐节点校验:列白名单 / 函数白名单 / 禁字面量、子查询、通配符、窗口函数等。 */
private void validateExpression(Expression expression, Map<String, TableRef> tables,
MysqlDataSourceDefinition dataSource) {
if (expression == null) {
@@ -282,6 +308,10 @@ public final class MysqlSqlValidator {
});
}
/**
* 列校验:限定表时查该表列白名单;未限定时必须在已注册表中唯一匹配
* (否则视为歧义或未授权)。
*/
private void validateColumn(Column column, Map<String, TableRef> tables,
MysqlDataSourceDefinition dataSource) {
String name = column.getColumnName();
@@ -290,12 +320,14 @@ public final class MysqlSqlValidator {
}
Table table = column.getTable();
if (table != null && table.getName() != null && !table.getName().isBlank()) {
// 限定表:必须在白名单内
TableRef ref = tables.get(table.getName().toLowerCase(Locale.ROOT));
if (ref == null || !dataSource.allowsColumn(ref.schema(), ref.table(), name)) {
throw new MysqlSecurityException("column is not allowlisted");
}
return;
}
// 未限定表:必须在已注册表中恰好一个白名单命中
List<TableRef> matches = tables.values().stream()
.filter(ref -> dataSource.allowsColumn(ref.schema(), ref.table(), name))
.toList();
@@ -304,6 +336,10 @@ public final class MysqlSqlValidator {
}
}
/**
* 统计 AST 中的占位符数(必须与 params 数量一致):
* 只在 AST 接受后按节点计数,引号内的问号不会被误计。
*/
private static int countPlaceholders(PlainSelect plainSelect) {
// Parser assigns JdbcParameter nodes; use the canonical SQL token count only after
// the AST has been accepted, so quoted question marks are not counted.
@@ -337,6 +373,7 @@ public final class MysqlSqlValidator {
return counter.count;
}
/** 占位符计数 visitor:统计 JdbcParameter;子查询继续拒绝。 */
private static final class PlaceholderCounter extends ExpressionVisitorAdapter {
private int count;
@@ -351,6 +388,7 @@ public final class MysqlSqlValidator {
}
}
/** 已注册表引用(schema + 表名),用于列白名单校验。 */
private record TableRef(String schema, String table) {
}
}
@@ -1,9 +1,17 @@
package com.superbiz.agent.harness.tool.mysql;
/**
* MySQL 工具限制:行数 / 单元格字符 / 结果字节 / 查询超时。
* 防止超大结果进 Agent 上下文,防长时间占用连接。
*/
public record MysqlToolLimits(
/** 最大返回行数。 */
int maxRows,
/** 单单元格最大字符数(超长截断)。 */
int maxCellChars,
/** 结果总字节上限。 */
int maxResultBytes,
/** 查询超时秒数。 */
int queryTimeoutSeconds) {
public MysqlToolLimits {
@@ -12,6 +20,7 @@ public record MysqlToolLimits(
}
}
/** 默认:100 行 / 2000 字单元 / 64KB 总字节 / 5 秒超时。 */
public static MysqlToolLimits defaults() {
return new MysqlToolLimits(100, 2_000, 64 * 1024, 5);
}
@@ -19,7 +19,17 @@ import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;
/** Projects legacy Mock log JSON into the frozen query_logs contract. */
/**
* 把 legacy Mock 日志 JSON 投影成冻结的 query_logs 契约。
*
* <p>关键职责:
* <ul>
* <li>脱敏:sanitize 抹掉消息里的密码/token/主机/Pod/IP/端口等敏感信息,不进入 Agent 上下文;</li>
* <li>有界:条数截断(maxEvents)+ 均匀采样 + 总字节兜底(fitBudget);</li>
* <li>聚合:按「级别+服务+规范化消息」压缩成 PatternAccumulator(模型看全貌不读每条);</li>
* <li>客观证据语义:allEvents 空不空 → NO_EVIDENCE / EVIDENCE_FOUND。</li>
* </ul>
*/
public final class QueryLogsResultProjector {
private static final Pattern SECRET = Pattern.compile(
@@ -41,6 +51,12 @@ public final class QueryLogsResultProjector {
this.limits = limits;
}
/**
* 主入口:legacy 日志 raw JSON → 冻结的 query_logs 契约。
*
* <p>流程:校验 raw(success 标记)→ 逐条脱敏+聚合 → 均匀采样截断 →
* 生成 patterns 聚合 → 判定 evidence status → fitBudget 总字节兜底。
*/
public ProjectedToolResult project(QueryLogsRequest request, String toolCallId,
LogQueryScope scope, String rawResponse) throws Exception {
if (request == null || toolCallId == null || toolCallId.isBlank() || scope == null) {
@@ -60,6 +76,7 @@ public final class QueryLogsResultProjector {
List<LogEvent> allEvents = new ArrayList<>();
Map<String, PatternAccumulator> aggregates = new LinkedHashMap<>();
for (JsonNode log : logs) {
// 逐条:字段截断 + 消息脱敏(敏感信息不进 Agent 上下文)
String timestamp = bounded(text(log, "timestamp"), limits.maxMessageChars());
String level = bounded(text(log, "level"), 32);
String service = bounded(text(log, "service"), 128);
@@ -70,13 +87,16 @@ public final class QueryLogsResultProjector {
}
String example = message;
allEvents.add(new LogEvent(nullable(timestamp), nullable(level), nullable(service), message));
// 聚合键:级别+服务+规范化消息(数字→<n>)
String patternKey = level + "\u0000" + service + "\u0000" + normalizePattern(message);
aggregates.computeIfAbsent(patternKey, ignored -> new PatternAccumulator(level, service, example))
.add(timestamp);
}
// 均匀采样截断(避免只保留头部事件,牺牲时间分布)
List<LogEvent> events = sample(allEvents, limits.maxEvents());
truncated |= events.size() < allEvents.size();
// 模式聚合按次数降序 + 示例升序,限制条数
List<LogPattern> patterns = aggregates.values().stream()
.sorted(Comparator.comparingLong(PatternAccumulator::count).reversed()
.thenComparing(PatternAccumulator::example))
@@ -86,6 +106,7 @@ public final class QueryLogsResultProjector {
truncated |= aggregates.size() > patterns.size();
long matchCount = root.path("total").canConvertToLong() ? root.path("total").asLong() : allEvents.size();
// 客观证据语义:事件空不空
QueryLogsToolResult result = new QueryLogsToolResult(
allEvents.isEmpty() ? EvidenceStatus.NO_EVIDENCE : EvidenceStatus.EVIDENCE_FOUND,
toolCallId,
@@ -100,6 +121,10 @@ public final class QueryLogsResultProjector {
return new ProjectedToolResult(objectMapper.writeValueAsString(result), result.evidenceStatus());
}
/**
* 总字节兜底:超过 maxAgentUtf8Bytes 时先裁事件尾部、再裁模式尾部;
* 仍超限则抛异常(fail closed)。
*/
private QueryLogsToolResult fitBudget(QueryLogsToolResult result) throws Exception {
QueryLogsToolResult current = result;
while (bytes(objectMapper.writeValueAsString(current)) > limits.maxAgentUtf8Bytes()
@@ -121,6 +146,9 @@ public final class QueryLogsResultProjector {
return current;
}
/**
* 均匀采样:超过 max 时按时间等距取 max 条(保留时间分布,而非只留头部)。
*/
private static List<LogEvent> sample(List<LogEvent> events, int max) {
if (events.size() <= max) {
return List.copyOf(events);
@@ -133,6 +161,11 @@ public final class QueryLogsResultProjector {
return sampled;
}
/**
* 消息脱敏(敏感信息不进 Agent 上下文/审计):
* 密码/token/secret/api-key → REDACTED;Pod/主机/端口/PID/IP → REDACTED_*;
* SQL 字符串字面量 → '[REDACTED_LITERAL]';去掉 Java 堆栈尾部。
*/
private static String sanitize(String message) {
String value = SECRET.matcher(message).replaceAll("$1=[REDACTED]");
value = POD.matcher(value).replaceAll("[REDACTED_POD]");
@@ -143,6 +176,7 @@ public final class QueryLogsResultProjector {
return STACK_SUFFIX.matcher(value).replaceAll("").trim();
}
/** 模式归一化:数字(含小数)→ &lt;n&gt;,压缩空白——让相似消息聚合到同一模式。 */
private static String normalizePattern(String message) {
return message.replaceAll("\\b\\d+(?:\\.\\d+)?\\b", "<n>")
.replaceAll("\\s+", " ").trim();
@@ -168,6 +202,7 @@ public final class QueryLogsResultProjector {
return value.getBytes(StandardCharsets.UTF_8).length;
}
/** 单个模式聚合器:同键(级别+服务+归一化消息)事件累加 count,记录首末时间。 */
private static final class PatternAccumulator {
private final String level;
private final String service;
@@ -182,6 +217,7 @@ public final class QueryLogsResultProjector {
this.example = example;
}
/** 累加一条事件:count++ 并更新首末时间窗。 */
private PatternAccumulator add(String timestamp) {
count++;
if (firstSeen == null || timestamp.compareTo(firstSeen) < 0) {
@@ -201,6 +237,7 @@ public final class QueryLogsResultProjector {
return example;
}
/** 转成冻结契约 LogPattern。 */
private LogPattern toPattern() {
return new LogPattern(count, firstSeen, lastSeen, nullable(level), nullable(service), example);
}
@@ -32,6 +32,12 @@ public final class RagResultProjector {
this.limits = limits;
}
/**
* 主入口:legacy raw JSON → 冻结的 Agent 可见 RAG 契约。
*
* <p>流程:解析 evidenceBlocks → 按 chunk 级身份去重 → 截断摘录/条数 →
* 判定 evidence status(空不空)→ 计算相关度 → fitBudget 总字节兜底。
*/
public ProjectedToolResult project(RagToolRequest request, String toolCallId,
String rawResponse) throws Exception {
if (request == null || toolCallId == null || toolCallId.isBlank()) {
@@ -52,10 +58,12 @@ public final class RagResultProjector {
int ordinal = 0;
for (JsonNode block : blocks) {
ordinal++;
// 条数上限:超出置 truncated 并停止
if (evidence.size() >= limits.maxEvidence()) {
truncated = true;
break;
}
// 摘录取 content(回退 excerpt),空则跳过该块
String excerpt = text(block, "content");
if (excerpt.isBlank()) {
excerpt = text(block, "excerpt");
@@ -66,6 +74,7 @@ public final class RagResultProjector {
String source = text(block, "source");
String title = text(block, "title");
// Chunk-scoped identity first; do not collapse on source alone.
// 证据身份优先级:evidenceKey → document_id → docId#chunk-idx → legacy 序号
String documentId = firstPresent(
text(block, "evidenceKey"),
text(block, "evidence_key"),
@@ -75,6 +84,7 @@ public final class RagResultProjector {
text(block, "chunkIndex"), text(block, "chunk_index")),
"legacy-evidence-" + ordinal);
if (!evidenceIds.add(documentId)) {
// 重复 chunk:跳过但标记 truncated(被去重)
truncated = true;
continue;
}
@@ -92,6 +102,7 @@ public final class RagResultProjector {
}
}
// 客观证据语义:证据数组空不空(不需要模型判断)
EvidenceStatus status = evidence.isEmpty()
? EvidenceStatus.NO_EVIDENCE
: EvidenceStatus.EVIDENCE_FOUND;
@@ -103,6 +114,7 @@ public final class RagResultProjector {
return new ProjectedToolResult(objectMapper.writeValueAsString(result), result.evidenceStatus());
}
/** 由 docId + chunkIndex 组合 chunk 级证据身份(两者都缺则返回 null)。 */
private static String composeChunkId(String docId, String docIdAlt, String chunkIndex, String chunkIndexAlt) {
String id = firstPresentOrNull(docId, docIdAlt);
String idx = firstPresentOrNull(chunkIndex, chunkIndexAlt);
@@ -112,11 +124,13 @@ public final class RagResultProjector {
return id + "#chunk-" + idx;
}
/** 取第一个非空值,全空回退 "unknown-document"。 */
private static String firstPresent(String... values) {
String found = firstPresentOrNull(values);
return found == null ? "unknown-document" : found;
}
/** 取第一个非空值,全空返回 null。 */
private static String firstPresentOrNull(String... values) {
if (values == null) {
return null;
@@ -129,6 +143,10 @@ public final class RagResultProjector {
return null;
}
/**
* 总字节兜底:投影结果超过 maxAgentUtf8Bytes 时逐条裁掉尾部证据
* (裁空则诚实降级为 NO_EVIDENCE);仍超限则抛异常(fail closed)。
*/
private RagToolResult fitBudget(RagToolResult result, boolean truncated) throws Exception {
RagToolResult current = result;
while (bytes(objectMapper.writeValueAsString(current)) > limits.maxAgentUtf8Bytes()
@@ -155,6 +173,7 @@ public final class RagResultProjector {
return value == null || value.isBlank() ? null : value;
}
/** 从 raw 读取相关度(relevance_level / relevanceLevel),无法解析返回 null。 */
private static RagRelevanceLevel relevanceLevel(JsonNode root) {
String value = text(root, "relevance_level");
if (value.isBlank()) {
@@ -170,6 +189,7 @@ public final class RagResultProjector {
}
}
/** 截断到 max 字符(null 视为空串)。 */
private static String bounded(String value, int max) {
if (value == null) {
return "";
@@ -177,6 +197,7 @@ public final class RagResultProjector {
return value.length() <= max ? value : value.substring(0, max);
}
/** UTF-8 字节数(投影总预算用)。 */
private static int bytes(String value) {
return value.getBytes(StandardCharsets.UTF_8).length;
}
@@ -1,13 +1,23 @@
package com.superbiz.agent.harness.tool.projection;
/** Bounds applied to Agent-facing projections. */
/**
* 投影层的有界限制:raw → agent 契约投影时的全部硬上限。
* 保证模型看到的任何结果都有界(数量/长度/字节),防止超大响应进入上下文。
*/
public record ToolProjectionLimits(
/** RAG 最大证据条数。 */
int maxEvidence,
/** RAG 单条摘录最大字符数。 */
int maxExcerptChars,
/** 日志最大模式聚合条数。 */
int maxPatterns,
/** 日志最大事件条数。 */
int maxEvents,
/** 日志单条消息最大字符数。 */
int maxMessageChars,
/** 查询关键词最大字符数。 */
int maxQueryChars,
/** 投影结果最大 UTF-8 字节数(总预算)。 */
int maxAgentUtf8Bytes) {
public ToolProjectionLimits {
@@ -18,6 +28,7 @@ public record ToolProjectionLimits(
}
}
/** 默认限制:8 条证据 / 1200 字摘录 / 12 模式 / 30 事件 / 1000 字消息 / 500 字查询 / 16KB 总字节。 */
public static ToolProjectionLimits defaults() {
return new ToolProjectionLimits(8, 1200, 12, 30, 1000, 500, 16_384);
}
@@ -4,9 +4,16 @@ import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Objects;
/**
* canonical 记录的大小与 TTL 限制:执行门禁(ToolBoundary)和 Redis 实现共用。
* 三条校验分别对应执行链的 request+raw、agent_result、序列化后的整条记录。
*/
public record CanonicalInvocationLimits(
/** 记录 TTL(过期后不可引用,ProgressProjector/EvidenceGuard 会排除)。 */
Duration ttl,
/** request + raw_response 合计字节上限。 */
long maxRecordBytes,
/** agent_result 字节上限(须 ≤ maxRecordBytes)。 */
long maxAgentResultBytes) {
public CanonicalInvocationLimits {
@@ -23,6 +30,7 @@ public record CanonicalInvocationLimits(
ttl.toMillis();
}
/** 校验 request + raw_response 未超上限(阶段三)。 */
public void validateRawCandidate(String request, String rawResponse) {
long actual = utf8Bytes(request) + utf8Bytes(rawResponse);
if (actual > maxRecordBytes) {
@@ -30,6 +38,7 @@ public record CanonicalInvocationLimits(
}
}
/** 校验 agent_result 未超上限(阶段五)。 */
public void validateAgentResult(String agentResult) {
long actual = utf8Bytes(agentResult);
if (actual > maxAgentResultBytes) {
@@ -37,6 +46,7 @@ public record CanonicalInvocationLimits(
}
}
/** 校验序列化后的整条记录未超上限(Redis 写入前)。 */
public void validateSerializedRecord(String json) {
long actual = utf8Bytes(json);
if (actual > maxRecordBytes) {
@@ -44,6 +54,7 @@ public record CanonicalInvocationLimits(
}
}
/** UTF-8 字节数(null 视为 0)。 */
public static long utf8Bytes(String value) {
return value == null ? 0 : value.getBytes(StandardCharsets.UTF_8).length;
}
@@ -5,20 +5,29 @@ import com.superbiz.agent.harness.contract.EvidenceStatus;
import java.time.Instant;
import java.util.Optional;
/**
* canonical 真相存储端口:一次 Tool 调用的完整生命周期(begin → markReady/markError → find)。
* RedisCanonicalInvocationStore 是唯一实现;ToolBoundary 通过它落库,ProgressProjector/EvidenceGuard 通过它回读验真。
*/
public interface CanonicalInvocationStore {
/** 记录大小/TTL 限制(执行门禁与写入前校验共用)。 */
CanonicalInvocationLimits limits();
/** 写入 PROJECTING 记录(同 key 重复 begin 抛 DuplicateInvocationException)。 */
void begin(String key, CanonicalToolInvocation invocation);
/** 按 key 查询记录(不存在或已过期返回 empty)。 */
Optional<CanonicalToolInvocation> find(String key);
/** PROJECTING → READY 迁移:携带 raw + agent_result + 客观 evidence status + 完成时间。 */
CanonicalToolInvocation markReady(String key,
String rawResponse,
String agentResult,
EvidenceStatus evidenceStatus,
Instant completedAt);
/** PROJECTING → ERROR 迁移:携带 raw(尽力而为)+ 稳定错误码 + 完成时间。 */
CanonicalToolInvocation markError(String key,
String rawResponse,
String errorCode,
@@ -1,5 +1,9 @@
package com.superbiz.agent.harness.tool.store;
/**
* canonical 存储基础设施异常(非业务规则):序列化/反序列化/值类型错误等。
* 由 ToolBoundary 映射为 STORE_ERROR 错误码。
*/
public class CanonicalStoreException extends RuntimeException {
public CanonicalStoreException(String message) {
@@ -1,5 +1,9 @@
package com.superbiz.agent.harness.tool.store;
/**
* 同 key 重复 begin(幂等拦截):同一 runId+toolCallId 只允许 begin 一次。
* ToolBoundary 映射为 DUPLICATE_TOOL_CALL 错误码。
*/
public final class DuplicateInvocationException extends CanonicalStoreException {
public DuplicateInvocationException() {
@@ -1,5 +1,9 @@
package com.superbiz.agent.harness.tool.store;
/**
* canonical 状态机违规:非法迁移(非 PROJECTING 时迁移)、记录缺失/已过期等。
* 属于防御性异常(正常流程不应触发)。
*/
public final class InvocationStateException extends CanonicalStoreException {
public InvocationStateException(String message) {
@@ -13,6 +13,16 @@ import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.function.UnaryOperator;
/**
* Redis 版 CanonicalInvocationStore:唯一真相源的持久化实现。
*
* <p>关键点:
* <ul>
* <li>begin 用 setIfAbsent(原子)实现「同 key 只 begin 一次」→ 幂等拦截;</li>
* <li>迁移(markReady/markError)用 update 读-改-写,保留剩余 TTL 续期;</li>
* <li>记录过期(TTL)后 find 返回 empty——ProgressProjector/EvidenceGuard 据此排除。</li>
* </ul>
*/
public final class RedisCanonicalInvocationStore implements CanonicalInvocationStore {
private final RedisTemplate<String, Object> redisTemplate;
@@ -43,6 +53,7 @@ public final class RedisCanonicalInvocationStore implements CanonicalInvocationS
}
String json = serialize(invocation);
limits.validateSerializedRecord(json);
// 原子 SETNX:同 key 已存在 → 幂等拒绝(DUPLICATE_TOOL_CALL)
Boolean created = values.setIfAbsent(
key, json, limits.ttl().toMillis(), TimeUnit.MILLISECONDS);
if (!Boolean.TRUE.equals(created)) {
@@ -55,6 +66,7 @@ public final class RedisCanonicalInvocationStore implements CanonicalInvocationS
requireKey(key);
Object stored = values.get(key);
if (stored == null) {
// 不存在或已过期(TTL 清除)
return Optional.empty();
}
if (!(stored instanceof String json)) {
@@ -71,6 +83,7 @@ public final class RedisCanonicalInvocationStore implements CanonicalInvocationS
Instant completedAt) {
CanonicalToolInvocation existing = find(key)
.orElseThrow(() -> new InvocationStateException("Canonical invocation is missing or expired"));
// 迁移前校验尺寸(raw + agent_result),不合法不落库
limits.validateRawCandidate(existing.request(), requireValue(rawResponse, "rawResponse"));
limits.validateAgentResult(requireValue(agentResult, "agentResult"));
return update(key, current -> current.markReady(
@@ -85,6 +98,7 @@ public final class RedisCanonicalInvocationStore implements CanonicalInvocationS
try {
return update(key, current -> current.markError(rawResponse, errorCode, completedAt));
} catch (ResultTooLargeException e) {
// raw 过大:降级为不保存 raw 的 ERROR(仍保留错误事实)
if (rawResponse == null) {
throw e;
}
@@ -93,6 +107,7 @@ public final class RedisCanonicalInvocationStore implements CanonicalInvocationS
}
}
/** 读-迁移-写:保留剩余 TTL(从 begin 起的生命周期,不被迁移重置)。 */
private CanonicalToolInvocation update(String key,
UnaryOperator<CanonicalToolInvocation> transition) {
CanonicalToolInvocation current = find(key)
@@ -105,6 +120,7 @@ public final class RedisCanonicalInvocationStore implements CanonicalInvocationS
return updated;
}
/** 剩余 TTL:不存在/已过期则抛错(防止迁移写入已失效记录)。 */
private long remainingTtlMillis(String key) {
Long remaining = redisTemplate.getExpire(key, TimeUnit.MILLISECONDS);
if (remaining == null || remaining <= 0) {
@@ -1,7 +1,12 @@
package com.superbiz.agent.harness.tool.store;
/**
* 结果超过字节上限:raw_response / agent_result / 序列化后的整条记录。
* ToolBoundary 映射为 RESULT_TOO_LARGE 错误码(或降级为不保存 raw 的 ERROR)。
*/
public final class ResultTooLargeException extends CanonicalStoreException {
/** 降级 ERROR 时使用的稳定错误码(raw 过大时替换为它)。 */
public static final String ERROR_CODE = "RESULT_TOO_LARGE";
public ResultTooLargeException(String field, long limit, long actual) {
@@ -2,6 +2,10 @@ package com.superbiz.agent.harness.tool.store;
import java.util.regex.Pattern;
/**
* canonical key 工厂:key = keyPrefix + ":" + runId + ":" + toolCallId。
* runId/toolCallId 必须匹配安全字符集(防 key 注入/分隔符混淆/超长 key)。
*/
public final class ToolCallKeyFactory {
private static final int MAX_SEGMENT_LENGTH = 128;
@@ -13,12 +17,14 @@ public final class ToolCallKeyFactory {
if (keyPrefix == null || keyPrefix.isBlank()) {
throw new IllegalArgumentException("keyPrefix must not be blank");
}
// 前缀不允许空段(防 "a::b" 式歧义)
if (keyPrefix.startsWith(":") || keyPrefix.endsWith(":") || keyPrefix.contains("::")) {
throw new IllegalArgumentException("keyPrefix contains an empty segment");
}
this.keyPrefix = keyPrefix;
}
/** 生成 canonical key(runId + toolCallId 双段定位一次调用)。 */
public String create(String runId, String toolCallId) {
requireSafeSegment(runId, "runId");
requireSafeSegment(toolCallId, "toolCallId");
@@ -29,6 +35,7 @@ public final class ToolCallKeyFactory {
return keyPrefix;
}
/** 段安全校验:非空、长度 ≤128、字符集 [A-Za-z0-9._-]。 */
private static void requireSafeSegment(String value, String name) {
if (value == null || value.isBlank()) {
throw new IllegalArgumentException(name + " must not be blank");