Skip to content

fix(service): mirror admitted personal chat events #3512

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
NewPeople-star wants to merge 1 commit into
base: main
Choose a base branch
from
+269 −8
Open

fix(service): mirror admitted personal chat events #3512

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 20 additions & 8 deletions ...ice/service-dataplane/src/main/java/io/agentscope/builder/control/ManagedEventOutbox.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
139 changes: 139 additions & 0 deletions ...aplane/src/test/java/io/agentscope/builder/control/ManagedEventOutboxPersistenceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,22 +17,161 @@

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;
import org.springframework.orm.jpa.SharedEntityManagerCreator;
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<Map<String, Object>> 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<String, Object> 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 =
Expand Down
106 changes: 106 additions & 0 deletions ...service-dataplane/src/test/java/io/agentscope/builder/control/ManagedEventOutboxTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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 -> {
Expand All @@ -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<Map<String, Object>> 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);
Expand Down
2 changes: 2 additions & 0 deletions docs/v2/en/service/sessions.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
Loading
Loading