agent-radar
2ed3415fb7
Publish Claude Code Binaries / Stage and publish binaries (push) Canceled after 0s
Publish Tutti App Runtime / Build darwin-arm64 (push) Canceled after 0s
Publish Tutti App Runtime / Build darwin-amd64 (push) Canceled after 0s
Publish Tutti App Runtime / Build linux-amd64 (push) Canceled after 0s
Publish Tutti App Runtime / Build windows-amd64 (push) Canceled after 0s
Sync Tutti Agent Skills / sync (push) Canceled after 0s
Windows Agent Adapters / Test Windows agent adapters (push) Canceled after 0s
Windows Daemon Adapters / Test Windows daemon adapters (push) Canceled after 0s
Publish Tutti App Runtime / Publish runtime catalog (push) Canceled after 0s
1171 lines
43 KiB
Go
1171 lines
43 KiB
Go
package agentruntime
|
|
|
|
import (
|
|
"encoding/json"
|
|
"strings"
|
|
"testing"
|
|
|
|
agentsessionstore "github.com/tutti-os/tutti/packages/agent/daemon/activity"
|
|
activityshared "github.com/tutti-os/tutti/packages/agent/daemon/activity/events"
|
|
"github.com/tutti-os/tutti/packages/agent/daemon/liveprotocol"
|
|
"github.com/tutti-os/tutti/packages/agent/store-sqlite/canonical"
|
|
)
|
|
|
|
func TestACPNormalizerProjectsSemanticMessageDeltaWithoutSnapshotDiff(t *testing.T) {
|
|
t.Parallel()
|
|
session := reportTestSession()
|
|
normalizer := newACPTurnNormalizer()
|
|
|
|
first := normalizer.AppendAssistantChunk(session, "turn-1", "Hel")
|
|
second := normalizer.AppendAssistantChunk(session, "turn-1", "Hello")
|
|
stream := ProjectActivityEventsToStreamEvents(session, append(first, second...))
|
|
if len(stream) != 2 || stream[0].EventType != StreamEventMessageDelta || stream[1].EventType != StreamEventMessageDelta {
|
|
t.Fatalf("stream = %#v, want two message deltas", stream)
|
|
}
|
|
firstDelta := stream[0].Data.(liveprotocol.Event)
|
|
secondDelta := stream[1].Data.(liveprotocol.Event)
|
|
var firstData, secondData liveprotocol.MessageDeltaData
|
|
if err := json.Unmarshal(firstDelta.Data, &firstData); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := json.Unmarshal(secondDelta.Data, &secondData); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if firstData.Content == nil || firstData.Content.Operation != "set" || string(firstData.Content.Value) != `"Hel"` {
|
|
t.Fatalf("first delta = %#v", firstData)
|
|
}
|
|
if secondData.Content == nil || secondData.Content.Operation != "append_text" || secondData.Content.Text != "lo" {
|
|
t.Fatalf("second delta = %#v", secondData)
|
|
}
|
|
|
|
terminal := normalizer.FinishCompleted(session, "turn-1")
|
|
if got := ProjectActivityEventsToStreamEvents(session, terminal); len(got) != 0 {
|
|
t.Fatalf("precommit terminal leaked to live stream: %#v", got)
|
|
}
|
|
}
|
|
|
|
func TestProjectActivityEventsUsesCanonicalSessionIdentityForMessageDelta(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
events := newACPTurnNormalizer().AppendAssistantChunk(session, "turn-1", "hello")
|
|
if len(events) != 1 {
|
|
t.Fatalf("normalized events = %#v, want one event", events)
|
|
}
|
|
// The projection derives the canonical identity from the session source when
|
|
// the provider event omits its identity. The live delta must use that same
|
|
// identity instead of serializing the raw, empty provider field.
|
|
events[0].AgentSessionID = ""
|
|
stream := ProjectActivityEventsToStreamEvents(session, events)
|
|
if len(stream) != 1 {
|
|
t.Fatalf("stream = %#v, want one message delta", stream)
|
|
}
|
|
delta, ok := stream[0].Data.(liveprotocol.Event)
|
|
if !ok {
|
|
t.Fatalf("stream data = %T, want live protocol event", stream[0].Data)
|
|
}
|
|
var data liveprotocol.MessageDeltaData
|
|
if err := json.Unmarshal(delta.Data, &data); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if data.AgentSessionID != session.AgentSessionID {
|
|
t.Fatalf("message delta agent session id = %q, want canonical %q", data.AgentSessionID, session.AgentSessionID)
|
|
}
|
|
}
|
|
|
|
func TestSnapshotNormalizerProjectsSuffixAppendAndFullRewrite(t *testing.T) {
|
|
t.Parallel()
|
|
session := reportTestSession()
|
|
|
|
tests := []struct {
|
|
name string
|
|
apply func(*acpTurnNormalizer, string) []activityshared.Event
|
|
role string
|
|
kind string
|
|
}{
|
|
{
|
|
name: "assistant",
|
|
apply: func(normalizer *acpTurnNormalizer, snapshot string) []activityshared.Event {
|
|
return normalizer.ApplyStreamingAssistantSnapshot(
|
|
session,
|
|
"turn-1",
|
|
snapshot,
|
|
"assistant-1",
|
|
)
|
|
},
|
|
role: RoleAssistant,
|
|
kind: "text",
|
|
},
|
|
{
|
|
name: "thinking",
|
|
apply: func(normalizer *acpTurnNormalizer, snapshot string) []activityshared.Event {
|
|
return normalizer.ApplyStreamingThinkingSnapshot(
|
|
session,
|
|
"turn-1",
|
|
snapshot,
|
|
"thinking-1",
|
|
)
|
|
},
|
|
role: RoleAssistantThinking,
|
|
kind: "reasoning",
|
|
},
|
|
}
|
|
|
|
for _, test := range tests {
|
|
t.Run(test.name, func(t *testing.T) {
|
|
t.Parallel()
|
|
normalizer := newACPTurnNormalizer()
|
|
|
|
assertSnapshotLiveOperation(
|
|
t,
|
|
test.apply(normalizer, "Hel"),
|
|
test.role,
|
|
test.kind,
|
|
"set",
|
|
"Hel",
|
|
)
|
|
assertSnapshotLiveOperation(
|
|
t,
|
|
test.apply(normalizer, "Hello"),
|
|
test.role,
|
|
test.kind,
|
|
"append_text",
|
|
"lo",
|
|
)
|
|
if duplicate := test.apply(normalizer, "Hello"); len(duplicate) != 0 {
|
|
t.Fatalf("duplicate snapshot events = %#v, want none", duplicate)
|
|
}
|
|
assertSnapshotLiveOperation(
|
|
t,
|
|
test.apply(normalizer, "Help"),
|
|
test.role,
|
|
test.kind,
|
|
"set",
|
|
"Help",
|
|
)
|
|
})
|
|
}
|
|
}
|
|
|
|
func assertSnapshotLiveOperation(
|
|
t *testing.T,
|
|
events []activityshared.Event,
|
|
wantRole string,
|
|
wantKind string,
|
|
wantOperation string,
|
|
wantContent string,
|
|
) {
|
|
t.Helper()
|
|
if len(events) != 1 {
|
|
t.Fatalf("events = %#v, want one snapshot event", events)
|
|
}
|
|
event := events[0]
|
|
if string(event.Payload.Role) != wantRole ||
|
|
event.Payload.Metadata[liveMessageKindMetadataKey] != wantKind {
|
|
t.Fatalf("snapshot identity = role %q metadata %#v", event.Payload.Role, event.Payload.Metadata)
|
|
}
|
|
operation, ok := event.Payload.Metadata[liveContentOperationMetadataKey].(*liveprotocol.MessageContentOperation)
|
|
if !ok || operation == nil || operation.Operation != wantOperation {
|
|
t.Fatalf("live operation = %#v, want %q", operation, wantOperation)
|
|
}
|
|
switch wantOperation {
|
|
case "append_text":
|
|
if operation.Text != wantContent || len(operation.Value) != 0 {
|
|
t.Fatalf("append operation = %#v, want suffix %q", operation, wantContent)
|
|
}
|
|
case "set":
|
|
var content string
|
|
if err := json.Unmarshal(operation.Value, &content); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if content != wantContent || operation.Text != "" {
|
|
t.Fatalf("set operation = %#v content %q, want %q", operation, content, wantContent)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestThinkingDeltaPreservesAssistantThinkingRole(t *testing.T) {
|
|
t.Parallel()
|
|
session := reportTestSession()
|
|
normalizer := newACPTurnNormalizer()
|
|
events := normalizer.AppendThinkingChunk(session, "turn-1", "inspect")
|
|
stream := ProjectActivityEventsToStreamEvents(session, events)
|
|
if len(stream) != 1 {
|
|
t.Fatalf("stream = %#v", stream)
|
|
}
|
|
event := stream[0].Data.(liveprotocol.Event)
|
|
var data liveprotocol.MessageDeltaData
|
|
if err := json.Unmarshal(event.Data, &data); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if data.Role != RoleAssistantThinking || data.Kind != "reasoning" {
|
|
t.Fatalf("thinking delta = %#v", data)
|
|
}
|
|
}
|
|
|
|
func TestExplicitToolOutputDeltaPersistsSnapshotAndProjectsOffsetAppend(t *testing.T) {
|
|
t.Parallel()
|
|
session := reportTestSession()
|
|
normalizer := newACPTurnNormalizer()
|
|
started, ok := normalizer.ToolCallEvents(session, "turn-1", map[string]any{
|
|
"sessionUpdate": "tool_call",
|
|
"toolCallId": "command-1",
|
|
"title": "printf",
|
|
"kind": "execute",
|
|
"status": "in_progress",
|
|
"rawInput": map[string]any{"command": "printf"},
|
|
})
|
|
if !ok || len(started) != 1 {
|
|
t.Fatalf("started = %#v, ok = %v", started, ok)
|
|
}
|
|
|
|
first := normalizer.AppendToolOutputDelta(session, "turn-1", "command-1", "你")
|
|
second := normalizer.AppendToolOutputDelta(session, "turn-1", "command-1", "好\n")
|
|
stream := ProjectActivityEventsToStreamEvents(session, append(first, second...))
|
|
if len(stream) != 2 {
|
|
t.Fatalf("stream = %#v, want two tool output deltas", stream)
|
|
}
|
|
var firstDelta, secondDelta liveprotocol.MessageDeltaData
|
|
for index, target := range []*liveprotocol.MessageDeltaData{&firstDelta, &secondDelta} {
|
|
event, ok := stream[index].Data.(liveprotocol.Event)
|
|
if !ok {
|
|
t.Fatalf("stream[%d] data = %T", index, stream[index].Data)
|
|
}
|
|
if err := json.Unmarshal(event.Data, target); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
if firstDelta.ToolOutput == nil ||
|
|
firstDelta.ToolOutput.Operation != "set" ||
|
|
firstDelta.ToolOutput.Text != "你" ||
|
|
firstDelta.ToolOutput.OffsetBytes != nil {
|
|
t.Fatalf("first tool output = %#v", firstDelta.ToolOutput)
|
|
}
|
|
if secondDelta.ToolOutput == nil ||
|
|
secondDelta.ToolOutput.Operation != "append_text" ||
|
|
secondDelta.ToolOutput.Text != "好\n" ||
|
|
secondDelta.ToolOutput.OffsetBytes == nil ||
|
|
*secondDelta.ToolOutput.OffsetBytes != int64(len("你")) {
|
|
t.Fatalf("second tool output = %#v", secondDelta.ToolOutput)
|
|
}
|
|
|
|
report := reportActivityInput(session, second)
|
|
if len(report.MessageUpdates) != 1 {
|
|
t.Fatalf("report = %#v, want one cumulative tool snapshot", report)
|
|
}
|
|
startReport := reportActivityInput(session, started)
|
|
if len(startReport.MessageUpdates) != 1 {
|
|
t.Fatalf("start report = %#v, want one canonical tool anchor", startReport)
|
|
}
|
|
if firstDelta.MessageID != startReport.MessageUpdates[0].MessageID ||
|
|
secondDelta.MessageID != startReport.MessageUpdates[0].MessageID ||
|
|
report.MessageUpdates[0].MessageID != startReport.MessageUpdates[0].MessageID {
|
|
t.Fatalf(
|
|
"tool message identity = start:%q first:%q second:%q snapshot:%q, want one canonical anchor",
|
|
startReport.MessageUpdates[0].MessageID,
|
|
firstDelta.MessageID,
|
|
secondDelta.MessageID,
|
|
report.MessageUpdates[0].MessageID,
|
|
)
|
|
}
|
|
output, _ := report.MessageUpdates[0].Payload["output"].(map[string]any)
|
|
if output["text"] != "你好\n" {
|
|
t.Fatalf("persisted output = %#v", output)
|
|
}
|
|
}
|
|
|
|
func TestExplicitToolOutputDeltaTruncatesOnceAndDropsLaterChunks(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
normalizer := newACPTurnNormalizer()
|
|
started, ok := normalizer.ToolCallEvents(session, "turn-1", map[string]any{
|
|
"sessionUpdate": "tool_call",
|
|
"toolCallId": "command-large",
|
|
"title": "printf",
|
|
"kind": "execute",
|
|
"status": "in_progress",
|
|
})
|
|
if !ok || len(started) != 1 {
|
|
t.Fatalf("started = %#v, ok = %v", started, ok)
|
|
}
|
|
|
|
firstLength := canonical.ToolOutputTextMaxBytes -
|
|
len(canonical.ToolOutputTruncationMarker) -
|
|
len("\n") -
|
|
8
|
|
if events := normalizer.AppendToolOutputDelta(
|
|
session,
|
|
"turn-1",
|
|
"command-large",
|
|
strings.Repeat("x", firstLength),
|
|
); len(events) != 1 {
|
|
t.Fatalf("first output events = %#v, want one", events)
|
|
}
|
|
truncated := normalizer.AppendToolOutputDelta(
|
|
session,
|
|
"turn-1",
|
|
"command-large",
|
|
strings.Repeat("y", 64),
|
|
)
|
|
if len(truncated) != 1 {
|
|
t.Fatalf("truncated output events = %#v, want one", truncated)
|
|
}
|
|
if later := normalizer.AppendToolOutputDelta(
|
|
session,
|
|
"turn-1",
|
|
"command-large",
|
|
"ignored",
|
|
); len(later) != 0 {
|
|
t.Fatalf("later output events = %#v, want dropped after truncation", later)
|
|
}
|
|
|
|
output := payloadMap(truncated[0].Payload.Metadata, "output")["text"].(string)
|
|
if len(output) > canonical.ToolOutputTextMaxBytes ||
|
|
!strings.HasSuffix(output, canonical.ToolOutputTruncationMarker) {
|
|
t.Fatalf("snapshot output was not bounded: %d bytes", len(output))
|
|
}
|
|
stream := ProjectActivityEventsToStreamEvents(session, truncated)
|
|
if len(stream) != 1 || stream[0].EventType != StreamEventMessageDelta {
|
|
t.Fatalf("stream = %#v, want one tool output delta", stream)
|
|
}
|
|
var delta liveprotocol.MessageDeltaData
|
|
if err := json.Unmarshal(stream[0].Data.(liveprotocol.Event).Data, &delta); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if delta.ToolOutput == nil ||
|
|
delta.ToolOutput.Operation != "append_text" ||
|
|
delta.ToolOutput.OffsetBytes == nil ||
|
|
*delta.ToolOutput.OffsetBytes != int64(firstLength) ||
|
|
!strings.HasSuffix(delta.ToolOutput.Text, canonical.ToolOutputTruncationMarker) {
|
|
t.Fatalf("truncated live operation = %#v", delta.ToolOutput)
|
|
}
|
|
|
|
report := reportActivityInput(session, truncated)
|
|
persisted := report.MessageUpdates[0].Payload["output"].(map[string]any)["text"].(string)
|
|
if persisted != output {
|
|
t.Fatal("live runtime snapshot and durable report used different truncated output")
|
|
}
|
|
}
|
|
|
|
func TestToolOutputDeltaWaitsForKnownStartAnchor(t *testing.T) {
|
|
t.Parallel()
|
|
session := reportTestSession()
|
|
normalizer := newACPTurnNormalizer()
|
|
if events := normalizer.AppendToolOutputDelta(
|
|
session,
|
|
"turn-1",
|
|
"missing-command",
|
|
"first",
|
|
); len(events) != 0 {
|
|
t.Fatalf("pre-anchor output events = %#v, want no invented anchor", events)
|
|
}
|
|
started, ok := normalizer.ToolCallEvents(session, "turn-1", map[string]any{
|
|
"sessionUpdate": "tool_call",
|
|
"toolCallId": "missing-command",
|
|
"title": "printf",
|
|
"kind": "execute",
|
|
"status": "in_progress",
|
|
})
|
|
if !ok || len(started) != 2 {
|
|
t.Fatalf("started = %#v, ok = %v, want anchor followed by buffered output", started, ok)
|
|
}
|
|
if started[0].Type != activityshared.EventCallStarted ||
|
|
started[1].Type != activityshared.EventCallStarted {
|
|
t.Fatalf("started events = %#v", started)
|
|
}
|
|
stream := ProjectActivityEventsToStreamEvents(session, started)
|
|
if len(stream) != 2 ||
|
|
stream[0].EventType != StreamEventMessageUpdate ||
|
|
stream[1].EventType != StreamEventMessageDelta {
|
|
t.Fatalf("stream = %#v, want canonical anchor then live output", stream)
|
|
}
|
|
var delta liveprotocol.MessageDeltaData
|
|
if err := json.Unmarshal(stream[1].Data.(liveprotocol.Event).Data, &delta); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if delta.ToolOutput == nil ||
|
|
delta.ToolOutput.Operation != "set" ||
|
|
delta.ToolOutput.Text != "first" {
|
|
t.Fatalf("buffered tool output = %#v", delta.ToolOutput)
|
|
}
|
|
}
|
|
|
|
func TestToolOutputDeltaTruncatesBufferedPreAnchorOutput(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
normalizer := newACPTurnNormalizer()
|
|
large := strings.Repeat("x", canonical.ToolOutputTextMaxBytes+64)
|
|
if events := normalizer.AppendToolOutputDelta(
|
|
session,
|
|
"turn-1",
|
|
"buffered-large-command",
|
|
large,
|
|
); len(events) != 0 {
|
|
t.Fatalf("pre-anchor output events = %#v, want buffered output", events)
|
|
}
|
|
started, ok := normalizer.ToolCallEvents(session, "turn-1", map[string]any{
|
|
"sessionUpdate": "tool_call",
|
|
"toolCallId": "buffered-large-command",
|
|
"title": "printf",
|
|
"kind": "execute",
|
|
"status": "in_progress",
|
|
})
|
|
if !ok || len(started) != 2 {
|
|
t.Fatalf("started = %#v, ok = %v, want anchor and truncated buffered output", started, ok)
|
|
}
|
|
output := payloadMap(started[1].Payload.Metadata, "output")["text"].(string)
|
|
if len(output) > canonical.ToolOutputTextMaxBytes ||
|
|
!strings.HasSuffix(output, canonical.ToolOutputTruncationMarker) {
|
|
t.Fatalf("buffered output was not bounded: %d bytes", len(output))
|
|
}
|
|
if later := normalizer.AppendToolOutputDelta(
|
|
session,
|
|
"turn-1",
|
|
"buffered-large-command",
|
|
"ignored",
|
|
); len(later) != 0 {
|
|
t.Fatalf("later output events = %#v, want dropped after buffered truncation", later)
|
|
}
|
|
}
|
|
|
|
func TestTruncatedBufferedToolOutputFitsLivePublisher(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
for _, test := range []struct {
|
|
name string
|
|
large string
|
|
}{
|
|
{
|
|
name: "ASCII",
|
|
large: strings.Repeat("x", canonical.ToolOutputTextMaxBytes+64),
|
|
},
|
|
{
|
|
name: "JSON escaping",
|
|
large: strings.Repeat("\x00", canonical.ToolOutputTextMaxBytes+64),
|
|
},
|
|
} {
|
|
test := test
|
|
t.Run(test.name, func(t *testing.T) {
|
|
session := reportTestSession()
|
|
normalizer := newACPTurnNormalizer()
|
|
if events := normalizer.AppendToolOutputDelta(
|
|
session,
|
|
"turn-1",
|
|
"buffered-large-command",
|
|
test.large,
|
|
); len(events) != 0 {
|
|
t.Fatalf("pre-anchor output events = %#v, want buffered output", events)
|
|
}
|
|
started, ok := normalizer.ToolCallEvents(session, "turn-1", map[string]any{
|
|
"sessionUpdate": "tool_call",
|
|
"toolCallId": "buffered-large-command",
|
|
"title": "printf",
|
|
"kind": "execute",
|
|
"status": "in_progress",
|
|
})
|
|
if !ok || len(started) != 2 {
|
|
t.Fatalf("started = %#v, ok = %v, want anchor and buffered output", started, ok)
|
|
}
|
|
expected := payloadMap(started[1].Payload.Metadata, "output")["text"].(string)
|
|
stream := ProjectActivityEventsToStreamEvents(session, started)
|
|
|
|
publisher, err := liveprotocol.NewPublisher(liveprotocol.PublisherConfig{
|
|
StreamID: "stream-1", BindingID: "binding-1", Epoch: 1,
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var delivered string
|
|
deltaCount := 0
|
|
for _, streamEvent := range stream {
|
|
if streamEvent.EventType != StreamEventMessageDelta {
|
|
continue
|
|
}
|
|
event, ok := streamEvent.Data.(liveprotocol.Event)
|
|
if !ok {
|
|
t.Fatalf("stream event data has type %T", streamEvent.Data)
|
|
}
|
|
frames, err := publisher.Publish(liveprotocol.PublishInput{
|
|
Event: &event, Immediate: true,
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(frames) != 1 || len(frames[0].Deliveries) != 1 {
|
|
t.Fatalf("published frames = %#v, want one delivery", frames)
|
|
}
|
|
delivery := frames[0].Deliveries[0]
|
|
if delivery.Kind != liveprotocol.DeliveryKindEvent {
|
|
t.Fatalf("delivery kind = %v, want event", delivery.Kind)
|
|
}
|
|
deliveredEvent, err := liveprotocol.DecodeEvent(delivery.Event)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var delta liveprotocol.MessageDeltaData
|
|
if err := json.Unmarshal(deliveredEvent.Data, &delta); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if delta.ToolOutput == nil {
|
|
t.Fatalf("delivered delta = %#v, want tool output", delta)
|
|
}
|
|
switch delta.ToolOutput.Operation {
|
|
case "set":
|
|
if deltaCount != 0 {
|
|
t.Fatalf("set operation arrived after %d deltas", deltaCount)
|
|
}
|
|
delivered = delta.ToolOutput.Text
|
|
case "append_text":
|
|
if delta.ToolOutput.OffsetBytes == nil ||
|
|
*delta.ToolOutput.OffsetBytes != int64(len(delivered)) {
|
|
t.Fatalf(
|
|
"append offset = %#v, delivered bytes = %d",
|
|
delta.ToolOutput.OffsetBytes,
|
|
len(delivered),
|
|
)
|
|
}
|
|
delivered += delta.ToolOutput.Text
|
|
default:
|
|
t.Fatalf("tool output operation = %#v", delta.ToolOutput)
|
|
}
|
|
deltaCount++
|
|
}
|
|
if deltaCount < 2 {
|
|
t.Fatalf("delta count = %d, want split delivery", deltaCount)
|
|
}
|
|
if delivered != expected ||
|
|
len(delivered) > canonical.ToolOutputTextMaxBytes ||
|
|
!strings.HasSuffix(delivered, canonical.ToolOutputTruncationMarker) {
|
|
t.Fatalf(
|
|
"delivered output mismatch: got %d bytes, want %d",
|
|
len(delivered),
|
|
len(expected),
|
|
)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestExtensionProviderProjectsTurnLifecycleEvents(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
session.Provider = "acp:gemini"
|
|
started := newTurnActivityEvent(session, EventTurnStarted, "turn-1", SessionStatusWorking, "", "", nil)
|
|
failed := newTurnActivityEvent(session, EventTurnFailed, "turn-1", SessionStatusFailed, "", "", map[string]any{
|
|
"error": "quota exceeded",
|
|
})
|
|
|
|
if started.Type != activityshared.EventTurnStarted {
|
|
t.Fatalf("extension turn started event = %#v, want %q", started, activityshared.EventTurnStarted)
|
|
}
|
|
if failed.Type != activityshared.EventTurnFailed {
|
|
t.Fatalf("extension turn failed event = %#v, want %q", failed, activityshared.EventTurnFailed)
|
|
}
|
|
if started.Provider != activityshared.Provider("acp:gemini") || failed.Provider != activityshared.Provider("acp:gemini") {
|
|
t.Fatalf("extension event providers = %q, %q; want acp:gemini", started.Provider, failed.Provider)
|
|
}
|
|
}
|
|
|
|
func TestEventSourceCarriesStableSessionIncarnation(t *testing.T) {
|
|
source := eventSourceFromSession(Session{Provider: "codex", AgentSessionID: "session", CreatedAtUnixMS: 4242})
|
|
if source.SessionCreatedAtUnixMS != 4242 {
|
|
t.Fatalf("session incarnation = %d", source.SessionCreatedAtUnixMS)
|
|
}
|
|
}
|
|
|
|
func TestProviderGlobalAuthEligibilityRequiresProviderNativeConfiguration(t *testing.T) {
|
|
tests := []struct {
|
|
name string
|
|
provider string
|
|
source string
|
|
want bool
|
|
}{
|
|
{name: "codex native", provider: "codex", source: "provider-native", want: true},
|
|
{name: "claude model plan", provider: "claude-code", source: "model-plan"},
|
|
{name: "cursor extension", provider: "cursor", source: "agent-extension"},
|
|
{name: "unknown provider", provider: "third-party", source: "provider-native"},
|
|
}
|
|
for _, test := range tests {
|
|
t.Run(test.name, func(t *testing.T) {
|
|
session := Session{Provider: test.provider, RuntimeContext: map[string]any{
|
|
"sessionRuntimeSnapshot": map[string]any{
|
|
"modelConfiguration": map[string]any{"source": test.source},
|
|
},
|
|
}}
|
|
if got := providerGlobalAuthEligible(session); got != test.want {
|
|
t.Fatalf("providerGlobalAuthEligible() = %v, want %v", got, test.want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestSessionAuditProjectsSeparatelyFromTurnMessages(t *testing.T) {
|
|
t.Parallel()
|
|
session := reportTestSession()
|
|
audit := newSessionAuditEventWithID(session, "goal-control:op-1", RoleUser, "/goal clear", map[string]any{"goalControl": true})
|
|
report := reportActivityInput(session, []activityshared.Event{audit})
|
|
if len(report.SessionAudits) != 1 || len(report.MessageUpdates) != 0 || len(report.StatePatches) != 0 {
|
|
t.Fatalf("report = %#v, want one standalone audit", report)
|
|
}
|
|
if report.SessionAudits[0].AuditID != "goal-control:op-1" || report.SessionAudits[0].Payload["goalControl"] != true {
|
|
t.Fatalf("audit = %#v", report.SessionAudits[0])
|
|
}
|
|
stream := ProjectActivityEventsToStreamEvents(session, []activityshared.Event{audit})
|
|
if len(stream) != 1 || stream[0].EventType != StreamEventSessionAudit {
|
|
t.Fatalf("stream events = %#v", stream)
|
|
}
|
|
}
|
|
|
|
func TestSessionFailureProjectsVisibleAuditWithoutTurn(t *testing.T) {
|
|
t.Parallel()
|
|
session := reportTestSession()
|
|
event := newSessionActivityEvent(session, EventSessionFailed, SessionStatusFailed, map[string]any{
|
|
"error": "provider failed before turn admission",
|
|
})
|
|
|
|
report := reportActivityInput(session, []activityshared.Event{event})
|
|
if len(report.StatePatches) != 1 || len(report.SessionAudits) != 1 || len(report.MessageUpdates) != 0 {
|
|
t.Fatalf("report = %#v, want state plus turnless visible audit", report)
|
|
}
|
|
audit := report.SessionAudits[0]
|
|
if audit.AuditID == "" || audit.Role != RoleAssistant || audit.Payload["kind"] != visibleErrorKind ||
|
|
audit.Payload["phase"] != "start" || audit.Payload["detail"] != "provider failed before turn admission" {
|
|
t.Fatalf("visible failure audit = %#v", audit)
|
|
}
|
|
|
|
stream := ProjectActivityEventsToStreamEvents(session, []activityshared.Event{event})
|
|
if len(stream) != 2 || stream[0].EventType != StreamEventStatePatch || stream[1].EventType != StreamEventSessionAudit {
|
|
t.Fatalf("stream = %#v, want state plus session audit", stream)
|
|
}
|
|
streamAudit, ok := stream[1].Data.(agentsessionstore.WorkspaceAgentSessionAuditUpdate)
|
|
if !ok || streamAudit.AuditID != audit.AuditID {
|
|
t.Fatalf("stream audit = %#v, want durable audit identity %q", stream[1].Data, audit.AuditID)
|
|
}
|
|
}
|
|
|
|
func TestGoalReconcileRequiredProjectsOnlyToInternalControlReport(t *testing.T) {
|
|
t.Parallel()
|
|
session := reportTestSession()
|
|
ctx, ok := activityEventContext(session, "goal-reconcile:req-1", "")
|
|
if !ok {
|
|
t.Fatal("activity event context unavailable")
|
|
}
|
|
event := activityshared.NewGoalReconcileRequired(ctx, map[string]any{
|
|
"requestId": "req-1", "providerTurnId": "provider-turn-old", "fenceMode": "operation",
|
|
"expectedGoalOperationId": "goal-op-2", "expectedGoalRevision": int64(2),
|
|
"expectedGoalRepairEpoch": int64(1), "quiesceSucceeded": true,
|
|
})
|
|
report := reportActivityInput(session, []activityshared.Event{event})
|
|
if len(report.GoalReconcileRequests) != 1 || len(report.SessionAudits) != 0 || len(report.MessageUpdates) != 0 || len(report.StatePatches) != 0 {
|
|
t.Fatalf("internal reconcile report = %#v", report)
|
|
}
|
|
request := report.GoalReconcileRequests[0]
|
|
if request.RequestID != "req-1" || request.ExpectedOperationID != "goal-op-2" || request.ExpectedRevision != 2 || request.ExpectedRepairEpoch != 1 || !request.QuiesceSucceeded {
|
|
t.Fatalf("internal reconcile request = %#v", request)
|
|
}
|
|
if stream := ProjectActivityEventsToStreamEvents(session, []activityshared.Event{event}); len(stream) != 0 {
|
|
t.Fatalf("internal reconcile evidence leaked to realtime stream: %#v", stream)
|
|
}
|
|
}
|
|
|
|
func TestReportableActivityEventsReportsOnlyCompletedAssistantSnapshots(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
session.AgentSessionID = "4e70b18d-b8b5-47a1-b293-3b98e4a23310"
|
|
streaming := newTurnActivityEvent(session, EventMessage, "turn-1", messageStreamStateStreaming, RoleAssistant, "hel", map[string]any{
|
|
"streamState": messageStreamStateStreaming,
|
|
})
|
|
completed := newTurnActivityEvent(session, EventMessage, "turn-1", messageStreamStateCompleted, RoleAssistant, "hello", map[string]any{
|
|
"streamState": messageStreamStateCompleted,
|
|
})
|
|
thinkingStreaming := newTurnActivityEvent(session, EventMessage, "turn-1", messageStreamStateStreaming, RoleAssistantThinking, "thinking", map[string]any{
|
|
"streamState": messageStreamStateStreaming,
|
|
})
|
|
thinkingCompleted := newTurnActivityEvent(session, EventMessage, "turn-1", messageStreamStateCompleted, RoleAssistantThinking, "thinking done", map[string]any{
|
|
"streamState": messageStreamStateCompleted,
|
|
})
|
|
|
|
events := ReportableActivityEvents([]activityshared.Event{
|
|
newTurnActivityEvent(session, EventMessage, "turn-1", "", RoleUser, "say hello", nil),
|
|
streaming,
|
|
completed,
|
|
thinkingStreaming,
|
|
thinkingCompleted,
|
|
})
|
|
|
|
if len(events) != 3 {
|
|
t.Fatalf("activity events = %#v, want user, completed assistant, and completed thinking snapshots", events)
|
|
}
|
|
if events[0].Type != activityshared.EventMessageAppended || events[1].Type != activityshared.EventMessageAppended || events[2].Type != activityshared.EventMessageAppended {
|
|
t.Fatalf("activity event types = %#v, want message events", events)
|
|
}
|
|
if events[2].Payload.Role != activityshared.MessageRoleAssistantThinking {
|
|
t.Fatalf("thinking role = %q, want assistant_thinking", events[2].Payload.Role)
|
|
}
|
|
if events[1].Payload.Metadata["streamState"] != messageStreamStateCompleted {
|
|
t.Fatalf("activity metadata streamState = %#v, want completed", events[1].Payload.Metadata)
|
|
}
|
|
}
|
|
|
|
func TestReportableActivityEventsIncludesRootProviderTurnLifecycle(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
ctx, ok := activityEventContext(session, "root-provider-turn", "root-turn-1")
|
|
if !ok {
|
|
t.Fatal("activityEventContext() returned !ok")
|
|
}
|
|
events := ReportableActivityEvents([]activityshared.Event{
|
|
activityshared.NewRootProviderTurnStarted(ctx, "root-turn-1", "provider-turn-1"),
|
|
activityshared.NewRootProviderTurnCompleted(ctx, "root-turn-1", "provider-turn-1", activityshared.TurnOutcomeCompleted),
|
|
})
|
|
|
|
if len(events) != 2 {
|
|
t.Fatalf("activity events = %#v, want root provider start and completion", events)
|
|
}
|
|
report := reportActivityInput(session, events)
|
|
if len(report.StatePatches) != 2 {
|
|
t.Fatalf("state patches = %#v, want root provider start and completion", report.StatePatches)
|
|
}
|
|
started := report.StatePatches[0].RootProviderTurn
|
|
if started == nil || started.RootTurnID != "root-turn-1" || started.ProviderTurnID != "provider-turn-1" || started.Phase != agentsessionstore.RootProviderTurnPhaseRunning {
|
|
t.Fatalf("started root provider turn = %#v, want running transition", started)
|
|
}
|
|
completed := report.StatePatches[1].RootProviderTurn
|
|
if completed == nil || completed.RootTurnID != "root-turn-1" || completed.ProviderTurnID != "provider-turn-1" || completed.Phase != agentsessionstore.RootProviderTurnPhaseCompleted || completed.Outcome != string(activityshared.TurnOutcomeCompleted) {
|
|
t.Fatalf("completed root provider turn = %#v, want completed transition", completed)
|
|
}
|
|
if completed.ErrorCode != "" {
|
|
t.Fatalf("completed root provider turn error code = %q, want empty", completed.ErrorCode)
|
|
}
|
|
}
|
|
|
|
func TestRootProviderTurnFailurePersistsVisibleErrorCode(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
session.Provider = ProviderTuttiAgent
|
|
ctx, ok := activityEventContext(session, "root-provider-turn-failed", "root-turn-1")
|
|
if !ok {
|
|
t.Fatal("activityEventContext() returned !ok")
|
|
}
|
|
const errorMessage = "You've hit your usage limit. Insufficient credits. View Tutti plans at https://tutti.sh/profile/plan, or try again later."
|
|
failed := activityshared.NewRootProviderTurnCompleted(ctx, "root-turn-1", "provider-turn-1", activityshared.TurnOutcomeFailed)
|
|
failed.Payload.Metadata = map[string]any{"error": errorMessage}
|
|
|
|
report := reportActivityInput(session, []activityshared.Event{failed})
|
|
if len(report.StatePatches) != 1 {
|
|
t.Fatalf("state patches = %#v, want one root provider failure", report.StatePatches)
|
|
}
|
|
completed := report.StatePatches[0].RootProviderTurn
|
|
if completed == nil {
|
|
t.Fatal("root provider turn transition is nil")
|
|
return
|
|
}
|
|
if completed.ErrorMessage != errorMessage {
|
|
t.Fatalf("root provider turn error message = %q, want %q", completed.ErrorMessage, errorMessage)
|
|
}
|
|
if completed.ErrorCode != "insufficient_credits" {
|
|
t.Fatalf("root provider turn error code = %q, want insufficient_credits", completed.ErrorCode)
|
|
}
|
|
}
|
|
|
|
func TestRootProviderTurnFailurePersistsUnknownCodeWithoutDiagnostics(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
ctx, ok := activityEventContext(session, "root-provider-turn-failed", "root-turn-1")
|
|
if !ok {
|
|
t.Fatal("activityEventContext() returned !ok")
|
|
}
|
|
failed := activityshared.NewRootProviderTurnCompleted(ctx, "root-turn-1", "provider-turn-1", activityshared.TurnOutcomeFailed)
|
|
|
|
report := reportActivityInput(session, []activityshared.Event{failed})
|
|
if len(report.StatePatches) != 1 {
|
|
t.Fatalf("state patches = %#v, want one root provider failure", report.StatePatches)
|
|
}
|
|
completed := report.StatePatches[0].RootProviderTurn
|
|
if completed == nil {
|
|
t.Fatal("root provider turn transition is nil")
|
|
}
|
|
if completed.ErrorCode != "unknown" || completed.ErrorMessage != "" {
|
|
t.Fatalf("root provider turn error = %q/%q, want unknown/empty", completed.ErrorCode, completed.ErrorMessage)
|
|
}
|
|
}
|
|
|
|
func TestSessionStatusFromActivityPreservesWaiting(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
got := sessionStatusFromActivity(string(activityshared.SessionStatusWaiting))
|
|
|
|
if got != SessionStatusWaiting {
|
|
t.Fatalf("sessionStatusFromActivity(waiting) = %q, want %q", got, SessionStatusWaiting)
|
|
}
|
|
}
|
|
|
|
func TestActivitySessionStatusFromControllerStatusPreservesWaiting(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
got := activitySessionStatusFromControllerStatus(SessionStatusWaiting)
|
|
|
|
if got != activityshared.SessionStatusWaiting {
|
|
t.Fatalf("activitySessionStatusFromControllerStatus(waiting) = %q, want %q", got, activityshared.SessionStatusWaiting)
|
|
}
|
|
}
|
|
|
|
func TestReportableActivityEventsReportsFailedAssistantFinalSnapshots(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
normalizer := newACPTurnNormalizer()
|
|
events := append([]activityshared.Event{},
|
|
normalizer.AppendThinkingChunk(session, "turn-1", "thinking")...,
|
|
)
|
|
events = append(events, normalizer.AppendAssistantChunk(session, "turn-1", "answer")...)
|
|
events = append(events, normalizer.Finish(session, "turn-1", messageStreamStateFailed)...)
|
|
|
|
report := reportActivityInput(session, events)
|
|
assistant := messageUpdatesWithKind(report, "text")
|
|
if len(assistant) != 2 {
|
|
t.Fatalf("assistant message updates = %#v, want streaming and failed final", assistant)
|
|
}
|
|
if assistant[1].Status != messageStreamStateFailed {
|
|
t.Fatalf("assistant status = %q, want failed", assistant[1].Status)
|
|
}
|
|
thinking := messageUpdatesWithKind(report, "reasoning")
|
|
if len(thinking) != 2 {
|
|
t.Fatalf("thinking message updates = %#v, want streaming and failed final", thinking)
|
|
}
|
|
if thinking[1].Status != messageStreamStateFailed {
|
|
t.Fatalf("thinking status = %q, want failed", thinking[1].Status)
|
|
}
|
|
}
|
|
|
|
func TestReportableActivityEventsReportsInterruptedOpenToolCalls(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
normalizer := newACPTurnNormalizer()
|
|
events, ok := normalizer.ToolCallEvents(session, "turn-1", map[string]any{
|
|
"toolCallId": "tool-1",
|
|
"title": "Run command",
|
|
"status": "in_progress",
|
|
"kind": "execute",
|
|
"rawInput": map[string]any{
|
|
"command": []any{"/bin/zsh", "-lc", "rg TODO"},
|
|
},
|
|
})
|
|
if !ok {
|
|
t.Fatal("ToolCallEvents() returned !ok")
|
|
}
|
|
|
|
events = append(events, normalizer.FinishInterrupted(session, "turn-1", "user_interrupt")...)
|
|
report := reportActivityInput(session, events)
|
|
failedCalls := messageUpdatesWithKind(report, "tool_call")
|
|
if len(failedCalls) != 2 {
|
|
t.Fatalf("failed call message updates = %#v, want start and interrupted final", failedCalls)
|
|
}
|
|
failedCall := failedCalls[1]
|
|
if failedCall.Status != "failed" {
|
|
t.Fatalf("failed call status = %q, want failed", failedCall.Status)
|
|
}
|
|
if got := payloadString(failedCall.Payload, "status"); got != SessionStatusCanceled {
|
|
t.Fatalf("failed call payload status = %q, want canceled", got)
|
|
}
|
|
if got := payloadString(payloadMap(failedCall.Payload, "error"), "reason"); got != "user_interrupt" {
|
|
t.Fatalf("failed call error payload = %#v, want interrupt reason", failedCall.Payload)
|
|
}
|
|
}
|
|
|
|
// TestReportableActivityEventsReportsFailedOpenToolCalls covers a tool call
|
|
// that is still open (no item/completed ever arrived for it) when its own
|
|
// turn otherwise reaches a normal terminal state - for example codex
|
|
// silently declining a spawnAgent delegation for a schema conflict, with no
|
|
// further notification tied to that call id for the rest of the turn
|
|
// (confirmed via exported session transcripts). Reporting it as a
|
|
// successful completion would paint a rejected/never-run call as having
|
|
// succeeded, which is exactly what previously left rejected sub-agent
|
|
// delegations rendered as stuck "running"/"queued" forever instead of
|
|
// failed. It must close out as failed, matching how an interrupted/failed
|
|
// turn already handles dangling calls.
|
|
func TestReportableActivityEventsReportsFailedOpenToolCalls(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
normalizer := newACPTurnNormalizer()
|
|
events, ok := normalizer.ToolCallEvents(session, "turn-1", map[string]any{
|
|
"toolCallId": "tool-1",
|
|
"title": "Run command",
|
|
"status": "in_progress",
|
|
"kind": "execute",
|
|
"rawInput": map[string]any{
|
|
"command": []any{"/bin/zsh", "-lc", "cat .env"},
|
|
},
|
|
})
|
|
if !ok {
|
|
t.Fatal("ToolCallEvents() returned !ok")
|
|
}
|
|
|
|
events = append(events, normalizer.FinishCompleted(session, "turn-1")...)
|
|
report := reportActivityInput(session, events)
|
|
completedCalls := messageUpdatesWithKind(report, "tool_call")
|
|
if len(completedCalls) != 2 {
|
|
t.Fatalf("completed call message updates = %#v, want start and failed final", completedCalls)
|
|
}
|
|
if completedCalls[1].MessageID != completedCalls[0].MessageID ||
|
|
completedCalls[1].CallID != completedCalls[0].CallID {
|
|
t.Fatalf("completed call identity = start:%#v completed:%#v, want same message and call IDs", completedCalls[0], completedCalls[1])
|
|
}
|
|
completedCall := completedCalls[1]
|
|
if completedCall.Status != messageStreamStateFailed {
|
|
t.Fatalf("dangling call status = %q, want failed", completedCall.Status)
|
|
}
|
|
if got := payloadString(completedCall.Payload, "status"); got != messageStreamStateFailed {
|
|
t.Fatalf("dangling call payload status = %q, want failed", got)
|
|
}
|
|
if got := payloadMap(completedCall.Payload, "error"); got == nil {
|
|
t.Fatalf("dangling call payload error = %#v, want a non-nil error detail", got)
|
|
}
|
|
}
|
|
|
|
func TestReportableActivityEventsIncludesCallStarted(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
session.AgentSessionID = "4e70b18d-b8b5-47a1-b293-3b98e4a23310"
|
|
|
|
events := ReportableActivityEvents([]activityshared.Event{
|
|
newTurnActivityEvent(session, EventCallStarted, "turn-1", messageStreamStateStreaming, "", "Read files", map[string]any{
|
|
"callId": "tool-1",
|
|
"callType": "tool",
|
|
"name": "Read files",
|
|
}),
|
|
})
|
|
|
|
if len(events) != 1 {
|
|
t.Fatalf("activity events = %#v, want call.started to be reportable", events)
|
|
}
|
|
if events[0].Type != activityshared.EventCallStarted {
|
|
t.Fatalf("activity event type = %q, want call.started", events[0].Type)
|
|
}
|
|
}
|
|
|
|
func TestNewTurnActivityEventDefaultsUserMessageMetadata(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
event := newTurnActivityEventWithID(session, "message-user-1", EventMessage, "turn-1", "", RoleUser, "hello", nil)
|
|
|
|
if event.Payload.Metadata["messageId"] != "message-user-1" {
|
|
t.Fatalf("message metadata = %#v, want message id", event.Payload.Metadata)
|
|
}
|
|
if event.Payload.Metadata["contentMode"] != messageContentModeSnapshot {
|
|
t.Fatalf("message metadata = %#v, want snapshot content mode", event.Payload.Metadata)
|
|
}
|
|
if event.Payload.Metadata["streamState"] != messageStreamStateCompleted {
|
|
t.Fatalf("message metadata = %#v, want completed streamState", event.Payload.Metadata)
|
|
}
|
|
}
|
|
|
|
func TestProjectActivityEventsToStreamEventsCarriesApprovalEnvelope(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
session.AgentSessionID = "4e70b18d-b8b5-47a1-b293-3b98e4a23310"
|
|
|
|
events := ProjectActivityEventsToStreamEvents(session, []activityshared.Event{
|
|
newTurnActivityEvent(session, EventCallStarted, "turn-approval", SessionStatusWaiting, "", "Read files", map[string]any{
|
|
"callId": "tool-approval-1",
|
|
"callType": "approval",
|
|
"name": "Read files",
|
|
"status": "waiting_approval",
|
|
"input": map[string]any{
|
|
"requestId": "permission-1",
|
|
"options": []map[string]any{
|
|
{"id": "allow_once", "label": "Allow once"},
|
|
},
|
|
},
|
|
}),
|
|
})
|
|
|
|
if len(events) != 1 {
|
|
t.Fatalf("stream events = %#v, want one approval call", events)
|
|
}
|
|
item, ok := events[0].Data.(agentsessionstore.WorkspaceAgentMessageUpdate)
|
|
if !ok {
|
|
t.Fatalf("stream event data = %#v, want message update", events[0].Data)
|
|
}
|
|
if got := payloadMap(item.Payload, "input"); got == nil || got["requestId"] != "permission-1" {
|
|
t.Fatalf("approval input payload = %#v, want requestId", item.Payload)
|
|
}
|
|
if item.Payload["callType"] != "approval" {
|
|
t.Fatalf("callType = %#v, want approval", item.Payload["callType"])
|
|
}
|
|
}
|
|
|
|
func TestProjectActivityEventsToStreamEventsCarriesInteractiveMetadata(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
session.AgentSessionID = "4e70b18d-b8b5-47a1-b293-3b98e4a23310"
|
|
|
|
events := ProjectActivityEventsToStreamEvents(session, []activityshared.Event{
|
|
newTurnActivityEvent(session, EventCallStarted, "turn-ask-user", SessionStatusWaiting, "", "AskUserQuestion", map[string]any{
|
|
"callId": "tool-ask-1",
|
|
"callType": "interactive",
|
|
"name": "AskUserQuestion",
|
|
"status": "waiting_input",
|
|
"input": map[string]any{
|
|
"questions": []map[string]any{
|
|
{"question": "Which approach should we use?"},
|
|
},
|
|
},
|
|
"metadata": map[string]any{
|
|
"interactiveKind": "ask-user",
|
|
},
|
|
}),
|
|
})
|
|
|
|
if len(events) != 1 {
|
|
t.Fatalf("stream events = %#v, want one interactive call", events)
|
|
}
|
|
item, ok := events[0].Data.(agentsessionstore.WorkspaceAgentMessageUpdate)
|
|
if !ok {
|
|
t.Fatalf("stream event data = %#v, want message update", events[0].Data)
|
|
}
|
|
if item.Payload["callType"] != "interactive" {
|
|
t.Fatalf("callType = %#v, want interactive", item.Payload["callType"])
|
|
}
|
|
if got := payloadMap(item.Payload, "input"); got == nil {
|
|
t.Fatalf("interactive input = %#v, want preserved questions", item.Payload)
|
|
}
|
|
if got := payloadString(item.Payload, "status"); got != "waiting_input" {
|
|
t.Fatalf("status = %q, want waiting_input", got)
|
|
}
|
|
}
|
|
|
|
func TestProjectActivityEventsToStreamEventsAddsVisibleTurnFailureMessage(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
session.Provider = hermesExtensionTestProvider
|
|
session.AgentSessionID = "4e70b18d-b8b5-47a1-b293-3b98e4a23310"
|
|
|
|
events := ProjectActivityEventsToStreamEvents(session, []activityshared.Event{
|
|
newTurnActivityEventWithID(session, "turn-failed-1", EventTurnFailed, "turn-1", SessionStatusFailed, "", "", map[string]any{
|
|
"error": "\x1b[33mAPI Error: 429 rate limit\x1b[39m",
|
|
}),
|
|
})
|
|
report := reportActivityInput(session, []activityshared.Event{
|
|
newTurnActivityEventWithID(session, "turn-failed-report-1", EventTurnFailed, "turn-1", SessionStatusFailed, "", "", map[string]any{
|
|
"error": "API Error: 429 rate limit",
|
|
}),
|
|
})
|
|
if len(report.MessageUpdates) != 1 || len(report.SessionAudits) != 0 {
|
|
t.Fatalf("turn failure report = %#v, want turn message only", report)
|
|
}
|
|
|
|
if len(events) != 2 {
|
|
t.Fatalf("stream events = %#v, want state patch and visible failure message", events)
|
|
}
|
|
item, ok := events[1].Data.(agentsessionstore.WorkspaceAgentMessageUpdate)
|
|
if !ok {
|
|
t.Fatalf("stream event data = %#v, want message update", events[1].Data)
|
|
}
|
|
if item.Kind != "text" || item.Status != messageStreamStateFailed {
|
|
t.Fatalf("visible failure item = %#v", item)
|
|
}
|
|
if item.Semantics == nil || !item.Semantics.UserVisibleAssistantResponse {
|
|
t.Fatalf("visible failure semantics = %#v, want explicit user-visible assistant response", item.Semantics)
|
|
}
|
|
if item.Payload["kind"] != visibleErrorKind ||
|
|
item.Payload["phase"] != "turn" ||
|
|
item.Payload["code"] != "quota_or_rate_limit" ||
|
|
item.Payload["detail"] != "API Error: 429 rate limit" {
|
|
t.Fatalf("visible failure payload = %#v", item.Payload)
|
|
}
|
|
}
|
|
|
|
func TestProjectActivityEventsToStreamEventsDoesNotDuplicateVisibleFailureMessage(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
session.AgentSessionID = "4e70b18d-b8b5-47a1-b293-3b98e4a23310"
|
|
|
|
events := ProjectActivityEventsToStreamEvents(session, []activityshared.Event{
|
|
newTurnActivityEventWithID(session, "assistant-failed-1", EventMessage, "turn-1", messageStreamStateFailed, RoleAssistant, "provider failure", map[string]any{
|
|
"streamState": messageStreamStateFailed,
|
|
}),
|
|
newTurnActivityEventWithID(session, "turn-failed-1", EventTurnFailed, "turn-1", SessionStatusFailed, "", "", map[string]any{
|
|
"error": "provider failure",
|
|
}),
|
|
})
|
|
|
|
var visibleFailures int
|
|
for _, event := range events {
|
|
item, ok := event.Data.(agentsessionstore.WorkspaceAgentMessageUpdate)
|
|
if ok && item.Payload["kind"] == visibleErrorKind {
|
|
visibleFailures++
|
|
}
|
|
}
|
|
if visibleFailures != 0 {
|
|
t.Fatalf("visible failure count = %d, want provider failed assistant message to suppress synthetic item", visibleFailures)
|
|
}
|
|
}
|
|
|
|
func TestNewActivityEventsCarryProjectionMetadata(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
session.AgentSessionID = "4e70b18d-b8b5-47a1-b293-3b98e4a23310"
|
|
|
|
event := newTurnActivityEventWithID(session, "stable-message-id", EventMessage, "turn-1", messageStreamStateCompleted, RoleAssistant, "done", map[string]any{
|
|
"messageId": "message-1",
|
|
"contentMode": messageContentModeSnapshot,
|
|
"streamState": messageStreamStateCompleted,
|
|
"adapterExtra": "kept-for-local-debug",
|
|
})
|
|
|
|
if event.EventID != "stable-message-id" {
|
|
t.Fatalf("activity event id = %q, want stable-message-id", event.EventID)
|
|
}
|
|
if event.Type != activityshared.EventMessageAppended {
|
|
t.Fatalf("activity type = %q, want %q", event.Type, activityshared.EventMessageAppended)
|
|
}
|
|
if event.Payload.Metadata["messageId"] != "message-1" {
|
|
t.Fatalf("activity metadata = %#v, want message id copied", event.Payload.Metadata)
|
|
}
|
|
}
|
|
|
|
func TestRuntimeSessionStartUsesSessionStartedEvent(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
session := reportTestSession()
|
|
session.AgentSessionID = "4e70b18d-b8b5-47a1-b293-3b98e4a23310"
|
|
|
|
events := ReportableActivityEvents([]activityshared.Event{
|
|
newSessionActivityEvent(session, EventSessionStarted, SessionStatusReady, nil),
|
|
})
|
|
|
|
if len(events) != 1 {
|
|
t.Fatalf("activity events = %#v, want exactly the session.started event", events)
|
|
}
|
|
if events[0].Type != activityshared.EventSessionStarted {
|
|
t.Fatalf("activity event type = %q, want %q", events[0].Type, activityshared.EventSessionStarted)
|
|
}
|
|
}
|
|
|
|
func messageUpdatesWithKind(report agentsessionstore.ReportActivityInput, kind string) []agentsessionstore.WorkspaceAgentMessageUpdate {
|
|
var out []agentsessionstore.WorkspaceAgentMessageUpdate
|
|
for _, update := range report.MessageUpdates {
|
|
if update.Kind == kind {
|
|
out = append(out, update)
|
|
}
|
|
}
|
|
return out
|
|
}
|