mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-15 03:55:09 +08:00
fix(goal): preserve authenticated identity for queued web input
This commit is contained in:
mateaix
parent
54e7b9b9e0
commit
83d965882d
@ -1115,7 +1115,7 @@ public class ChatController {
|
||||
// Commit the payload before publishing acceptance. The stream tracker is
|
||||
// only a wake signal; the database row remains authoritative on restart.
|
||||
var stored = inputQueue.enqueue(conversationId, agentId, username, message, contentParts,
|
||||
LocalDateTime.now());
|
||||
requesterUserIdOf(auth), LocalDateTime.now());
|
||||
boolean queued = streamTracker.notifyQueuedInput(conversationId);
|
||||
if (!queued) {
|
||||
inputQueue.cancel(stored.id(), "stream_finished_before_queue_registration",
|
||||
@ -1526,12 +1526,15 @@ public class ChatController {
|
||||
streamTracker.incrementFlux(conversationId);
|
||||
// RFC-063r §2.5: queued messages land in the same conversation; carry
|
||||
// a web-origin ChatOrigin so any cron job created during the queued
|
||||
// turn keeps a consistent (null-channel) binding.
|
||||
// turn keeps a consistent (null-channel) binding. The account id comes
|
||||
// from the authenticated enqueue, never from the previous stream
|
||||
// username. Managed operations revalidate this account and scope.
|
||||
vip.mate.agent.context.ChatOrigin queuedOrigin =
|
||||
vip.mate.agent.context.ChatOrigin.web(conversationId, requesterId, null, null)
|
||||
.withBaseUrl(baseUrl)
|
||||
vip.mate.agent.context.ChatOrigin.web(conversationId, preConsumedInput.createdBy(),
|
||||
queuedConversation.getWorkspaceId(), null, baseUrl, preConsumedInput.requesterUserId())
|
||||
.withAgent(agentId)
|
||||
.withOriginMessageId(queuedOriginMessageId);
|
||||
Disposable disposable = agentService.chatStructuredStream(agentId, queuedMessage, conversationId, requesterId, null, queuedOrigin)
|
||||
Disposable disposable = agentService.chatStructuredStream(agentId, queuedMessage, conversationId, preConsumedInput.createdBy(), null, queuedOrigin)
|
||||
.doOnNext(delta -> {
|
||||
if (emitterDone.get()) return;
|
||||
try {
|
||||
|
||||
@ -31,14 +31,20 @@ public class ConversationInputQueueStore {
|
||||
public QueuedInput enqueue(String conversationId, Long agentId, String createdBy,
|
||||
String message, List<MessageContentPart> contentParts,
|
||||
LocalDateTime now) {
|
||||
return enqueue(conversationId, agentId, createdBy, message, contentParts, null, now);
|
||||
}
|
||||
|
||||
public QueuedInput enqueue(String conversationId, Long agentId, String createdBy,
|
||||
String message, List<MessageContentPart> contentParts,
|
||||
Long requesterUserId, LocalDateTime now) {
|
||||
long id = IdWorker.getId();
|
||||
jdbc.update("""
|
||||
INSERT INTO mate_conversation_input_queue(
|
||||
id,conversation_id,agent_id,created_by,message,content_parts,state,
|
||||
created_at,updated_at)
|
||||
VALUES(?,?,?,?,?,?,'queued',?,?)
|
||||
created_at,updated_at,requester_user_id)
|
||||
VALUES(?,?,?,?,?,?,'queued',?,?,?)
|
||||
""", id, conversationId, agentId, createdBy, message == null ? "" : message,
|
||||
writeParts(contentParts), now, now);
|
||||
writeParts(contentParts), now, now, requesterUserId);
|
||||
return get(id);
|
||||
}
|
||||
|
||||
@ -137,7 +143,7 @@ public class ConversationInputQueueStore {
|
||||
rs.getString("message"), readParts(rs.getString("content_parts")),
|
||||
rs.getString("state"), rs.getString("claimed_by_attempt_id"),
|
||||
nullableLong(rs, "persisted_message_id"), rs.getString("cancel_reason"),
|
||||
time(rs, "created_at"), time(rs, "updated_at"));
|
||||
time(rs, "created_at"), time(rs, "updated_at"), nullableLong(rs, "requester_user_id"));
|
||||
}
|
||||
|
||||
private String writeParts(List<MessageContentPart> parts) {
|
||||
@ -183,5 +189,14 @@ public class ConversationInputQueueStore {
|
||||
Long persistedMessageId,
|
||||
String cancelReason,
|
||||
LocalDateTime createdAt,
|
||||
LocalDateTime updatedAt) {}
|
||||
LocalDateTime updatedAt,
|
||||
Long requesterUserId) {
|
||||
public QueuedInput(Long id, String conversationId, Long agentId, String createdBy,
|
||||
String message, List<MessageContentPart> contentParts, String state,
|
||||
String claimedByAttemptId, Long persistedMessageId, String cancelReason,
|
||||
LocalDateTime createdAt, LocalDateTime updatedAt) {
|
||||
this(id, conversationId, agentId, createdBy, message, contentParts, state,
|
||||
claimedByAttemptId, persistedMessageId, cancelReason, createdAt, updatedAt, null);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -0,0 +1,3 @@
|
||||
-- Preserve authenticated account identity across durable queue replay.
|
||||
-- Legacy entries deliberately remain unasserted; never infer identity from a display name.
|
||||
ALTER TABLE mate_conversation_input_queue ADD COLUMN requester_user_id BIGINT NULL;
|
||||
@ -0,0 +1,3 @@
|
||||
-- Preserve authenticated account identity across durable queue replay.
|
||||
-- Legacy entries deliberately remain unasserted; never infer identity from a display name.
|
||||
ALTER TABLE mate_conversation_input_queue ADD COLUMN requester_user_id BIGINT NULL;
|
||||
@ -0,0 +1,3 @@
|
||||
-- Preserve authenticated account identity across durable queue replay.
|
||||
-- Legacy entries deliberately remain unasserted; never infer identity from a display name.
|
||||
ALTER TABLE mate_conversation_input_queue ADD COLUMN requester_user_id BIGINT NULL;
|
||||
@ -56,3 +56,5 @@ V198 also stores absolute scheduler lease deadlines. Existing leases expire duri
|
||||
Validation snapshot (2026-09-14): the full default backend test run passed 5,249 executed tests with 33 conditional skips; the frontend passed 386 tests. The managed contract also has real compiled ReAct/Plan graph tests for account and scheduled-owner execution. Their model choices and semantic verdicts are controlled fixtures, not online-model benchmarks. Specialized integration profiles and the proprietary Kingbase engine are outside that full-default-suite claim.
|
||||
|
||||
Built-in shell/code execution is not OS-isolated from the service host. Selecting JSON acceptance does not sandbox those tools, and the protocol cannot defend against host code that can access database credentials or files. Environment-name filtering and workspace path checks do not replace that isolation. Lease deadlines are calculated from absolute instants, including daylight-saving clock rollback; scheduling display fields remain local timestamps.
|
||||
|
||||
From V199, queued Web input stores the authenticated account ID at enqueue time, and ordinary Web replay carries the conversation workspace. Managed operations still recheck the account, ownership and current requirements. Legacy queue items do not gain an asserted identity from a username; users must resend an authenticated request for managed JSON operations. Persistent Goal workers retain their existing attempt-owner validation when consuming input; this does not introduce an account path without a lease check.
|
||||
|
||||
@ -56,3 +56,5 @@ V198 同样以绝对时间保存调度租约截止。升级时旧租约失效，
|
||||
验证快照（2026-09-14）：默认后端完整测试集实际通过 5,249 项、条件跳过 33 项，前端通过 386 项。受管契约还覆盖真实编译的 ReAct/Plan 图及账户/调度 owner 四种组合；模型选择和语义评价是受控夹具，不是在线模型基准。专门集成 profile 与 Kingbase 专有引擎不包含在“默认完整测试集”结论中。
|
||||
|
||||
内置 shell/code 执行没有与服务宿主做操作系统隔离。选择 JSON 验收不会把这些工具变成沙箱；此协议不能抵抗能访问数据库凭据或文件的宿主代码，环境变量名称过滤和工作区路径检查也不能替代隔离。租约截止从绝对时刻计算，覆盖夏令时回拨；调度显示字段仍使用本地时间戳。
|
||||
|
||||
Web排队消息从V199起保存入队时已认证账户的内部ID，普通Web续跑同时携带当前会话工作区；受管工具执行时仍重新校验账户、归属和当前要求。旧队列项不按用户名补造身份，不能用于受管JSON操作；需要用户重新发送已认证请求。持久Goal工作器消费输入时继续使用原有attempt owner校验，没有转换成免租约的账户路径。
|
||||
|
||||
@ -33,6 +33,7 @@ class ChatControllerDurableQueueTest {
|
||||
ConversationInputQueueStore queue = mock(ConversationInputQueueStore.class);
|
||||
Authentication authentication = mock(Authentication.class);
|
||||
when(authentication.getName()).thenReturn("alice");
|
||||
when(authentication.getDetails()).thenReturn(42L);
|
||||
when(conversations.isConversationOwner("conv", "alice")).thenReturn(true);
|
||||
when(streams.isRunning("conv")).thenReturn(true);
|
||||
when(streams.notifyQueuedInput("conv")).thenReturn(true);
|
||||
@ -42,6 +43,9 @@ class ChatControllerDurableQueueTest {
|
||||
when(queue.enqueue(eq("conv"), eq(2L), eq("alice"), eq("follow-up"),
|
||||
eq(List.of()), any())).thenReturn(stored);
|
||||
|
||||
when(queue.enqueue(eq("conv"), eq(2L), eq("alice"), eq("follow-up"),
|
||||
eq(List.of()), eq(42L), any())).thenReturn(stored);
|
||||
|
||||
ChatController controller = new ChatController(agents, conversations, approvals, streams,
|
||||
new ObjectMapper(), mock(ConversationCompletionPublisher.class),
|
||||
mock(MemoryOwnerResolver.class), mock(ChatUploadLocationResolver.class),
|
||||
@ -57,7 +61,36 @@ class ChatControllerDurableQueueTest {
|
||||
.containsEntry("queueItemId", "91");
|
||||
var order = inOrder(queue, streams);
|
||||
order.verify(queue).enqueue(eq("conv"), eq(2L), eq("alice"),
|
||||
eq("follow-up"), eq(List.of()), any());
|
||||
eq("follow-up"), eq(List.of()), eq(42L), any());
|
||||
order.verify(streams).notifyQueuedInput("conv");
|
||||
}
|
||||
@Test
|
||||
void queuedStreamCarriesThePersistedAccountInsteadOfThePreviousTurnDisplayName() {
|
||||
AgentService agents = mock(AgentService.class);
|
||||
ConversationService conversations = mock(ConversationService.class);
|
||||
ChatStreamTracker streams = mock(ChatStreamTracker.class);
|
||||
ConversationInputQueueStore queue = mock(ConversationInputQueueStore.class);
|
||||
var input = new QueuedInput(91L, "conv", 2L, "alice", "queued",
|
||||
List.of(), "claimed", "claim", 100L, null, LocalDateTime.now(), LocalDateTime.now(), 42L);
|
||||
when(queue.claimNext(eq("conv"), any(), any())).thenReturn(java.util.Optional.of(input));
|
||||
when(queue.consume(eq(91L), any(), any())).thenReturn(true);
|
||||
var conversation = new vip.mate.workspace.conversation.model.ConversationEntity();
|
||||
conversation.setConversationId("conv"); conversation.setAgentId(2L); conversation.setWorkspaceId(3L);
|
||||
when(conversations.findByConversationId("conv")).thenReturn(conversation);
|
||||
when(agents.chatStructuredStream(eq(2L), eq("queued"), eq("conv"), any(), any(), any()))
|
||||
.thenReturn(reactor.core.publisher.Flux.never());
|
||||
ChatController controller = new ChatController(agents, conversations, mock(ApprovalWorkflowService.class), streams,
|
||||
new ObjectMapper(), mock(ConversationCompletionPublisher.class), mock(MemoryOwnerResolver.class),
|
||||
mock(ChatUploadLocationResolver.class), mock(OfficePreviewService.class), queue);
|
||||
org.springframework.test.util.ReflectionTestUtils.invokeMethod(controller, "startQueuedMessage", "conv",
|
||||
new org.springframework.web.servlet.mvc.method.annotation.SseEmitter(),
|
||||
new java.util.concurrent.atomic.AtomicBoolean(true), "previous-turn-user", "http://localhost");
|
||||
var origin = org.mockito.ArgumentCaptor.forClass(vip.mate.agent.context.ChatOrigin.class);
|
||||
org.mockito.Mockito.verify(agents).chatStructuredStream(eq(2L), eq("queued"), eq("conv"), any(), any(), origin.capture());
|
||||
assertThat(origin.getValue().requesterUserId()).isEqualTo(42L);
|
||||
assertThat(origin.getValue().requesterId()).isEqualTo("alice");
|
||||
assertThat(origin.getValue().workspaceId()).isEqualTo(3L);
|
||||
assertThat(origin.getValue().originMessageId()).isEqualTo(100L);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -33,6 +33,8 @@ class ConversationInputQueueStoreTest {
|
||||
new ClassPathResource("db/migration/h2/V189__goal_attempt_and_input_queue.sql"))
|
||||
.execute(dataSource);
|
||||
jdbc = new JdbcTemplate(dataSource);
|
||||
jdbc.update("INSERT INTO mate_conversation_input_queue(id,conversation_id,agent_id,created_by,message,content_parts,state,created_at,updated_at) VALUES (7,'legacy-conv',1,'mate','pre-upgrade input','[]','queued',?,?)", now, now);
|
||||
new ResourceDatabasePopulator(new ClassPathResource("db/migration/h2/V199__queued_input_account_identity.sql")).execute(dataSource);
|
||||
mapper = new ObjectMapper();
|
||||
store = new ConversationInputQueueStore(jdbc, mapper);
|
||||
}
|
||||
@ -57,6 +59,18 @@ class ConversationInputQueueStoreTest {
|
||||
assertThat(restarted.get(first.id()).state()).isEqualTo("consumed");
|
||||
}
|
||||
|
||||
@Test
|
||||
void accountIdentitySurvivesReconstructionAndLegacyEntriesStayUnasserted() {
|
||||
assertThat(store.get(7L).message()).isEqualTo("pre-upgrade input");
|
||||
assertThat(store.get(7L).requesterUserId()).isNull();
|
||||
var known = store.enqueue("conv", 1L, "mate", "known", List.of(), 9223372036854775801L, now);
|
||||
var legacy = store.enqueue("conv", 1L, "mate", "legacy", List.of(), now);
|
||||
var restarted = new ConversationInputQueueStore(jdbc, mapper);
|
||||
assertThat(restarted.claimNext("conv", "worker", now).orElseThrow().requesterUserId()).isEqualTo(9223372036854775801L);
|
||||
assertThat(restarted.get(known.id()).requesterUserId()).isEqualTo(9223372036854775801L);
|
||||
assertThat(restarted.get(legacy.id()).requesterUserId()).isNull();
|
||||
}
|
||||
|
||||
@Test
|
||||
void claimReleaseAndCancellationAreFencedByAttempt() {
|
||||
QueuedInput input = store.enqueue("conv", 1L, "mate", "queued", List.of(), now);
|
||||
|
||||
@ -285,6 +285,23 @@ class GoalJsonAcceptanceIntegrationTest {
|
||||
.withAgent(goal.getAgentId());
|
||||
}
|
||||
|
||||
@Test void persistedQueuedAccountCanPublishButCannotBecomeARecreatedUsername() {
|
||||
GoalEntity goal = goal(false);
|
||||
acceptance.configure(goal.getId(), "r", request(0, "summary"), alice);
|
||||
var queue = new vip.mate.channel.web.ConversationInputQueueStore(jdbc, new com.fasterxml.jackson.databind.ObjectMapper());
|
||||
Long accountId = jdbc.queryForObject("SELECT id FROM mate_user WHERE username=?", Long.class, alice);
|
||||
var input = queue.enqueue(goal.getConversationId(), 1L, alice, "publish", List.of(), accountId, java.time.LocalDateTime.now());
|
||||
var restored = new vip.mate.channel.web.ConversationInputQueueStore(jdbc, new com.fasterxml.jackson.databind.ObjectMapper()).get(input.id());
|
||||
var origin = vip.mate.agent.context.ChatOrigin.web(restored.conversationId(), restored.createdBy(), 1L, null, null, restored.requesterUserId()).withAgent(restored.agentId());
|
||||
assertEquals(1, artifacts.publishForRuntime(origin, "report", publication(0, "{\"summary\":false}")).generation());
|
||||
jdbc.update("DELETE FROM mate_user WHERE id=?", accountId);
|
||||
jdbc.update("INSERT INTO mate_user(id,username,password,enabled,role,create_time,update_time,deleted) VALUES (?,?,?,TRUE,'user',CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,0)", IdWorker.getId(), alice, "replacement-fixture");
|
||||
assertThrows(MateClawException.class, () -> artifacts.publishForRuntime(origin, "report", publication(1, "{}")));
|
||||
var legacy = queue.enqueue(goal.getConversationId(), 1L, alice, "legacy", List.of(), java.time.LocalDateTime.now());
|
||||
var unasserted = vip.mate.agent.context.ChatOrigin.web(legacy.conversationId(), legacy.createdBy(), 1L, null, null, legacy.requesterUserId()).withAgent(1L);
|
||||
assertThrows(MateClawException.class, () -> artifacts.publishForRuntime(unasserted, "report", publication(1, "{}")));
|
||||
}
|
||||
|
||||
@Test void actualManagedToolUsesServerAccountContextAndExposesRequirements() throws Exception {
|
||||
GoalEntity goal = goal(false);
|
||||
acceptance.configure(goal.getId(), "r", request(0, "summary"), alice);
|
||||
|
||||
@ -36,7 +36,8 @@ class GoalRecoveryServiceTest {
|
||||
new ResourceDatabasePopulator(new ClassPathResource("db/migration/h2/V120__agent_goal.sql"),
|
||||
new ClassPathResource("db/migration/h2/V188__goal_continuation.sql"),
|
||||
new ClassPathResource("db/migration/h2/V189__goal_attempt_and_input_queue.sql"),
|
||||
new ClassPathResource("db/migration/h2/V198__goal_absolute_owner_leases.sql")).execute(ds);
|
||||
new ClassPathResource("db/migration/h2/V198__goal_absolute_owner_leases.sql"),
|
||||
new ClassPathResource("db/migration/h2/V199__queued_input_account_identity.sql")).execute(ds);
|
||||
jdbc=new JdbcTemplate(ds);attempts=new GoalAttemptStore(jdbc);continuations=new GoalContinuationStore(jdbc);
|
||||
inputs=new ConversationInputQueueStore(jdbc,new ObjectMapper());
|
||||
coordinator=new GoalRunCoordinator(continuations,attempts,goals,new vip.mate.goal.config.GoalProperties(),java.time.Clock.fixed(now.atZone(java.time.ZoneId.systemDefault()).toInstant(), java.time.ZoneId.systemDefault()));
|
||||
|
||||
Loading…
Reference in New Issue
Block a user