From 68c906a20ca8ace359a695830d4759f7de91ea39 Mon Sep 17 00:00:00 2001 From: itcast Date: Mon, 12 Oct 2026 08:46:44 +0800 Subject: [PATCH] fix(service): mirror admitted personal chat events --- .../builder/control/ManagedEventOutbox.java | 28 +++- .../ManagedEventOutboxPersistenceTest.java | 139 ++++++++++++++++++ .../control/ManagedEventOutboxTest.java | 106 +++++++++++++ docs/v2/en/service/sessions.md | 2 + docs/v2/zh/service/sessions.md | 2 + 5 files changed, 269 insertions(+), 8 deletions(-) diff --git a/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/control/ManagedEventOutbox.java b/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/control/ManagedEventOutbox.java index d9ea2fabd5..661ef1b01e 100644 --- a/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/control/ManagedEventOutbox.java +++ b/agentscope-service/service-dataplane/src/main/java/io/agentscope/builder/control/ManagedEventOutbox.java @@ -69,27 +69,39 @@ public ManagedEventOutbox( /** Called inside the source-event transaction. Failure must roll it back. */ public void enqueue(SessionEventDto event, SessionEventScope scope) { - if (scope == null || !scope.managed()) return; + SessionEventScope captured = scope == null ? SessionEventScope.NONE : scope; + if (!captured.managed()) { + // Generic events from task sessions must not block the queue with unfenced reports. + // A durable admission without an Attempt identifies an admitted personal Chat. + if (!captured.forNativeFact().equals(SessionEventScope.NONE.forNativeFact()) + || scopes.findBySessionIdAndAttemptId(event.sessionId(), null).isEmpty()) + return; + } if (!TransactionSynchronizationManager.isActualTransactionActive()) throw new IllegalStateException("Mirror outbox requires the source event transaction"); - String report = json(client.sessionEventReport(event, scope)); + // Keep existing NOT NULL outbox schemas compatible. The empty value is local only; + // personal Chat reports omit the orchestration fence entirely. + String attemptId = captured.managed() ? captured.attemptId() : ""; + String report = json(client.sessionEventReport(event, captured)); var old = events.findById(event.id()); if (old.isPresent()) { if (!old.get().sessionId.equals(event.sessionId()) - || !old.get().attemptId.equals(scope.attemptId()) + || !old.get().attemptId.equals(attemptId) || !tree(old.get().reportJson).equals(tree(report))) throw new IllegalStateException( "Mirror event identity or fence changed: " + event.id()); return; } - var fence = lockFence(event.sessionId(), scope.attemptId()); - if (fence.sealed || (fence.closing && !scope.nativeFact())) - throw new IllegalStateException( - "Managed Attempt events are permanently sealed: " + scope.attemptId()); + if (captured.managed()) { + var fence = lockFence(event.sessionId(), attemptId); + if (fence.sealed || (fence.closing && !captured.nativeFact())) + throw new IllegalStateException( + "Managed Attempt events are permanently sealed: " + attemptId); + } var row = new SessionEventOutboxEntity(); row.eventId = event.id(); row.sessionId = event.sessionId(); - row.attemptId = scope.attemptId(); + row.attemptId = attemptId; row.eventSeq = event.seq(); row.reportJson = report; row.createdAt = System.currentTimeMillis(); diff --git a/agentscope-service/service-dataplane/src/test/java/io/agentscope/builder/control/ManagedEventOutboxPersistenceTest.java b/agentscope-service/service-dataplane/src/test/java/io/agentscope/builder/control/ManagedEventOutboxPersistenceTest.java index ff91415aeb..992d44b8ee 100644 --- a/agentscope-service/service-dataplane/src/test/java/io/agentscope/builder/control/ManagedEventOutboxPersistenceTest.java +++ b/agentscope-service/service-dataplane/src/test/java/io/agentscope/builder/control/ManagedEventOutboxPersistenceTest.java @@ -17,15 +17,33 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.agentscope.builder.web.managed.service.DeletedSessionRegistry; +import io.agentscope.builder.web.managed.service.ManagedJsonHelper; +import io.agentscope.builder.web.managed.service.SessionEventLog; +import io.agentscope.builder.web.managed.service.SessionEventNotifier; +import io.agentscope.builder.web.managed.service.SessionEventScope; +import io.agentscope.builder.web.persistence.jpa.SessionEventEntity; +import io.agentscope.builder.web.persistence.jpa.SessionEventEntityRepository; import io.agentscope.builder.web.persistence.jpa.SessionEventFenceEntity; import io.agentscope.builder.web.persistence.jpa.SessionEventFenceRepository; import io.agentscope.builder.web.persistence.jpa.SessionEventOutboxEntity; import io.agentscope.builder.web.persistence.jpa.SessionEventOutboxRepository; import io.agentscope.builder.web.persistence.jpa.SessionExecutionScopeEntity; import io.agentscope.builder.web.persistence.jpa.SessionExecutionScopeRepository; +import java.util.List; +import java.util.Map; import org.hibernate.cfg.Configuration; import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; import org.springframework.data.domain.PageRequest; import org.springframework.data.jpa.repository.support.JpaRepositoryFactory; import org.springframework.orm.jpa.JpaTransactionManager; @@ -33,6 +51,127 @@ import org.springframework.transaction.support.TransactionTemplate; class ManagedEventOutboxPersistenceTest { + @Test + @SuppressWarnings("unchecked") + void personalChatCommitsWithTheSourceLogAndDeliversInOrderAfterRestart() { + try (var factory = + new Configuration() + .addAnnotatedClass(SessionEventEntity.class) + .addAnnotatedClass(SessionEventOutboxEntity.class) + .addAnnotatedClass(SessionEventFenceEntity.class) + .addAnnotatedClass(SessionExecutionScopeEntity.class) + .setProperty( + "hibernate.connection.url", + "jdbc:h2:mem:chat-outbox;DB_CLOSE_DELAY=-1") + .setProperty("hibernate.hbm2ddl.auto", "create-drop") + .buildSessionFactory()) { + var repositories = + new JpaRepositoryFactory( + SharedEntityManagerCreator.createSharedEntityManager(factory)); + var events = repositories.getRepository(SessionEventOutboxRepository.class); + var sources = repositories.getRepository(SessionEventEntityRepository.class); + var scopes = repositories.getRepository(SessionExecutionScopeRepository.class); + var fences = repositories.getRepository(SessionEventFenceRepository.class); + var tx = new TransactionTemplate(new JpaTransactionManager(factory)); + var mapper = new ObjectMapper(); + var client = spy(new ControlPlaneClient("http://unused.invalid", "test", mapper)); + var outbox = new ManagedEventOutbox(events, scopes, fences, tx, mapper, client); + var log = + new SessionEventLog( + sources, + new ManagedJsonHelper(mapper), + tx, + new DeletedSessionRegistry(), + mock(SessionEventNotifier.class), + List.of(new ControlPlaneSessionEventMirror(outbox)), + 1000); + outbox.beginAdmission("chat", "chat-admission", SessionEventScope.NONE); + outbox.bindRun("chat-admission", "chat-run", 0); + assertThatThrownBy( + () -> + tx.executeWithoutResult( + status -> { + log.append( + "chat", + "message.created", + Map.of("text", "rolled back"), + "rolled-back"); + throw new IllegalStateException( + "source transaction failed"); + })) + .isInstanceOf(IllegalStateException.class); + assertThat(sources.findAll()).isEmpty(); + assertThat(events.findAll()).isEmpty(); + var user = + log.appendIdempotentScoped( + "chat", + "item.completed", + message("user", "hello"), + "user-event", + SessionEventScope.NONE); + var assistant = + log.appendIdempotentScoped( + "chat", + "item.completed", + message("assistant", "reply"), + "assistant-event", + outbox.scopeForRun("chat", "chat-run").forNativeFact()); + log.appendIdempotentScoped( + "chat", + "item.completed", + assistant.payload(), + assistant.id(), + SessionEventScope.NONE.forNativeFact()); + assertThat(sources.count()).isEqualTo(2); + assertThat(events.count()).isEqualTo(2); + assertThat(fences.count()).isZero(); + assertThat(events.findDeliverable(System.currentTimeMillis(), PageRequest.of(0, 8))) + .extracting(event -> event.eventId) + .containsExactly(user.id()); + doThrow(new IllegalStateException("CP unavailable")) + .doNothing() + .when(client) + .sendSessionEventReport(anyString(), any()); + assertThat(outbox.deliver(user.id())).isFalse(); + assertThat(events.findDeliverable(System.currentTimeMillis(), PageRequest.of(0, 8))) + .isEmpty(); + tx.executeWithoutResult( + status -> { + var row = events.lockById(user.id()).orElseThrow(); + row.nextAttemptAt = 0; + events.saveAndFlush(row); + }); + var restarted = new ManagedEventOutbox(events, scopes, fences, tx, mapper, client); + assertThat(restarted.deliver(user.id())).isTrue(); + assertThat(events.findDeliverable(System.currentTimeMillis(), PageRequest.of(0, 8))) + .extracting(event -> event.eventId) + .containsExactly(assistant.id()); + assertThat(restarted.deliver(assistant.id())).isTrue(); + assertThat(events.findDeliverable(System.currentTimeMillis(), PageRequest.of(0, 8))) + .isEmpty(); + ArgumentCaptor> reports = ArgumentCaptor.forClass(Map.class); + verify(client, times(3)).sendSessionEventReport(anyString(), reports.capture()); + assertThat(reports.getAllValues()) + .extracting(report -> report.get("type")) + .containsExactly("user.message", "user.message", "agent.message"); + assertThat(reports.getAllValues()) + .extracting(report -> (String) ((Map) report.get("payload")).get("text")) + .containsExactly("hello", "hello", "reply"); + } + } + + private static Map message(String role, String text) { + return Map.of( + "item", + Map.of( + "id", + role + "-item", + "role", + role, + "content", + List.of(Map.of("type", "text", "text", text)))); + } + @Test void rollbackLeavesNoDeliveryAndTheQueryOnlyClaimsTheEarliestPendingSessionEvent() { try (var factory = diff --git a/agentscope-service/service-dataplane/src/test/java/io/agentscope/builder/control/ManagedEventOutboxTest.java b/agentscope-service/service-dataplane/src/test/java/io/agentscope/builder/control/ManagedEventOutboxTest.java index 1d7366bb8f..7639a7bb66 100644 --- a/agentscope-service/service-dataplane/src/test/java/io/agentscope/builder/control/ManagedEventOutboxTest.java +++ b/agentscope-service/service-dataplane/src/test/java/io/agentscope/builder/control/ManagedEventOutboxTest.java @@ -24,6 +24,7 @@ import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; import com.fasterxml.jackson.databind.ObjectMapper; @@ -36,7 +37,9 @@ import io.agentscope.builder.web.persistence.jpa.SessionExecutionScopeEntity; import io.agentscope.builder.web.persistence.jpa.SessionExecutionScopeRepository; import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -108,6 +111,17 @@ void prepare() { savedScopes.values().stream() .filter(row -> call.getArgument(0).equals(row.runId)) .findFirst()); + when(scopes.findBySessionIdAndAttemptId(anyString(), any())) + .thenAnswer( + call -> + savedScopes.values().stream() + .filter( + row -> + row.sessionId.equals(call.getArgument(0)) + && Objects.equals( + row.attemptId, + call.getArgument(1))) + .toList()); when(scopes.saveAndFlush(any())) .thenAnswer( call -> { @@ -124,6 +138,98 @@ void clearTransaction() { TransactionSynchronizationManager.clear(); } + @Test + @SuppressWarnings("unchecked") + void personalChatEventsSurviveRestartAndRetryWithoutAnAttemptFence() { + outbox.beginAdmission("session", "chat-admission", SessionEventScope.NONE); + outbox.bindRun("chat-admission", "chat-run", 0); + outbox.finishAdmission("chat-admission"); + var restarted = new ManagedEventOutbox(events, scopes, fences, tx, mapper, client); + var event = + new SessionEventDto( + "chat-event", + "session", + 4, + "item.completed", + Map.of( + "item", + Map.of( + "id", + "reply", + "role", + "assistant", + "content", + List.of(Map.of("type", "text", "text", "hello")))), + null, + 7); + var scope = restarted.scopeForRun("session", "chat-run").forNativeFact(); + restarted.enqueue(event, scope); + restarted.enqueue(event, scope); + assertThat(savedEvents).containsOnlyKeys("chat-event"); + assertThatThrownBy(() -> restarted.enqueue(event, original)) + .isInstanceOf(IllegalStateException.class); + doThrow(new IllegalStateException("CP unavailable")) + .doNothing() + .when(client) + .sendSessionEventReport(anyString(), any()); + assertThat(restarted.deliver(event.id())).isFalse(); + assertThat(savedEvents.get(event.id()).lastError).contains("CP unavailable"); + savedEvents.get(event.id()).nextAttemptAt = 0; + var retry = new ManagedEventOutbox(events, scopes, fences, tx, mapper, client); + assertThat(retry.deliver(event.id())).isTrue(); + assertThat(retry.deliver(event.id())).isFalse(); + ArgumentCaptor> reports = ArgumentCaptor.forClass(Map.class); + verify(client, times(2)).sendSessionEventReport(anyString(), reports.capture()); + assertThat(reports.getAllValues()) + .allSatisfy( + report -> + assertThat(report) + .doesNotContainKeys( + "agentTaskId", + "attemptId", + "dispatchGeneration", + "turnId") + .containsEntry("id", event.id()) + .containsEntry("type", "agent.message")); + assertThat(reports.getAllValues().get(0)).isEqualTo(reports.getAllValues().get(1)); + verifyNoInteractions(fences); + } + + @Test + void genericTaskEventsCannotEnterThePersonalChatOutbox() { + outbox.beginAdmission("personal", "chat-admission", SessionEventScope.NONE); + var event = new SessionEventDto("event", "session", 1, "run.ended", Map.of(), null, 7); + outbox.enqueue(event, SessionEventScope.NONE); + assertThat(savedEvents).isEmpty(); + outbox.beginAdmission("session", "task-admission", original); + outbox.enqueue(event, SessionEventScope.NONE); + assertThat(savedEvents).isEmpty(); + outbox.enqueue(event, original); + assertThat(savedEvents.get("event").attemptId).isEqualTo("attempt-a"); + outbox.enqueue(event, SessionEventScope.NONE); + assertThat(savedEvents.get("event").attemptId).isEqualTo("attempt-a"); + } + + @Test + void incompleteTaskScopeIsNotTreatedAsPersonalChat() { + outbox.beginAdmission("session", "chat-admission", SessionEventScope.NONE); + var event = new SessionEventDto("event", "session", 1, "run.ended", Map.of(), null, 7); + outbox.enqueue(event, new SessionEventScope("tenant", "task", null, 1, "turn")); + outbox.enqueue(event, new SessionEventScope("tenant", "task", "attempt", 0, "turn")); + assertThat(savedEvents).isEmpty(); + verifyNoInteractions(fences); + } + + @Test + void personalChatWritesStillRequireTheSourceTransaction() { + outbox.beginAdmission("session", "chat-admission", SessionEventScope.NONE); + TransactionSynchronizationManager.clear(); + var event = new SessionEventDto("event", "session", 1, "run.ended", Map.of(), null, 7); + assertThatThrownBy(() -> outbox.enqueue(event, SessionEventScope.NONE)) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("source event transaction"); + } + @Test void nativeReplayKeepsOriginalAttemptAfterReplacementAndRestart() { outbox.beginAdmission("session", "admission-a", original); diff --git a/docs/v2/en/service/sessions.md b/docs/v2/en/service/sessions.md index e5b411ca52..709c8540fb 100644 --- a/docs/v2/en/service/sessions.md +++ b/docs/v2/en/service/sessions.md @@ -68,6 +68,8 @@ Reopen the original work and query saved events and current state. SSE ending is Diagnostic endpoints include `/messages` (`offset`, `limit`, `fromEnd`), `/context`, `/events`, `/events/stream`, and `/turns`. Availability depends on the runtime's observation and query capabilities. Registering an Agent does not imply full context access or recovery support. For business-facing session management, use the Agent API below. +Managed Agent personal Chat also synchronizes messages and execution events to the control plane after execution is admitted, even though it has no orchestration Attempt. Delivery is asynchronous and retries after a temporary control-plane outage. Refresh the existing conversation to read delivered events instead of resubmitting the same message. + ## Managed Session log and recovery entry points Agent API correlates session_id → turn_id → run_id; run_id is not an orchestration Run ID. Snapshot supplies items/tools/turns/required_actions and an as_of cursor for stream resumption. GET turns returns command states keyed by the submitted turn ID. diff --git a/docs/v2/zh/service/sessions.md b/docs/v2/zh/service/sessions.md index c1a27f68cb..dbc6a11c65 100644 --- a/docs/v2/zh/service/sessions.md +++ b/docs/v2/zh/service/sessions.md @@ -68,6 +68,8 @@ Pause 不冻结已经开始的外部进程。Cancel 也不自动回滚已经产 诊断接口包括 `/messages`（`offset`、`limit`、`fromEnd`）、`/context`、`/events`、`/events/stream`、`/turns`。内容是否可用取决于运行时提供的观测与查询能力，注册为 Agent 不代表都支持完整上下文或恢复。该组接口用于执行诊断；业务侧会话管理优先使用下述 Agent API。 +Managed Agent 的个人 Chat 在执行被接纳后，也会将消息和执行事件同步到控制面，即使该会话没有编排 Attempt。同步为异步投递，控制面暂时不可用时会重试。刷新原有对话即可读取已投递的事件，无需重复发送同一条消息。 + ## Managed Session 的日志与恢复入口 Agent API 使用 session_id → turn_id → run_id 关联推理工作；run_id 不是编排 Run ID。GET snapshot 返回已保存的 items/tools/turns/required_actions 与公共 as_of cursor，再从 events/stream 续传。GET turns 按提交返回的 turn ID 查询队列和结果。