Skip to content

Route relay /new through thread actor #3613

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

Merged
louis4li merged 2 commits into from
Sep 9, 2026
Merged

Route relay /new through thread actor #3613

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7e2a485
Route relay new conversations through thread actor.
louis4li Sep 9, 2026
fc93c44
Update relay endpoint thread actor tests.
louis4li Sep 9, 2026
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
115 changes: 102 additions & 13 deletions agents/Aevatar.GAgents.Channel.Runtime/Conversation/ChannelConversationThreadGAgent.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,17 +29,42 @@ public async Task HandleInboundActivityAsync(ChatActivity activity)
if (string.IsNullOrWhiteSpace(sourceConversationCanonicalKey))
return;

var activeConversationCanonicalKey = ResolveActiveConversationCanonicalKey(sourceConversationCanonicalKey);
if (ShouldStartNewConversation(activity))
{
activeConversationCanonicalKey = await RotateAsync(
sourceConversationCanonicalKey,
activity.Id ?? string.Empty,
DateTimeOffset.UtcNow.ToUnixTimeMilliseconds())
.ConfigureAwait(false);
}
var activeConversationCanonicalKey = await ResolveTurnConversationCanonicalKeyAsync(
activity,
sourceConversationCanonicalKey,
DateTimeOffset.UtcNow.ToUnixTimeMilliseconds())
.ConfigureAwait(false);

await DispatchToConversationAsync(
activity,
activeConversationCanonicalKey,
conversationActorScopeKey: string.Empty,
CancellationToken.None)
.ConfigureAwait(false);
}

[EventHandler]
public async Task HandleNyxRelayInboundActivityAsync(NyxRelayInboundActivity relayActivity)
{
ArgumentNullException.ThrowIfNull(relayActivity);
var activity = relayActivity.Activity?.Clone() ?? new ChatActivity();
var sourceConversationCanonicalKey = activity.Conversation?.CanonicalKey?.Trim();
if (string.IsNullOrWhiteSpace(sourceConversationCanonicalKey))
return;

var activeConversationCanonicalKey = await ResolveTurnConversationCanonicalKeyAsync(
activity,
sourceConversationCanonicalKey,
relayActivity.CallbackObservedAtUnixMs)
.ConfigureAwait(false);
var scopedRelayActivity = relayActivity.Clone();
scopedRelayActivity.Activity = BuildActivityForLogicalConversation(activity, activeConversationCanonicalKey);

await DispatchToConversationAsync(activity, activeConversationCanonicalKey, CancellationToken.None)
await DispatchToConversationAsync(
scopedRelayActivity,
activeConversationCanonicalKey,
relayActivity.ConversationActorScopeKey,
CancellationToken.None)
.ConfigureAwait(false);
}

Expand Down Expand Up @@ -71,6 +96,22 @@ private string ResolveActiveConversationCanonicalKey(string sourceConversationCa
? sourceConversationCanonicalKey.Trim()
: State.ActiveConversationCanonicalKey;

private async Task<string> ResolveTurnConversationCanonicalKeyAsync(
ChatActivity activity,
string sourceConversationCanonicalKey,
long requestedAtUnixMs)
{
var activeConversationCanonicalKey = ResolveActiveConversationCanonicalKey(sourceConversationCanonicalKey);
if (!ShouldStartNewConversation(activity))
return activeConversationCanonicalKey;

return await RotateAsync(
sourceConversationCanonicalKey,
activity.Id ?? string.Empty,
requestedAtUnixMs)
.ConfigureAwait(false);
}

private async Task<string> RotateAsync(
string sourceConversationCanonicalKey,
string requestedActivityId,
Expand Down Expand Up @@ -101,26 +142,74 @@ await PersistDomainEventAsync(new ConversationThreadRotatedEvent
private async Task DispatchToConversationAsync(
ChatActivity activity,
string activeConversationCanonicalKey,
string conversationActorScopeKey,
CancellationToken ct) =>
await DispatchToConversationAsync(
BuildActivityForLogicalConversation(activity, activeConversationCanonicalKey),
activeConversationCanonicalKey,
conversationActorScopeKey,
activity.Id ?? string.Empty,
ct)
.ConfigureAwait(false);

private async Task DispatchToConversationAsync(
NyxRelayInboundActivity relayActivity,
string activeConversationCanonicalKey,
string conversationActorScopeKey,
CancellationToken ct) =>
await DispatchToConversationAsync(
relayActivity,
activeConversationCanonicalKey,
conversationActorScopeKey,
relayActivity.Activity?.Id ?? string.Empty,
ct)
.ConfigureAwait(false);

private async Task DispatchToConversationAsync(
IMessage payload,
string activeConversationCanonicalKey,
string conversationActorScopeKey,
string correlationId,
CancellationToken ct)
{
var actorRuntime = Services.GetRequiredService<IActorRuntime>();
var dispatchPort = Services.GetRequiredService<IActorDispatchPort>();
var actorId = ConversationGAgent.BuildActorId(activeConversationCanonicalKey);
var actorId = BuildConversationActorId(activeConversationCanonicalKey, conversationActorScopeKey);
var actor = await actorRuntime.CreateAsync<ConversationGAgent>(actorId, ct).ConfigureAwait(false);
await dispatchPort.DispatchAsync(
actor.Id,
new EventEnvelope
{
Id = Guid.NewGuid().ToString("N"),
Timestamp = Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow),
Payload = Any.Pack(activity),
Payload = Any.Pack(payload),
Route = EnvelopeRouteSemantics.CreateDirect(PublisherActorId, actor.Id),
Propagation = new EnvelopePropagation { CorrelationId = activity.Id ?? string.Empty },
Propagation = new EnvelopePropagation { CorrelationId = correlationId },
},
ct)
.ConfigureAwait(false);
}

private static ChatActivity BuildActivityForLogicalConversation(
ChatActivity activity,
string activeConversationCanonicalKey)
{
var logicalActivity = activity.Clone();
if (logicalActivity.Conversation is not null)
logicalActivity.Conversation.CanonicalKey = activeConversationCanonicalKey;
return logicalActivity;
}

private static string BuildConversationActorId(
string activeConversationCanonicalKey,
string conversationActorScopeKey)
{
var actorId = ConversationGAgent.BuildActorId(activeConversationCanonicalKey);
return string.IsNullOrWhiteSpace(conversationActorScopeKey)
? actorId
: $"{actorId}:scope:{conversationActorScopeKey.Trim()}";
}

private static bool ShouldStartNewConversation(ChatActivity activity) =>
activity.Conversation?.Scope == ConversationScope.DirectMessage
&& TryParseSlashCommand(activity.Content?.Text, out var commandName)
Expand Down
3 changes: 3 additions & 0 deletions agents/Aevatar.GAgents.Channel.Runtime/protos/conversation_events.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,6 +163,9 @@ message NyxRelayInboundActivity {
string callback_jti = 6;
int64 callback_observed_at_unix_ms = 7;
int64 callback_replay_expires_at_unix_ms = 8;
// Stable owner-scope discriminator for relay-backed conversation actors. This
// is routing state, not a credential; the raw scope id stays at the HTTP edge.
string conversation_actor_scope_key = 9;
}

// Refactor (iter17/cluster-038): Old pattern: callback_jti replay claims were process-local guard entries. New principle: admission is a persisted typed actor event before any business turn work.
Expand Down
26 changes: 19 additions & 7 deletions agents/Aevatar.GAgents.NyxidChat/NyxIdRelayIngressPort.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,8 +63,11 @@ public async Task<NyxIdRelayIngressAccepted> AcceptAsync(
throw new InvalidOperationException("Relay payload did not resolve to a canonical conversation key.");
}

var actorId = BuildScopedRelayConversationActorId(request.ScopeId, activity.Conversation.CanonicalKey);
var actor = await _actorRuntime.CreateAsync<ConversationGAgent>(actorId, ct);
var conversationActorScopeKey = BuildConversationActorScopeKey(request.ScopeId);
var actorId = BuildScopedRelayConversationThreadActorId(
activity.Conversation.CanonicalKey,
conversationActorScopeKey);
var actor = await _actorRuntime.CreateAsync<ChannelConversationThreadGAgent>(actorId, ct);
Comment thread
louis4li marked this conversation as resolved.
var relayInbound = new NyxRelayInboundActivity
{
Activity = activity,
Expand All @@ -75,6 +78,7 @@ public async Task<NyxIdRelayIngressAccepted> AcceptAsync(
CallbackJti = request.CallbackJti ?? string.Empty,
CallbackObservedAtUnixMs = request.CallbackObservedAtUnixMs,
CallbackReplayExpiresAtUnixMs = request.CallbackReplayExpiresAtUnixMs,
ConversationActorScopeKey = conversationActorScopeKey,
};
var command = new EventEnvelope
{
Expand All @@ -87,7 +91,7 @@ public async Task<NyxIdRelayIngressAccepted> AcceptAsync(
await _actorDispatchPort.DispatchAsync(actor.Id, command, ct);

_logger.LogInformation(
"Accepted relay callback into channel conversation backbone: message={MessageId}, actor={ActorId}, platform={Platform}, activity={ActivityType}",
"Accepted relay callback into channel conversation thread: message={MessageId}, actor={ActorId}, platform={Platform}, activity={ActivityType}",
activity.Id,
actorId,
activity.ChannelId?.Value,
Expand All @@ -96,13 +100,21 @@ public async Task<NyxIdRelayIngressAccepted> AcceptAsync(
return new NyxIdRelayIngressAccepted(activity.Id, actorId);
}

private static string BuildScopedRelayConversationActorId(string? scopeId, string canonicalKey)
private static string BuildConversationActorScopeKey(string? scopeId)
{
ArgumentException.ThrowIfNullOrWhiteSpace(scopeId);
ArgumentException.ThrowIfNullOrWhiteSpace(canonicalKey);

var scopeHash = Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(scopeId.Trim())))
return Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(scopeId.Trim())))
.ToLowerInvariant();
return $"{ConversationGAgent.BuildActorId(canonicalKey)}:scope:{scopeHash}";
}

private static string BuildScopedRelayConversationThreadActorId(
string canonicalKey,
string conversationActorScopeKey)
{
ArgumentException.ThrowIfNullOrWhiteSpace(canonicalKey);
ArgumentException.ThrowIfNullOrWhiteSpace(conversationActorScopeKey);

return $"{ChannelConversationThreadGAgent.BuildActorId(canonicalKey)}:scope:{conversationActorScopeKey.Trim()}";
}
}
62 changes: 32 additions & 30 deletions test/Aevatar.AI.Tests/NyxIdChatEndpointsCoverageTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1941,8 +1941,11 @@ public async Task HandleRelayWebhookAsync_ShouldDispatchCardAction_ToConversatio
response.Body.Should().Contain("msg-card-builder-1");
response.Body.Should().NotContain("unsupported_card_action");

runtime.CreateCalls.Should().ContainSingle(call => call.Type == typeof(ConversationGAgent));
var actor = (StubActor)runtime.Actors.Values.Single();
var threadActorId = runtime.CreateCalls.Should()
.ContainSingle(call => call.Type == typeof(ChannelConversationThreadGAgent))
.Subject.Id!;
threadActorId.Should().Be(BuildScopedRelayConversationThreadActorId("scope-card", "lark:dm:ou_user_b"));
var actor = (StubActor)runtime.Actors[threadActorId];
actor.HandledEnvelopes.Should().ContainSingle(envelope =>
envelope.Payload != null &&
envelope.Payload.Is(NyxRelayInboundActivity.Descriptor));
Expand Down Expand Up @@ -2002,8 +2005,11 @@ public async Task HandleRelayWebhookAsync_ShouldDispatchCardAction_ToConversatio
response.StatusCode.Should().Be(StatusCodes.Status202Accepted);
response.Body.Should().Contain("accepted");

runtime.CreateCalls.Should().ContainSingle(call => call.Type == typeof(ConversationGAgent));
var actor = (StubActor)runtime.Actors.Values.Single();
var threadActorId = runtime.CreateCalls.Should()
.ContainSingle(call => call.Type == typeof(ChannelConversationThreadGAgent))
.Subject.Id!;
threadActorId.Should().Be(BuildScopedRelayConversationThreadActorId("scope-card", "lark:dm:ou_user_wf"));
var actor = (StubActor)runtime.Actors[threadActorId];
var relayInbound = actor.HandledEnvelopes.Should().ContainSingle().Subject.Payload.Unpack<NyxRelayInboundActivity>();
var activity = relayInbound.Activity;
activity.Type.Should().Be(ActivityType.CardAction);
Expand Down Expand Up @@ -2129,16 +2135,16 @@ public async Task HandleRelayWebhookAsync_ShouldAcceptAndDispatchChatActivity_Wh
response.StatusCode.Should().Be(StatusCodes.Status202Accepted);
response.Body.Should().Contain("accepted");
response.Body.Should().Contain("msg-1");
var expectedActorId = BuildScopedRelayConversationActorId("scope-a", "slack:group:room-1");
var expectedThreadActorId = BuildScopedRelayConversationThreadActorId("scope-a", "slack:group:room-1");
runtime.CreateCalls.Should().ContainSingle(call =>
call.Type == typeof(ConversationGAgent) &&
call.Id == expectedActorId);
runtime.Actors.Should().ContainKey(expectedActorId);
var actor = (StubActor)runtime.Actors[expectedActorId];
actor.HandledEnvelopes.Should().ContainSingle(envelope =>
call.Type == typeof(ChannelConversationThreadGAgent) &&
call.Id == expectedThreadActorId);
runtime.Actors.Should().ContainKey(expectedThreadActorId);
var threadActor = (StubActor)runtime.Actors[expectedThreadActorId];
threadActor.HandledEnvelopes.Should().ContainSingle(envelope =>
envelope.Payload != null &&
envelope.Payload.Is(NyxRelayInboundActivity.Descriptor));
var relayInbound = actor.HandledEnvelopes.Single().Payload.Unpack<NyxRelayInboundActivity>();
var relayInbound = threadActor.HandledEnvelopes.Single().Payload.Unpack<NyxRelayInboundActivity>();
relayInbound.ReplyToken.Should().Be("reply-token-1");
relayInbound.CorrelationId.Should().Be("corr-1");
relayInbound.RelayApiKeyId.Should().Be(relay.RelayApiKeyId);
Expand Down Expand Up @@ -2202,17 +2208,17 @@ public async Task HandleRelayWebhookAsync_ShouldDispatchLarkPrivateSummarySlashC
var response = await ExecuteResultAsync(result);
response.StatusCode.Should().Be(StatusCodes.Status202Accepted);
response.Body.Should().Contain("accepted");
var expectedActorId = BuildScopedRelayConversationActorId("scope-summary", "lark:dm:ou_user_1");
var expectedThreadActorId = BuildScopedRelayConversationThreadActorId("scope-summary", "lark:dm:ou_user_1");
runtime.CreateCalls.Should().ContainSingle(call =>
call.Type == typeof(ConversationGAgent) &&
call.Id == expectedActorId);
runtime.Actors.Should().ContainKey(expectedActorId);
call.Type == typeof(ChannelConversationThreadGAgent) &&
call.Id == expectedThreadActorId);
runtime.Actors.Should().ContainKey(expectedThreadActorId);

var actor = (StubActor)runtime.Actors[expectedActorId];
actor.HandledEnvelopes.Should().ContainSingle(envelope =>
var threadActor = (StubActor)runtime.Actors[expectedThreadActorId];
threadActor.HandledEnvelopes.Should().ContainSingle(envelope =>
envelope.Payload != null &&
envelope.Payload.Is(NyxRelayInboundActivity.Descriptor));
var relayInbound = actor.HandledEnvelopes.Single().Payload.Unpack<NyxRelayInboundActivity>();
var relayInbound = threadActor.HandledEnvelopes.Single().Payload.Unpack<NyxRelayInboundActivity>();
relayInbound.ReplyToken.Should().Be("reply-token-summary-1");
relayInbound.CorrelationId.Should().Be("corr-summary-1");
relayInbound.RelayApiKeyId.Should().Be(relay.RelayApiKeyId);
Expand Down Expand Up @@ -2276,7 +2282,7 @@ public async Task HandleRelayWebhookAsync_ShouldStashSenderNyxUserIdOnTransportE

var response = await ExecuteResultAsync(result);
response.StatusCode.Should().Be(StatusCodes.Status202Accepted);
var expectedActorId = BuildScopedRelayConversationActorId("scope-summary", "lark:dm:ou_user_2");
var expectedActorId = BuildScopedRelayConversationThreadActorId("scope-summary", "lark:dm:ou_user_2");
var actor = (StubActor)runtime.Actors[expectedActorId];
var relayInbound = actor.HandledEnvelopes.Single().Payload.Unpack<NyxRelayInboundActivity>();
relayInbound.Activity.TransportExtras.NyxSenderUserId.Should().Be(
Expand Down Expand Up @@ -2331,7 +2337,7 @@ public async Task HandleRelayWebhookAsync_ShouldLeaveSenderNyxUserIdEmpty_WhenRe
response.StatusCode.Should().Be(
StatusCodes.Status202Accepted,
"an unreliable /me must not break ingress; routing falls back to scope-only / default policies");
var expectedActorId = BuildScopedRelayConversationActorId("scope-summary", "lark:dm:ou_user_3");
var expectedActorId = BuildScopedRelayConversationThreadActorId("scope-summary", "lark:dm:ou_user_3");
var actor = (StubActor)runtime.Actors[expectedActorId];
var relayInbound = actor.HandledEnvelopes.Single().Payload.Unpack<NyxRelayInboundActivity>();
relayInbound.Activity.TransportExtras.NyxSenderUserId.Should().BeEmpty();
Expand Down Expand Up @@ -2382,9 +2388,9 @@ public async Task HandleRelayWebhookAsync_ShouldResolveScopeIdFromRegistration_W
var response = await ExecuteResultAsync(result);
response.StatusCode.Should().Be(StatusCodes.Status202Accepted);
scopeResolver.LastNyxAgentApiKeyId.Should().Be("nyx-key-1");
var expectedActorId = BuildScopedRelayConversationActorId("scope-from-registration", "lark:dm:ou_user_1");
var expectedActorId = BuildScopedRelayConversationThreadActorId("scope-from-registration", "lark:dm:ou_user_1");
runtime.CreateCalls.Should().ContainSingle(call =>
call.Type == typeof(ConversationGAgent) &&
call.Type == typeof(ChannelConversationThreadGAgent) &&
call.Id == expectedActorId);
runtime.Actors.Should().ContainKey(expectedActorId);
}
Expand Down Expand Up @@ -2608,9 +2614,9 @@ public async Task HandleRelayWebhookAsync_ShouldUseConversationId_WhenPresent()

var response = await ExecuteResultAsync(result);
response.StatusCode.Should().Be(StatusCodes.Status202Accepted);
var expectedActorId = BuildScopedRelayConversationActorId("scope-b", "discord:channel:conv-1");
var expectedActorId = BuildScopedRelayConversationThreadActorId("scope-b", "discord:channel:conv-1");
runtime.CreateCalls.Should().ContainSingle(call =>
call.Type == typeof(ConversationGAgent) &&
call.Type == typeof(ChannelConversationThreadGAgent) &&
call.Id == expectedActorId);
runtime.Actors.Should().ContainKey(expectedActorId);
}
Expand Down Expand Up @@ -3136,12 +3142,8 @@ private sealed class EmptyServiceProvider : IServiceProvider
public object? GetService(Type serviceType) => null;
}

private static string BuildScopedRelayConversationActorId(string scopeId, string canonicalKey)
{
var scopeHash = Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(scopeId.Trim())))
.ToLowerInvariant();
return $"channel-conversation:{canonicalKey}:scope:{scopeHash}";
}
private static string BuildScopedRelayConversationThreadActorId(string scopeId, string canonicalKey) =>
$"channel-conversation-thread:{canonicalKey}:scope:{Convert.ToHexString(SHA256.HashData(Encoding.UTF8.GetBytes(scopeId.Trim()))).ToLowerInvariant()}";

private static string GetRepositoryRoot()
{
Expand Down
Loading
Loading