feat(agent): bind chat sessions to a fixed datasource

data-agent-backend/src/main/java/io/github/malonetalk/agent/AgentService.java

@@ -40,6 +40,7 @@
 import io.github.malonetalk.enums.ChatStreamEventType;
 import io.github.malonetalk.exception.ErrorResponse;
 import io.github.malonetalk.exception.ExceptionResponseMapper;
+import io.github.malonetalk.service.DatasourceService;
 import io.github.malonetalk.web.TraceIdFilter;
 import jakarta.annotation.PostConstruct;
 import java.util.List;
@@ -62,6 +63,7 @@ public class AgentService {
     private final SkillLoaderService skillLoaderService;
     private final ExceptionResponseMapper exceptionResponseMapper;
     private final EventConverter eventConverter;
+    private final DatasourceService datasourceService;
     private Toolkit toolkit;
     private SkillBox skillBox;
 
@@ -73,13 +75,23 @@ public void init() {
     }
 
     public Flux<ChatStreamEvent> chatStream(
-            String sessionId, String userInput, List<ChatRequest.ToolResultInput> toolResults) {
-        return Flux.defer(() -> streamAgent(sessionId, userInput, toolResults))
+            String sessionId,
+            String userInput,
+            List<ChatRequest.ToolResultInput> toolResults,
+            Integer datasourceId) {
+        return Flux.defer(() -> streamAgent(sessionId, userInput, toolResults, datasourceId))
                 .onErrorResume(this::toErrorEvent);
     }
 
     private Flux<ChatStreamEvent> streamAgent(
-            String sessionId, String userInput, List<ChatRequest.ToolResultInput> toolResults) {
+            String sessionId,
+            String userInput,
+            List<ChatRequest.ToolResultInput> toolResults,
+            Integer datasourceId) {
+        if (datasourceId != null) {
+            // 首次绑定；已绑定会被 INSERT IGNORE 忽略，锁定语义在 mapper 层保证。
+            datasourceService.bindSessionDatasource(sessionId, datasourceId);
+        }
         ReActAgent agent = createAgent(ToolCallContext.builder().sessionId(sessionId).build());
 
         Session session = sessionService.getOrCreateSession(sessionId);

data-agent-backend/src/main/java/io/github/malonetalk/agent/SessionService.java

@@ -190,6 +190,16 @@ public void clearSession(String sessionId) {
         if (session != null) {
             session.delete(SimpleSessionKey.of(sessionId));
         }
+        // 一并清理会话的数据源绑定，避免孤儿绑定。
+        try (Connection conn = dataSource.getConnection();
+                PreparedStatement ps =
+                        conn.prepareStatement(
+                                "DELETE FROM session_datasource WHERE session_id = ?")) {
+            ps.setString(1, sessionId);
+            ps.executeUpdate();
+        } catch (Exception e) {
+            log.error("Error deleting session datasource binding", e);
+        }
     }
 
     public void clearAllSessions() {
@@ -219,6 +229,21 @@ public List<SessionInfo> listSessions() {
             log.error("Error listing session timestamps", e);
         }
 
+        Map<String, String[]> bindings = new HashMap<>();
+        try (Connection conn = dataSource.getConnection();
+                PreparedStatement ps =
+                        conn.prepareStatement(
+                                "SELECT sd.session_id, sd.datasource_id, d.name"
+                                        + " FROM session_datasource sd"
+                                        + " LEFT JOIN datasource d ON sd.datasource_id = d.id");
+                ResultSet rs = ps.executeQuery()) {
+            while (rs.next()) {
+                bindings.put(rs.getString(1), new String[] {rs.getString(2), rs.getString(3)});
+            }
+        } catch (Exception e) {
+            log.error("Error listing session datasource bindings", e);
+        }
+
         List<SessionInfo> result = new ArrayList<>();
         for (SessionKey key : keys) {
             String sid = key.toIdentifier();
@@ -241,7 +266,12 @@ public List<SessionInfo> listSessions() {
             }
 
             String[] times = timestamps.getOrDefault(sid, new String[] {"", ""});
-            result.add(new SessionInfo(sid, title, times[0], times[1]));
+            String[] binding = bindings.get(sid);
+            Integer datasourceId =
+                    binding == null || binding[0] == null ? null : Integer.valueOf(binding[0]);
+            String datasourceName = binding == null ? null : binding[1];
+            result.add(
+                    new SessionInfo(sid, title, times[0], times[1], datasourceId, datasourceName));
         }
 
         result.sort((a, b) -> b.lastActiveAt().compareTo(a.lastActiveAt()));

data-agent-backend/src/main/java/io/github/malonetalk/agent/tools/ExecuteSqlTool.java

@@ -20,11 +20,10 @@
 import io.agentscope.core.message.ToolResultBlock;
 import io.agentscope.core.tool.Tool;
 import io.agentscope.core.tool.ToolParam;
+import io.github.malonetalk.agent.ToolCallContext;
 import io.github.malonetalk.agent.datasource.QueryResult;
 import io.github.malonetalk.agent.datasource.SqlExecutor;
-import io.github.malonetalk.common.ErrorCode;
 import io.github.malonetalk.entity.Datasource;
-import io.github.malonetalk.exception.BusinessException;
 import io.github.malonetalk.exception.ToolExceptionMapper;
 import io.github.malonetalk.service.DatasourceService;
 import lombok.AllArgsConstructor;
@@ -46,18 +45,12 @@ public class ExecuteSqlTool implements MarkAgentTool {
                         + " other modification operations.")
     public ToolResultBlock executeSql(
             @ToolParam(name = "sql", description = "The SELECT SQL query statement to execute")
-                    String sql) {
+                    String sql,
+            ToolCallContext ctx) {
         return toolExceptionMapper.run(
                 () -> {
                     Datasource datasource =
-                            dataSourceService
-                                    .getActiveDatasource()
-                                    .orElseThrow(
-                                            () ->
-                                                    BusinessException.of(
-                                                            ErrorCode.NO_ACTIVE_DATASOURCE,
-                                                            "No active datasource is available."
-                                                                    + " Unable to execute SQL."));
+                            dataSourceService.getDatasourceForSession(ctx.sessionId());
                     QueryResult result = sqlExecutor.execute(datasource, sql);
                     return ToolResultBlock.text(formatResult(result));
                 });

data-agent-backend/src/main/java/io/github/malonetalk/agent/tools/GetTableSchemaTool.java

@@ -20,10 +20,9 @@
 import io.agentscope.core.message.ToolResultBlock;
 import io.agentscope.core.tool.Tool;
 import io.agentscope.core.tool.ToolParam;
-import io.github.malonetalk.common.ErrorCode;
+import io.github.malonetalk.agent.ToolCallContext;
 import io.github.malonetalk.dto.prompt.ColumnPromptResponse;
 import io.github.malonetalk.entity.Datasource;
-import io.github.malonetalk.exception.BusinessException;
 import io.github.malonetalk.exception.ToolExceptionMapper;
 import io.github.malonetalk.service.DatasourceService;
 import io.github.malonetalk.service.semantic.column.ColumnSemanticService;
@@ -53,19 +52,12 @@ public class GetTableSchemaTool implements MarkAgentTool {
                     """)
     public ToolResultBlock getTableSchema(
             @ToolParam(name = "table_name", description = "The table name to query schema for")
-                    String tableName) {
+                    String tableName,
+            ToolCallContext ctx) {
         return toolExceptionMapper.run(
                 () -> {
                     Datasource datasource =
-                            dataSourceService
-                                    .getActiveDatasource()
-                                    .orElseThrow(
-                                            () ->
-                                                    BusinessException.of(
-                                                            ErrorCode.NO_ACTIVE_DATASOURCE,
-                                                            "No active datasource is available."
-                                                                    + " Unable to retrieve the"
-                                                                    + " table schema."));
+                            dataSourceService.getDatasourceForSession(ctx.sessionId());
                     List<ColumnPromptResponse> columns =
                             columnSemanticService.getMergedTableSchema(
                                     datasource.getId(), tableName);

data-agent-backend/src/main/java/io/github/malonetalk/agent/tools/GetTablesTool.java

@@ -21,9 +21,8 @@
 import io.agentscope.core.tool.Tool;
 import io.agentscope.core.tool.ToolParam;
 import io.agentscope.core.util.JsonUtils;
-import io.github.malonetalk.common.ErrorCode;
+import io.github.malonetalk.agent.ToolCallContext;
 import io.github.malonetalk.entity.Datasource;
-import io.github.malonetalk.exception.BusinessException;
 import io.github.malonetalk.exception.ToolExceptionMapper;
 import io.github.malonetalk.service.DatasourceService;
 import io.github.malonetalk.service.semantic.table.TableSemanticService;
@@ -58,19 +57,12 @@ public ToolResultBlock getTables(
                                     returns all tables.\
                                     """,
                             required = false)
-                    List<String> domains) {
+                    List<String> domains,
+            ToolCallContext ctx) {
         return toolExceptionMapper.run(
                 () -> {
                     Datasource dataSource =
-                            dataSourceService
-                                    .getActiveDatasource()
-                                    .orElseThrow(
-                                            () ->
-                                                    BusinessException.of(
-                                                            ErrorCode.NO_ACTIVE_DATASOURCE,
-                                                            "No active datasource is available."
-                                                                    + " Unable to retrieve"
-                                                                    + " tables."));
+                            dataSourceService.getDatasourceForSession(ctx.sessionId());
                     return ToolResultBlock.text(
                             JsonUtils.getJsonCodec()
                                     .toJson(

data-agent-backend/src/main/java/io/github/malonetalk/common/ErrorCode.java

@@ -72,6 +72,12 @@ public enum ErrorCode {
             HttpStatus.CONFLICT,
             "The operation conflicts with the current data state."),
 
+    /** 会话绑定的数据源已被删除，会话无法继续使用。 */
+    BOUND_DATASOURCE_UNAVAILABLE(
+            "BOUND_DATASOURCE_UNAVAILABLE",
+            HttpStatus.CONFLICT,
+            "The datasource bound to this session no longer exists. Please start a new session."),
+
     /** HTTP 方法不支持。 */
     METHOD_NOT_ALLOWED(
             "METHOD_NOT_ALLOWED",

data-agent-backend/src/main/java/io/github/malonetalk/controller/AgentController.java

@@ -54,7 +54,11 @@ public Flux<ServerSentEvent<ChatStreamEvent>> chatStream(
             @Valid @RequestBody ChatRequest request) {
         log.info("SSE chat stream started: sessionId={}", request.sessionId());
         return agentService
-                .chatStream(request.sessionId(), request.message(), request.toolResults())
+                .chatStream(
+                        request.sessionId(),
+                        request.message(),
+                        request.toolResults(),
+                        request.datasourceId())
                 .map(
                         event ->
                                 ServerSentEvent.<ChatStreamEvent>builder()

data-agent-backend/src/main/java/io/github/malonetalk/dto/ChatRequest.java

@@ -23,7 +23,8 @@
 public record ChatRequest(
         @NotBlank(message = "sessionId 不能为空") String sessionId,
         String message,
-        List<ToolResultInput> toolResults) {
+        List<ToolResultInput> toolResults,
+        Integer datasourceId) {
 
     public record ToolResultInput(String toolCallId, String toolName, String output) {}
 }

data-agent-backend/src/main/java/io/github/malonetalk/dto/SessionInfo.java

@@ -17,4 +17,11 @@
  */
 package io.github.malonetalk.dto;
 
-public record SessionInfo(String sessionId, String title, String createdAt, String lastActiveAt) {}
+/** 会话摘要；datasourceId/datasourceName 为绑定的数据源信息，未绑定为 null。 */
+public record SessionInfo(
+        String sessionId,
+        String title,
+        String createdAt,
+        String lastActiveAt,
+        Integer datasourceId,
+        String datasourceName) {}

data-agent-backend/src/main/java/io/github/malonetalk/entity/SessionDatasource.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright (C) 2026 github.com/MaloneTalk
+ *
+ * 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 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/>.
+ * limitations under the License.
+ */
+package io.github.malonetalk.entity;
+
+import java.time.LocalDateTime;
+import lombok.Data;
+
+/** 会话与数据源绑定关系（session_id → datasource_id，一会话一源）。 */
+@Data
+public class SessionDatasource {
+
+    private String sessionId;
+    private Integer datasourceId;
+    private LocalDateTime createTime;
+    private LocalDateTime updateTime;
+}

data-agent-backend/src/main/java/io/github/malonetalk/mapper/SessionDatasourceMapper.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright (C) 2026 github.com/MaloneTalk
+ *
+ * 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 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/>.
+ * limitations under the License.
+ */
+package io.github.malonetalk.mapper;
+
+import io.github.malonetalk.entity.SessionDatasource;
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+
+@Mapper
+public interface SessionDatasourceMapper {
+
+    /** 首次绑定；主键冲突时保留已有绑定（锁定语义）。 */
+    int insertIfAbsent(SessionDatasource sessionDatasource);
+
+    SessionDatasource selectBySessionId(@Param("sessionId") String sessionId);
+}

data-agent-backend/src/main/java/io/github/malonetalk/service/DatasourceService.java

@@ -37,6 +37,12 @@ public interface DatasourceService {
 
     Optional<Datasource> getActiveDatasource();
 
+    /** 会话维度解析数据源：有绑定用绑定（无视 status），无绑定回退激活源。 */
+    Datasource getDatasourceForSession(String sessionId);
+
+    /** 绑定会话到数据源；数据源不存在抛 400，已绑定则保留首次绑定。 */
+    void bindSessionDatasource(String sessionId, Integer datasourceId);
+
     List<Datasource> findByType(String type);
 
     boolean updateStatus(Integer id, String status);

data-agent-backend/src/main/java/io/github/malonetalk/service/DatasourceServiceImpl.java

@@ -17,9 +17,13 @@
  */
 package io.github.malonetalk.service;
 
+import io.github.malonetalk.common.ErrorCode;
 import io.github.malonetalk.entity.Datasource;
+import io.github.malonetalk.entity.SessionDatasource;
 import io.github.malonetalk.enums.Status;
+import io.github.malonetalk.exception.BusinessException;
 import io.github.malonetalk.mapper.DatasourceMapper;
+import io.github.malonetalk.mapper.SessionDatasourceMapper;
 import java.time.LocalDateTime;
 import java.util.List;
 import java.util.Optional;
@@ -33,6 +37,7 @@
 public class DatasourceServiceImpl implements DatasourceService {
 
     private final DatasourceMapper dataSourceMapper;
+    private final SessionDatasourceMapper sessionDatasourceMapper;
 
     @Override
     public List<Datasource> findAll() {
@@ -76,6 +81,36 @@ public Optional<Datasource> getActiveDatasource() {
         return active.isEmpty() ? Optional.empty() : Optional.of(active.get(0));
     }
 
+    @Override
+    public Datasource getDatasourceForSession(String sessionId) {
+        SessionDatasource binding = sessionDatasourceMapper.selectBySessionId(sessionId);
+        if (binding != null) {
+            Datasource datasource = findExistingDatasource(binding.getDatasourceId());
+            if (datasource == null) {
+                throw BusinessException.of(ErrorCode.BOUND_DATASOURCE_UNAVAILABLE);
+            }
+            return datasource;
+        }
+        return getActiveDatasource()
+                .orElseThrow(() -> BusinessException.of(ErrorCode.NO_ACTIVE_DATASOURCE));
+    }
+
+    @Override
+    public void bindSessionDatasource(String sessionId, Integer datasourceId) {
+        if (findExistingDatasource(datasourceId) == null) {
+            throw BusinessException.of(
+                    ErrorCode.BAD_REQUEST, "Datasource does not exist: " + datasourceId);
+        }
+        SessionDatasource binding = new SessionDatasource();
+        binding.setSessionId(sessionId);
+        binding.setDatasourceId(datasourceId);
+        sessionDatasourceMapper.insertIfAbsent(binding);
+    }
+
+    private Datasource findExistingDatasource(Integer id) {
+        return dataSourceMapper.selectById(id);
+    }
+
     @Override
     public List<Datasource> findByType(String type) {
         return dataSourceMapper.selectByType(type);