mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-16 04:18:17 +08:00
persist selected Goal identity for queued Web input
This commit is contained in:
mateaix
parent
d080bf09f6
commit
cbe04d9515
@ -1142,8 +1142,17 @@ 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,
|
||||
requesterUserIdOf(auth), LocalDateTime.now());
|
||||
var enqueueConversation = conversationService.findByConversationId(conversationId);
|
||||
Long queueAgentId = agentId == null && enqueueConversation != null
|
||||
? enqueueConversation.getAgentId() : agentId;
|
||||
if (enqueueConversation == null || queueAgentId == null
|
||||
|| !java.util.Objects.equals(queueAgentId, enqueueConversation.getAgentId())) {
|
||||
return R.fail(409, "会话助手已变化，请刷新后重试");
|
||||
}
|
||||
var queuedSelection = captureWebGoal(vip.mate.agent.context.ChatOrigin.web(conversationId,
|
||||
username, enqueueConversation.getWorkspaceId(), null, null, requesterUserIdOf(auth)), queueAgentId);
|
||||
var stored = inputQueue.enqueue(conversationId, queueAgentId, username, message, contentParts,
|
||||
requesterUserIdOf(auth), queuedSelection.selectedGoalId(), LocalDateTime.now());
|
||||
boolean queued = streamTracker.notifyQueuedInput(conversationId);
|
||||
if (!queued) {
|
||||
inputQueue.cancel(stored.id(), "stream_finished_before_queue_registration",
|
||||
@ -1524,6 +1533,30 @@ public class ChatController {
|
||||
return;
|
||||
}
|
||||
|
||||
// A row queued by an older binary has no selection snapshot. If this
|
||||
// conversation has managed Goal history, execution could turn an old
|
||||
// selected request into an explicitly unselected approval after a Goal
|
||||
// ended. Keep the user's text and require a fresh authenticated turn.
|
||||
if (preConsumedInput.selectedGoalId() == null && goalApprovalRuns != null
|
||||
&& goalApprovalRuns.hasManagedGoalHistory(conversationId, String.valueOf(agentId))) {
|
||||
if (preConsumedInput.persistedMessageId() == null) {
|
||||
MessageEntity saved = conversationService.saveMessage(conversationId, "user",
|
||||
preConsumedInput.message(), preConsumedInput.contentParts(), "queued");
|
||||
if (saved == null || !inputQueue.bindMessage(preConsumedInput.id(), queueClaimId,
|
||||
saved.getId(), LocalDateTime.now())) {
|
||||
inputQueue.release(preConsumedInput.id(), queueClaimId, LocalDateTime.now());
|
||||
throw new IllegalStateException("Legacy queued input could not be preserved");
|
||||
}
|
||||
}
|
||||
if (!inputQueue.consume(preConsumedInput.id(), queueClaimId, LocalDateTime.now()))
|
||||
throw new IllegalStateException("Legacy queued input claim was lost");
|
||||
broadcastEvent(conversationId, "warning", Map.of(
|
||||
"message", "排队消息缺少Goal选择快照，内容已保存，请重新发送"));
|
||||
conversationService.updateStreamStatus(conversationId, "idle");
|
||||
completeEmitterQuietly(emitter, emitterDone);
|
||||
return;
|
||||
}
|
||||
|
||||
// Rate Limit 防护：如果上一轮以 rate limit 错误结束，不立即续跑排队消息（必然再次 429）。
|
||||
// 改为持久化用户消息 + 通知前端"稍后重试"，避免连锁 429 浪费配额。
|
||||
String lastMessage = conversationService.getLastMessage(conversationId);
|
||||
@ -1595,7 +1628,8 @@ public class ChatController {
|
||||
vip.mate.agent.context.ChatOrigin queuedOrigin =
|
||||
vip.mate.agent.context.ChatOrigin.web(conversationId, preConsumedInput.createdBy(),
|
||||
queuedConversation.getWorkspaceId(), null, baseUrl, preConsumedInput.requesterUserId())
|
||||
.withOriginMessageId(queuedOriginMessageId);
|
||||
.withOriginMessageId(queuedOriginMessageId)
|
||||
.withSelectedGoalId(preConsumedInput.selectedGoalId());
|
||||
queuedOrigin = captureWebGoal(queuedOrigin, agentId);
|
||||
Disposable disposable = agentService.chatStructuredStream(agentId, queuedMessage, conversationId, preConsumedInput.createdBy(), null, queuedOrigin)
|
||||
.doOnNext(delta -> {
|
||||
|
||||
@ -37,14 +37,21 @@ public class ConversationInputQueueStore {
|
||||
public QueuedInput enqueue(String conversationId, Long agentId, String createdBy,
|
||||
String message, List<MessageContentPart> contentParts,
|
||||
Long requesterUserId, LocalDateTime now) {
|
||||
return enqueue(conversationId, agentId, createdBy, message, contentParts,
|
||||
requesterUserId, null, now);
|
||||
}
|
||||
|
||||
public QueuedInput enqueue(String conversationId, Long agentId, String createdBy,
|
||||
String message, List<MessageContentPart> contentParts,
|
||||
Long requesterUserId, Long selectedGoalId, 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,requester_user_id)
|
||||
VALUES(?,?,?,?,?,?,'queued',?,?,?)
|
||||
created_at,updated_at,requester_user_id,selected_goal_id)
|
||||
VALUES(?,?,?,?,?,?,'queued',?,?,?,?)
|
||||
""", id, conversationId, agentId, createdBy, message == null ? "" : message,
|
||||
writeParts(contentParts), now, now, requesterUserId);
|
||||
writeParts(contentParts), now, now, requesterUserId, selectedGoalId);
|
||||
return get(id);
|
||||
}
|
||||
|
||||
@ -143,7 +150,8 @@ 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"), nullableLong(rs, "requester_user_id"));
|
||||
time(rs, "created_at"), time(rs, "updated_at"), nullableLong(rs, "requester_user_id"),
|
||||
nullableLong(rs, "selected_goal_id"));
|
||||
}
|
||||
|
||||
private String writeParts(List<MessageContentPart> parts) {
|
||||
@ -190,13 +198,22 @@ public class ConversationInputQueueStore {
|
||||
String cancelReason,
|
||||
LocalDateTime createdAt,
|
||||
LocalDateTime updatedAt,
|
||||
Long requesterUserId) {
|
||||
Long requesterUserId,
|
||||
Long selectedGoalId) {
|
||||
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, Long requesterUserId) {
|
||||
this(id, conversationId, agentId, createdBy, message, contentParts, state,
|
||||
claimedByAttemptId, persistedMessageId, cancelReason, createdAt, updatedAt,
|
||||
requesterUserId, null);
|
||||
}
|
||||
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);
|
||||
claimedByAttemptId, persistedMessageId, cancelReason, createdAt, updatedAt, null, null);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -0,0 +1,3 @@
|
||||
-- Snapshot the selected managed Goal when an authenticated Web follow-up is queued.
|
||||
-- NULL is an old, unknown selection; 0 explicitly means no managed Goal was selected.
|
||||
ALTER TABLE mate_conversation_input_queue ADD COLUMN selected_goal_id BIGINT NULL;
|
||||
@ -0,0 +1,3 @@
|
||||
-- Snapshot the selected managed Goal when an authenticated Web follow-up is queued.
|
||||
-- NULL is an old, unknown selection; 0 explicitly means no managed Goal was selected.
|
||||
ALTER TABLE mate_conversation_input_queue ADD COLUMN selected_goal_id BIGINT NULL;
|
||||
@ -0,0 +1,3 @@
|
||||
-- Snapshot the selected managed Goal when an authenticated Web follow-up is queued.
|
||||
-- NULL is an old, unknown selection; 0 explicitly means no managed Goal was selected.
|
||||
ALTER TABLE mate_conversation_input_queue ADD COLUMN selected_goal_id BIGINT NULL;
|
||||
@ -18,7 +18,7 @@ Prefix: `/api/v1/goals/{goalId}/json-acceptance`. An enabled account with conver
|
||||
|
||||
Publication requires an active or paused goal and a slot referenced by a current requirement. Content must be a strict JSON object: duplicate keys, trailing documents, nesting beyond 32 levels and UTF-8 content over 1 MiB are rejected. Each goal can retain at most 32 versions; the limit rejects new publication instead of overwriting history. Each version expires after 24 hours. Republishing identical bytes still creates a new version. Reload after a generation conflict rather than automatically overwriting another publication. Retries can reuse a suitable current version and refresh its check binding; reaching the quota still permits checking and completing with that version. If the version is expired or its content must change and all 32 versions are used, further publication remains unavailable.
|
||||
|
||||
Managed bodies live independently in the database. Ordinary workspace files, cache paths and hashes in text are not substitutes. No publication API edits historical bodies; bodies and pointers commit together. SHA-256 identifies content and supports integrity checks; it does not isolate an attacker with database credentials or host privileges. The database and service host are trusted foundations of this limited protocol. The JSON service contract has been exercised on H2, MySQL 8.0.46 and PostgreSQL 16.14. MySQL and PostgreSQL each passed 72 JSON protocol cases on cycle060 and 11 opt-in HTTP approval/authentication cases on cycle061. The MySQL run isolated an existing V192 migration failure using a test-only migration copy; PostgreSQL used the original Kingbase migration tree while skipping an unrelated bundled-skill import failure. These are protocol tests, not confirmation that an unmodified full installation succeeds. The proprietary Kingbase engine has not been tested.
|
||||
Managed bodies live independently in the database. Ordinary workspace files, cache paths and hashes in text are not substitutes. No publication API edits historical bodies; bodies and pointers commit together. SHA-256 identifies content and supports integrity checks; it does not isolate an attacker with database credentials or host privileges. The database and service host are trusted foundations of this limited protocol. The JSON service contract has been exercised on H2, MySQL 8.0.46 and PostgreSQL 16.14. MySQL and PostgreSQL each passed 72 JSON protocol cases and 13 opt-in HTTP approval/authentication cases with V201 on cycle062. The MySQL run isolated an existing V192 migration failure using a test-only migration copy; PostgreSQL used the original Kingbase migration tree while skipping an unrelated bundled-skill import failure. These are protocol tests, not confirmation that an unmodified full installation succeeds. The proprietary Kingbase engine has not been tested.
|
||||
|
||||
## Agent publication
|
||||
|
||||
@ -59,7 +59,7 @@ Built-in shell/code execution is not OS-isolated from the service host. Selectin
|
||||
|
||||
Recovery attempts receive guidance to inspect existing evidence before repeating work. If the first recovered segment is deferred before execution, its recovery context is retained for the next claim. Ordinary continuation after an executed segment does not become a new recovery.
|
||||
|
||||
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.
|
||||
From V199, queued Web input stores the authenticated account ID at enqueue time, and ordinary Web replay carries the conversation workspace. V201 also stores the selected managed Goal ID at enqueue time. A queued turn keeps that identity if the Goal ends before dequeue; approval then refuses to execute the stale selection. An old queue row without a selection snapshot is saved as user text and requires a fresh request when its conversation has managed Goal history. 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.
|
||||
|
||||
Approval replay restores the persisted runtime identity; approval does not renew an expired attempt lease or override account revocation. Legacy snapshots without an authenticated account ID cannot gain managed JSON access from a display username alone.
|
||||
|
||||
|
||||
@ -18,7 +18,7 @@
|
||||
|
||||
仅当前要求引用的槽可发布，Goal 必须 active 或 paused。正文必须是严格 JSON 对象，拒绝重复键、尾随文档、超过 32 层的嵌套及超过 1 MiB 的 UTF-8 内容。每个 Goal 最多保存 32 个版本；达到配额拒绝继续发布，不覆盖旧版本。每版有效期 24 小时，重复发布同样正文也产生新版本。客户端遇到 generation 冲突应重新读取，不自动覆盖他人发布。重试可以复用适用的当前版本并更新检查绑定；达到配额后仍可检查当前版本并在合格时完成。如果版本已过期或正文必须修改且32个版本均已使用，则不能继续发布。
|
||||
|
||||
这些版本独立存储在数据库，不能用普通工作区文件、缓存路径或文字中的 hash 替代。发布接口不支持更新历史正文；所有版本与槽指针同事务保存。SHA-256 用于标识及完整性核对，不能隔离拥有数据库凭据或宿主权限的攻击者；数据库和服务宿主是此有限协议的可信基础。JSON 服务契约已在 H2、MySQL 8.0.46 和 PostgreSQL 16.14 上实测。MySQL与PostgreSQL在cycle060各通过72项JSON协议案例，在cycle061各通过11项显式启用的HTTP审批/认证案例。MySQL 使用仅修正既有 V192 失败的临时迁移副本；PostgreSQL 使用原始 Kingbase 迁移树，跳过无关的内置技能导入失败。这是协议验证，不能代表未修改的完整安装成功；尚未实测 Kingbase 专有引擎。
|
||||
这些版本独立存储在数据库，不能用普通工作区文件、缓存路径或文字中的 hash 替代。发布接口不支持更新历史正文；所有版本与槽指针同事务保存。SHA-256 用于标识及完整性核对，不能隔离拥有数据库凭据或宿主权限的攻击者；数据库和服务宿主是此有限协议的可信基础。JSON 服务契约已在 H2、MySQL 8.0.46 和 PostgreSQL 16.14 上实测。MySQL与PostgreSQL在带V201的cycle062源码上各通过72项JSON协议案例及13项显式启用的HTTP审批/认证案例。MySQL 使用仅修正既有 V192 失败的临时迁移副本；PostgreSQL 使用原始 Kingbase 迁移树，跳过无关的内置技能导入失败。这是协议验证，不能代表未修改的完整安装成功；尚未实测 Kingbase 专有引擎。
|
||||
|
||||
## 代理发布
|
||||
|
||||
@ -57,7 +57,7 @@ V198 同样以绝对时间保存调度租约截止。升级时旧租约失效，
|
||||
|
||||
内置 shell/code 执行没有与服务宿主做操作系统隔离。选择 JSON 验收不会把这些工具变成沙箱；此协议不能抵抗能访问数据库凭据或文件的宿主代码，环境变量名称过滤和工作区路径检查也不能替代隔离。租约截止从绝对时刻计算，覆盖夏令时回拨；调度显示字段仍使用本地时间戳。
|
||||
|
||||
Web排队消息从V199起保存入队时已认证账户的内部ID，普通Web续跑同时携带当前会话工作区；受管工具执行时仍重新校验账户、归属和当前要求。旧队列项不按用户名补造身份，不能用于受管JSON操作；需要用户重新发送已认证请求。持久Goal工作器消费输入时继续使用原有attempt owner校验，没有转换成免租约的账户路径。
|
||||
Web排队消息从V199起保存入队时已认证账户的内部ID，普通Web续跑同时携带当前会话工作区；V201还在入队时保存选定的受管Goal ID。若Goal在出队前终结，排队请求仍保留其身份，随后批准会拒绝旧选择。升级前没有选择快照的队列行，若会话有受管Goal历史，就保存为用户文字并要求重新发送。受管工具执行时仍重新校验账户、归属和当前要求。旧队列项不按用户名补造身份，不能用于受管JSON操作；需要用户重新发送已认证请求。持久Goal工作器消费输入时继续使用原有attempt owner校验，没有转换成免租约的账户路径。
|
||||
|
||||
恢复执行会收到先核实已有证据、不要重放未知副作用的提示。首次恢复执行若在实际运行前延期，下一次领取仍保留恢复关联；已经执行过后的普通续跑不会因此变成新恢复。
|
||||
|
||||
|
||||
@ -18,6 +18,7 @@ import java.util.List;
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.eq;
|
||||
import static org.mockito.ArgumentMatchers.isNull;
|
||||
import static org.mockito.Mockito.inOrder;
|
||||
import static org.mockito.Mockito.mock;
|
||||
import static org.mockito.Mockito.when;
|
||||
@ -35,16 +36,16 @@ class ChatControllerDurableQueueTest {
|
||||
when(authentication.getName()).thenReturn("alice");
|
||||
when(authentication.getDetails()).thenReturn(42L);
|
||||
when(conversations.isConversationOwner("conv", "alice")).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(streams.isRunning("conv")).thenReturn(true);
|
||||
when(streams.notifyQueuedInput("conv")).thenReturn(true);
|
||||
QueuedInput stored = new QueuedInput(91L, "conv", 2L, "alice", "follow-up",
|
||||
List.of(), "queued", null, null, null,
|
||||
LocalDateTime.now(), LocalDateTime.now());
|
||||
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);
|
||||
eq(List.of()), eq(42L), isNull(), any())).thenReturn(stored);
|
||||
|
||||
ChatController controller = new ChatController(agents, conversations, approvals, streams,
|
||||
new ObjectMapper(), mock(ConversationCompletionPublisher.class),
|
||||
@ -52,7 +53,7 @@ class ChatControllerDurableQueueTest {
|
||||
mock(OfficePreviewService.class), queue);
|
||||
ChatController.InterruptRequest request = new ChatController.InterruptRequest();
|
||||
request.setMessage("follow-up");
|
||||
request.setAgentId(2L);
|
||||
request.setAgentId(null); // resolved from the current conversation before snapshotting
|
||||
request.setContentParts(List.of());
|
||||
|
||||
var response = controller.interruptStream("conv", request, authentication);
|
||||
@ -61,7 +62,7 @@ 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()), eq(42L), any());
|
||||
eq("follow-up"), eq(List.of()), eq(42L), isNull(), any());
|
||||
order.verify(streams).notifyQueuedInput("conv");
|
||||
}
|
||||
@Test
|
||||
@ -71,7 +72,7 @@ class ChatControllerDurableQueueTest {
|
||||
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);
|
||||
List.of(), "claimed", "claim", 100L, null, LocalDateTime.now(), LocalDateTime.now(), 42L, 7L);
|
||||
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();
|
||||
@ -91,6 +92,42 @@ class ChatControllerDurableQueueTest {
|
||||
assertThat(origin.getValue().requesterId()).isEqualTo("alice");
|
||||
assertThat(origin.getValue().workspaceId()).isEqualTo(3L);
|
||||
assertThat(origin.getValue().originMessageId()).isEqualTo(100L);
|
||||
assertThat(origin.getValue().selectedGoalId()).isEqualTo(7L);
|
||||
}
|
||||
|
||||
@Test
|
||||
void oldQueuedInputWithManagedHistoryDoesNotStartAnUnselectedTurn() {
|
||||
AgentService agents = mock(AgentService.class);
|
||||
ConversationService conversations = mock(ConversationService.class);
|
||||
ConversationInputQueueStore queue = mock(ConversationInputQueueStore.class);
|
||||
var input = new QueuedInput(92L, "conv", 2L, "alice", "old queued text",
|
||||
List.of(), "claimed", "claim", null, null,
|
||||
LocalDateTime.now(), LocalDateTime.now(), 42L, null);
|
||||
when(queue.claimNext(eq("conv"), any(), any())).thenReturn(java.util.Optional.of(input));
|
||||
var saved = new vip.mate.workspace.conversation.model.MessageEntity(); saved.setId(101L);
|
||||
when(conversations.saveMessage("conv", "user", "old queued text", List.of(), "queued"))
|
||||
.thenReturn(saved);
|
||||
when(queue.bindMessage(eq(92L), any(), eq(101L), any())).thenReturn(true);
|
||||
when(queue.consume(eq(92L), 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);
|
||||
var runs = mock(vip.mate.goal.service.GoalApprovalRunService.class);
|
||||
when(runs.hasManagedGoalHistory("conv", "2")).thenReturn(true);
|
||||
ChatController controller = new ChatController(agents, conversations, mock(ApprovalWorkflowService.class),
|
||||
mock(ChatStreamTracker.class), new ObjectMapper(), mock(ConversationCompletionPublisher.class),
|
||||
mock(MemoryOwnerResolver.class), mock(ChatUploadLocationResolver.class),
|
||||
mock(OfficePreviewService.class), queue);
|
||||
org.springframework.test.util.ReflectionTestUtils.setField(controller, "goalApprovalRuns", runs);
|
||||
|
||||
org.springframework.test.util.ReflectionTestUtils.invokeMethod(controller, "startQueuedMessage", "conv",
|
||||
new org.springframework.web.servlet.mvc.method.annotation.SseEmitter(),
|
||||
new java.util.concurrent.atomic.AtomicBoolean(false), "alice", "http://localhost");
|
||||
|
||||
org.mockito.Mockito.verify(conversations).saveMessage("conv", "user", "old queued text", List.of(), "queued");
|
||||
org.mockito.Mockito.verify(queue).bindMessage(eq(92L), any(), eq(101L), any());
|
||||
org.mockito.Mockito.verify(queue).consume(eq(92L), any(), any());
|
||||
org.mockito.Mockito.verifyNoInteractions(agents);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@ -34,7 +34,10 @@ class ConversationInputQueueStoreTest {
|
||||
.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);
|
||||
new ResourceDatabasePopulator(
|
||||
new ClassPathResource("db/migration/h2/V199__queued_input_account_identity.sql"),
|
||||
new ClassPathResource("db/migration/h2/V201__queued_input_selected_goal.sql"))
|
||||
.execute(dataSource);
|
||||
mapper = new ObjectMapper();
|
||||
store = new ConversationInputQueueStore(jdbc, mapper);
|
||||
}
|
||||
@ -63,11 +66,15 @@ class ConversationInputQueueStoreTest {
|
||||
void accountIdentitySurvivesReconstructionAndLegacyEntriesStayUnasserted() {
|
||||
assertThat(store.get(7L).message()).isEqualTo("pre-upgrade input");
|
||||
assertThat(store.get(7L).requesterUserId()).isNull();
|
||||
assertThat(store.get(7L).selectedGoalId()).isNull();
|
||||
var known = store.enqueue("conv", 1L, "mate", "known", List.of(), 9223372036854775801L, now);
|
||||
var selected = store.enqueue("conv", 1L, "mate", "selected", List.of(),
|
||||
9223372036854775801L, 9223372036854775799L, 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(selected.id()).selectedGoalId()).isEqualTo(9223372036854775799L);
|
||||
assertThat(restarted.get(legacy.id()).requesterUserId()).isNull();
|
||||
}
|
||||
|
||||
|
||||
@ -18,7 +18,8 @@ class GoalJsonExternalApprovalIntegrationTest extends GoalJsonHttpRuntimeIntegra
|
||||
"false,terminal-approval,true", "true,terminal-approval,true",
|
||||
"false,legacy-terminal-approval,true", "true,legacy-terminal-approval,true",
|
||||
"false,originless-terminal-approval,true", "true,originless-terminal-approval,true",
|
||||
"false,late-terminal-approval,true", "true,late-terminal-approval,true"})
|
||||
"false,late-terminal-approval,true", "true,late-terminal-approval,true",
|
||||
"false,queued-terminal-approval,true", "true,queued-terminal-approval,true"})
|
||||
void authenticatedGoalCompletesThroughHttpOrScheduledProductionRuntime(
|
||||
boolean plan, String entry, boolean accepted) throws Exception {
|
||||
super.authenticatedGoalCompletesThroughHttpOrScheduledProductionRuntime(plan, entry, accepted);
|
||||
|
||||
@ -93,20 +93,22 @@ class GoalJsonHttpRuntimeIntegrationTest {
|
||||
"false,terminal-approval,true", "true,terminal-approval,true",
|
||||
"false,legacy-terminal-approval,true", "true,legacy-terminal-approval,true",
|
||||
"false,originless-terminal-approval,true", "true,originless-terminal-approval,true",
|
||||
"false,late-terminal-approval,true", "true,late-terminal-approval,true"})
|
||||
"false,late-terminal-approval,true", "true,late-terminal-approval,true",
|
||||
"false,queued-terminal-approval,true", "true,queued-terminal-approval,true"})
|
||||
void authenticatedGoalCompletesThroughHttpOrScheduledProductionRuntime(boolean plan, String entry, boolean accepted) throws Exception {
|
||||
boolean approval = entry.endsWith("approval");
|
||||
boolean doubleApproval = entry.equals("scheduled-double-approval");
|
||||
boolean reassigned = entry.contains("reassigned");
|
||||
boolean terminal = entry.contains("terminal-");
|
||||
boolean lateTerminal = entry.startsWith("late-");
|
||||
boolean queuedTerminal = entry.startsWith("queued-terminal-");
|
||||
boolean detached = entry.contains("detached");
|
||||
boolean foreign = entry.contains("foreign");
|
||||
boolean supervised = entry.startsWith("supervised");
|
||||
boolean scheduled = entry.startsWith("scheduled") || entry.equals("recovered") || supervised;
|
||||
boolean reuse = entry.equals("reuse");
|
||||
boolean recheck = entry.equals("recheck");
|
||||
boolean queued = entry.equals("queued");
|
||||
boolean queued = entry.equals("queued") || queuedTerminal;
|
||||
boolean recovered = entry.equals("recovered") || entry.equals("supervised-recovered");
|
||||
String username = "http-json-" + UUID.randomUUID();
|
||||
String conversation = UUID.randomUUID().toString();
|
||||
@ -325,6 +327,39 @@ class GoalJsonHttpRuntimeIntegrationTest {
|
||||
assertTrue(coordinator.settle(run, outcome, java.time.LocalDateTime.now()));
|
||||
assertEquals("waiting_approval", continuations.get(goal.getId()).state());
|
||||
waiting = outcome.toString();
|
||||
} else if (queuedTerminal) {
|
||||
String queuedToken = token;
|
||||
var initialTurn = java.util.concurrent.CompletableFuture.supplyAsync(() -> {
|
||||
try {
|
||||
return requestBody("POST", "/api/v1/chat/stream", queuedToken,
|
||||
Map.of("agentId", String.valueOf(agentId), "conversationId", conversation,
|
||||
"message", "Wait for a queued approval fixture."));
|
||||
} catch (Exception failure) {
|
||||
throw new java.util.concurrent.CompletionException(failure);
|
||||
}
|
||||
});
|
||||
try {
|
||||
assertTrue(firstSubscribed.await(10, java.util.concurrent.TimeUnit.SECONDS));
|
||||
JsonNode enqueue = request("POST", "/api/v1/chat/" + conversation + "/interrupt", token,
|
||||
Map.of("agentId", String.valueOf(agentId), "message", message));
|
||||
assertTrue(enqueue.path("data").path("queued").asBoolean(), enqueue.toString());
|
||||
long queueId = Long.parseLong(enqueue.path("data").path("queueItemId").asText());
|
||||
assertEquals(userId, jdbc.queryForObject("SELECT requester_user_id FROM mate_conversation_input_queue WHERE id=?", Long.class, queueId));
|
||||
assertEquals(goal.getId(), jdbc.queryForObject(
|
||||
"SELECT selected_goal_id FROM mate_conversation_input_queue WHERE id=?", Long.class, queueId));
|
||||
goals.abandon(goal.getId(), username);
|
||||
assertEquals(GoalStatus.ABANDONED, goals.getById(goal.getId()).getStatus());
|
||||
String initialAnswer = plan
|
||||
? "{\"needs_planning\":false,\"direct_answer\":\"Initial fixture turn finished.\"}"
|
||||
: "Initial fixture turn finished.";
|
||||
assertEquals(reactor.core.publisher.Sinks.EmitResult.OK, initialResponse.tryEmitValue(
|
||||
new ChatResponse(List.of(new Generation(new AssistantMessage(initialAnswer))))));
|
||||
waiting = initialTurn.get(45, java.util.concurrent.TimeUnit.SECONDS);
|
||||
assertEquals("consumed", jdbc.queryForObject("SELECT state FROM mate_conversation_input_queue WHERE id=?", String.class, queueId));
|
||||
} finally {
|
||||
initialResponse.tryEmitEmpty();
|
||||
initialTurn.cancel(true);
|
||||
}
|
||||
} else if (lateTerminal) {
|
||||
String inFlightToken = token;
|
||||
var inFlight = java.util.concurrent.CompletableFuture.supplyAsync(() -> {
|
||||
@ -349,7 +384,7 @@ class GoalJsonHttpRuntimeIntegrationTest {
|
||||
assertEquals(1, pending.size(), waiting);
|
||||
String pendingId = pending.get(0).path("pendingId").asText();
|
||||
assertEquals("getManagedGoalJsonSlots", pending.get(0).path("toolName").asText());
|
||||
assertEquals(lateTerminal ? GoalStatus.ABANDONED : GoalStatus.ACTIVE,
|
||||
assertEquals(lateTerminal || queuedTerminal ? GoalStatus.ABANDONED : GoalStatus.ACTIVE,
|
||||
goals.getById(goal.getId()).getStatus());
|
||||
assertEquals(0, jdbc.queryForObject("SELECT COUNT(*) FROM mate_goal_json_artifact WHERE goal_id=?", Integer.class, goal.getId()));
|
||||
String persistedOrigin = jdbc.queryForObject("SELECT chat_origin FROM mate_tool_approval WHERE pending_id=?", String.class, pendingId);
|
||||
@ -363,7 +398,7 @@ class GoalJsonHttpRuntimeIntegrationTest {
|
||||
else {
|
||||
assertEquals(userId, approvals.restoreChatOrigin(persistedOrigin).requesterUserId());
|
||||
assertEquals(goal.getId(), approvals.restoreChatOrigin(persistedOrigin).selectedGoalId());
|
||||
if (!lateTerminal) {
|
||||
if (!lateTerminal && !queuedTerminal) {
|
||||
var approvalReplayOrigin = approvals.restoreChatOrigin(persistedOrigin)
|
||||
.withSelectedGoalId(null).withApprovalId(pendingId);
|
||||
assertEquals(goal.getId(), approvalRuns.captureSelectedGoal(approvalReplayOrigin).selectedGoalId(),
|
||||
@ -382,7 +417,7 @@ class GoalJsonHttpRuntimeIntegrationTest {
|
||||
approvals.getPending(pendingId).orElseThrow().setChatOrigin(null);
|
||||
jdbc.update("UPDATE mate_tool_approval SET chat_origin=NULL WHERE pending_id=?", pendingId);
|
||||
}
|
||||
if (!lateTerminal) goals.abandon(goal.getId(), username);
|
||||
if (!lateTerminal && !queuedTerminal) goals.abandon(goal.getId(), username);
|
||||
assertEquals(GoalStatus.ABANDONED, goals.getById(goal.getId()).getStatus());
|
||||
if (entry.startsWith("legacy-")) {
|
||||
var oldApprovalOrigin = approvals.restoreChatOrigin(
|
||||
|
||||
@ -38,7 +38,8 @@ class GoalRecoveryServiceTest {
|
||||
new ClassPathResource("db/migration/h2/V189__goal_attempt_and_input_queue.sql"),
|
||||
new ClassPathResource("db/migration/h2/V198__goal_absolute_owner_leases.sql"),
|
||||
new ClassPathResource("db/migration/h2/V200__goal_approval_attempt_handoff.sql"),
|
||||
new ClassPathResource("db/migration/h2/V199__queued_input_account_identity.sql")).execute(ds);
|
||||
new ClassPathResource("db/migration/h2/V199__queued_input_account_identity.sql"),
|
||||
new ClassPathResource("db/migration/h2/V201__queued_input_selected_goal.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