Skip to content

fix: strip every Astra directive from streamed chat deltas, not only metrics #494

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
thomasluizon merged 3 commits into from
Aug 24, 2026
Merged

fix: strip every Astra directive from streamed chat deltas, not only metrics #494

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
81371f1
chore: start #357
thomasluizon Aug 24, 2026
527da14
fix: strip all directives from chat deltas (#357)
thomasluizon Aug 24, 2026
35e4db8
Merge branch 'main' into fix/ticket-357-strip-all-directives
thomasluizon Aug 24, 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
54 changes: 44 additions & 10 deletions src/Orbit.Application/Chat/Commands/ProcessUserChatCommand.Ai.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -156,16 +156,24 @@ private static AgentContextSnapshot BuildAgentContextSnapshot(
ClientContext: clientContext);
}

private static MetricsDirectiveStreamFilter? BuildAiStreamFilter(Func<ChatStreamEvent, Task>? streamSink)
private static ResponseDirectiveStreamFilter? BuildAiStreamFilter(Func<ChatStreamEvent, Task>? streamSink)
{
if (streamSink is null)
return null;

return new MetricsDirectiveStreamFilter(streamSink);
return new ResponseDirectiveStreamFilter(streamSink);
}

private sealed class MetricsDirectiveStreamFilter(Func<ChatStreamEvent, Task> streamSink)
private sealed class ResponseDirectiveStreamFilter(Func<ChatStreamEvent, Task> streamSink)
{
private static readonly string[] Directives =
[
"[[orbit:habits:today]]",
"[[orbit:habits:all]]",
"[[orbit:goals]]",
MetricsCardBuilder.Directive
];

private string _pending = string.Empty;

public async Task HandleAsync(AiStreamEvent aiEvent)
Expand All @@ -187,11 +195,11 @@ private async Task DrainAsync(bool flush)
{
while (_pending.Length > 0)
{
var directiveIndex = _pending.IndexOf(MetricsCardBuilder.Directive, StringComparison.OrdinalIgnoreCase);
var (directiveIndex, directiveLength) = FindDirective(_pending);
if (directiveIndex >= 0)
{
await EmitAsync(_pending[..directiveIndex]);
_pending = _pending[(directiveIndex + MetricsCardBuilder.Directive.Length)..];
_pending = _pending[(directiveIndex + directiveLength)..];
continue;
}

Expand All @@ -213,14 +221,40 @@ private async Task EmitAsync(string text)

private static int DirectivePrefixSuffixLength(string text)
{
var maximumLength = Math.Min(text.Length, MetricsCardBuilder.Directive.Length - 1);
for (var length = maximumLength; length > 0; length--)
var retainedCharacters = 0;

foreach (var directive in Directives)
{
var maximumLength = Math.Min(text.Length, directive.Length - 1);
for (var length = maximumLength; length > retainedCharacters; length--)
{
if (!text.EndsWith(directive[..length], StringComparison.OrdinalIgnoreCase))
continue;

retainedCharacters = length;
break;
}
}

return retainedCharacters;
}

private static (int Index, int Length) FindDirective(string text)
{
var earliestIndex = -1;
var matchedLength = 0;

foreach (var directive in Directives)
{
if (text.EndsWith(MetricsCardBuilder.Directive[..length], StringComparison.OrdinalIgnoreCase))
return length;
var index = text.IndexOf(directive, StringComparison.OrdinalIgnoreCase);
if (index < 0 || earliestIndex >= 0 && index >= earliestIndex)
continue;

earliestIndex = index;
matchedLength = directive.Length;
}

return 0;
return (earliestIndex, matchedLength);
}
}

Expand Down
161 changes: 143 additions & 18 deletions tests/Orbit.Application.Tests/Commands/Chat/ProcessUserChatCommandHandlerTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -603,33 +603,47 @@ public async Task Handle_MetricsDirectiveWithoutCapability_StripsTokenAndSkipsRe
}

[Fact]
public async Task Handle_MetricsAndHabitDirectives_ReturnsBothCards()
public async Task Handle_StreamedMetricsAndHabitDirectives_StripsBothAndReturnsBothCards()
{
SetupUserAndPayGate();
_habitRepo.FindAsync(
Arg.Any<Expression<Func<Habit, bool>>>(),
Arg.Any<Func<IQueryable<Habit>, IQueryable<Habit>>?>(),
Arg.Any<CancellationToken>())
.Returns(new List<Habit> { CreateHabit("Meditate") }.AsReadOnly());
SetupAiResponse(new AiResponse
{
TextMessage = "Here is today and your week:\n[[orbit:habits:today]]\n[[orbit:metrics]]",
ToolCalls = null
});
const string responseText = "Here is today and your week:\n[[orbit:habits:today]]\n[[orbit:metrics]]";
_aiIntentService.SendWithToolsAsync(
Arg.Any<AiToolRequest>(),
Arg.Any<Func<AiStreamEvent, Task>?>(),
Arg.Any<CancellationToken>())
.Returns(async callInfo =>
{
var sink = callInfo.ArgAt<Func<AiStreamEvent, Task>?>(1);
await sink!(AiStreamEvent.Delta(responseText));
return Result.Success(new AiResponse { TextMessage = responseText });
});
SetupRecap(Metrics(totalScheduled: 1));
var streamEvents = new List<ChatStreamEvent>();
var handler = CreateHandler();

var result = await handler.Handle(
new ProcessUserChatCommand(
UserId,
"Show today and my progress",
ClientContext: new AgentClientContext(SupportsHabitListCard: true, SupportsMetricsCard: true)),
ClientContext: new AgentClientContext(SupportsHabitListCard: true, SupportsMetricsCard: true),
StreamSink: streamEvent =>
{
streamEvents.Add(streamEvent);
return Task.CompletedTask;
}),
CancellationToken.None);

result.IsSuccess.Should().BeTrue();
result.Value.AiMessage.Should().Be("Here is today and your week:");
result.Value.HabitList.Should().NotBeNull();
result.Value.MetricsCard.Should().NotBeNull();
string.Concat(streamEvents.Where(streamEvent => streamEvent.Type == "delta").Select(streamEvent => streamEvent.Text))
.Should().Be("Here is today and your week:\n\n");
}

[Fact]
Expand Down Expand Up @@ -969,30 +983,108 @@ public async Task Handle_WithStreamSink_EmitsRoundPerIterationAndBridgesAiEvents
streamEvents[^1].Type.Should().Be("reset");
}

[Theory]
[InlineData("[[orbit:habits:today]]")]
[InlineData("[[orbit:habits:all]]")]
[InlineData("[[orbit:goals]]")]
[InlineData("[[orbit:metrics]]")]
public async Task Handle_StreamedDirective_NeverEmitsToken(string directive)
{
SetupUserAndPayGate();
_aiIntentService.SendWithToolsAsync(
Arg.Any<AiToolRequest>(),
Arg.Any<Func<AiStreamEvent, Task>?>(),
Arg.Any<CancellationToken>())
.Returns(async callInfo =>
{
var sink = callInfo.ArgAt<Func<AiStreamEvent, Task>?>(1);
await sink!(AiStreamEvent.Delta($"Ready:\n{directive}"));
return Result.Success(new AiResponse { TextMessage = $"Ready:\n{directive}" });
});
var streamEvents = new List<ChatStreamEvent>();
var handler = CreateHandler();

var result = await handler.Handle(
new ProcessUserChatCommand(
UserId,
"Show me",
StreamSink: streamEvent =>
{
streamEvents.Add(streamEvent);
return Task.CompletedTask;
}),
CancellationToken.None);

result.IsSuccess.Should().BeTrue();
string.Concat(streamEvents.Where(streamEvent => streamEvent.Type == "delta").Select(streamEvent => streamEvent.Text))
.Should().Be("Ready:\n");
}

[Fact]
public async Task Handle_StreamedMetricsDirectiveAcrossChunks_NeverEmitsToken()
public async Task Handle_StreamedHabitDirectiveAcrossChunks_StripsTokenAndReturnsCard()
{
SetupUserAndPayGate();
SetupRecap(Metrics(totalScheduled: 1));
_habitRepo.FindAsync(
Arg.Any<Expression<Func<Habit, bool>>>(),
Arg.Any<Func<IQueryable<Habit>, IQueryable<Habit>>?>(),
Arg.Any<CancellationToken>())
.Returns(new List<Habit> { CreateHabit("Meditate") }.AsReadOnly());
_aiIntentService.SendWithToolsAsync(
Arg.Any<AiToolRequest>(),
Arg.Any<Func<AiStreamEvent, Task>?>(),
Arg.Any<CancellationToken>())
.Returns(async callInfo =>
{
var sink = callInfo.ArgAt<Func<AiStreamEvent, Task>?>(1);
await sink!(AiStreamEvent.Delta("Your week:\n[[orbit:met"));
await sink(AiStreamEvent.Delta("rics]]"));
return Result.Success(new AiResponse { TextMessage = "Your week:\n[[orbit:metrics]]" });
await sink!(AiStreamEvent.Delta("Your habits:\n[[orbit:habi"));
await sink(AiStreamEvent.Delta("ts:today]]"));
return Result.Success(new AiResponse { TextMessage = "Your habits:\n[[orbit:habits:today]]" });
});
var streamEvents = new List<ChatStreamEvent>();
var handler = CreateHandler();

var result = await handler.Handle(
new ProcessUserChatCommand(
UserId,
"How did my week go?",
ClientContext: new AgentClientContext(SupportsMetricsCard: true),
"Show today's habits",
ClientContext: new AgentClientContext(SupportsHabitListCard: true),
StreamSink: streamEvent =>
{
streamEvents.Add(streamEvent);
return Task.CompletedTask;
}),
CancellationToken.None);

result.IsSuccess.Should().BeTrue();
string.Concat(streamEvents.Where(streamEvent => streamEvent.Type == "delta").Select(streamEvent => streamEvent.Text))
.Should().Be("Your habits:\n");
result.Value.AiMessage.Should().Be("Your habits:");
result.Value.HabitList.Should().NotBeNull();
result.Value.HabitList!.Scope.Should().Be("today");
}

[Fact]
public async Task Handle_StreamedUppercaseHabitDirective_IsRemoved()
{
SetupUserAndPayGate();
const string directive = "[[ORBIT:HABITS:ALL]]";
_aiIntentService.SendWithToolsAsync(
Arg.Any<AiToolRequest>(),
Arg.Any<Func<AiStreamEvent, Task>?>(),
Arg.Any<CancellationToken>())
.Returns(async callInfo =>
{
var sink = callInfo.ArgAt<Func<AiStreamEvent, Task>?>(1);
await sink!(AiStreamEvent.Delta($"All habits:\n{directive}"));
return Result.Success(new AiResponse { TextMessage = $"All habits:\n{directive}" });
});
var streamEvents = new List<ChatStreamEvent>();
var handler = CreateHandler();

var result = await handler.Handle(
new ProcessUserChatCommand(
UserId,
"Show all habits",
StreamSink: streamEvent =>
{
streamEvents.Add(streamEvent);
Expand All @@ -1002,10 +1094,43 @@ public async Task Handle_StreamedMetricsDirectiveAcrossChunks_NeverEmitsToken()

result.IsSuccess.Should().BeTrue();
string.Concat(streamEvents.Where(streamEvent => streamEvent.Type == "delta").Select(streamEvent => streamEvent.Text))
.Should().Be("Your week:\n");
streamEvents
.Where(streamEvent => streamEvent.Text?.Contains("orbit:metrics", StringComparison.OrdinalIgnoreCase) == true)
.Should().BeEmpty();
.Should().Be("All habits:\n");
}

[Fact]
public async Task Handle_StreamedReplyWithoutDirective_PreservesDeltas()
{
SetupUserAndPayGate();
string[] chunks = ["Hello, ", "world!"];
_aiIntentService.SendWithToolsAsync(
Arg.Any<AiToolRequest>(),
Arg.Any<Func<AiStreamEvent, Task>?>(),
Arg.Any<CancellationToken>())
.Returns(async callInfo =>
{
var sink = callInfo.ArgAt<Func<AiStreamEvent, Task>?>(1);
foreach (var chunk in chunks)
await sink!(AiStreamEvent.Delta(chunk));

return Result.Success(new AiResponse { TextMessage = string.Concat(chunks) });
});
var streamEvents = new List<ChatStreamEvent>();
var handler = CreateHandler();

var result = await handler.Handle(
new ProcessUserChatCommand(
UserId,
"Hello",
StreamSink: streamEvent =>
{
streamEvents.Add(streamEvent);
return Task.CompletedTask;
}),
CancellationToken.None);

result.IsSuccess.Should().BeTrue();
streamEvents.Where(streamEvent => streamEvent.Type == "delta").Select(streamEvent => streamEvent.Text)
.Should().Equal(chunks);
}

[Fact]
Expand Down
Loading