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
525 lines
17 KiB
Go
525 lines
17 KiB
Go
package agentruntime
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
activityshared "github.com/tutti-os/tutti/packages/agent/daemon/activity/events"
|
|
)
|
|
|
|
func TestCodexAppServerAdapterExecSteersActiveTurn(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
adapter, transport, session := startedAppServerAdapter(t)
|
|
transport.server.holdTurn = true
|
|
|
|
execDone := make(chan struct{})
|
|
go func() {
|
|
_, _ = adapter.Exec(context.Background(), session, []PromptContentBlock{{
|
|
Type: "text", Text: "long task",
|
|
}}, "", "turn-local-1", nil, nil)
|
|
close(execDone)
|
|
}()
|
|
waitForCondition(t, func() bool {
|
|
return adapter.sessionActiveTurnID(session.AgentSessionID) == "turn-1"
|
|
})
|
|
|
|
dispatches := make(chan ProviderDispatchResult, 1)
|
|
events, err := adapter.ExecWithProviderAcceptance(
|
|
context.Background(),
|
|
session,
|
|
[]PromptContentBlock{{
|
|
Type: "text", Text: "also update the docs",
|
|
}},
|
|
"",
|
|
"turn-local-2",
|
|
nil,
|
|
nil,
|
|
func(result ProviderDispatchResult) { dispatches <- result },
|
|
nil,
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("steer Exec: %v", err)
|
|
}
|
|
dispatch := <-dispatches
|
|
if dispatch.Disposition != DispatchDispositionAppliedWithoutProviderTurn ||
|
|
dispatch.Acceptance != nil {
|
|
t.Fatalf("steer dispatch = %#v", dispatch)
|
|
}
|
|
steer := appServerRequestParams(t, transport.conn, appServerMethodTurnSteer)
|
|
if asString(steer["expectedTurnId"]) != "turn-1" {
|
|
t.Fatalf("turn/steer params = %#v", steer)
|
|
}
|
|
messages := eventsOfType(events, activityshared.EventMessageAppended)
|
|
if len(messages) != 1 || messages[0].Payload.Role != activityshared.MessageRoleUser {
|
|
t.Fatalf("steer events = %#v, want single user message", events)
|
|
}
|
|
// The controller relies on this marker (turnSteeredIntoActiveTurn) to
|
|
// settle the steer submission's turn record: no terminal event will ever
|
|
// arrive for a steered turn id.
|
|
if steered, ok := messages[0].Payload.Metadata["steered"].(bool); !ok || !steered {
|
|
t.Fatalf("steer message metadata = %#v, want steered=true", messages[0].Payload.Metadata)
|
|
}
|
|
if guidance, ok := messages[0].Payload.Metadata["guidance"].(bool); !ok || !guidance {
|
|
t.Fatalf("steer message metadata = %#v, want guidance=true", messages[0].Payload.Metadata)
|
|
}
|
|
|
|
transport.server.completePendingTurn()
|
|
select {
|
|
case <-execDone:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatalf("original Exec did not finish")
|
|
}
|
|
}
|
|
|
|
func TestCodexAppServerAdapterGuideActiveTurnUsesTurnSteer(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
adapter, transport, session := startedAppServerAdapter(t)
|
|
transport.server.holdTurn = true
|
|
|
|
execDone := make(chan struct{})
|
|
go func() {
|
|
_, _ = adapter.Exec(context.Background(), session, textPrompt("long task"), "", "turn-local-1", nil, nil)
|
|
close(execDone)
|
|
}()
|
|
waitForCondition(t, func() bool {
|
|
return adapter.sessionActiveTurnID(session.AgentSessionID) == "turn-1"
|
|
})
|
|
|
|
returned, err := adapter.GuideActiveTurn(
|
|
context.Background(), session, textPrompt("guide current turn"), "", "turn-local-1", nil, nil,
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("GuideActiveTurn: %v", err)
|
|
}
|
|
steer := appServerRequestParams(t, transport.conn, appServerMethodTurnSteer)
|
|
if asString(steer["threadId"]) != "codex-thread-1" || asString(steer["expectedTurnId"]) != "turn-1" {
|
|
t.Fatalf("turn/steer params = %#v", steer)
|
|
}
|
|
if interrupts := appServerRequestParamsList(t, transport.conn, appServerMethodTurnInterrupt); len(interrupts) != 0 {
|
|
t.Fatalf("turn/interrupt requests = %#v, want non-interrupting guidance", interrupts)
|
|
}
|
|
if starts := appServerRequestParamsList(t, transport.conn, appServerMethodTurnStart); len(starts) != 1 {
|
|
t.Fatalf("turn/start requests = %#v, want only the original provider turn", starts)
|
|
}
|
|
messages := eventsOfType(returned, activityshared.EventMessageAppended)
|
|
if len(messages) != 1 || messages[0].Payload.Role != activityshared.MessageRoleUser {
|
|
t.Fatalf("guidance events = %#v, want one user message", returned)
|
|
}
|
|
if messages[0].Payload.Metadata["guidance"] != true || messages[0].Payload.Metadata["steered"] != true {
|
|
t.Fatalf("guidance metadata = %#v, want guidance+steered", messages[0].Payload.Metadata)
|
|
}
|
|
|
|
transport.server.completePendingTurn()
|
|
select {
|
|
case <-execDone:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatalf("original Exec did not finish")
|
|
}
|
|
}
|
|
|
|
func TestCodexAppServerAdapterGuidancePreflightFailureIsNotDispatched(t *testing.T) {
|
|
adapter := NewCodexAppServerAdapter(nil)
|
|
dispatches := make(chan ProviderDispatchResult, 1)
|
|
_, err := adapter.GuideActiveTurnWithProviderDispatch(
|
|
t.Context(),
|
|
standardTestSession(ProviderCodex),
|
|
textPrompt("guide current turn"),
|
|
"",
|
|
"turn-guidance",
|
|
nil,
|
|
nil,
|
|
func(result ProviderDispatchResult) { dispatches <- result },
|
|
)
|
|
if !errors.Is(err, ErrSessionDisconnected) {
|
|
t.Fatalf("GuideActiveTurn error = %v, want ErrSessionDisconnected", err)
|
|
}
|
|
if dispatch := <-dispatches; dispatch.Disposition != DispatchDispositionNotDispatched {
|
|
t.Fatalf("guidance dispatch = %#v, want not dispatched", dispatch)
|
|
}
|
|
}
|
|
|
|
func TestCodexAppServerAdapterGuideMaterializesRemoteImageAtProviderBoundary(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
adapter, transport, session := startedAppServerAdapter(t)
|
|
imageURL, materializer := testRemotePromptImageMaterializer(t)
|
|
adapter.promptImageMaterializer = materializer
|
|
transport.server.holdTurn = true
|
|
|
|
execDone := make(chan struct{})
|
|
go func() {
|
|
_, _ = adapter.Exec(context.Background(), session, textPrompt("long task"), "", "turn-local-1", nil, nil)
|
|
close(execDone)
|
|
}()
|
|
waitForCondition(t, func() bool {
|
|
return adapter.sessionActiveTurnID(session.AgentSessionID) == "turn-1"
|
|
})
|
|
|
|
if _, err := adapter.GuideActiveTurn(context.Background(), session, []PromptContentBlock{
|
|
{Type: "text", Text: "use this screenshot"},
|
|
{Type: "image", MimeType: "image/png", URL: imageURL},
|
|
}, "", "turn-local-1", nil, nil); err != nil {
|
|
t.Fatalf("GuideActiveTurn: %v", err)
|
|
}
|
|
|
|
steer := appServerRequestParams(t, transport.conn, appServerMethodTurnSteer)
|
|
input, _ := steer["input"].([]any)
|
|
if len(input) != 2 {
|
|
t.Fatalf("guidance turn/steer input = %#v, want text+image", steer["input"])
|
|
}
|
|
image := payloadObject(input[1])
|
|
if got := asString(image["url"]); got != "data:image/png;base64,aGk=" {
|
|
t.Fatalf("guidance turn/steer image URL = %q, want inline data URL", got)
|
|
}
|
|
if interrupts := appServerRequestParamsList(t, transport.conn, appServerMethodTurnInterrupt); len(interrupts) != 0 {
|
|
t.Fatalf("turn/interrupt requests = %#v, want non-interrupting guidance", interrupts)
|
|
}
|
|
if starts := appServerRequestParamsList(t, transport.conn, appServerMethodTurnStart); len(starts) != 1 {
|
|
t.Fatalf("turn/start requests = %#v, want only the original provider turn", starts)
|
|
}
|
|
|
|
transport.server.completePendingTurn()
|
|
select {
|
|
case <-execDone:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("original Exec did not finish after guidance steer")
|
|
}
|
|
waitForCondition(t, func() bool {
|
|
return adapter.sessionActiveTurnID(session.AgentSessionID) == ""
|
|
})
|
|
}
|
|
|
|
func TestCodexAppServerAdapterGuidanceStartsProviderContinuationOnSameRootTurn(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
adapter, transport, session := startedAppServerAdapter(t)
|
|
if _, err := adapter.Exec(context.Background(), session, textPrompt("delegate work"), "", "root-turn-1", nil, nil); err != nil {
|
|
t.Fatalf("initial Exec: %v", err)
|
|
}
|
|
if got := adapter.sessionActiveTurnID(session.AgentSessionID); got != "" {
|
|
t.Fatalf("provider turn id after completion = %q, want empty", got)
|
|
}
|
|
_, childEvents := adapter.rememberAppServerChildThreads(
|
|
session,
|
|
session.ProviderSessionID,
|
|
session.AgentSessionID,
|
|
"root-turn-1",
|
|
session.AgentSessionID,
|
|
"root-turn-1",
|
|
map[string]any{
|
|
"type": "collabAgentToolCall",
|
|
"id": "spawn-child-1",
|
|
"tool": "spawnAgent",
|
|
"receiverThreadIds": []any{"child-thread-1"},
|
|
},
|
|
)
|
|
if len(childEvents) != 1 || childEvents[0].Type != activityshared.EventSessionStarted {
|
|
t.Fatalf("child creation events = %#v", childEvents)
|
|
}
|
|
|
|
var mu sync.Mutex
|
|
var streamed []activityshared.Event
|
|
returned, err := adapter.GuideActiveTurn(
|
|
context.Background(),
|
|
session,
|
|
textPrompt("include the child's findings"),
|
|
"",
|
|
"root-turn-1",
|
|
func(events []activityshared.Event) {
|
|
mu.Lock()
|
|
streamed = append(streamed, events...)
|
|
mu.Unlock()
|
|
},
|
|
nil,
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("GuideActiveTurn: %v", err)
|
|
}
|
|
if len(returned) != 1 || returned[0].Type != activityshared.EventRootProviderTurnStarted ||
|
|
returned[0].Payload.TurnID != "root-turn-1" {
|
|
t.Fatalf("guidance return events = %#v, want provisional provider continuation", returned)
|
|
}
|
|
|
|
waitForCondition(t, func() bool {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
for _, event := range streamed {
|
|
if event.Type == activityshared.EventRootProviderTurnCompleted && event.Payload.TurnID == "root-turn-1" {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
})
|
|
if requests := appServerRequestParamsList(t, transport.conn, appServerMethodTurnStart); len(requests) < 2 {
|
|
t.Fatalf("turn/start requests = %#v, want initial turn and same-root continuation", requests)
|
|
}
|
|
}
|
|
|
|
func TestCodexAppServerAdapterGuidanceContinuationMaterializationFailureClosesProvisionalProviderTurn(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
adapter, transport, session := startedAppServerAdapter(t)
|
|
if _, err := adapter.Exec(context.Background(), session, textPrompt("delegate work"), "", "root-turn-1", nil, nil); err != nil {
|
|
t.Fatalf("initial Exec: %v", err)
|
|
}
|
|
_, childEvents := adapter.rememberAppServerChildThreads(
|
|
session,
|
|
session.ProviderSessionID,
|
|
session.AgentSessionID,
|
|
"root-turn-1",
|
|
session.AgentSessionID,
|
|
"root-turn-1",
|
|
map[string]any{
|
|
"type": "collabAgentToolCall",
|
|
"id": "spawn-child-1",
|
|
"tool": "spawnAgent",
|
|
"receiverThreadIds": []any{"child-thread-1"},
|
|
},
|
|
)
|
|
if len(childEvents) != 1 || childEvents[0].Type != activityshared.EventSessionStarted {
|
|
t.Fatalf("child creation events = %#v", childEvents)
|
|
}
|
|
adapter.promptImageMaterializer = func(context.Context, []PromptContentBlock) ([]PromptContentBlock, error) {
|
|
return nil, errors.New("signed image expired")
|
|
}
|
|
|
|
var mu sync.Mutex
|
|
var streamed []activityshared.Event
|
|
returned, err := adapter.GuideActiveTurn(
|
|
context.Background(),
|
|
session,
|
|
[]PromptContentBlock{
|
|
{Type: "text", Text: "include this image"},
|
|
{Type: "image", MimeType: "image/png", URL: "https://public.example/image.png"},
|
|
},
|
|
"",
|
|
"root-turn-1",
|
|
func(events []activityshared.Event) {
|
|
mu.Lock()
|
|
streamed = append(streamed, events...)
|
|
mu.Unlock()
|
|
},
|
|
nil,
|
|
)
|
|
if err != nil {
|
|
t.Fatalf("GuideActiveTurn: %v", err)
|
|
}
|
|
if len(returned) != 1 || returned[0].Type != activityshared.EventRootProviderTurnStarted {
|
|
t.Fatalf("guidance return events = %#v, want provisional provider continuation", returned)
|
|
}
|
|
attemptID := returned[0].Payload.ProviderTurnID
|
|
waitForCondition(t, func() bool {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
for _, event := range streamed {
|
|
if event.Type == activityshared.EventRootProviderTurnCompleted &&
|
|
event.Payload.ProviderTurnID == attemptID &&
|
|
event.Payload.TurnOutcome == string(activityshared.TurnOutcomeFailed) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
})
|
|
if requests := appServerRequestParamsList(t, transport.conn, appServerMethodTurnStart); len(requests) != 1 {
|
|
t.Fatalf("turn/start requests = %#v, want no continuation provider request", requests)
|
|
}
|
|
}
|
|
|
|
func TestCodexAppServerAdapterConcurrentGuidancePublishesSingleProvisionalContinuation(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
adapter, transport, session := startedAppServerAdapter(t)
|
|
if _, err := adapter.Exec(context.Background(), session, textPrompt("delegate work"), "", "root-turn-1", nil, nil); err != nil {
|
|
t.Fatalf("initial Exec: %v", err)
|
|
}
|
|
_, childEvents := adapter.rememberAppServerChildThreads(
|
|
session,
|
|
session.ProviderSessionID,
|
|
session.AgentSessionID,
|
|
"root-turn-1",
|
|
session.AgentSessionID,
|
|
"root-turn-1",
|
|
map[string]any{
|
|
"type": "collabAgentToolCall",
|
|
"id": "spawn-child-1",
|
|
"tool": "spawnAgent",
|
|
"receiverThreadIds": []any{"child-thread-1"},
|
|
},
|
|
)
|
|
if len(childEvents) != 1 || childEvents[0].Type != activityshared.EventSessionStarted {
|
|
t.Fatalf("child creation events = %#v", childEvents)
|
|
}
|
|
transport.server.holdTurn = true
|
|
|
|
type guidanceCallResult struct {
|
|
events []activityshared.Event
|
|
err error
|
|
}
|
|
var streamedMu sync.Mutex
|
|
var streamed []activityshared.Event
|
|
emit := func(events []activityshared.Event) {
|
|
streamedMu.Lock()
|
|
defer streamedMu.Unlock()
|
|
streamed = append(streamed, events...)
|
|
}
|
|
start := make(chan struct{})
|
|
results := make(chan guidanceCallResult, 2)
|
|
for index := 0; index < 2; index++ {
|
|
index := index
|
|
go func() {
|
|
<-start
|
|
events, err := adapter.GuideActiveTurn(
|
|
context.Background(),
|
|
session,
|
|
textPrompt(fmt.Sprintf("guidance %d", index)),
|
|
"",
|
|
"root-turn-1",
|
|
emit,
|
|
nil,
|
|
)
|
|
results <- guidanceCallResult{events: events, err: err}
|
|
}()
|
|
}
|
|
close(start)
|
|
|
|
provisionalStarts := 0
|
|
for index := 0; index < 2; index++ {
|
|
select {
|
|
case result := <-results:
|
|
for _, event := range result.events {
|
|
if event.Type == activityshared.EventRootProviderTurnStarted &&
|
|
strings.HasPrefix(event.Payload.ProviderTurnID, "continuation:") {
|
|
provisionalStarts++
|
|
}
|
|
}
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatal("timed out waiting for concurrent guidance")
|
|
}
|
|
}
|
|
if provisionalStarts != 1 {
|
|
t.Fatalf("provisional provider starts = %d, want exactly 1", provisionalStarts)
|
|
}
|
|
waitForCondition(t, func() bool {
|
|
streamedMu.Lock()
|
|
defer streamedMu.Unlock()
|
|
started := 0
|
|
for _, event := range streamed {
|
|
if event.Type == activityshared.EventTurnStarted {
|
|
started++
|
|
}
|
|
}
|
|
return started >= 1
|
|
})
|
|
streamedMu.Lock()
|
|
started := 0
|
|
for _, event := range streamed {
|
|
if event.Type == activityshared.EventTurnStarted {
|
|
started++
|
|
}
|
|
}
|
|
streamedMu.Unlock()
|
|
if started != 1 {
|
|
t.Fatalf("streamed turn starts = %d, want exactly 1 admitted continuation", started)
|
|
}
|
|
|
|
waitForCondition(t, func() bool {
|
|
return adapter.sessionActiveTurnID(session.AgentSessionID) != ""
|
|
})
|
|
transport.server.completePendingTurn()
|
|
waitForCondition(t, func() bool {
|
|
return adapter.sessionActiveTurnID(session.AgentSessionID) == ""
|
|
})
|
|
}
|
|
|
|
func TestCodexAppServerAdapterRejectedContinuationAdmissionEmitsNoTurnStart(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
adapter, _, session := startedAppServerAdapter(t)
|
|
blockingTurn := &codexAppServerActiveTurn{turnID: "blocking-turn"}
|
|
if !adapter.beginActiveTurn(session.AgentSessionID, blockingTurn) {
|
|
t.Fatal("failed to reserve blocking turn")
|
|
}
|
|
defer adapter.endActiveTurn(session.AgentSessionID, blockingTurn)
|
|
|
|
var streamed []activityshared.Event
|
|
continuation := newCodexGuidanceContinuationAdmission("continuation:rejected")
|
|
events, err := adapter.execBlocking(
|
|
context.Background(),
|
|
session,
|
|
textPrompt("guidance"),
|
|
"",
|
|
"root-turn-1",
|
|
func(next []activityshared.Event) {
|
|
streamed = append(streamed, next...)
|
|
},
|
|
nil,
|
|
codexTurnExecOptions{continuation: continuation},
|
|
)
|
|
if !errors.Is(err, ErrSessionActiveTurn) {
|
|
t.Fatalf("execBlocking error = %v, want ErrSessionActiveTurn", err)
|
|
}
|
|
if admittedErr := <-continuation.admitted; !errors.Is(admittedErr, ErrSessionActiveTurn) {
|
|
t.Fatalf("admission error = %v, want ErrSessionActiveTurn", admittedErr)
|
|
}
|
|
if len(events) != 0 || len(streamed) != 0 {
|
|
t.Fatalf("rejected continuation events = %#v, streamed = %#v; want none", events, streamed)
|
|
}
|
|
}
|
|
|
|
func TestCodexAppServerAdapterSteerRoutesAgentTargetMention(t *testing.T) {
|
|
t.Parallel()
|
|
|
|
adapter, transport, session := startedAppServerAdapter(t)
|
|
transport.server.holdTurn = true
|
|
|
|
execDone := make(chan struct{})
|
|
go func() {
|
|
_, _ = adapter.Exec(context.Background(), session, textPrompt("long task"), "", "turn-local-1", nil, nil)
|
|
close(execDone)
|
|
}()
|
|
waitForCondition(t, func() bool {
|
|
return adapter.sessionActiveTurnID(session.AgentSessionID) == "turn-1"
|
|
})
|
|
|
|
prompt := "让 [@Codex](mention://agent-target/local:codex?workspaceId=workspace-1) 来看下"
|
|
events, err := adapter.Exec(context.Background(), session, textPrompt(prompt), "", "turn-agent-target-steer", nil, nil)
|
|
if err != nil {
|
|
t.Fatalf("steer Exec: %v", err)
|
|
}
|
|
steer := appServerRequestParams(t, transport.conn, appServerMethodTurnSteer)
|
|
if asString(steer["expectedTurnId"]) != "turn-1" {
|
|
t.Fatalf("turn/steer params = %#v", steer)
|
|
}
|
|
input, _ := steer["input"].([]any)
|
|
if len(input) != 2 {
|
|
t.Fatalf("turn/steer input = %#v, want user prompt plus internal routing prompt", steer["input"])
|
|
}
|
|
first, _ := input[0].(map[string]any)
|
|
if asString(first["text"]) != prompt {
|
|
t.Fatalf("turn/steer user text = %q, want %q", asString(first["text"]), prompt)
|
|
}
|
|
last, _ := input[len(input)-1].(map[string]any)
|
|
if asString(last["text"]) != tuttiAgentMentionRoutingReminder {
|
|
t.Fatalf("turn/steer routing text = %q, want %q", asString(last["text"]), tuttiAgentMentionRoutingReminder)
|
|
}
|
|
|
|
userContent := firstUserMessageContent(t, events)
|
|
if userContent != prompt || strings.Contains(userContent, "system-reminder") {
|
|
t.Fatalf("user activity content = %q, want original prompt only", userContent)
|
|
}
|
|
|
|
transport.server.completePendingTurn()
|
|
select {
|
|
case <-execDone:
|
|
case <-time.After(5 * time.Second):
|
|
t.Fatalf("original Exec did not finish")
|
|
}
|
|
}
|