package wire import ( "context" "encoding/json" "errors" "fmt" "io" "net" "net/http" "net/http/httptest" "net/url" "os" "path/filepath" "strings" "sync" "sync/atomic" "testing" "time" "github.com/gorilla/websocket" "github.com/pingcap/failpoint" "github.com/stretchr/testify/require" "go.yaml.in/yaml/v3" "github.com/tidbcloud/AgentTeam/desktop/internal/agent" "github.com/tidbcloud/AgentTeam/desktop/internal/agent/runtimeconfig" "github.com/tidbcloud/AgentTeam/desktop/internal/agent/runtimeevents" "github.com/tidbcloud/AgentTeam/desktop/internal/consts" "github.com/tidbcloud/AgentTeam/desktop/internal/profile" "github.com/tidbcloud/AgentTeam/desktop/internal/promptcompiler" "github.com/tidbcloud/AgentTeam/desktop/internal/types" "github.com/tidbcloud/AgentTeam/desktop/internal/wire/apiclient" "github.com/tidbcloud/AgentTeam/desktop/internal/wire/mcpformat" "github.com/tidbcloud/AgentTeam/desktop/internal/wire/workspacefs" "github.com/tidbcloud/AgentTeam/internal/core/domain" ) func deprecatedUploadScopeAliasFormKeyForNegativeTest() string { return "space" + "Id" } type agentteamTestDispatcher struct { session types.SessionSnapshot sessions []types.SessionSnapshot hasSession bool sessionAgentID string dispatchCalls int startCalls int startErr error lastStartConfig types.AgentStartConfig startConfigs []types.AgentStartConfig lastStartWake *types.Message startWakes []*types.Message onStart func(call int) restoreCalls []agentteamTestRestoreCall setActiveTaskTargetCalls int lastActiveTaskTargetAgent string lastActiveTaskTarget types.ActiveTaskTarget ackedSuppressions []string ackedReconcileCalls int lastAckedReconcileAgent string lastAckedMessageIDs []string trajectoryItems []types.TrajectoryLogItem lastDispatchedMessage *types.Message dispatchedMessages []*types.Message dispatchErr error deliveryAckDeadline time.Time deliveryAckHasDeadline bool wakeToken string prepareVisibleReplyAgent string prepareVisibleReplySelectors []string confirmVisibleReplyCalls int lastConfirmedWakeAgent string lastConfirmedWakeToken string lastConfirmedWakeTarget string lastConfirmedReplyMessage string confirmVisibleReplyResult bool confirmReactionReplyCalls int lastReactionWakeAgent string lastReactionWakeToken string lastReactionWakeMessageID string confirmReactionReplyResult bool inboxMessages []types.InboxMessage interruptCalls int lastInterruptedAgent string events []string runtimeDeliveryAgent string runtimeDeliveryMessages []types.InboxMessage runtimeDeliveryPrepareCalls int refreshedAgentConfigs []agent.Config identityUpdates []identityUpdateCall refreshStartCalls int lastRefreshStartAgent string lastRefreshStartConfig types.AgentStartConfig sessionStateBus runtimeevents.Bus clearSessionAgent string clearSessionCalls int clearSessionErr error activeRunInterruptCalls int lastActiveRunInterruptAgent string } type identityUpdateCall struct { AgentID string Name string DisplayName string } type blockingManagedSkillLaunchDispatcher struct { agentteamTestDispatcher managedSkillReady atomic.Bool managedSkillRefreshes atomic.Int32 startCallsObserved atomic.Int32 managedSkillRefreshSeen chan struct{} managedSkillRelease <-chan struct{} } type blockingStartStopDispatcher struct { agentteamTestDispatcher startEntered chan struct{} startRelease <-chan struct{} startAdmissions atomic.Int32 running atomic.Bool stopCalls atomic.Int32 } type freshnessAgentTeamTestDispatcher struct { agentteamTestDispatcher snapshotTargetKey string snapshotAgentID string snapshot types.MessageFreshnessFrontier ackedAgentID string ackedMessages []types.InboxMessage heldAgentID string heldTargetKey string heldSeenUpToSeq int64 heldLatestSeq int64 freshnessInboxAgentID string freshnessInboxTargetKey string freshnessInboxThroughSeq int64 freshnessInboxLimit int freshnessInboxMessagesError error freshnessRuntimeDeliveryClaims []types.RuntimeDeliveryClaim freshnessRuntimeDeliveryOwner types.MCPToolResultDeliveryOwner pendingRuntimeInputAgentID string pendingRuntimeInputTargetKey string pendingRuntimeInputThroughSeq int64 pendingRuntimeInputCovers bool } type agentteamTestDeliveryOwner struct{} func (*agentteamTestDeliveryOwner) AcknowledgeMCPToolResultDelivery([]string, types.MCPToolDeliveryOutcome) { } func installAgentTeamTestRuntime(t *testing.T, name string) { t.Helper() binDir := t.TempDir() binaryPath := filepath.Join(binDir, name) require.NoError(t, os.WriteFile(binaryPath, []byte("#!/bin/sh\nexit 0\n"), 0o755)) t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH")) } func (d *blockingManagedSkillLaunchDispatcher) ManagedSkillLaunchReady(string) bool { return d.managedSkillReady.Load() } func (d *blockingManagedSkillLaunchDispatcher) RefreshSkillAccess(ctx context.Context, agentIDs []string) { d.managedSkillRefreshes.Add(1) if d.managedSkillRefreshSeen != nil { select { case d.managedSkillRefreshSeen <- struct{}{}: default: } } if d.managedSkillRelease != nil { select { case <-d.managedSkillRelease: case <-ctx.Done(): } } d.managedSkillReady.Store(true) } func (d *blockingManagedSkillLaunchDispatcher) StartAgentForAgentTeam(channelID, serverAgentID string, cfg types.AgentStartConfig, wake *types.Message) error { d.startCallsObserved.Add(1) return d.agentteamTestDispatcher.StartAgentForAgentTeam(channelID, serverAgentID, cfg, wake) } func (d *blockingStartStopDispatcher) StartAgentForAgentTeam(channelID, serverAgentID string, cfg types.AgentStartConfig, wake *types.Message) error { if d.startAdmissions.Add(1) == 1 { close(d.startEntered) <-d.startRelease } d.running.Store(true) return d.agentteamTestDispatcher.StartAgentForAgentTeam(channelID, serverAgentID, cfg, wake) } func (d *blockingStartStopDispatcher) StopByServerAgentID(string) bool { d.stopCalls.Add(1) return d.running.Swap(false) } func (d *blockingStartStopDispatcher) Session(string) (types.SessionSnapshot, bool) { if !d.running.Load() { return types.SessionSnapshot{}, false } return types.SessionSnapshot{Status: "running"}, true } func (d *freshnessAgentTeamTestDispatcher) MessageFreshnessSnapshot(agentID string, targetKey string) (types.MessageFreshnessFrontier, bool) { d.snapshotAgentID = agentID d.snapshotTargetKey = targetKey return d.snapshot, true } func (d *freshnessAgentTeamTestDispatcher) RememberMessageFreshnessObserved(agentID string, messages []types.InboxMessage) bool { d.ackedAgentID = agentID d.ackedMessages = append([]types.InboxMessage(nil), messages...) return true } func (d *freshnessAgentTeamTestDispatcher) RememberMessageFreshnessHeld(agentID string, targetKey string, seenUpToSeq int64, latestSeq int64) bool { d.heldAgentID = agentID d.heldTargetKey = targetKey d.heldSeenUpToSeq = seenUpToSeq d.heldLatestSeq = latestSeq return true } func (d *freshnessAgentTeamTestDispatcher) InboxMessagesForFreshnessPage(agentID string, targetKey string, throughSeq int64, limit int) (types.InboxReadPage, error) { d.freshnessInboxAgentID = agentID d.freshnessInboxTargetKey = targetKey d.freshnessInboxThroughSeq = throughSeq d.freshnessInboxLimit = limit if d.freshnessInboxMessagesError != nil { return types.InboxReadPage{}, d.freshnessInboxMessagesError } messages := append([]types.InboxMessage(nil), d.inboxMessages...) if limit <= 0 || limit > len(messages) { limit = len(messages) } return types.InboxReadPage{ Messages: append([]types.InboxMessage(nil), messages[:limit]...), RuntimeDeliveryClaims: append([]types.RuntimeDeliveryClaim(nil), d.freshnessRuntimeDeliveryClaims...), RuntimeDeliveryCommitOwner: d.freshnessRuntimeDeliveryOwner, HasMore: len(messages) > limit, RemainingCount: len(messages) - limit, }, nil } func (d *freshnessAgentTeamTestDispatcher) PendingRuntimeInputCoversFreshness(agentID string, targetKey string, throughSeq int64) bool { d.pendingRuntimeInputAgentID = agentID d.pendingRuntimeInputTargetKey = targetKey d.pendingRuntimeInputThroughSeq = throughSeq return d.pendingRuntimeInputCovers } type agentteamTestRestoreCall struct { ChannelID string AgentID string RequestID string Runtime string Model string SessionID string Strategy string RecoveryRef string } func (d *agentteamTestDispatcher) Session(agentID string) (types.SessionSnapshot, bool) { if d.sessionAgentID != "" && agentID != d.sessionAgentID { return types.SessionSnapshot{}, false } return d.session, d.hasSession } func (d *agentteamTestDispatcher) Sessions() []types.SessionSnapshot { return append([]types.SessionSnapshot(nil), d.sessions...) } func (d *agentteamTestDispatcher) SubscribeSessionStateChanges(subscriberName string) runtimeevents.Subscription { if d.sessionStateBus == nil { return nil } return d.sessionStateBus.Subscribe(subscriberName, runtimeevents.FilterKind(runtimeevents.KindSessionStateChanged), 8, runtimeevents.DropOldest) } func (d *agentteamTestDispatcher) publishSessionStateChanged(agentID string) { if d.sessionStateBus == nil { return } d.sessionStateBus.Publish(context.Background(), runtimeevents.NewEvent( runtimeevents.KindSessionStateChanged, runtimeevents.WithContext(runtimeevents.EventContext{AgentID: agentID}), )) } func (d *agentteamTestDispatcher) SetActiveTaskTarget(agentID string, target types.ActiveTaskTarget) bool { target = target.Normalize() if !target.Valid() { return false } d.setActiveTaskTargetCalls++ d.lastActiveTaskTargetAgent = agentID d.lastActiveTaskTarget = target d.session.ActiveTaskTarget = target return true } func (d *agentteamTestDispatcher) History(agentID string, limit int, before, after int64) ([]types.HistoryMessage, error) { return nil, nil } func (d *agentteamTestDispatcher) TrajectoryHistory(agentID string, limit int, before int64) []types.TrajectoryLogItem { return append([]types.TrajectoryLogItem(nil), d.trajectoryItems...) } func (d *agentteamTestDispatcher) PrepareVisibleReply(agentID string, selectors ...string) string { d.prepareVisibleReplyAgent = agentID d.prepareVisibleReplySelectors = append([]string(nil), selectors...) return d.wakeToken } func (d *agentteamTestDispatcher) ConfirmReactionReply(agentID, wakeToken, messageID string) bool { d.confirmReactionReplyCalls++ d.lastReactionWakeAgent = agentID d.lastReactionWakeToken = wakeToken d.lastReactionWakeMessageID = messageID return d.confirmReactionReplyResult } func (d *agentteamTestDispatcher) Dispatch(ctx context.Context, ch types.ServerConnection, msg *types.Message) { d.dispatchCalls++ d.events = append(d.events, "dispatch") d.lastDispatchedMessage = msg d.dispatchedMessages = append(d.dispatchedMessages, msg) } func (d *agentteamTestDispatcher) DispatchForDeliveryAck(ctx context.Context, ch types.ServerConnection, msg *types.Message) error { d.deliveryAckDeadline, d.deliveryAckHasDeadline = ctx.Deadline() d.Dispatch(ctx, ch, msg) return d.dispatchErr } func (d *agentteamTestDispatcher) PrepareRuntimeMessageToolDeliveries(agentID string, messages []types.InboxMessage) int { d.runtimeDeliveryAgent = agentID d.runtimeDeliveryMessages = append([]types.InboxMessage(nil), messages...) d.runtimeDeliveryPrepareCalls++ return len(messages) } func (d *agentteamTestDispatcher) StopByServerAgentID(serverAgentID string) bool { return false } func (d *agentteamTestDispatcher) ClearSessionResumeStateByAgent(agentID string) error { d.clearSessionCalls++ d.clearSessionAgent = agentID return d.clearSessionErr } func (d *agentteamTestDispatcher) InterruptByServerAgentID(serverAgentID string) bool { d.interruptCalls++ d.lastInterruptedAgent = serverAgentID d.events = append(d.events, "interrupt") return true } func (d *agentteamTestDispatcher) InterruptActiveRunByServerAgentID(serverAgentID string) bool { d.activeRunInterruptCalls++ d.lastActiveRunInterruptAgent = serverAgentID d.events = append(d.events, "active-run-interrupt") return true } func (d *agentteamTestDispatcher) SleepByServerAgentID(serverAgentID string) bool { return false } func TestHandleAgentStopClearSessionClearsDaemonLocalResumeState(t *testing.T) { disp := &agentteamTestDispatcher{} ch := &Protocol{id: "agentteam-test", disp: disp} ch.handleAgentStop(context.Background(), []byte(`{"type":"agent:stop","agentId":"server-agent-clear","clearSession":true}`)) require.Equal(t, 1, disp.clearSessionCalls) require.Equal(t, "server-agent-clear", disp.clearSessionAgent) } func TestHandleAgentInterruptKeepsSessionMetadata(t *testing.T) { disp := &agentteamTestDispatcher{} ch := &Protocol{id: "agentteam-test", disp: disp} ch.handleAgentInterrupt(context.Background(), []byte(`{"type":"agent:interrupt","agentId":"server-agent-interrupt"}`)) require.Equal(t, 1, disp.activeRunInterruptCalls) require.Equal(t, "server-agent-interrupt", disp.lastActiveRunInterruptAgent) require.Zero(t, disp.interruptCalls) require.Zero(t, disp.clearSessionCalls) } func (d *agentteamTestDispatcher) StartAgentForAgentTeam(channelID, serverAgentID string, cfg types.AgentStartConfig, wake *types.Message) error { d.startCalls++ d.lastStartConfig = cfg d.startConfigs = append(d.startConfigs, cfg) if wake != nil { d.lastStartWake = wake d.startWakes = append(d.startWakes, wake) } else { d.startWakes = append(d.startWakes, nil) } if d.onStart != nil { d.onStart(d.startCalls) } return d.startErr } func (d *agentteamTestDispatcher) UpdateAgentIdentity(serverAgentID, name, displayName string) { d.identityUpdates = append(d.identityUpdates, identityUpdateCall{ AgentID: serverAgentID, Name: name, DisplayName: displayName, }) } func (d *agentteamTestDispatcher) RefreshRunningAgentConfig(cfg agent.Config) bool { d.refreshedAgentConfigs = append(d.refreshedAgentConfigs, cfg) return true } func (d *agentteamTestDispatcher) RefreshRunningAgentStartConfig(serverAgentID string, cfg types.AgentStartConfig) bool { d.refreshStartCalls++ d.lastRefreshStartAgent = serverAgentID d.lastRefreshStartConfig = cfg return true } type agentteamTestRestartingStartConfigDispatcher struct { agentteamTestDispatcher } func (d *agentteamTestRestartingStartConfigDispatcher) RefreshRunningAgentStartConfig(serverAgentID string, cfg types.AgentStartConfig) bool { d.agentteamTestDispatcher.RefreshRunningAgentStartConfig(serverAgentID, cfg) d.hasSession = false return true } func (d *agentteamTestDispatcher) RestoreAgentRecoveryPoint(channelID, serverAgentID, requestID, runtime, model, sessionID, strategy, recoveryRef string) error { d.restoreCalls = append(d.restoreCalls, agentteamTestRestoreCall{ ChannelID: channelID, AgentID: serverAgentID, RequestID: requestID, Runtime: runtime, Model: model, SessionID: sessionID, Strategy: strategy, RecoveryRef: recoveryRef, }) return nil } func (d *agentteamTestDispatcher) GCRecoveryPoints(serverAgentID string, requests []types.RecoveryArtifactRequest) error { return nil } func (d *agentteamTestDispatcher) ReconcileDeferredDeliveries(serverAgentID string, messageIDs []string) ([]string, error) { return nil, nil } func (d *agentteamTestDispatcher) ReconcileRuntimeAckedDeliveries(serverAgentID string, messageIDs []string) ([]string, error) { d.ackedReconcileCalls++ d.lastAckedReconcileAgent = serverAgentID d.lastAckedMessageIDs = append([]string(nil), messageIDs...) return append([]string(nil), d.ackedSuppressions...), nil } func (d *agentteamTestDispatcher) HasDeferredDeliveries(serverAgentID string) bool { return false } func (d *agentteamTestDispatcher) AppendDeferredDelivery(serverAgentID string, seq int, raw json.RawMessage) error { return nil } func (d *agentteamTestDispatcher) SubscribeTrajectory(agentID string) <-chan types.TrajectoryLogItem { return make(chan types.TrajectoryLogItem) } func (d *agentteamTestDispatcher) UnsubscribeTrajectory(agentID string, ch <-chan types.TrajectoryLogItem) { } func (d *agentteamTestDispatcher) ServerConnections() types.ServerConnectionAccessor { return nil } func TestListWorkspaceDir_HidesRootAgentTeamManagedCarrierFile(t *testing.T) { configDir := t.TempDir() require.NoError(t, failpoint.Enable("agentteam-config-dir", `return("`+configDir+`")`)) t.Cleanup(func() { require.NoError(t, failpoint.Disable("agentteam-config-dir")) }) agentID := "agent-hide-test" workDir := filepath.Join(configDir, "agents", agentID) require.NoError(t, os.MkdirAll(workDir, 0o755)) require.NoError(t, os.WriteFile(filepath.Join(workDir, "AGENTS.md"), []byte("hidden"), 0o644)) require.NoError(t, os.WriteFile(filepath.Join(workDir, "visible.txt"), []byte("shown"), 0o644)) files, err := workspacefs.ListDir(agentID, "") require.NoError(t, err) require.Len(t, files, 1) require.Equal(t, "visible.txt", files[0].Name) } func TestListWorkspaceDir_DoesNotHideNestedRuntimeCarrierFiles(t *testing.T) { configDir := t.TempDir() require.NoError(t, failpoint.Enable("agentteam-config-dir", `return("`+configDir+`")`)) t.Cleanup(func() { require.NoError(t, failpoint.Disable("agentteam-config-dir")) }) agentID := "agent-nested-visible-test" workDir := filepath.Join(configDir, "agents", agentID) nestedDir := filepath.Join(workDir, "docs") require.NoError(t, os.MkdirAll(nestedDir, 0o755)) require.NoError(t, os.WriteFile(filepath.Join(nestedDir, "AGENTS.md"), []byte("nested user file"), 0o644)) require.NoError(t, os.WriteFile(filepath.Join(nestedDir, "README.md"), []byte("nested user file"), 0o644)) files, err := workspacefs.ListDir(agentID, "docs") require.NoError(t, err) require.Len(t, files, 2) require.Equal(t, "AGENTS.md", files[0].Name) require.Equal(t, "README.md", files[1].Name) } func TestWorkspaceTargetDir_UsesConfiguredTargetPath(t *testing.T) { configDir := t.TempDir() require.NoError(t, failpoint.Enable("agentteam-config-dir", `return("`+configDir+`")`)) t.Cleanup(func() { require.NoError(t, failpoint.Disable("agentteam-config-dir")) }) agentID := "agent-target-path-test" workDir := filepath.Join(configDir, "agents", agentID) require.NoError(t, os.MkdirAll(workDir, 0o755)) require.NoError(t, os.WriteFile(filepath.Join(workDir, "workspace.txt"), []byte("workspace"), 0o644)) targetDir := filepath.Join(configDir, "review-output") require.NoError(t, os.MkdirAll(targetDir, 0o755)) require.NoError(t, os.WriteFile(filepath.Join(targetDir, "report.md"), []byte("# Review output\n"), 0o644)) files, err := workspacefs.ListTargetDir(agentID, targetDir, "") require.NoError(t, err) require.Len(t, files, 1) require.Equal(t, "report.md", files[0].Name) content, binary, err := workspacefs.ReadTargetFile(agentID, targetDir, "report.md") require.NoError(t, err) require.False(t, binary) require.Equal(t, "# Review output\n", content) } func TestWorkspaceTargetReadRejectsSymlinkEscape(t *testing.T) { root := t.TempDir() targetDir := filepath.Join(root, "workspace") outsideDir := filepath.Join(root, "outside") require.NoError(t, os.MkdirAll(targetDir, 0o755)) require.NoError(t, os.MkdirAll(outsideDir, 0o755)) require.NoError(t, os.WriteFile(filepath.Join(outsideDir, "secret.md"), []byte("secret"), 0o644)) if err := os.Symlink(outsideDir, filepath.Join(targetDir, "link")); err != nil { t.Skipf("symlink unavailable: %v", err) } _, _, err := workspacefs.ReadTargetFile("agent-symlink-test", targetDir, "link/secret.md") require.Error(t, err) require.Contains(t, err.Error(), "symlink") } func TestApplyWorkspaceSeedFilesRejectsSymlinkParentEscape(t *testing.T) { configDir := t.TempDir() require.NoError(t, failpoint.Enable("agentteam-config-dir", `return("`+configDir+`")`)) t.Cleanup(func() { require.NoError(t, failpoint.Disable("agentteam-config-dir")) }) agentID := "agent-seed-symlink-test" workDir := filepath.Join(configDir, "agents", agentID) outsideDir := filepath.Join(configDir, "outside") require.NoError(t, os.MkdirAll(workDir, 0o755)) require.NoError(t, os.MkdirAll(outsideDir, 0o755)) if err := os.Symlink(outsideDir, filepath.Join(workDir, "link")); err != nil { t.Skipf("symlink unavailable: %v", err) } err := workspacefs.ApplySeedFiles(agentID, []types.WorkspaceSeedFile{{ Path: "link/owned.md", Content: "owned", }}) require.Error(t, err) require.NoFileExists(t, filepath.Join(outsideDir, "owned.md")) } // --------------------------------------------------------------------------- // dedupRing tests // --------------------------------------------------------------------------- func TestDedupRing_Basic(t *testing.T) { d := newDedupRing(5) if d.Len() != 0 { t.Fatalf("expected empty ring, got %d", d.Len()) } // First insert should not be a duplicate. if dup := d.Add(100); dup { t.Fatal("expected first insert to not be a duplicate") } if d.Len() != 1 { t.Fatalf("expected len=1, got %d", d.Len()) } // Second insert of the same hash should be a duplicate. if dup := d.Add(100); !dup { t.Fatal("expected second insert of same hash to be a duplicate") } if d.Len() != 1 { t.Fatalf("expected len=1 after dup insert, got %d", d.Len()) } // Insert different hashes. for i := uint64(200); i < 205; i++ { if dup := d.Add(i); dup { t.Fatalf("unexpected duplicate for hash %d", i) } } } func TestDeliveryDedupHash_PrefersMessageIDWhenPresent(t *testing.T) { first := deliveryDedupHash("agent-1", 8, "msg-thread-a") second := deliveryDedupHash("agent-1", 8, "msg-thread-b") require.NotEqual(t, first, second) ring := newDedupRing(4) require.False(t, ring.Add(first)) require.False(t, ring.Add(second)) require.Equal(t, 2, ring.Len()) } func TestDeliveryDedupHash_FallsBackToSeqWithoutMessageID(t *testing.T) { first := deliveryDedupHash("agent-1", 8, "") second := deliveryDedupHash("agent-1", 8, " ") require.Equal(t, first, second) ring := newDedupRing(4) require.False(t, ring.Add(first)) require.True(t, ring.Add(second)) require.Equal(t, 1, ring.Len()) } func TestDedupRing_Eviction(t *testing.T) { cap := 4 d := newDedupRing(cap) // Fill the ring: [1, 2, 3, 4] for i := uint64(1); i <= 4; i++ { d.Add(i) } if d.Len() != 4 { t.Fatalf("expected len=4, got %d", d.Len()) } // Insert 5: should evict the oldest (1), ring becomes [5, 2, 3, 4] d.Add(5) if d.Len() != 4 { t.Fatalf("expected len=4 after eviction, got %d", d.Len()) } if d.Contains(1) { t.Fatal("expected hash 1 to be evicted") } if !d.Contains(5) { t.Fatal("expected hash 5 to be present") } // Insert 6: should evict 2 d.Add(6) if d.Contains(2) { t.Fatal("expected hash 2 to be evicted") } if !d.Contains(6) { t.Fatal("expected hash 6 to be present") } // Verify all remaining entries: 3, 4, 5, 6 for _, h := range []uint64{3, 4, 5, 6} { if !d.Contains(h) { t.Fatalf("expected hash %d to be present", h) } } } func TestDedupRing_FullCycleEviction(t *testing.T) { cap := 3 d := newDedupRing(cap) // Insert more than 2*cap entries to ensure full wrap-around. for i := uint64(1); i <= 10; i++ { d.Add(i) } // Only the last 3 entries should remain: 8, 9, 10. if d.Len() != 3 { t.Fatalf("expected len=3, got %d", d.Len()) } for _, h := range []uint64{8, 9, 10} { if !d.Contains(h) { t.Fatalf("expected hash %d to be present after full cycle", h) } } for _, h := range []uint64{1, 2, 3, 4, 5, 6, 7} { if d.Contains(h) { t.Fatalf("expected hash %d to be evicted after full cycle", h) } } } func TestDedupRing_DuplicateAfterEviction(t *testing.T) { d := newDedupRing(3) d.Add(1) d.Add(2) d.Add(3) // Evict 1 by inserting 4 d.Add(4) // Hash 1 was evicted, so re-inserting should succeed (not be duplicate). if dup := d.Add(1); dup { t.Fatal("expected re-insert of evicted hash 1 to not be duplicate") } } // --------------------------------------------------------------------------- // Utility function tests // --------------------------------------------------------------------------- func TestBuildWSURL(t *testing.T) { tests := []struct { serverURL string apiKey string workspaceID string want string }{ {"https://example.com/", "key123", "wsp_1", "wss://example.com/api/v1/workspaces/wsp_1/daemon/connect?key=key123"}, {"https://example.com", "a+b&c%d=e?x y", "ws id", "wss://example.com/api/v1/workspaces/ws%20id/daemon/connect?key=a%2Bb%26c%25d%3De%3Fx+y"}, } for _, tt := range tests { got := buildWSURL(tt.serverURL, tt.apiKey, tt.workspaceID) require.Equal(t, tt.want, got, "buildWSURL(%q, %q, %q)", tt.serverURL, tt.apiKey, tt.workspaceID) parsed, err := url.Parse(got) require.NoError(t, err) require.Equal(t, "/api/v1/workspaces/"+tt.workspaceID+"/daemon/connect", parsed.Path) query := parsed.Query() require.Len(t, query, 1) require.Equal(t, []string{tt.apiKey}, query["key"]) } } func TestNewRequiresWorkspaceIDWithReconnectRemediation(t *testing.T) { _, err := New(profile.AgentTeamServer{ ID: "legacy-server", Enabled: true, ServerURL: "https://agentteam.example.com", APIKey: "machine-key", }, &agentteamTestDispatcher{}) require.ErrorContains(t, err, "workspace_id is required") require.ErrorContains(t, err, "reconnect or log in again") } func TestNewRequiresResolvedWorkspaceID(t *testing.T) { validMaximum := strings.Repeat("a", 128) for _, test := range []struct { name string workspaceID string want string wantError string }{ {name: "minimum", workspaceID: "a", want: "a"}, {name: "maximum", workspaceID: validMaximum, want: validMaximum}, {name: "trimmed", workspaceID: " wsp_1 ", want: "wsp_1"}, {name: "too long", workspaceID: strings.Repeat("a", 129), wantError: "workspace_id must be a resolved workspace identifier"}, {name: "dot", workspaceID: ".", wantError: "workspace_id must be a resolved workspace identifier"}, {name: "percent", workspaceID: "%2F", wantError: "workspace_id must be a resolved workspace identifier"}, {name: "unicode", workspaceID: "workspace-工作区", wantError: "workspace_id must be a resolved workspace identifier"}, {name: "control", workspaceID: "workspace\nname", wantError: "workspace_id must be a resolved workspace identifier"}, } { t.Run(test.name, func(t *testing.T) { connection, err := New(profile.AgentTeamServer{ ID: "agentteam", Enabled: true, ServerURL: "https://agentteam.example.com", APIKey: "machine-key", WorkspaceID: test.workspaceID, }, &agentteamTestDispatcher{}) if test.wantError != "" { require.ErrorContains(t, err, test.wantError) require.Nil(t, connection) return } require.NoError(t, err) protocol := connection.(*Protocol) require.Equal(t, test.want, protocol.WorkspaceID()) require.Equal(t, test.want, protocol.APIClient().RouteWorkspaceID) }) } } func TestReconnectBackoffCapsAndResets(t *testing.T) { backoff := &reconnectBackoff{min: 10 * time.Millisecond, max: 25 * time.Millisecond} require.Equal(t, 0, backoff.Attempt()) assertJitteredDelay(t, backoff.Next(), 10*time.Millisecond) require.Equal(t, 1, backoff.Attempt()) assertJitteredDelay(t, backoff.Next(), 20*time.Millisecond) assertJitteredDelay(t, backoff.Next(), 25*time.Millisecond) backoff.Reset() require.Equal(t, 0, backoff.Attempt()) assertJitteredDelay(t, backoff.Next(), 10*time.Millisecond) } // assertJitteredDelay checks a Next() result stays within the ±20% jitter band // of its deterministic envelope value. func assertJitteredDelay(t *testing.T, got time.Duration, envelope time.Duration) { t.Helper() spread := envelope / 5 require.GreaterOrEqual(t, got, envelope-spread) require.LessOrEqual(t, got, envelope+spread) } func TestClampDelayKeepsJitteredDelayInsideEnvelope(t *testing.T) { // The jitter band assertion alone cannot distinguish the clamp from an // unclamped ±20% randomization: lock the [min, max] invariant itself. minDelay := 10 * time.Millisecond maxDelay := 25 * time.Millisecond require.Equal(t, minDelay, clampDelay(minDelay-1, minDelay, maxDelay)) require.Equal(t, minDelay, clampDelay(0, minDelay, maxDelay)) require.Equal(t, 17*time.Millisecond, clampDelay(17*time.Millisecond, minDelay, maxDelay)) require.Equal(t, maxDelay, clampDelay(maxDelay+1, minDelay, maxDelay)) require.Equal(t, maxDelay, clampDelay(time.Hour, minDelay, maxDelay)) } func TestReconnectBackoffSkipsJitterForFixedDelay(t *testing.T) { backoff := &reconnectBackoff{min: time.Millisecond, max: time.Millisecond} for i := 0; i < 20; i++ { require.Equal(t, time.Millisecond, backoff.Next()) } } func TestServeRetriesAfterHandshakeFailureWithBackoff(t *testing.T) { var hits int32 server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, "/api/v1/workspaces/wsp_1/daemon/connect", request.URL.Path) atomic.AddInt32(&hits, 1) writer.WriteHeader(http.StatusUnauthorized) })) defer server.Close() channel := &Protocol{ id: "agentteam-retry", serverURL: server.URL, apiKey: "bad-key", workspaceID: "wsp_1", backoffFactory: func() *reconnectBackoff { return &reconnectBackoff{min: time.Millisecond, max: time.Millisecond} }, } ctx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel() done := make(chan struct{}) go func() { channel.serve(ctx) close(done) }() require.Eventually(t, func() bool { return atomic.LoadInt32(&hits) >= 2 }, time.Second, 10*time.Millisecond) cancel() select { case <-done: case <-time.After(time.Second): t.Fatal("serve did not stop after context cancellation") } } func TestServeStopsAfterInvalidMachineAPIKeyHandshake(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) var hits int32 server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, "/api/v1/workspaces/wsp_1/daemon/connect", request.URL.Path) atomic.AddInt32(&hits, 1) writer.Header().Set("Content-Type", "application/json") writer.WriteHeader(http.StatusUnauthorized) _, _ = writer.Write([]byte(`{"error":"invalid machine api key"}`)) })) defer server.Close() rawConfig, err := yaml.Marshal(Config{ ServerURL: server.URL, APIKey: "bad-key", WorkspaceID: "wsp_1", }) require.NoError(t, err) require.NoError(t, profile.Default().Servers().Add(&profile.AgentTeamServer{ ID: "agentteam-invalid-key", Enabled: true, Config: types.RawMessage(rawConfig), })) channel := &Protocol{ id: "agentteam-invalid-key", serverURL: server.URL, apiKey: "bad-key", workspaceID: "wsp_1", backoffFactory: func() *reconnectBackoff { return &reconnectBackoff{min: time.Millisecond, max: time.Millisecond} }, } ctx, cancel := context.WithCancel(context.Background()) defer cancel() done := make(chan struct{}) go func() { channel.serve(ctx) close(done) }() select { case <-done: case <-time.After(time.Second): t.Fatal("serve did not stop after rejected machine API key") } require.Equal(t, int32(1), atomic.LoadInt32(&hits)) chCfg, ok := profile.Default().Servers().Get("agentteam-invalid-key") require.True(t, ok) persistedCfg, err := decodeConfig(chCfg.Config) require.NoError(t, err) require.Equal(t, connectionStatusInvalidMachineAPIKey, persistedCfg.ConnectionStatus) } func TestWaitConnectedBlocksUntilWebSocketHandshakeCompletes(t *testing.T) { upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} release := make(chan struct{}) server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { <-release conn, err := upgrader.Upgrade(writer, request, nil) require.NoError(t, err) defer conn.Close() <-request.Context().Done() })) defer server.Close() channel := &Protocol{ id: "agentteam-readiness", enabled: true, serverURL: server.URL, apiKey: "machine-key", sendCh: make(chan []byte, 1), agents: make(map[string]*agentState), dedupRing: newDedupRing(10), disp: &agentteamTestDispatcher{}, } ctx, cancel := context.WithCancel(context.Background()) defer cancel() require.NoError(t, channel.Start(ctx)) t.Cleanup(func() { _ = channel.Stop(context.Background()) }) done := make(chan error, 1) go func() { done <- channel.WaitConnected(context.Background()) }() select { case err := <-done: t.Fatalf("readiness returned before handshake: %v", err) case <-time.After(20 * time.Millisecond): } close(release) select { case err := <-done: require.NoError(t, err) case <-time.After(time.Second): t.Fatal("readiness did not observe completed handshake") } } func TestWaitConnectedReturnsTerminalHandshakeFailure(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) { writer.Header().Set("Content-Type", "application/json") writer.WriteHeader(http.StatusUnauthorized) _, _ = writer.Write([]byte(`{"error":"invalid machine api key"}`)) })) defer server.Close() channel := &Protocol{ id: "agentteam-readiness-auth-failure", enabled: true, serverURL: server.URL, apiKey: "invalid-key", sendCh: make(chan []byte, 1), agents: make(map[string]*agentState), dedupRing: newDedupRing(10), disp: &agentteamTestDispatcher{}, } require.NoError(t, channel.Start(context.Background())) t.Cleanup(func() { _ = channel.Stop(context.Background()) }) waitCtx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel() err := channel.WaitConnected(waitCtx) require.ErrorIs(t, err, errMachineAPIKeyRejected) require.NotContains(t, err.Error(), "invalid-key") } func TestWaitConnectedIncludesLatestDialFailureAtDeadline(t *testing.T) { listener, err := net.Listen("tcp", "127.0.0.1:0") require.NoError(t, err) unusedAddress := listener.Addr().String() require.NoError(t, listener.Close()) channel := &Protocol{ id: "agentteam-readiness-dial-failure", enabled: true, serverURL: "http://" + unusedAddress, apiKey: "secret-key", sendCh: make(chan []byte, 1), agents: make(map[string]*agentState), dedupRing: newDedupRing(10), disp: &agentteamTestDispatcher{}, } require.NoError(t, channel.Start(context.Background())) t.Cleanup(func() { _ = channel.Stop(context.Background()) }) waitCtx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond) defer cancel() err = channel.WaitConnected(waitCtx) require.ErrorIs(t, err, context.DeadlineExceeded) require.Contains(t, err.Error(), "last WebSocket connection error") require.Contains(t, err.Error(), unusedAddress) require.NotContains(t, err.Error(), "secret-key") } func TestServeSendsReadyAfterReconnectWithFullOutboundQueue(t *testing.T) { t.Setenv("PATH", t.TempDir()) readyCh := make(chan map[string]any, 1) upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, "/api/v1/workspaces/wsp_1/daemon/connect", request.URL.Path) conn, err := upgrader.Upgrade(writer, request, nil) require.NoError(t, err) defer conn.Close() require.NoError(t, conn.SetReadDeadline(time.Now().Add(time.Second))) var message map[string]any require.NoError(t, conn.ReadJSON(&message)) readyCh <- message <-request.Context().Done() })) defer server.Close() channel := &Protocol{ id: "agentteam-ready-reconnect", serverURL: server.URL, apiKey: "machine-key", workspaceID: "wsp_1", sendCh: make(chan []byte, 1), agents: make(map[string]*agentState), dedupRing: newDedupRing(10), disp: &agentteamTestDispatcher{}, } channel.sendCh <- []byte(`{"type":"stale_telemetry"}`) ctx, cancel := context.WithCancel(context.Background()) defer cancel() done := make(chan struct{}) go func() { channel.serve(ctx) close(done) }() select { case message := <-readyCh: require.Equal(t, "ready", message["type"]) case <-time.After(time.Second): t.Fatal("server did not receive ready after reconnect") } cancel() select { case <-done: case <-time.After(time.Second): t.Fatal("serve did not stop after context cancellation") } } func TestStartStopLifecycleIsIdempotentAndClearsRunState(t *testing.T) { t.Setenv("PATH", t.TempDir()) readyCh := make(chan map[string]any, 1) releaseServer := make(chan struct{}) upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, "/api/v1/workspaces/wsp_1/daemon/connect", request.URL.Path) conn, err := upgrader.Upgrade(writer, request, nil) require.NoError(t, err) defer conn.Close() var message map[string]any require.NoError(t, conn.SetReadDeadline(time.Now().Add(time.Second))) require.NoError(t, conn.ReadJSON(&message)) readyCh <- message <-releaseServer })) t.Cleanup(func() { close(releaseServer) server.Close() }) ch := &Protocol{ id: "agentteam-lifecycle", enabled: true, serverURL: server.URL, apiKey: "machine-key", workspaceID: "wsp_1", sendCh: make(chan []byte, sendBufSize), agents: make(map[string]*agentState), dedupRing: newDedupRing(10), disp: &agentteamTestDispatcher{}, } ctx := context.Background() require.NoError(t, ch.Start(ctx)) ch.mu.Lock() firstCancel := ch.cancel firstDone := ch.done ch.mu.Unlock() require.NotNil(t, firstCancel) require.NotNil(t, firstDone) require.NoError(t, ch.Start(ctx), "second Start must not replace the running lifecycle") ch.mu.Lock() require.Equal(t, firstDone, ch.done) require.NotNil(t, ch.cancel) ch.mu.Unlock() select { case message := <-readyCh: require.Equal(t, "ready", message["type"]) case <-time.After(time.Second): t.Fatal("server did not receive ready message from started channel") } stopCtx, cancelStop := context.WithTimeout(context.Background(), time.Second) defer cancelStop() require.NoError(t, ch.Stop(stopCtx)) ch.mu.Lock() require.Nil(t, ch.cancel) require.Nil(t, ch.done) ch.mu.Unlock() require.NoError(t, ch.Stop(context.Background()), "second Stop after cleanup must be a no-op") } func TestConnectDialFailureIncludesContextAndRedactsKey(t *testing.T) { channel := &Protocol{ serverURL: "http://127.0.0.1:1", apiKey: "secret-key", workspaceID: "wsp_1", } err := channel.connect(context.Background()) require.Error(t, err) require.Contains(t, err.Error(), "websocket dial failed for") require.Contains(t, err.Error(), "key=%3Credacted%3E") require.NotContains(t, err.Error(), "secret-key") } func TestConnectIncludesHandshakeResponseDetails(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, "/api/v1/workspaces/wsp_1/daemon/connect", request.URL.Path) require.Equal(t, "bad-key", request.URL.Query().Get("key")) writer.Header().Set("Content-Type", "application/json") writer.WriteHeader(http.StatusUnauthorized) _, _ = writer.Write([]byte(`{"error":"invalid machine api key"}`)) })) defer server.Close() channel := &Protocol{ serverURL: server.URL, apiKey: "bad-key", workspaceID: "wsp_1", } err := channel.connect(context.Background()) require.Error(t, err) require.True(t, errors.Is(err, errMachineAPIKeyRejected)) require.Contains(t, err.Error(), "websocket handshake failed for") require.Contains(t, err.Error(), "401 Unauthorized") require.NotContains(t, err.Error(), "bad-key") require.Contains(t, err.Error(), `{"error":"invalid machine api key"}`) } func TestConnectSendsRouteWorkspaceHeaderWhenConfigured(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { require.Equal(t, "/api/v1/workspaces/wsp_1/daemon/connect", request.URL.Path) require.Equal(t, "wsp_1", request.Header.Get(routeWorkspaceHeader)) writer.WriteHeader(http.StatusUnauthorized) })) defer server.Close() channel := &Protocol{ serverURL: server.URL, apiKey: "machine-key", workspaceID: "wsp_1", } require.Error(t, channel.connect(context.Background())) } func TestAPIClientRefreshesAgentToken(t *testing.T) { var agentCalls int32 var refreshCalls int32 server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { switch request.URL.Path { case "/api/v1/workspaces/wsp_1/agents/server-agent/runtime/resolve-channel": require.Equal(t, http.MethodPost, request.Method) require.Empty(t, request.Header.Get(routeWorkspaceHeader)) switch atomic.AddInt32(&agentCalls, 1) { case 1: require.Equal(t, "Bearer", request.Header.Get("Authorization")) writer.WriteHeader(http.StatusUnauthorized) _, _ = writer.Write([]byte(`{"error":"missing agent token"}`)) case 2: require.Equal(t, "Bearer fresh-agent-token", request.Header.Get("Authorization")) writer.Header().Set("Content-Type", "application/json") _, _ = writer.Write([]byte(`{"scopeId":"spc_1"}`)) default: t.Fatalf("unexpected agent request %d", agentCalls) } case "/api/v1/workspaces/wsp_1/daemon/agents/server-agent/auth-token": require.Equal(t, http.MethodPost, request.Method) require.Equal(t, "Bearer machine-key", request.Header.Get("Authorization")) require.Equal(t, "wsp_1", request.Header.Get(routeWorkspaceHeader)) require.Equal(t, "connection-token", request.Header.Get(daemonTokenRefreshProofHeader)) atomic.AddInt32(&refreshCalls, 1) writer.Header().Set("Content-Type", "application/json") _, _ = writer.Write([]byte(`{"authToken":"fresh-agent-token"}`)) default: t.Fatalf("unexpected path %s", request.URL.Path) } })) defer server.Close() channel := &Protocol{ serverURL: server.URL, apiKey: "machine-key", workspaceID: "wsp_1", authToken: "machine-token", httpClient: server.Client(), agents: map[string]*agentState{"server-agent": {}}, agentTokenRefresh: map[string]*agentTokenRefreshState{}, tokenRefreshProof: "connection-token", } api := channel.APIClient("server-agent") var result struct { ScopeID string `json:"scopeId"` } err := api.Post(context.Background(), api.AgentRuntimePath("/resolve-channel"), map[string]string{"target": "#ops"}, &result) require.NoError(t, err) require.Equal(t, "spc_1", result.ScopeID) require.Equal(t, "fresh-agent-token", api.AuthToken) require.Equal(t, int32(2), atomic.LoadInt32(&agentCalls)) require.Equal(t, int32(1), atomic.LoadInt32(&refreshCalls)) channel.mu.Lock() require.Equal(t, "fresh-agent-token", channel.agents["server-agent"].Token) channel.mu.Unlock() } func TestRefreshAgentTokenReusesRefreshedCacheForStaggeredHolders(t *testing.T) { // Multiple holders of superseded snapshots that 401 at different times // must all reuse the newest issued token: each additional issuance // invalidates the token another holder just obtained. var calls int32 var issued int32 server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) { atomic.AddInt32(&calls, 1) require.Equal(t, "/api/v1/workspaces/ws-1/daemon/agents/server-agent/auth-token", request.URL.Path) writer.Header().Set("Content-Type", "application/json") _, err := fmt.Fprintf(writer, `{"authToken":"issued-%d"}`, atomic.AddInt32(&issued, 1)) require.NoError(t, err) })) defer server.Close() channel := &Protocol{ serverURL: server.URL, apiKey: "machine-key", workspaceID: "ws-1", httpClient: server.Client(), agents: map[string]*agentState{"server-agent": {Token: "snapshot-a"}}, agentTokenRefresh: map[string]*agentTokenRefreshState{}, } // First staggered holder: the cache equals its own stale token, so one // server issuance refreshes the cache. token, err := channel.refreshAgentToken(context.Background(), "server-agent", "snapshot-a") require.NoError(t, err) require.Equal(t, "issued-1", token) // Later staggered holders with different superseded snapshots reuse the // refreshed cache without another issuance. token, err = channel.refreshAgentToken(context.Background(), "server-agent", "snapshot-b") require.NoError(t, err) require.Equal(t, "issued-1", token) token, err = channel.refreshAgentToken(context.Background(), "server-agent", "snapshot-c") require.NoError(t, err) require.Equal(t, "issued-1", token) require.Equal(t, int32(1), atomic.LoadInt32(&calls)) // A caller whose stale token IS the cached one falls through to a real // re-issue (the bounded second-round case). token, err = channel.refreshAgentToken(context.Background(), "server-agent", "issued-1") require.NoError(t, err) require.Equal(t, "issued-2", token) require.Equal(t, int32(2), atomic.LoadInt32(&calls)) } func TestFormatTarget(t *testing.T) { tests := []struct { chType, chName, parentType, parentName string want string }{ {"channel", "general", "", "", "#general"}, {"dm", "alice", "", "", "dm:@alice"}, {"thread", "thread-abc123", "channel", "general", "#general:abc123"}, {"thread", "thread-abc123", "dm", "bob", "dm:@bob:abc123"}, {"thread", "abc123", "channel", "ops", "#ops:abc123"}, } for _, tt := range tests { got := mcpformat.FormatTarget(tt.chType, tt.chName, tt.parentType, tt.parentName) if got != tt.want { t.Errorf("mcpfmt.FormatTarget(%q,%q,%q,%q) = %q, want %q", tt.chType, tt.chName, tt.parentType, tt.parentName, got, tt.want) } } } func TestDeliverMessageTargetFallsBackToTaskDispatchTargetKey(t *testing.T) { require.Equal(t, "#tsk_scope_1", deliverMessageTarget(deliverMessage{ MessageType: "task_dispatch", TargetKey: "tsk_scope_1", })) require.Equal(t, "#tsk_scope_1", deliverMessageTarget(deliverMessage{ MessageType: "task_dispatch", TargetKey: "#tsk_scope_1", })) require.Equal(t, "#", deliverMessageTarget(deliverMessage{ MessageType: "text", TargetKey: "tsk_scope_1", })) require.Equal(t, "#", deliverMessageTarget(deliverMessage{ MessageType: "task_dispatch", TargetKey: "#tsk_Bad", })) require.Equal(t, "#", deliverMessageTarget(deliverMessage{ MessageType: "task_dispatch", TargetKey: "#team_abcdefghijklmn", })) } func TestHandleAgentDeliverDispatchesFreshCodexRuntimeWithoutStartToken(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_server_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "memory-reviewer", Runtime: "codex", })) dispatcher := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 1), disp: dispatcher, } channel.handleDeliver(context.Background(), []byte(`{ "type":"agent:deliver", "agentId":`+fmt.Sprintf("%q", agentID)+`, "seq":1, "message":{ "message_id":"msg_1", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"please review", "timestamp":"2026-04-19T12:00:00Z", "teamContexts":[{ "teamId":"team_core", "teamHandle":"core-team", "teamDisplayName":"Core Team", "teamDescription":"Act as core.", "teamMemberId":"wtm_agent", "role":"leader", "roleLabel":"Builder", "memberRoster":[{"teamMemberId":"wtm_agent","actorKind":"agent","actorId":"mem_agent","displayName":"AgentTeam Agent","handle":"agentteam-agent","role":"leader","deliveryKind":"wake","self":true}] }] } }`)) require.Equal(t, 1, dispatcher.dispatchCalls) require.NotNil(t, dispatcher.lastDispatchedMessage) meta, ok := dispatcher.lastDispatchedMessage.Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Len(t, meta.TeamContexts, 1) require.Equal(t, "team_core", meta.TeamContexts[0].TeamID) require.Equal(t, "Act as core.", meta.TeamContexts[0].TeamDescription) require.Equal(t, "leader", meta.TeamContexts[0].Role) require.Len(t, meta.TeamContexts[0].MemberRoster, 1) require.Equal(t, "wtm_agent", meta.TeamContexts[0].MemberRoster[0].TeamMemberID) require.Equal(t, "leader", meta.TeamContexts[0].MemberRoster[0].Role) require.True(t, meta.TeamContexts[0].MemberRoster[0].Self) select { case raw := <-channel.sendCh: var ack deliverAckMsg require.NoError(t, json.Unmarshal(raw, &ack)) require.Equal(t, "agent:deliver:ack", ack.Type) require.Equal(t, agentID, ack.AgentID) require.True(t, ack.Accepted) require.Equal(t, "msg_1", ack.MessageID) default: t.Fatal("expected an agent:deliver:ack message") } } func TestHandleAgentDeliverExtractsWorkflowContextAndStripsRawMetadata(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_server_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "workflow-helper", Runtime: "codex", })) dispatcher := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 1), disp: dispatcher, } channel.handleDeliver(context.Background(), []byte(`{ "type":"agent:deliver", "agentId":`+fmt.Sprintf("%q", agentID)+`, "seq":1, "message":{ "message_id":"msg_workflow_context", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"workflow wake", "timestamp":"2026-07-08T08:30:00Z", "metadata":{ "workflow_workspace_id":"wsp_1", "workflow_plan_id":"pln_1", "workflow_plan_version_id":"plv_1", "workflow_run_id":"wri_1", "workflow_effect_id":"wfx_1", "workflow_node_id":"post_reminder", "workflow_source_kind":"schedule", "workflow_delivery_scope":{"kind":"channel","scopeId":"chn_ops"}, "workflow_callback_expected":true, "custom":"visible" } } }`)) require.Equal(t, 1, dispatcher.dispatchCalls) require.NotNil(t, dispatcher.lastDispatchedMessage) meta, ok := dispatcher.lastDispatchedMessage.Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Equal(t, types.WorkflowContext{ WorkspaceID: "wsp_1", PlanID: "pln_1", PlanVersionID: "plv_1", RunID: "wri_1", EffectID: "wfx_1", NodeID: "post_reminder", SourceKind: "schedule", DeliveryScope: domain.MessageScopeValue{ Kind: domain.MessageScopeKindChannel, ScopeID: "chn_ops", }, CallbackExpected: true, }, meta.WorkflowContext) require.Equal(t, map[string]any{"custom": "visible"}, meta.Metadata) require.NotContains(t, meta.Metadata, types.WorkflowContextMetadataRunIDKey) } func TestHandleAgentDeliverInterruptsBeforeDispatchWhenRequested(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_interrupt_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "interruptible", Runtime: "codex", })) dispatcher := &agentteamTestDispatcher{ hasSession: true, session: types.SessionSnapshot{Status: "active"}, } channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 1), disp: dispatcher, } channel.handleDeliver(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:deliver", "agentId":%q, "seq":5, "interruptPolicy":"interrupt_current_run", "message":{ "message_id":"msg_interrupt_submit", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"please switch now", "timestamp":"2026-04-19T12:00:00Z" } }`, agentID))) require.Equal(t, 0, dispatcher.interruptCalls) require.Equal(t, []string{"dispatch"}, dispatcher.events) require.NotNil(t, dispatcher.lastDispatchedMessage) agentteamMeta, ok := dispatcher.lastDispatchedMessage.Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Equal(t, "interrupt_current_run", agentteamMeta.InterruptPolicy) select { case raw := <-channel.sendCh: var ack deliverAckMsg require.NoError(t, json.Unmarshal(raw, &ack)) require.True(t, ack.Accepted) require.Equal(t, "msg_interrupt_submit", ack.MessageID) require.Equal(t, "interrupt_current_run", ack.InterruptPolicy) default: t.Fatal("expected an agent:deliver:ack message") } } func TestHandleAgentDeliverUrgentRejectsAckWhenDispatchFails(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_interrupt_fail_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "interrupt-fail", Runtime: "codex", })) dispatcher := &agentteamTestDispatcher{ hasSession: true, session: types.SessionSnapshot{Status: "active"}, dispatchErr: errors.New("interrupt delivery failed"), } channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 1), disp: dispatcher, } channel.handleDeliver(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:deliver", "agentId":%q, "seq":6, "interruptPolicy":"interrupt_current_run", "message":{ "message_id":"msg_interrupt_fail", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"please switch now", "timestamp":"2026-04-19T12:00:00Z" } }`, agentID))) require.Equal(t, 1, dispatcher.dispatchCalls) require.False(t, channel.dedupRing.Contains(deliveryDedupHashWithInterruptPolicy(agentID, 6, "msg_interrupt_fail", agentInterruptPolicyCurrentRun))) select { case raw := <-channel.sendCh: var ack deliverAckMsg require.NoError(t, json.Unmarshal(raw, &ack)) require.False(t, ack.Accepted) require.Equal(t, "msg_interrupt_fail", ack.MessageID) require.Equal(t, "interrupt_current_run", ack.InterruptPolicy) require.Contains(t, ack.Detail, "interrupt delivery failed") default: t.Fatal("expected a rejected agent:deliver:ack message") } } func TestHandleAgentDeliverOrdinaryRejectsAckWhenDispatchFails(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_deliver_fail_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "deliver-fail", Runtime: "codex", })) dispatcher := &agentteamTestDispatcher{ dispatchErr: errors.New("local handoff failed"), } channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 1), disp: dispatcher, } channel.handleDeliver(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:deliver", "agentId":%q, "seq":8, "message":{ "message_id":"msg_deliver_fail", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"please process after local handoff", "timestamp":"2026-04-19T12:00:00Z" } }`, agentID))) require.Equal(t, 1, dispatcher.dispatchCalls) require.False(t, channel.dedupRing.Contains(deliveryDedupHash(agentID, 8, "msg_deliver_fail"))) select { case raw := <-channel.sendCh: var ack deliverAckMsg require.NoError(t, json.Unmarshal(raw, &ack)) require.False(t, ack.Accepted) require.Equal(t, "msg_deliver_fail", ack.MessageID) require.Contains(t, ack.Detail, "local handoff failed") default: t.Fatal("expected a rejected agent:deliver:ack message") } } func TestHandleAgentDeliverUsesBoundedDispatchContextForAck(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_deliver_deadline_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "deliver-deadline", Runtime: "codex", })) dispatcher := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 1), disp: dispatcher, } channel.handleDeliver(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:deliver", "agentId":%q, "seq":9, "message":{ "message_id":"msg_deliver_deadline", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"please process within ack window", "timestamp":"2026-04-19T12:00:00Z" } }`, agentID))) require.Equal(t, 1, dispatcher.dispatchCalls) require.True(t, dispatcher.deliveryAckHasDeadline, "DispatchForDeliveryAck context must be bounded by the server ACK wait") require.Positive(t, time.Until(dispatcher.deliveryAckDeadline)) require.LessOrEqual(t, time.Until(dispatcher.deliveryAckDeadline), 2*time.Second) } func TestHandleAgentDeliverInterruptReplayBypassesDedupAndFreshness(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_interrupt_replay_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "interrupt-replay", Runtime: "codex", })) dispatcher := &freshnessAgentTeamTestDispatcher{} channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 2), disp: dispatcher, } payload := fmt.Sprintf(`{ "type":"agent:deliver", "agentId":%q, "delivery_seq":222, "message":{ "message_id":"msg_interrupt_replay", "target_key":"scp_thread", "target_seq":43, "channel_type":"thread", "channel_name":"d5c85c79", "parent_channel_type":"channel", "parent_channel_name":"agentteam-in-agentteam", "sender_type":"human", "sender_name":"alice", "content":"normal delivery", "timestamp":"2026-04-19T12:00:00Z" } }`, agentID) channel.handleDeliver(context.Background(), []byte(payload)) require.Equal(t, 1, dispatcher.dispatchCalls) require.True(t, channel.dedupRing.Contains(deliveryDedupHash(agentID, 222, "msg_interrupt_replay"))) dispatcher.snapshot = types.MessageFreshnessFrontier{ TargetKey: "scp_thread", ObservedSeq: 43, } interruptPayload := fmt.Sprintf(`{ "type":"agent:deliver", "agentId":%q, "delivery_seq":222, "interruptPolicy":"interrupt_current_run", "message":{ "message_id":"msg_interrupt_replay", "target_key":"scp_thread", "target_seq":43, "channel_type":"thread", "channel_name":"d5c85c79", "parent_channel_type":"channel", "parent_channel_name":"agentteam-in-agentteam", "sender_type":"human", "sender_name":"alice", "content":"urgent replay", "timestamp":"2026-04-19T12:00:01Z" } }`, agentID) channel.handleDeliver(context.Background(), []byte(interruptPayload)) require.Equal(t, 2, dispatcher.dispatchCalls) require.Len(t, dispatcher.dispatchedMessages, 2) meta, ok := dispatcher.dispatchedMessages[1].Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Equal(t, "interrupt_current_run", meta.InterruptPolicy) require.True(t, channel.dedupRing.Contains(deliveryDedupHashWithInterruptPolicy(agentID, 222, "msg_interrupt_replay", agentInterruptPolicyCurrentRun))) for i := 0; i < 2; i++ { select { case raw := <-channel.sendCh: var ack deliverAckMsg require.NoError(t, json.Unmarshal(raw, &ack)) require.True(t, ack.Accepted) require.Equal(t, "msg_interrupt_replay", ack.MessageID) require.NotEqual(t, "duplicate", ack.Detail) require.NotEqual(t, "stale delivery suppressed", ack.Detail) default: t.Fatalf("expected ack %d", i+1) } } } func TestHandleAgentRestoreRecoveryPointPassesChannelContext(t *testing.T) { dispatcher := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam-main", sendCh: make(chan []byte, 1), disp: dispatcher, } channel.handleAgentRestoreRecoveryPoint(context.Background(), []byte(`{ "type":"agent:restore_recovery_point", "agentId":"agt_server", "requestId":"req_restore", "runtime":"builtin", "model":"gpt-test", "sessionId":"sess_restore", "strategy":"builtin_message_count", "recoveryRef":"ref_restore" }`)) require.Len(t, dispatcher.restoreCalls, 1) require.Equal(t, agentteamTestRestoreCall{ ChannelID: "agentteam-main", AgentID: "agt_server", RequestID: "req_restore", Runtime: "builtin", Model: "gpt-test", SessionID: "sess_restore", Strategy: "builtin_message_count", RecoveryRef: "ref_restore", }, dispatcher.restoreCalls[0]) var response agentRestoreRecoveryPointResultMsg require.NoError(t, json.Unmarshal(<-channel.sendCh, &response)) require.True(t, response.Success) require.Equal(t, "req_restore", response.RequestID) } func TestHandleAgentDeliverDedupsAfterAcceptedAckEnqueueFails(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_ack_enqueue_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "ack-retry", Runtime: "codex", })) dispatcher := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 1), disp: dispatcher, } raw := []byte(fmt.Sprintf(`{ "type":"agent:deliver", "agentId":%q, "seq":1, "message":{ "message_id":"msg_ack_retry", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"please retry after ack failure", "timestamp":"2026-04-19T12:00:00Z" } }`, agentID)) channel.sendCh <- []byte(`{"type":"blocked"}`) channel.handleDeliver(context.Background(), raw) require.Equal(t, 1, dispatcher.dispatchCalls) hash := deliveryDedupHash(agentID, 1, "msg_ack_retry") require.True(t, channel.dedupRing.Contains(hash)) require.Equal(t, []byte(`{"type":"blocked"}`), <-channel.sendCh) channel.handleDeliver(context.Background(), raw) require.Equal(t, 1, dispatcher.dispatchCalls) require.True(t, channel.dedupRing.Contains(hash)) select { case rawAck := <-channel.sendCh: var ack deliverAckMsg require.NoError(t, json.Unmarshal(rawAck, &ack)) require.Equal(t, "agent:deliver:ack", ack.Type) require.Equal(t, agentID, ack.AgentID) require.Equal(t, "msg_ack_retry", ack.MessageID) require.True(t, ack.Accepted) require.Equal(t, "duplicate", ack.Detail) default: t.Fatal("expected an agent:deliver:ack message") } } func TestHandleAgentDeliverKeepsDeliverySeqSeparateFromTargetSeq(t *testing.T) { configDir := t.TempDir() require.NoError(t, failpoint.Enable("agentteam-config-dir", `return("`+configDir+`")`)) t.Cleanup(func() { require.NoError(t, failpoint.Disable("agentteam-config-dir")) }) agentID := fmt.Sprintf("agt_freshness_delivery_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "freshness-reviewer", Runtime: "codex", })) dispatcher := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 1), disp: dispatcher, } channel.handleDeliver(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:deliver", "agentId":%q, "delivery_seq":111, "message":{ "message_id":"msg_fresh", "target_key":"scp_ops", "target_seq":42, "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"please review", "timestamp":"2026-04-19T12:00:00Z" } }`, agentID))) require.Equal(t, 1, dispatcher.dispatchCalls) require.NotNil(t, dispatcher.lastDispatchedMessage) meta, ok := dispatcher.lastDispatchedMessage.Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Equal(t, "scp_ops", meta.TargetKey) require.Equal(t, int64(42), meta.TargetSeq) select { case raw := <-channel.sendCh: var ack deliverAckMsg require.NoError(t, json.Unmarshal(raw, &ack)) require.Equal(t, "agent:deliver:ack", ack.Type) require.Equal(t, agentID, ack.AgentID) require.True(t, ack.Accepted) require.Equal(t, "msg_fresh", ack.MessageID) require.Equal(t, 111, ack.Seq) default: t.Fatal("expected an agent:deliver:ack message") } } func TestHandleAgentDeliverPreservesNativeServiceEventSource(t *testing.T) { configDir := t.TempDir() require.NoError(t, failpoint.Enable("agentteam-config-dir", `return("`+configDir+`")`)) t.Cleanup(func() { require.NoError(t, failpoint.Disable("agentteam-config-dir")) }) agentID := fmt.Sprintf("agt_source_delivery_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "source-reviewer", Runtime: "codex", })) dispatcher := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 1), disp: dispatcher, } channel.handleDeliver(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:deliver", "agentId":%q, "delivery_seq":118, "message":{ "message_id":"msg_triggered", "target_key":"scp_ops", "target_seq":45, "channel_type":"channel", "channel_name":"ops", "sender_type":"service", "sender_name":"gitea", "content":"PR synchronized", "timestamp":"2026-05-28T12:00:00Z", "source":{ "kind":"provider_event", "refType":"service_event_delivery", "refId":"sed_1", "runId":"sed_1", "eventId":"evt_1", "eventType":"gitea.pull_request.synchronized", "label":"Gitea PR watch" } } }`, agentID))) require.Equal(t, 1, dispatcher.dispatchCalls) require.NotNil(t, dispatcher.lastDispatchedMessage) meta, ok := dispatcher.lastDispatchedMessage.Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Equal(t, "provider_event", meta.Source.Kind) require.Equal(t, "service_event_delivery", meta.Source.RefType) require.Equal(t, "sed_1", meta.Source.RunID) } func TestNormalizedDeliverTargetSeqRejectsTaskActivitySequence(t *testing.T) { t.Parallel() require.Zero(t, normalizedDeliverTargetSeq(deliverMessage{ MessageID: "tev_checkpoint", TargetKey: "tsk_scope", TargetSeq: 99, MessageType: "task.checkpoint_added", Source: types.ContributionSource{ RefType: "task_activity", RefID: "tev_checkpoint", }, }, 123)) } func TestHandleAgentDeliverDoesNotMixDeliverySeqIntoTargetFreshness(t *testing.T) { configDir := t.TempDir() require.NoError(t, failpoint.Enable("agentteam-config-dir", `return("`+configDir+`")`)) t.Cleanup(func() { require.NoError(t, failpoint.Disable("agentteam-config-dir")) }) agentID := fmt.Sprintf("agt_freshness_backfill_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "freshness-domain", Runtime: "codex", })) dispatcher := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 1), disp: dispatcher, } channel.handleDeliver(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:deliver", "agentId":%q, "delivery_seq":70, "message":{ "message_id":"msg_missing_target_seq", "target_key":"scp_ops", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"please review", "timestamp":"2026-04-19T12:00:00Z" } }`, agentID))) require.Equal(t, 1, dispatcher.dispatchCalls) require.NotNil(t, dispatcher.lastDispatchedMessage) meta, ok := dispatcher.lastDispatchedMessage.Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Equal(t, "scp_ops", meta.TargetKey) require.Zero(t, meta.TargetSeq) } func TestHandleAgentDeliverClearsStartedReplyThreadRootTargetSeq(t *testing.T) { configDir := t.TempDir() require.NoError(t, failpoint.Enable("agentteam-config-dir", `return("`+configDir+`")`)) t.Cleanup(func() { require.NoError(t, failpoint.Disable("agentteam-config-dir")) }) agentID := fmt.Sprintf("agt_thread_root_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "thread-root-reviewer", Runtime: "codex", })) dispatcher := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 1), disp: dispatcher, } channel.handleDeliver(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:deliver", "agentId":%q, "delivery_seq":119, "message":{ "message_id":"7ab5e11d114e93823e92e32dc4de1685", "target_key":"chn_thread", "target_seq":5, "channel_type":"thread", "channel_name":"thread-7ab5e11d", "parent_channel_type":"channel", "parent_channel_name":"agent-session-andc5vq", "sender_type":"human", "sender_name":"alice", "content":"start this thread", "timestamp":"2026-06-22T13:19:58+08:00" } }`, agentID))) require.Equal(t, 1, dispatcher.dispatchCalls) require.NotNil(t, dispatcher.lastDispatchedMessage) meta, ok := dispatcher.lastDispatchedMessage.Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Equal(t, "chn_thread", meta.TargetKey) require.Zero(t, meta.TargetSeq) require.Equal(t, "#agent-session-andc5vq:7ab5e11d", meta.Target) } func TestMessageFromWakePayloadClearsStartedReplyThreadRootTargetSeq(t *testing.T) { channel := &Protocol{id: "agentteam"} raw := json.RawMessage(`{ "message_id":"7ab5e11d114e93823e92e32dc4de1685", "target_key":"chn_thread", "target_seq":5, "channel_type":"thread", "channel_name":"thread-7ab5e11d", "parent_channel_type":"channel", "parent_channel_name":"agent-session-andc5vq", "sender_type":"human", "sender_name":"alice", "content":"start this thread", "timestamp":"2026-06-22T13:19:58+08:00" }`) msg, messageID, _, err := channel.messageFromWakePayload("agt_1", raw, types.AgentStartConfig{Runtime: "codex"}, nil) require.NoError(t, err) require.Equal(t, "7ab5e11d114e93823e92e32dc4de1685", messageID) require.NotNil(t, msg) meta, ok := msg.Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Equal(t, "chn_thread", meta.TargetKey) require.Zero(t, meta.TargetSeq) require.Equal(t, "#agent-session-andc5vq:7ab5e11d", meta.Target) } func TestMessageFromWakePayloadExtractsWorkflowContextAndStripsRawMetadata(t *testing.T) { channel := &Protocol{id: "agentteam"} raw := json.RawMessage(`{ "message_id":"msg_workflow_start", "target_key":"chn_ops", "target_seq":8, "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"workflow start", "timestamp":"2026-07-08T08:35:00Z", "metadata":{ "workflow_workspace_id":"wsp_1", "workflow_plan_id":"pln_1", "workflow_plan_version_id":"plv_1", "workflow_run_id":"wri_1", "workflow_effect_id":"wfx_1", "workflow_node_id":"notify_agent", "workflow_source_kind":"manual", "workflow_callback_expected":true, "custom":"visible" } }`) msg, _, _, err := channel.messageFromWakePayload("agt_1", raw, types.AgentStartConfig{Runtime: "codex"}, nil) require.NoError(t, err) require.NotNil(t, msg) meta, ok := msg.Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Equal(t, types.WorkflowContext{ WorkspaceID: "wsp_1", PlanID: "pln_1", PlanVersionID: "plv_1", RunID: "wri_1", EffectID: "wfx_1", NodeID: "notify_agent", SourceKind: "manual", CallbackExpected: true, }, meta.WorkflowContext) require.Equal(t, map[string]any{"custom": "visible"}, meta.Metadata) require.NotContains(t, meta.Metadata, types.WorkflowContextMetadataEffectIDKey) } func TestHandleAgentDeliverPreservesReplyThreadMessageTargetSeq(t *testing.T) { configDir := t.TempDir() require.NoError(t, failpoint.Enable("agentteam-config-dir", `return("`+configDir+`")`)) t.Cleanup(func() { require.NoError(t, failpoint.Disable("agentteam-config-dir")) }) agentID := fmt.Sprintf("agt_thread_reply_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "thread-reply-reviewer", Runtime: "codex", })) dispatcher := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 1), disp: dispatcher, } channel.handleDeliver(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:deliver", "agentId":%q, "delivery_seq":120, "message":{ "message_id":"745bd74915d18db24e9118e1abd7a31f", "target_key":"chn_thread", "target_seq":2, "channel_type":"thread", "channel_name":"thread-7ab5e11d", "parent_channel_type":"channel", "parent_channel_name":"agent-session-andc5vq", "sender_type":"human", "sender_name":"alice", "content":"DXF 是什么？", "timestamp":"2026-06-22T13:20:58+08:00" } }`, agentID))) require.Equal(t, 1, dispatcher.dispatchCalls) require.NotNil(t, dispatcher.lastDispatchedMessage) meta, ok := dispatcher.lastDispatchedMessage.Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Equal(t, "chn_thread", meta.TargetKey) require.Equal(t, int64(2), meta.TargetSeq) require.Equal(t, "#agent-session-andc5vq:7ab5e11d", meta.Target) } func TestHandleAgentDeliverSuppressesStaleTargetSeqBeforeDispatch(t *testing.T) { configDir := t.TempDir() require.NoError(t, failpoint.Enable("agentteam-config-dir", `return("`+configDir+`")`)) t.Cleanup(func() { require.NoError(t, failpoint.Disable("agentteam-config-dir")) }) agentID := fmt.Sprintf("agt_stale_delivery_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "freshness-reviewer", Runtime: "codex", })) dispatcher := &freshnessAgentTeamTestDispatcher{ snapshot: types.MessageFreshnessFrontier{ TargetKey: "scp_thread", ObservedSeq: 43, }, } channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 1), disp: dispatcher, } channel.handleDeliver(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:deliver", "agentId":%q, "delivery_seq":222, "message":{ "message_id":"msg_stale", "target_key":"scp_thread", "target_seq":36, "channel_type":"thread", "channel_name":"d5c85c79", "parent_channel_type":"channel", "parent_channel_name":"agentteam-in-agentteam", "sender_type":"human", "sender_name":"alice", "content":"stale replay", "timestamp":"2026-04-19T12:00:00Z" } }`, agentID))) require.Zero(t, dispatcher.dispatchCalls) require.Equal(t, agentID, dispatcher.snapshotAgentID) require.Equal(t, "scp_thread", dispatcher.snapshotTargetKey) require.True(t, channel.dedupRing.Contains(deliveryDedupHash(agentID, 222, "msg_stale"))) select { case raw := <-channel.sendCh: var ack deliverAckMsg require.NoError(t, json.Unmarshal(raw, &ack)) require.Equal(t, "agent:deliver:ack", ack.Type) require.Equal(t, agentID, ack.AgentID) require.True(t, ack.Accepted) require.Equal(t, "msg_stale", ack.MessageID) require.Equal(t, 222, ack.Seq) require.Equal(t, "stale delivery suppressed", ack.Detail) default: t.Fatal("expected an agent:deliver:ack message") } } func TestHandleAgentDeliverRejectedProcessHandoffDoesNotEnterDedup(t *testing.T) { agentID := fmt.Sprintf("agt_rejected_deliver_%d", time.Now().UnixNano()) dispatcher := &agentteamTestDispatcher{ dispatchErr: errors.New("agent delivery requires an existing process accepting deliveries"), } channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{ agentID: { Config: types.AgentStartConfig{ Runtime: "unknown-runtime", }, Sleeping: true, }, }, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 2), disp: dispatcher, } raw := []byte(fmt.Sprintf(`{ "type":"agent:deliver", "agentId":%q, "seq":7, "message":{ "message_id":"msg_rejected_retry", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"please fall back to server start", "timestamp":"2026-04-19T12:00:00Z" } }`, agentID)) channel.handleDeliver(context.Background(), raw) channel.handleDeliver(context.Background(), raw) require.Equal(t, 2, dispatcher.dispatchCalls) require.Zero(t, dispatcher.startCalls) channel.mu.Lock() require.True(t, channel.agents[agentID].Sleeping, "rejected handoff leaves wake state for server agent:start") require.False(t, channel.agentListSyncInFlight) channel.mu.Unlock() hash := deliveryDedupHash(agentID, 7, "msg_rejected_retry") require.False(t, channel.dedupRing.Contains(hash)) for i := 0; i < 2; i++ { select { case rawAck := <-channel.sendCh: var ack deliverAckMsg require.NoError(t, json.Unmarshal(rawAck, &ack)) require.Equal(t, agentID, ack.AgentID) require.Equal(t, "msg_rejected_retry", ack.MessageID) require.False(t, ack.Accepted) require.Contains(t, ack.Detail, "existing process accepting deliveries") default: t.Fatalf("expected rejected ack %d", i+1) } } } func TestHandleAgentDeliverBypassesAgentListSyncForCurrentProcessHandoff(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_missing_profile_live_%d", time.Now().UnixNano()) _, found := profile.Default().Agents().Get(agentID) require.False(t, found) dispatcher := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 4), disp: dispatcher, } channel.handleDeliver(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:deliver", "agentId":%q, "seq":1, "message":{ "message_id":"msg_missing_profile_live", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"please review", "timestamp":"2026-04-19T12:00:00Z" } }`, agentID))) require.Equal(t, 1, dispatcher.dispatchCalls) require.True(t, channel.dedupRing.Contains(deliveryDedupHash(agentID, 1, "msg_missing_profile_live"))) channel.mu.Lock() require.False(t, channel.agentListSyncInFlight) require.Empty(t, channel.agentListSyncWaiters) channel.mu.Unlock() require.Len(t, channel.sendCh, 1, "delivery must not enqueue agent-list sync or wake-failure events") raw := <-channel.sendCh var ack deliverAckMsg require.NoError(t, json.Unmarshal(raw, &ack)) require.True(t, ack.Accepted) require.Equal(t, "msg_missing_profile_live", ack.MessageID) } func TestHandleAgentDeliverUsesProcessAuthorityWithoutMutatingSleepingOrRuntimeState(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_stale_runtime_live_%d", time.Now().UnixNano()) _, found := profile.Default().Agents().Get(agentID) require.False(t, found) dispatcher := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{ agentID: { Config: types.AgentStartConfig{Runtime: "unknown-stale-runtime", Model: "stale-model"}, Sleeping: true, }, }, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 4), disp: dispatcher, } channel.handleDeliver(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:deliver", "agentId":%q, "seq":2, "message":{ "message_id":"msg_stale_runtime_live", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"deliver to the published process", "timestamp":"2026-04-19T12:00:00Z" } }`, agentID))) require.Equal(t, 1, dispatcher.dispatchCalls) require.Zero(t, dispatcher.startCalls) channel.mu.Lock() require.True(t, channel.agents[agentID].Sleeping, "agent:deliver must not perform agent:start wake-state mutation") require.Equal(t, "unknown-stale-runtime", channel.agents[agentID].Config.Runtime) require.Equal(t, "stale-model", channel.agents[agentID].Config.Model) require.False(t, channel.agentListSyncInFlight) channel.mu.Unlock() _, found = profile.Default().Agents().Get(agentID) require.False(t, found, "agent:deliver must not synchronize or mutate the cold-start catalog") select { case raw := <-channel.sendCh: var ack deliverAckMsg require.NoError(t, json.Unmarshal(raw, &ack)) require.Equal(t, "agent:deliver:ack", ack.Type) require.Equal(t, agentID, ack.AgentID) require.Equal(t, "msg_stale_runtime_live", ack.MessageID) require.Equal(t, 2, ack.Seq) require.True(t, ack.Accepted) default: t.Fatal("expected accepted agent:deliver:ack from the current Process handoff") } require.Empty(t, channel.sendCh, "agent:deliver must not emit a synthetic active status") } func TestHandleAgentListResponseUpdatesExistingAgentPermissionMode(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) installAgentTeamTestRuntime(t, "codex") agentID := fmt.Sprintf("agt_permission_sync_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Permission Reviewer", Runtime: "codex", PermissionMode: types.AgentPermissionModeFullAccess, Source: agent.NewAgentTeamSource("agentteam-main"), })) channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 1), } channel.handleAgentListResponse([]byte(fmt.Sprintf(`{ "type":"machine:agent:list:response", "agents":[{ "id":%q, "name":"permission-reviewer", "displayName":"Permission Reviewer", "runtime":"codex", "permissionMode":"agent_workspace_write", "runtimePolicy":{ "schemaVersion":1, "generation":3, "fingerprint":"rp_permission_sync", "runtime":"codex", "runtimeType":"codex", "permissionMode":"agent_workspace_write" }, "status":"idle", "activity":"" }] }`, agentID))) ag, found := profile.Default().Agents().Get(agentID) require.True(t, found) require.Equal(t, types.AgentPermissionModeWorkspaceWrite, ag.PermissionMode) require.Equal(t, int64(3), ag.RuntimePolicy.Generation) require.Equal(t, "rp_permission_sync", ag.RuntimePolicy.Fingerprint) require.Equal(t, "codex", ag.RuntimePolicy.RuntimeType) require.Equal(t, types.AgentPermissionModeWorkspaceWrite, ag.RuntimePolicy.PermissionMode) } func TestHandleAgentListResponseObservesThenRefreshesExtensionCatalog(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) installAgentTeamTestRuntime(t, "codex") agentID := fmt.Sprintf("agt_extension_catalog_%d", time.Now().UnixNano()) dispatcher := &extensionCatalogAgentTeamTestDispatcher{ agentteamTestDispatcher: agentteamTestDispatcher{}, refreshed: make(chan extensionCatalogRefreshCall, 1), } channel := &Protocol{ id: "agentteam-main", disp: dispatcher, agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 1), } channel.handleAgentListResponse([]byte(fmt.Sprintf(`{ "type":"machine:agent:list:response", "agents":[{ "id":%q, "name":"extension-agent", "displayName":"Extension Agent", "runtime":"codex", "status":"idle", "activity":"", "extensionCatalog":{"version":7} }] }`, agentID))) require.Len(t, dispatcher.observed, 1) require.Equal(t, "agentteam-main", dispatcher.observedServerID) require.Equal(t, agentID, dispatcher.observed[0].ID) require.NotNil(t, dispatcher.observed[0].ExtensionCatalog) require.Equal(t, int64(7), dispatcher.observed[0].ExtensionCatalog.Version) select { case call := <-dispatcher.refreshed: require.Equal(t, "agentteam-main", call.serverID) require.Equal(t, agentID, call.agentID) require.NotNil(t, call.api) require.Equal(t, agentID, call.api.AgentID) case <-time.After(time.Second): t.Fatal("Extension Catalog refresh task did not run") } } type extensionCatalogRefreshCall struct { serverID string agentID string api *apiclient.Client } type extensionCatalogAgentTeamTestDispatcher struct { agentteamTestDispatcher observedServerID string observed []types.MachineAgentInfo refreshed chan extensionCatalogRefreshCall } func (dispatcher *extensionCatalogAgentTeamTestDispatcher) ObserveAgentExtensionCatalogHints( serverID string, agents []types.MachineAgentInfo, ) { dispatcher.observedServerID = serverID dispatcher.observed = append([]types.MachineAgentInfo(nil), agents...) } func (dispatcher *extensionCatalogAgentTeamTestDispatcher) RefreshAgentExtensionCatalog( _ context.Context, serverID string, agentID string, api *apiclient.Client, ) error { dispatcher.refreshed <- extensionCatalogRefreshCall{ serverID: serverID, agentID: agentID, api: api, } return nil } func TestHandleAgentListResponseRefreshesRunningSessionConfig(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) installAgentTeamTestRuntime(t, "codex") agentID := fmt.Sprintf("agt_name_sync_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Old Agent", Runtime: "codex", Source: agent.NewAgentTeamSource("agentteam-main"), })) require.NoError(t, profile.Default().Servers().Add(&profile.AgentTeamServer{ ID: "agentteam-main", Enabled: true, Name: "agentteam", })) disp := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam-main", disp: disp, agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 1), } channel.handleAgentListResponse([]byte(fmt.Sprintf(`{ "type":"machine:agent:list:response", "agents":[{ "id":%q, "name":"new-agent", "displayName":"New Agent", "runtime":"codex", "permissionMode":"agent_workspace_write", "status":"idle", "activity":"" }] }`, agentID))) require.Len(t, disp.refreshedAgentConfigs, 1) require.Equal(t, agentID, disp.refreshedAgentConfigs[0].ID) require.Equal(t, "New Agent", disp.refreshedAgentConfigs[0].Name) require.Equal(t, []identityUpdateCall{{ AgentID: agentID, Name: "new-agent", DisplayName: "New Agent", }}, disp.identityUpdates) } func TestHandleAgentListResponseReconcilesUnchangedIdentity(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) installAgentTeamTestRuntime(t, "codex") agentID := fmt.Sprintf("agt_unchanged_name_sync_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Same Agent", Runtime: "codex", Source: agent.NewAgentTeamSource("agentteam-main"), })) require.NoError(t, profile.Default().Servers().Add(&profile.AgentTeamServer{ ID: "agentteam-main", Enabled: true, Name: "agentteam", })) disp := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam-main", disp: disp, agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 1), } channel.handleAgentListResponse([]byte(fmt.Sprintf(`{ "type":"machine:agent:list:response", "agents":[{ "id":%q, "name":"same-agent", "displayName":"Same Agent", "runtime":"codex", "status":"idle", "activity":"" }] }`, agentID))) require.Equal(t, []identityUpdateCall{{ AgentID: agentID, Name: "same-agent", DisplayName: "Same Agent", }}, disp.identityUpdates) require.Len(t, disp.refreshedAgentConfigs, 1) } func TestHandleAgentStartQueuesReplacementWhileStarting(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_start_replace_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Permission Reviewer", Runtime: "codex", PermissionMode: types.AgentPermissionModeFullAccess, Source: agent.NewAgentTeamSource("agentteam-main"), })) dispatcher := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 8), disp: dispatcher, } dispatcher.onStart = func(call int) { if call != 1 { return } channel.handleAgentStart(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{ "name":"permission-reviewer", "runtime":"codex", "permissionMode":"agent_workspace_write", "authToken":"agent-token-replacement" }, "wakeMessage":{ "message_id":"msg_replacement_wake", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"queued wake", "timestamp":"2026-05-27T07:39:00Z" }, "unreadSummary":{ "ops":2 } }`, agentID))) } channel.handleAgentStart(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{ "name":"permission-reviewer", "runtime":"codex", "permissionMode":"full_access", "authToken":"agent-token-original" } }`, agentID))) require.Equal(t, 2, dispatcher.startCalls) require.Len(t, dispatcher.startConfigs, 2) require.Equal(t, types.AgentPermissionModeFullAccess, dispatcher.startConfigs[0].PermissionMode) require.Equal(t, types.AgentPermissionModeWorkspaceWrite, dispatcher.startConfigs[1].PermissionMode) require.Equal(t, 0, dispatcher.dispatchCalls) require.Len(t, dispatcher.startWakes, 2) require.NotNil(t, dispatcher.startWakes[1]) require.Equal(t, "queued wake", dispatcher.startWakes[1].Input.Prompt) meta, ok := dispatcher.startWakes[1].Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Equal(t, "msg_replacement_wake", meta.ID) require.Equal(t, map[string]int{"ops": 2}, meta.UnreadSummary) } func TestHandleAgentStartWithWakeForRunningSessionKeepsDeliverDedup(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_start_running_dedup_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Running Reviewer", Runtime: "codex", PermissionMode: types.AgentPermissionModeFullAccess, Source: agent.NewAgentTeamSource("agentteam-main"), })) dispatcher := &agentteamTestDispatcher{ hasSession: true, session: types.SessionSnapshot{Status: "running"}, } channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 8), disp: dispatcher, } hash := deliveryDedupHash(agentID, 0, "msg_start_fallback") require.False(t, channel.dedupRing.AddWithKey(hash, agentID)) channel.handleAgentStart(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{ "name":"running-reviewer", "runtime":"codex", "permissionMode":"full_access", "authToken":"agent-token" }, "wakeMessage":{ "message_id":"msg_start_fallback", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"wake already accepted by deliver", "timestamp":"2026-05-27T07:39:00Z" } }`, agentID))) require.Equal(t, 0, dispatcher.startCalls) require.Equal(t, 0, dispatcher.dispatchCalls) require.True(t, channel.dedupRing.Contains(hash)) var ack deliverAckMsg foundAck := false for i := 0; i < 2; i++ { select { case raw := <-channel.sendCh: if json.Unmarshal(raw, &ack) == nil && ack.Type == "agent:deliver:ack" { foundAck = true } default: } } require.True(t, foundAck, "expected duplicate agent:deliver:ack") require.True(t, ack.Accepted) require.Equal(t, "duplicate", ack.Detail) require.Equal(t, "msg_start_fallback", ack.MessageID) } func TestHandleAgentStartWithWakeForRunningSessionRefreshesStartConfig(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_start_running_refresh_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Running Reviewer", Runtime: "codex", PermissionMode: types.AgentPermissionModeFullAccess, Source: agent.NewAgentTeamSource("agentteam-main"), })) dispatcher := &agentteamTestDispatcher{ hasSession: true, session: types.SessionSnapshot{Status: "running"}, } channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 8), disp: dispatcher, } channel.handleAgentStart(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{ "name":"running-reviewer", "runtime":"codex", "permissionMode":"full_access", "authToken":"agent-token" }, "wakeMessage":{ "message_id":"msg_start_policy_refresh", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"wake with policy refresh", "timestamp":"2026-05-27T07:39:00Z" } }`, agentID))) require.Zero(t, dispatcher.startCalls) require.Equal(t, 1, dispatcher.refreshStartCalls) require.Equal(t, agentID, dispatcher.lastRefreshStartAgent) require.Equal(t, 1, dispatcher.dispatchCalls) } func TestHandleAgentStartWithWakeFallsBackWhenRunningSessionRejectsHandoff(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_start_running_reject_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Running Reviewer", Runtime: "codex", PermissionMode: types.AgentPermissionModeFullAccess, Source: agent.NewAgentTeamSource("agentteam-main"), })) dispatcher := &agentteamTestDispatcher{ hasSession: true, session: types.SessionSnapshot{Status: "running"}, dispatchErr: errors.New("Process generation retired before handoff commit"), } channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 8), disp: dispatcher, } dispatcher.onStart = func(int) { for len(channel.sendCh) > 0 { raw := <-channel.sendCh var status struct { Type string `json:"type"` Status string `json:"status"` } require.NoError(t, json.Unmarshal(raw, &status)) require.False(t, status.Type == "agent:status" && status.Status == "active", "a rejected shortcut must not report active before replacement succeeds") } } channel.handleAgentStart(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{ "name":"running-reviewer", "runtime":"codex", "permissionMode":"full_access", "authToken":"agent-token" }, "wakeMessage":{ "message_id":"msg_start_rejected_handoff", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"wake must survive retirement", "timestamp":"2026-05-27T07:39:00Z" } }`, agentID))) require.Equal(t, 1, dispatcher.dispatchCalls) require.Equal(t, 1, dispatcher.startCalls) require.Len(t, dispatcher.startWakes, 1) require.NotNil(t, dispatcher.startWakes[0]) require.Equal(t, "wake must survive retirement", dispatcher.startWakes[0].Input.Prompt) meta, ok := dispatcher.startWakes[0].Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Equal(t, "msg_start_rejected_handoff", meta.ID) } func TestHandleAgentStartUrgentWakeReplacesRunningSession(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_start_urgent_replace_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Urgent Reviewer", Runtime: "codex", PermissionMode: types.AgentPermissionModeFullAccess, Source: agent.NewAgentTeamSource("agentteam-main"), })) dispatcher := &agentteamTestDispatcher{ hasSession: true, session: types.SessionSnapshot{Status: "running"}, } channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 8), disp: dispatcher, } channel.handleAgentStart(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{"name":"urgent-reviewer","runtime":"codex"}, "wakeMessage":{ "message_id":"msg_urgent_start_fallback", "interruptPolicy":"interrupt_current_run", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"replace the failed native interrupt", "timestamp":"2026-05-27T07:39:00Z" } }`, agentID))) require.Zero(t, dispatcher.dispatchCalls, "agent:start owns replacement after an urgent handoff failure") require.Equal(t, 1, dispatcher.startCalls) require.Len(t, dispatcher.startWakes, 1) require.NotNil(t, dispatcher.startWakes[0]) meta, ok := dispatcher.startWakes[0].Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Equal(t, "interrupt_current_run", meta.InterruptPolicy) require.Equal(t, "msg_urgent_start_fallback", meta.ID) } func TestHandleAgentStartWithWakeRestartsWhenStartConfigRefreshStopsSession(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_start_running_restart_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Running Reviewer", Runtime: "codex", PermissionMode: types.AgentPermissionModeFullAccess, Source: agent.NewAgentTeamSource("agentteam-main"), })) dispatcher := &agentteamTestRestartingStartConfigDispatcher{} dispatcher.hasSession = true dispatcher.session = types.SessionSnapshot{Status: "running"} channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 8), disp: dispatcher, } channel.handleAgentStart(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{ "name":"running-reviewer", "runtime":"codex", "permissionMode":"full_access", "authToken":"agent-token" }, "wakeMessage":{ "message_id":"msg_start_policy_restart", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"wake after policy restart", "timestamp":"2026-05-27T07:39:00Z" } }`, agentID))) require.Equal(t, 1, dispatcher.refreshStartCalls) require.Equal(t, 1, dispatcher.startCalls) require.Zero(t, dispatcher.dispatchCalls) require.Len(t, dispatcher.startWakes, 1) require.NotNil(t, dispatcher.startWakes[0]) require.Equal(t, "wake after policy restart", dispatcher.startWakes[0].Input.Prompt) } func TestHandleAgentStartWithoutWakeForRunningSessionReplacesProcess(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_start_running_replace_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Running Reviewer", Runtime: "codex", PermissionMode: types.AgentPermissionModeFullAccess, Source: agent.NewAgentTeamSource("agentteam-main"), })) dispatcher := &agentteamTestDispatcher{ hasSession: true, session: types.SessionSnapshot{Status: "running"}, } channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 8), disp: dispatcher, } hash := deliveryDedupHash(agentID, 1, "msg-before-restart") require.False(t, channel.dedupRing.AddWithKey(hash, agentID)) channel.handleAgentStart(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{ "name":"running-reviewer", "runtime":"codex", "permissionMode":"full_access", "model":"gpt-test-next", "authToken":"agent-token-refresh" } }`, agentID))) require.Equal(t, 1, dispatcher.startCalls) require.Equal(t, 0, dispatcher.dispatchCalls) require.False(t, channel.dedupRing.Contains(hash)) require.Empty(t, dispatcher.refreshedAgentConfigs) require.Len(t, dispatcher.startConfigs, 1) require.Equal(t, types.AgentPermissionModeFullAccess, dispatcher.startConfigs[0].PermissionMode) require.Equal(t, "gpt-test-next", dispatcher.startConfigs[0].Model) } func TestHandleAgentStartWaitsForManagedSkillReadinessAfterAgentListSync(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_start_skill_ready_%d", time.Now().UnixNano()) releaseRefresh := make(chan struct{}) dispatcher := &blockingManagedSkillLaunchDispatcher{ managedSkillRefreshSeen: make(chan struct{}, 4), managedSkillRelease: releaseRefresh, } channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 8), disp: dispatcher, } done := make(chan struct{}) go func() { channel.handleAgentStart(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{ "name":"skill-ready-reviewer", "runtime":"codex", "permissionMode":"full_access", "authToken":"agent-token" } }`, agentID))) close(done) }() require.Eventually(t, func() bool { return len(channel.sendCh) == 1 }, time.Second, 5*time.Millisecond) raw := <-channel.sendCh require.JSONEq(t, `{"type":"machine:agent:list"}`, string(raw)) channel.handleAgentListResponse([]byte(fmt.Sprintf(`{ "type":"machine:agent:list:response", "agents":[{ "id":%q, "name":"skill-ready-reviewer", "displayName":"Skill Ready Reviewer", "runtime":"codex", "status":"idle", "activity":"" }] }`, agentID))) select { case <-dispatcher.managedSkillRefreshSeen: case <-time.After(time.Second): t.Fatal("managed skill refresh did not start") } require.Never(t, func() bool { return dispatcher.startCallsObserved.Load() > 0 }, 100*time.Millisecond, 5*time.Millisecond, "agent:start must not launch before managed skill refresh completes on cache miss") close(releaseRefresh) require.Eventually(t, func() bool { return dispatcher.startCallsObserved.Load() == 1 }, time.Second, 5*time.Millisecond) select { case <-done: case <-time.After(time.Second): t.Fatal("handleAgentStart did not finish after managed skill refresh completed") } require.GreaterOrEqual(t, dispatcher.managedSkillRefreshes.Load(), int32(1)) } func TestAgentIdentityReceivedDuringStartPreflightFoldsIntoFirstLaunch(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_identity_preflight_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Identity A", Runtime: "codex", Source: agent.NewAgentTeamSource("agentteam-main"), })) releaseRefresh := make(chan struct{}) dispatcher := &blockingManagedSkillLaunchDispatcher{ managedSkillRefreshSeen: make(chan struct{}, 1), managedSkillRelease: releaseRefresh, } channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 8), disp: dispatcher, } run := newConnectionRun(context.Background()) ctx := run.context() channel.dispatch(ctx, []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{"name":"identity-a","displayName":"Identity A","runtime":"codex"} }`, agentID))) select { case <-dispatcher.managedSkillRefreshSeen: case <-time.After(time.Second): t.Fatal("agent:start did not reach managed-skill preflight") } channel.dispatch(ctx, []byte(fmt.Sprintf(`{ "type":"agent:update_identity", "agentId":%q, "name":"identity-b", "displayName":"Identity B" }`, agentID))) channel.mu.Lock() require.Equal(t, "identity-b", channel.agents[agentID].Config.Name) require.Equal(t, "Identity B", channel.agents[agentID].Config.DisplayName) channel.mu.Unlock() close(releaseRefresh) require.Eventually(t, func() bool { channel.mu.Lock() defer channel.mu.Unlock() state := channel.agents[agentID] return dispatcher.startCallsObserved.Load() == 1 && state != nil && state.lifecycleTail == nil }, time.Second, 5*time.Millisecond) require.Len(t, dispatcher.startConfigs, 1) require.Equal(t, "identity-b", dispatcher.startConfigs[0].Name) require.Equal(t, "Identity B", dispatcher.startConfigs[0].DisplayName) require.Equal(t, []identityUpdateCall{{AgentID: agentID, Name: "identity-b", DisplayName: "Identity B"}}, dispatcher.identityUpdates) owners := run.closeAdmission() require.NoError(t, waitForConnectionRunOwners(context.Background(), owners)) } func TestAgentIdentityReceivedDuringDispatcherStartRunsAfterPublicationOwner(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_identity_launch_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Identity A", Runtime: "codex", Source: agent.NewAgentTeamSource("agentteam-main"), })) startEntered := make(chan struct{}) releaseStart := make(chan struct{}) dispatcher := &agentteamTestDispatcher{} dispatcher.onStart = func(call int) { if call == 1 { close(startEntered) <-releaseStart } } channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 8), disp: dispatcher, } run := newConnectionRun(context.Background()) ctx := run.context() channel.dispatch(ctx, []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{"name":"identity-a","displayName":"Identity A","runtime":"codex"} }`, agentID))) select { case <-startEntered: case <-time.After(time.Second): t.Fatal("agent:start did not enter Dispatcher") } channel.dispatch(ctx, []byte(fmt.Sprintf(`{ "type":"agent:update_identity", "agentId":%q, "name":"identity-b", "displayName":"Identity B" }`, agentID))) require.Empty(t, dispatcher.identityUpdates, "identity mutation must wait for the older start publication owner") close(releaseStart) require.Eventually(t, func() bool { channel.mu.Lock() defer channel.mu.Unlock() state := channel.agents[agentID] return state != nil && state.lifecycleTail == nil }, time.Second, 5*time.Millisecond) require.Len(t, dispatcher.startConfigs, 1) require.Equal(t, "identity-a", dispatcher.startConfigs[0].Name) require.Len(t, dispatcher.identityUpdates, 1) require.Equal(t, []identityUpdateCall{{AgentID: agentID, Name: "identity-b", DisplayName: "Identity B"}}, dispatcher.identityUpdates) owners := run.closeAdmission() require.NoError(t, waitForConnectionRunOwners(context.Background(), owners)) } func TestAgentStopReceivedDuringDispatcherStartRunsAfterPublicationOwner(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_stop_during_start_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Stop During Start", Runtime: "codex", Source: agent.NewAgentTeamSource("agentteam-main"), })) releaseStart := make(chan struct{}) dispatcher := &blockingStartStopDispatcher{ startEntered: make(chan struct{}), startRelease: releaseStart, } channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 8), disp: dispatcher, } run := newConnectionRun(context.Background()) ctx := run.context() channel.dispatch(ctx, []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{"name":"stop-during-start","runtime":"codex"} }`, agentID))) select { case <-dispatcher.startEntered: case <-time.After(time.Second): t.Fatal("agent:start did not enter Dispatcher") } channel.dispatch(ctx, []byte(fmt.Sprintf(`{ "type":"agent:stop", "agentId":%q }`, agentID))) require.Zero(t, dispatcher.stopCalls.Load(), "later stop must wait for the earlier start publication owner") close(releaseStart) require.Eventually(t, func() bool { channel.mu.Lock() state := channel.agents[agentID] tailDone := state == nil || state.lifecycleTail == nil channel.mu.Unlock() return dispatcher.stopCalls.Load() == 1 && tailDone }, time.Second, 5*time.Millisecond) require.False(t, dispatcher.running.Load(), "receipt-ordered stop must retire the Process published by the earlier start") owners := run.closeAdmission() require.NoError(t, waitForConnectionRunOwners(context.Background(), owners)) } func TestAgentStopPreservesLaterStartReceiptAuthority(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_start_stop_start_authority_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Start A", Runtime: "codex", Source: agent.NewAgentTeamSource("agentteam-main"), })) releaseStart := make(chan struct{}) dispatcher := &blockingStartStopDispatcher{ startEntered: make(chan struct{}), startRelease: releaseStart, } channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 8), disp: dispatcher, } run := newConnectionRun(context.Background()) ctx := run.context() channel.dispatch(ctx, []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{"name":"start-a","runtime":"codex","model":"gpt-a","authToken":"token-a"} }`, agentID))) select { case <-dispatcher.startEntered: case <-time.After(time.Second): t.Fatal("start A did not enter Dispatcher") } channel.dispatch(ctx, []byte(fmt.Sprintf(`{ "type":"agent:stop", "agentId":%q }`, agentID))) channel.dispatch(ctx, []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{ "name":"start-c", "displayName":"Start C", "runtime":"claude", "model":"claude-c", "permissionMode":"agent_workspace_write", "authToken":"token-c" } }`, agentID))) channel.mu.Lock() receivedState := channel.agents[agentID] var receivedConfig types.AgentStartConfig var receivedToken string if receivedState != nil { receivedConfig = receivedState.Config receivedToken = receivedState.Token } channel.mu.Unlock() require.NotNil(t, receivedState) require.Equal(t, "start-c", receivedConfig.Name) require.Equal(t, "claude", receivedConfig.Runtime) require.Equal(t, "token-c", receivedToken) require.Zero(t, dispatcher.stopCalls.Load()) close(releaseStart) require.Eventually(t, func() bool { channel.mu.Lock() defer channel.mu.Unlock() state := channel.agents[agentID] return dispatcher.startAdmissions.Load() == 2 && dispatcher.stopCalls.Load() == 1 && state != nil && state.lifecycleTail == nil }, time.Second, 5*time.Millisecond) require.True(t, dispatcher.running.Load(), "start C must publish after stop B retires start A") require.Len(t, dispatcher.startConfigs, 2) require.Equal(t, "start-a", dispatcher.startConfigs[0].Name) startC := dispatcher.startConfigs[1] require.Equal(t, "start-c", startC.Name) require.Equal(t, "Start C", startC.DisplayName) require.Equal(t, "claude", startC.Runtime) require.Equal(t, "claude-c", startC.Model) require.Equal(t, types.AgentPermissionModeWorkspaceWrite, startC.PermissionMode) require.Equal(t, "token-c", startC.AuthToken) channel.mu.Lock() finalState := channel.agents[agentID] var finalConfig types.AgentStartConfig var finalToken string if finalState != nil { finalConfig = finalState.Config finalToken = finalState.Token } channel.mu.Unlock() require.NotNil(t, finalState) require.Equal(t, "token-c", finalToken) require.Equal(t, startC, finalConfig) owners := run.closeAdmission() require.NoError(t, waitForConnectionRunOwners(context.Background(), owners)) } func TestInvalidAgentStartResolutionWaitsForEarlierPublicationOwner(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_invalid_start_after_publication_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Invalid Start After Publication", Runtime: "codex", Source: agent.NewAgentTeamSource("agentteam-main"), })) releaseStart := make(chan struct{}) dispatcher := &blockingStartStopDispatcher{ startEntered: make(chan struct{}), startRelease: releaseStart, } channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 8), disp: dispatcher, } run := newConnectionRun(context.Background()) ctx := run.context() channel.dispatch(ctx, []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{"name":"valid-publication","runtime":"codex"} }`, agentID))) select { case <-dispatcher.startEntered: case <-time.After(time.Second): t.Fatal("first agent:start did not enter Dispatcher") } channel.dispatch(ctx, []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{"name":"invalid-wake","runtime":"codex"}, "wakeMessage":{ "message_id":"msg_invalid_after_publication", "channel_type":"channel", "channel_name":"ops", "content":"reject after the older publication", "interruptPolicy":"replace_everything" } }`, agentID))) var ack deliverAckMsg select { case raw := <-channel.sendCh: require.NoError(t, json.Unmarshal(raw, &ack)) require.Equal(t, "agent:deliver:ack", ack.Type) require.False(t, ack.Accepted) case <-time.After(time.Second): t.Fatal("invalid wake did not receive its immediate compatibility ACK") } require.Empty(t, channel.sendCh, "authoritative invalid-start resolution must wait for the older publication owner") close(releaseStart) require.Eventually(t, func() bool { channel.mu.Lock() defer channel.mu.Unlock() state := channel.agents[agentID] return state != nil && state.lifecycleTail == nil }, time.Second, 5*time.Millisecond) require.True(t, dispatcher.running.Load()) var frames []map[string]any for len(channel.sendCh) > 0 { var frame map[string]any require.NoError(t, json.Unmarshal(<-channel.sendCh, &frame)) frames = append(frames, frame) } require.Len(t, frames, 4) require.Equal(t, "agent:status", frames[0]["type"]) require.Equal(t, "active", frames[0]["status"], "the older valid start publishes first") require.Equal(t, "agent:wake_status", frames[1]["type"]) require.Equal(t, "agent:runtime_event", frames[2]["type"]) require.Equal(t, "agent:status", frames[3]["type"]) require.Equal(t, "active", frames[3]["status"], "the sequenced rejection must retain the healthy Process authority") owners := run.closeAdmission() require.NoError(t, waitForConnectionRunOwners(context.Background(), owners)) } func TestHandleDeliverBypassesManagedSkillLaunchReadiness(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_deliver_skill_ready_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Skill Ready Delivery", Runtime: "codex", Source: agent.NewAgentTeamSource("agentteam-main"), })) releaseRefresh := make(chan struct{}) dispatcher := &blockingManagedSkillLaunchDispatcher{ managedSkillRefreshSeen: make(chan struct{}, 2), managedSkillRelease: releaseRefresh, } channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 8), disp: dispatcher, } done := make(chan struct{}) go func() { channel.handleDeliver(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:deliver", "agentId":%q, "seq":1, "message":{ "message_id":"msg_skill_ready_deliver", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"please wake", "timestamp":"2026-07-02T00:00:00Z" } }`, agentID))) close(done) }() select { case <-done: case <-time.After(time.Second): t.Fatal("handleDeliver should reject quickly instead of blocking on managed skill refresh") } require.Equal(t, 1, dispatcher.dispatchCalls, "Process admission, not cold-start readiness, decides agent:deliver") require.Zero(t, dispatcher.managedSkillRefreshes.Load(), "agent:deliver must not block reader on managed skill refresh") require.Zero(t, dispatcher.startCallsObserved.Load(), "agent:deliver must not launch a Process") select { case raw := <-channel.sendCh: var ack deliverAckMsg require.NoError(t, json.Unmarshal(raw, &ack)) require.True(t, ack.Accepted) require.Equal(t, "msg_skill_ready_deliver", ack.MessageID) case <-time.After(time.Second): t.Fatal("expected Process handoff acknowledgement") } close(releaseRefresh) } func TestHandleAgentStartRejectsUnsupportedWakeWithoutStarting(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_start_invalid_wake_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Invalid Wake Reviewer", Runtime: "codex", Source: agent.NewAgentTeamSource("agentteam-main"), })) dispatcher := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 4), disp: dispatcher, } channel.handleAgentStart(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{ "name":"invalid-wake-reviewer", "runtime":"codex", "authToken":"rotated-token" }, "wakeMessage":{ "message_id":"msg_invalid_policy", "channel_type":"channel", "channel_name":"ops", "content":"must not become a nil-input start", "interruptPolicy":"replace_everything" } }`, agentID))) require.Zero(t, dispatcher.startCalls) require.Zero(t, dispatcher.dispatchCalls) channel.mu.Lock() state := channel.agents[agentID] require.NotNil(t, state) require.Equal(t, "rotated-token", state.Token, "config receipt remains authoritative despite wake rejection") channel.mu.Unlock() select { case raw := <-channel.sendCh: var ack deliverAckMsg require.NoError(t, json.Unmarshal(raw, &ack)) require.False(t, ack.Accepted) require.Equal(t, "msg_invalid_policy", ack.MessageID) require.Equal(t, "replace_everything", ack.InterruptPolicy) require.Contains(t, ack.Detail, "unsupported interruptPolicy") case <-time.After(time.Second): t.Fatal("invalid agent:start wake did not receive a terminal rejection") } var wakeStatus types.AgentWakeStatusMsg require.NoError(t, json.Unmarshal(<-channel.sendCh, &wakeStatus)) require.Equal(t, "agent:wake_status", wakeStatus.Type) require.Equal(t, agentID, wakeStatus.AgentID) require.Equal(t, "msg_invalid_policy", wakeStatus.MessageID) require.Equal(t, "agent-start:msg_invalid_policy", wakeStatus.WakeToken) require.Equal(t, "#ops", wakeStatus.Target) require.Equal(t, consts.WakeStatusFailed, wakeStatus.Status) require.Contains(t, wakeStatus.Detail, "unsupported interruptPolicy") var runtimeEvent agentRuntimeEventMsg require.NoError(t, json.Unmarshal(<-channel.sendCh, &runtimeEvent)) require.Equal(t, "agent:runtime_event", runtimeEvent.Type) require.Equal(t, agentID, runtimeEvent.AgentID) require.Equal(t, "msg_invalid_policy", runtimeEvent.MessageID) require.Equal(t, "#ops", runtimeEvent.Target) require.Equal(t, consts.RuntimeEventClassRuntimeStartFailed, runtimeEvent.EventClass) require.Contains(t, runtimeEvent.Detail, "unsupported interruptPolicy") var status map[string]string require.NoError(t, json.Unmarshal(<-channel.sendCh, &status)) require.Equal(t, "agent:status", status["type"]) require.Equal(t, agentID, status["agentId"]) require.Equal(t, "inactive", status["status"]) require.Empty(t, channel.sendCh) } func TestHandleAgentStartRejectsMalformedWakeWithAuthoritativeFailure(t *testing.T) { dispatcher := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 4), disp: dispatcher, } channel.handleAgentStart(context.Background(), []byte(`{ "type":"agent:start", "agentId":"agt_malformed_start_wake", "config":{"name":"malformed-wake","runtime":"codex","authToken":"rotated-token"}, "wakeMessage":"not-an-object" }`)) require.Zero(t, dispatcher.startCalls) require.Zero(t, dispatcher.dispatchCalls) channel.mu.Lock() require.Equal(t, "rotated-token", channel.agents["agt_malformed_start_wake"].Token) channel.mu.Unlock() var ack deliverAckMsg require.NoError(t, json.Unmarshal(<-channel.sendCh, &ack)) require.Equal(t, "agent:deliver:ack", ack.Type) require.Equal(t, "agt_malformed_start_wake", ack.AgentID) require.Empty(t, ack.MessageID) require.False(t, ack.Accepted) require.Contains(t, ack.Detail, "parse error") var runtimeEvent agentRuntimeEventMsg require.NoError(t, json.Unmarshal(<-channel.sendCh, &runtimeEvent)) require.Equal(t, "agent:runtime_event", runtimeEvent.Type) require.Equal(t, "agt_malformed_start_wake", runtimeEvent.AgentID) require.Empty(t, runtimeEvent.MessageID) require.Equal(t, consts.RuntimeEventClassRuntimeStartFailed, runtimeEvent.EventClass) require.Contains(t, runtimeEvent.Detail, "parse error") var status map[string]string require.NoError(t, json.Unmarshal(<-channel.sendCh, &status)) require.Equal(t, "agent:status", status["type"]) require.Equal(t, "agt_malformed_start_wake", status["agentId"]) require.Equal(t, "inactive", status["status"]) require.Empty(t, channel.sendCh, "malformed wakes cannot emit an invalid empty-message wake status") } func TestHandleAgentStartInvalidWakeKeepsHealthyProcessActive(t *testing.T) { dispatcher := &agentteamTestDispatcher{ hasSession: true, session: types.SessionSnapshot{Status: "running"}, } channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 4), disp: dispatcher, } channel.handleAgentStart(context.Background(), []byte(`{ "type":"agent:start", "agentId":"agt_healthy_invalid_wake", "wakeMessage":{ "message_id":"msg_healthy_invalid_wake", "channel_type":"channel", "channel_name":"ops", "content":"reject only this payload", "interruptPolicy":"replace_everything" } }`)) require.Zero(t, dispatcher.startCalls) require.Zero(t, dispatcher.dispatchCalls) require.Len(t, channel.sendCh, 4) var ack deliverAckMsg require.NoError(t, json.Unmarshal(<-channel.sendCh, &ack)) require.False(t, ack.Accepted) var wakeStatus types.AgentWakeStatusMsg require.NoError(t, json.Unmarshal(<-channel.sendCh, &wakeStatus)) require.Equal(t, consts.WakeStatusFailed, wakeStatus.Status) var runtimeEvent agentRuntimeEventMsg require.NoError(t, json.Unmarshal(<-channel.sendCh, &runtimeEvent)) require.Equal(t, consts.RuntimeEventClassRuntimeStartFailed, runtimeEvent.EventClass) var status map[string]string require.NoError(t, json.Unmarshal(<-channel.sendCh, &status)) require.Equal(t, "active", status["status"], "rejecting one invalid fallback payload must not mark a healthy older Process inactive") } func TestHandleAgentStartQueuesEveryWakePayloadWhileStarting(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_start_wake_queue_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Wake Reviewer", Runtime: "codex", PermissionMode: types.AgentPermissionModeFullAccess, Source: agent.NewAgentTeamSource("agentteam-main"), })) dispatcher := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 8), disp: dispatcher, } dispatcher.onStart = func(call int) { if call != 1 { return } channel.handleAgentStart(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{ "name":"wake-reviewer", "runtime":"codex", "permissionMode":"agent_workspace_write", "authToken":"agent-token-replacement" }, "wakeMessage":{ "message_id":"msg_wake_b", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"wake B", "timestamp":"2026-05-27T07:40:00Z" } }`, agentID))) } channel.handleAgentStart(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{ "name":"wake-reviewer", "runtime":"codex", "permissionMode":"full_access", "authToken":"agent-token-original" }, "wakeMessage":{ "message_id":"msg_wake_a", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"wake A", "timestamp":"2026-05-27T07:39:00Z" } }`, agentID))) require.Equal(t, 2, dispatcher.startCalls) require.Len(t, dispatcher.startWakes, 2) require.NotNil(t, dispatcher.startWakes[1]) firstMeta, ok := dispatcher.startWakes[1].Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Equal(t, "msg_wake_a", firstMeta.ID) require.Equal(t, "wake A", dispatcher.startWakes[1].Input.Prompt) require.Equal(t, 1, dispatcher.dispatchCalls) require.Len(t, dispatcher.dispatchedMessages, 1) secondMeta, ok := dispatcher.dispatchedMessages[0].Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Equal(t, "msg_wake_b", secondMeta.ID) require.Equal(t, "wake B", dispatcher.dispatchedMessages[0].Input.Prompt) } func TestHandleAgentStartQueuedWakeFallsBackWhenPostStartHandoffRejected(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_start_wake_reject_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Wake Reviewer", Runtime: "codex", PermissionMode: types.AgentPermissionModeFullAccess, Source: agent.NewAgentTeamSource("agentteam-main"), })) dispatcher := &agentteamTestDispatcher{ dispatchErr: errors.New("Process generation retired before handoff commit"), } channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 16), disp: dispatcher, } dispatcher.onStart = func(call int) { if call != 1 { return } channel.handleAgentStart(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{ "name":"wake-reviewer", "runtime":"codex", "permissionMode":"agent_workspace_write", "authToken":"agent-token-replacement" }, "wakeMessage":{ "message_id":"msg_wake_b_rejected", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"wake B must become lifecycle-owned", "timestamp":"2026-05-27T07:40:00Z" }, "unreadSummary":{"ops":3} }`, agentID))) } channel.handleAgentStart(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{ "name":"wake-reviewer", "runtime":"codex", "permissionMode":"full_access", "authToken":"agent-token-original" }, "wakeMessage":{ "message_id":"msg_wake_a_rejected", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"wake A", "timestamp":"2026-05-27T07:39:00Z" } }`, agentID))) require.Equal(t, 3, dispatcher.startCalls) require.Equal(t, 1, dispatcher.dispatchCalls) require.Len(t, dispatcher.startWakes, 3) finalWake := dispatcher.startWakes[2] require.NotNil(t, finalWake) require.Equal(t, "wake B must become lifecycle-owned", finalWake.Input.Prompt) meta, ok := finalWake.Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Equal(t, "msg_wake_b_rejected", meta.ID) require.Equal(t, map[string]int{"ops": 3}, meta.UnreadSummary) require.Equal(t, types.AgentPermissionModeWorkspaceWrite, dispatcher.startConfigs[2].PermissionMode) require.True(t, channel.dedupRing.Contains(deliveryDedupHash(agentID, 0, "msg_wake_b_rejected"))) channel.mu.Lock() state := channel.agents[agentID] require.NotNil(t, state) require.False(t, state.Starting) require.False(t, state.RestartQueued) require.Empty(t, state.PendingStarts) channel.mu.Unlock() } func TestHandleAgentStartDuplicateQueuedWakeDoesNotLeakUnreadSummary(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) agentID := fmt.Sprintf("agt_start_wake_dedup_%d", time.Now().UnixNano()) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: agentID, Name: "Wake Reviewer", Runtime: "codex", PermissionMode: types.AgentPermissionModeFullAccess, Source: agent.NewAgentTeamSource("agentteam-main"), })) dispatcher := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam-main", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 8), disp: dispatcher, } dispatcher.onStart = func(call int) { if call != 1 { return } channel.handleAgentStart(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{ "name":"wake-reviewer", "runtime":"codex", "permissionMode":"full_access", "authToken":"agent-token-duplicate" }, "wakeMessage":{ "message_id":"msg_wake_a", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"wake A duplicate", "timestamp":"2026-05-27T07:40:00Z" }, "unreadSummary":{"ops":9} }`, agentID))) channel.handleAgentStart(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{ "name":"wake-reviewer", "runtime":"codex", "permissionMode":"agent_workspace_write", "authToken":"agent-token-replacement" }, "wakeMessage":{ "message_id":"msg_wake_b", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"wake B", "timestamp":"2026-05-27T07:41:00Z" } }`, agentID))) } channel.handleAgentStart(context.Background(), []byte(fmt.Sprintf(`{ "type":"agent:start", "agentId":%q, "config":{ "name":"wake-reviewer", "runtime":"codex", "permissionMode":"full_access", "authToken":"agent-token-original" }, "wakeMessage":{ "message_id":"msg_wake_a", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"wake A", "timestamp":"2026-05-27T07:39:00Z" }, "unreadSummary":{"ops":2} }`, agentID))) require.Equal(t, 2, dispatcher.startCalls) require.Len(t, dispatcher.startWakes, 2) require.NotNil(t, dispatcher.startWakes[1]) firstMeta, ok := dispatcher.startWakes[1].Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Equal(t, "msg_wake_a", firstMeta.ID) require.Equal(t, map[string]int{"ops": 2}, firstMeta.UnreadSummary) require.Equal(t, 1, dispatcher.dispatchCalls) require.Len(t, dispatcher.dispatchedMessages, 1) secondMeta, ok := dispatcher.dispatchedMessages[0].Input.Meta.(*types.AgentTeamMeta) require.True(t, ok) require.Equal(t, "msg_wake_b", secondMeta.ID) require.Empty(t, secondMeta.UnreadSummary) } func TestHandleAgentStartInvalidConfigFailsFast(t *testing.T) { dispatcher := &agentteamTestDispatcher{} channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 4), disp: dispatcher, } channel.handleAgentStart(context.Background(), []byte(`{ "type":"agent:start", "agentId":"agt_server", "config":"not-an-object", "wakeMessage":{ "message_id":"msg_start", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"wake up", "timestamp":"2026-05-16T07:39:00Z" } }`)) require.Zero(t, dispatcher.startCalls) require.Len(t, channel.sendCh, 3) var wakeStatus types.AgentWakeStatusMsg require.NoError(t, json.Unmarshal(<-channel.sendCh, &wakeStatus)) require.Equal(t, "agent:wake_status", wakeStatus.Type) require.Equal(t, "agt_server", wakeStatus.AgentID) require.Equal(t, "msg_start", wakeStatus.MessageID) require.Equal(t, "#ops", wakeStatus.Target) require.Equal(t, consts.WakeStatusFailed, wakeStatus.Status) require.Contains(t, wakeStatus.Detail, "invalid agent start config") var runtimeEvent agentRuntimeEventMsg require.NoError(t, json.Unmarshal(<-channel.sendCh, &runtimeEvent)) require.Equal(t, "agent:runtime_event", runtimeEvent.Type) require.Equal(t, "agt_server", runtimeEvent.AgentID) require.Equal(t, "msg_start", runtimeEvent.MessageID) require.Equal(t, "#ops", runtimeEvent.Target) require.Equal(t, consts.RuntimeEventClassRuntimeStartFailed, runtimeEvent.EventClass) require.Contains(t, runtimeEvent.Detail, "invalid agent start config") var status map[string]string require.NoError(t, json.Unmarshal(<-channel.sendCh, &status)) require.Equal(t, "agent:status", status["type"]) require.Equal(t, "agt_server", status["agentId"]) require.Equal(t, "inactive", status["status"]) } func TestHandleAgentStartReportsSpawnFailureActivity(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: "agt_server", Name: "memory-reviewer", Runtime: "claude", })) startErr := errors.New("claude: AgentTeam generated Claude startup arguments exceed the Windows process launch limit") dispatcher := &agentteamTestDispatcher{startErr: startErr} channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 4), disp: dispatcher, } channel.handleAgentStart(context.Background(), []byte(`{ "type":"agent:start", "agentId":"agt_server", "config":{ "name":"memory-reviewer", "runtime":"claude", "authToken":"agent-token" } }`)) require.Equal(t, 1, dispatcher.startCalls) require.Len(t, channel.sendCh, 2) var status map[string]string require.NoError(t, json.Unmarshal(<-channel.sendCh, &status)) require.Equal(t, "agent:status", status["type"]) require.Equal(t, "agt_server", status["agentId"]) require.Equal(t, "inactive", status["status"]) var activity map[string]string require.NoError(t, json.Unmarshal(<-channel.sendCh, &activity)) require.Equal(t, "agent:activity", activity["type"]) require.Equal(t, "agt_server", activity["agentId"]) require.Equal(t, "offline", activity["activity"]) require.Contains(t, activity["detail"], startErr.Error()) } func TestHandleAgentStartSpawnFailureEmitsWakeStatusForWakeMessage(t *testing.T) { t.Setenv("HOME", t.TempDir()) t.Setenv(profile.ConfigDirEnvKey, filepath.Join(t.TempDir(), ".agentteam")) require.NoError(t, profile.Default().Agents().Add(agent.Config{ ID: "agt_server", Name: "memory-reviewer", Runtime: "claude", })) startErr := errors.New("thread/start: failed to load configuration: Model provider `ccrs` not found") dispatcher := &agentteamTestDispatcher{startErr: startErr} channel := &Protocol{ id: "agentteam", agents: map[string]*agentState{}, dedupRing: newDedupRing(10), sendCh: make(chan []byte, 4), disp: dispatcher, } channel.handleAgentStart(context.Background(), []byte(`{ "type":"agent:start", "agentId":"agt_server", "config":{ "name":"memory-reviewer", "runtime":"claude", "authToken":"agent-token" }, "wakeMessage":{ "message_id":"msg_start_failure", "channel_type":"channel", "channel_name":"ops", "sender_type":"human", "sender_name":"alice", "content":"wake up", "timestamp":"2026-05-16T07:39:00Z" } }`)) require.Equal(t, 1, dispatcher.startCalls) require.Len(t, channel.sendCh, 3) var wakeStatus types.AgentWakeStatusMsg require.NoError(t, json.Unmarshal(<-channel.sendCh, &wakeStatus)) require.Equal(t, "agent:wake_status", wakeStatus.Type) require.Equal(t, "agt_server", wakeStatus.AgentID) require.Equal(t, "msg_start_failure", wakeStatus.MessageID) require.Equal(t, "agent-start:msg_start_failure", wakeStatus.WakeToken) require.Equal(t, "#ops", wakeStatus.Target) require.Equal(t, consts.WakeStatusFailed, wakeStatus.Status) require.Contains(t, wakeStatus.Detail, startErr.Error()) var status map[string]string require.NoError(t, json.Unmarshal(<-channel.sendCh, &status)) require.Equal(t, "agent:status", status["type"]) require.Equal(t, "inactive", status["status"]) var activity map[string]string require.NoError(t, json.Unmarshal(<-channel.sendCh, &activity)) require.Equal(t, "agent:activity", activity["type"]) require.Equal(t, "offline", activity["activity"]) require.Contains(t, activity["detail"], startErr.Error()) } func TestKnowledgeBaseScopeTargetUsesParentConversationForReplyThreadTargets(t *testing.T) { tests := []struct { target string want string }{ {target: "#agentteam:00000102", want: "#agentteam"}, {target: "#agentteam", want: "#agentteam"}, {target: "DM:@alice:00000102", want: "dm:@alice"}, {target: "dm:@alice", want: "dm:@alice"}, {target: "#tsk_scope_1", want: "#tsk_scope_1"}, {target: "#team_abcdefghijklmn:00000102", want: "#team_abcdefghijklmn"}, {target: "#team_abcdefghijklmn", want: "#team_abcdefghijklmn"}, {target: "#agentteam:bad", want: "#agentteam:bad"}, {target: "chn_current_scope", want: "chn_current_scope"}, {target: "", want: ""}, } for _, tt := range tests { require.Equal(t, tt.want, knowledgeBaseScopeTarget(tt.target)) } } func TestSplitTextChunks(t *testing.T) { tests := []struct { text string max int wantLen int wantAll string // all chunks joined should equal input }{ {"hello", 10, 1, "hello"}, {"abc\ndef\nghi", 4, 3, "abc\ndef\nghi"}, {"你好世界", 7, 2, "你好世界"}, {"", 10, 1, ""}, } for _, tt := range tests { chunks := apiclient.SplitMessageText(tt.text, tt.max) if len(chunks) != tt.wantLen { t.Errorf("splitTextChunks(%q, %d) returned %d chunks, want %d", tt.text, tt.max, len(chunks), tt.wantLen) } joined := "" for _, c := range chunks { joined += c } if joined != tt.wantAll { t.Errorf("splitTextChunks(%q, %d) joined = %q, want %q", tt.text, tt.max, joined, tt.wantAll) } } } func TestSplitTextChunksRespectsByteBudget(t *testing.T) { text := strings.Repeat("a", textMaxBytes+1) chunks := apiclient.SplitMessageText(text, textMaxBytes) require.Len(t, chunks, 2) for _, chunk := range chunks { require.LessOrEqual(t, len(chunk), textMaxBytes) } require.Equal(t, text, strings.Join(chunks, "")) } func TestPreviewText(t *testing.T) { if got := previewText("hello", 10); got != "hello" { t.Errorf("previewText short: got %q", got) } if got := previewText("hello world", 5); got != "hello..." { t.Errorf("previewText long: got %q", got) } } func TestFnvHash64_Deterministic(t *testing.T) { h1 := fnvHash64("agent-abc", 42) h2 := fnvHash64("agent-abc", 42) if h1 != h2 { t.Fatal("fnvHash64 not deterministic") } h3 := fnvHash64("agent-abc", 43) if h1 == h3 { t.Fatal("fnvHash64 collision on different seq") } } func TestFormatReceiveMessagesIncludesPreloadedThreadRoot(t *testing.T) { formatted := mcpformat.FormatReceiveMessages([]types.InboxMessage{ { MessageID: "00000014fdde160dc7325109eed9bf54", Target: "#agentteam:00000014", ChannelType: "thread", ChannelName: "chn_48CroYy2iZem", SenderName: "vwjokl4", Content: "@agentteam1-claude fix this", Timestamp: "2026-04-08T02:40:02+08:00", Metadata: map[string]any{ "replyThreadRoot": map[string]any{ "messageId": "49ac6c63bc45e2bf8e1e08868239e247", "senderType": "human", "senderName": "andc5vq", "messageType": "text", "content": "@agentteam1-claude 请基于 ~/agentteam 最新的 origin/main，定位 agent 看不到 thread 里根消息的问题", "timestamp": "2026-05-17T15:09:05+08:00", }, }, }, }, nil) require.Contains(t, formatted, "target: #agentteam:00000014") require.Contains(t, formatted, "messageId: 00000014fdde160dc7325109eed9bf54") require.Contains(t, formatted, "content:\n@agentteam1-claude fix this") require.Contains(t, formatted, "threadRoot:") require.Contains(t, formatted, "messageId: 49ac6c63bc45e2bf8e1e08868239e247") require.Contains(t, formatted, "from: @andc5vq") require.Contains(t, formatted, "@agentteam1-claude 请基于 ~/agentteam 最新的 origin/main，定位 agent 看不到 thread 里根消息的问题") require.NotContains(t, formatted, "contextHint:") } func TestFormatReceiveMessagesThreadMessageGuidesHistoryWhenUnclear(t *testing.T) { formatted := mcpformat.FormatReceiveMessages([]types.InboxMessage{ { MessageID: "00000014fdde160dc7325109eed9bf54", Target: "#agentteam:00000014", ChannelType: "thread", ChannelName: "chn_48CroYy2iZem", SenderName: "vwjokl4", Content: "@agentteam1-claude", Timestamp: "2026-04-08T02:40:02+08:00", }, }, nil) require.Contains(t, formatted, "Thread context guidance:") require.Contains(t, formatted, "If the message is ambiguous") require.Contains(t, formatted, `read recent thread history (target="#agentteam:00000014", limit=20)`) require.Contains(t, formatted, "If the message is already clear and actionable") } func TestFormatReceiveMessagesChannelMessageDoesNotRenderThreadContextGuidance(t *testing.T) { formatted := mcpformat.FormatReceiveMessages([]types.InboxMessage{ { MessageID: "00000014fdde160dc7325109eed9bf54", Target: "#agentteam", ChannelType: "channel", ChannelName: "agentteam", SenderName: "vwjokl4", Content: "@agentteam1-claude fix this", Timestamp: "2026-04-08T02:40:02+08:00", }, }, nil) require.NotContains(t, formatted, "Thread context guidance:") } func TestFormatReceiveMessagesSkipsThreadRootSectionWhenNotPreloaded(t *testing.T) { formatted := mcpformat.FormatReceiveMessages([]types.InboxMessage{ { MessageID: "00000014fdde160dc7325109eed9bf54", Target: "#agentteam:00000014", ChannelType: "thread", ChannelName: "chn_48CroYy2iZem", SenderName: "vwjokl4", Content: "@agentteam1-claude 请基于 ~/agentteam 最新的 origin/main，定位 agent 看不到 thread 里根消息的问题", Timestamp: "2026-04-08T02:40:02+08:00", }, }, nil) require.NotContains(t, formatted, "threadRoot:") require.NotContains(t, formatted, "contextHint:") } func TestFormatReceiveMessagesIncludesChannelDisplayName(t *testing.T) { formatted := mcpformat.FormatReceiveMessages([]types.InboxMessage{ { MessageID: "00000014fdde160dc7325109eed9bf54", Target: "#1", ChannelType: "channel", ChannelName: "1", ChannelDisplayName: "中文频道 1", SenderType: "human", SenderName: "alice", Content: "can you see the channel name?", Timestamp: "2026-04-30T15:28:00Z", }, }, nil) require.Contains(t, formatted, "target: #1") require.Contains(t, formatted, "channel: 1") require.Contains(t, formatted, "channelDisplayName: 中文频道 1") } func TestFormatReceiveMessagesIncludesTeamContexts(t *testing.T) { formatted := mcpformat.FormatReceiveMessages([]types.InboxMessage{ { MessageID: "msg_team_context", Target: "#agentteam", ChannelType: "channel", ChannelName: "agentteam", SenderType: "human", SenderName: "alice", Content: "@core-team please prepare a complete responsibility proposal for this Team and submit it for approval", Timestamp: "2026-06-18T19:00:00Z", TeamContexts: []types.TeamContext{{ TeamID: "team_core", TeamHandle: "core-team", TeamDisplayName: "Core Team", TeamDescription: "Act as core.\nKeep responses concise.", TeamMemberID: "wtm_agent", Role: "leader", RoleLabel: "Coordinator", RoleOverlayPrompt: "Build it.", MemberRoster: []types.TeamRosterMember{{ TeamMemberID: "wtm_agent", ActorKind: "agent", ActorID: "mem_agent", DisplayName: "AgentTeam Agent", Handle: "agentteam-agent", Role: "leader", RoleLabel: "Coordinator", RoleOverlayPrompt: "Build it.", WakeOnTeamMention: true, DeliveryKind: "wake", Self: true, }, { TeamMemberID: "wtm_reviewer", ActorKind: "agent", ActorID: "mem_reviewer", DisplayName: "Review Agent", Handle: "review-agent", Role: "member", RoleLabel: "Reviewer", RoleOverlayPrompt: "", WakeOnTeamMention: false, DeliveryKind: "agent", }}, }}, }, }, nil) require.Contains(t, formatted, "team contexts:") require.Contains(t, formatted, "- team: @core-team") require.Contains(t, formatted, "team id: team_core") require.Contains(t, formatted, "team invocation: you were awakened as this Team member. Other Team members may not have been awakened automatically.") require.NotContains(t, formatted, "collaboration rules:") require.NotContains(t, formatted, "team_invoke_team_id") require.NotContains(t, formatted, "workflow-plan-author Skill") require.Contains(t, formatted, "Team responsibility updates:") require.Contains(t, formatted, "with `agentteam exec team update`; it will request Human approval") require.Contains(t, formatted, "When replacing a rejected proposal, use a new requestId.") require.Contains(t, formatted, "team duty:\n Act as core.\n Keep responses concise.") require.NotContains(t, formatted, "description:") require.Contains(t, formatted, "your Team role: leader") require.Contains(t, formatted, "your role label: Coordinator") require.Contains(t, formatted, "your duty:\n Build it.") require.Contains(t, formatted, "team member roster:\n - @agentteam-agent (AgentTeam Agent) (agent, Coordinator, active participant, self)") require.Contains(t, formatted, " teamMemberId: wtm_agent") require.Contains(t, formatted, " Team role: leader") require.Contains(t, formatted, " duty:\n Build it.") require.Contains(t, formatted, " - @review-agent (Review Agent) (agent, Reviewer, as-needed participant)") require.Contains(t, formatted, " teamMemberId: wtm_reviewer") require.Contains(t, formatted, " Team role: member") require.Contains(t, formatted, " duty: not configured") } func TestFormatReceiveMessagesKeepsCompleteProposalIDsBeyondRosterLimit(t *testing.T) { const rosterLimit = 20 roster := make([]types.TeamRosterMember, rosterLimit+1) for index := range roster { roster[index] = types.TeamRosterMember{ TeamMemberID: fmt.Sprintf("wtm_%02d", index), ActorKind: "agent", ActorID: fmt.Sprintf("mem_%02d", index), DisplayName: fmt.Sprintf("Agent %02d", index), Handle: fmt.Sprintf("agent-%02d", index), Role: "member", RoleLabel: fmt.Sprintf("Role %02d", index), RoleOverlayPrompt: "Configured duty.", } } roster[0].Role = "leader" roster[1].RoleOverlayPrompt = "" roster[len(roster)-1].RoleOverlayPrompt = "Overflow duty must stay compact." format := func(role string) string { return mcpformat.FormatReceiveMessages([]types.InboxMessage{{ MessageID: "msg_large_team", Target: "#agentteam", Content: "Please prepare the Team responsibilities.", TeamContexts: []types.TeamContext{{ TeamID: "team_large", Role: role, MemberRoster: roster, }}, }}, nil) } leaderPrompt := format("leader") require.Contains(t, leaderPrompt, " - ... (1 more)") require.NotContains(t, leaderPrompt, "Overflow duty must stay compact.") require.Contains(t, leaderPrompt, "complete responsibility proposal member index (use every teamMemberId exactly once):") require.Contains(t, leaderPrompt, "- wtm_00: @agent-00 (Agent 00); Team role: leader; role label: Role 00") require.Contains(t, leaderPrompt, "- wtm_20: @agent-20 (Agent 20); Team role: member; role label: Role 20") require.NotContains(t, format("member"), "complete responsibility proposal member index") roster[1].RoleOverlayPrompt = "Configured duty." require.NotContains(t, format("leader"), "complete responsibility proposal member index") } func TestFormatReceiveMessagesUsesAgentHandleForPromptAuthor(t *testing.T) { formatted := mcpformat.FormatReceiveMessages([]types.InboxMessage{ { MessageID: "00000014fdde160dc7325109eed9bf54", Target: "#agentteam", ChannelType: "channel", ChannelName: "agentteam", SenderType: "agent", SenderName: "legacy-agent", AuthorProfile: types.AuthorProfile{ ActorType: "agent", PrincipalID: "agt_worker", MemberID: "mem_worker", TargetID: "hh-worker1", DisplayName: "研发", }, Content: "done", Timestamp: "2026-04-30T15:28:00Z", }, }, nil) require.Contains(t, formatted, "from: @hh-worker1") require.Contains(t, formatted, "fromDisplayName: 研发") require.NotContains(t, formatted, "from: @研发") } func TestFormatReceiveMessagesUsesWorkflowAuthor(t *testing.T) { msg := types.InboxMessage{ MessageID: "00000014fdde160dc7325109eed9bf55", Target: "#tsk_calculation", ChannelType: "task", ChannelName: "task-10", AuthorProfile: types.AuthorProfile{ ActorType: "human", PrincipalID: "usr_owner", MemberID: "mem_owner", DisplayName: "Arithmetic E2E Owner", }, Source: types.ContributionSource{ Kind: "workflow_run", RefType: "workflow_run", RefID: "wri_calculation", RunID: "wri_calculation", EventID: "wrf_calculation_effect", Label: "Parallel calculator", }, Content: "Compute 23 + 19.", Timestamp: "2026-07-31T15:30:00Z", } formatted := mcpformat.FormatReceiveMessages([]types.InboxMessage{msg}, nil) require.Contains(t, formatted, "from: @Parallel calculator") require.NotContains(t, formatted, "Arithmetic E2E Owner") // The resolved author is the Workflow service author, never the frozen // Human authorization/audit principal. require.Equal(t, "Parallel calculator", msg.AuthorIdentity().DisplayName) require.Equal(t, "service", msg.AuthorIdentity().ActorType) } func TestFormatReceiveMessagesFormatsReactionDelivery(t *testing.T) { formatted := mcpformat.FormatReceiveMessages([]types.InboxMessage{ { MessageID: "rxdlv_00000014fdde160dc7325109eed9bf54", Target: "#agentteam", ChannelType: "channel", ChannelName: "agentteam", SenderType: "human", SenderName: "alice", AuthorProfile: types.AuthorProfile{ ActorType: "human", PrincipalID: "usr_profile", MemberID: "mem_profile", DisplayName: "Profile Reactor", }, Content: "reacted 👍 to your message", Timestamp: "2026-04-15T10:00:00Z", MessageType: "reaction", Metadata: map[string]any{ "reaction": map[string]any{ "reactionId": "rxn_1234", "targetMessageId": "msg_1234", "targetMessagePreview": "Looks good to me", "emoji": "👍", }, }, }, }, nil) require.Contains(t, formatted, "[Reaction]") require.Contains(t, formatted, "from: @Profile Reactor") require.NotContains(t, formatted, "from: @alice") require.Contains(t, formatted, "reaction: 👍") require.Contains(t, formatted, "targetMessageId: msg_1234") require.Contains(t, formatted, "targetMessagePreview:\nLooks good to me") } func TestFormatReceiveMessagesIncludesTaskDiscussionDetails(t *testing.T) { formatted := mcpformat.FormatReceiveMessages([]types.InboxMessage{ { MessageID: "tsk_123456789", Target: "#agentteam", ChannelType: "channel", ChannelName: "agentteam", SenderType: "human", SenderName: "alice", Content: "Task discussion update for #t9", Timestamp: "2026-04-20T10:00:00Z", MessageType: "task_dispatch", Metadata: map[string]any{ "task": map[string]any{ "taskNumber": 9, "title": "Fix reply routing", "channelRef": "#agentteam", }, "taskDiscussion": map[string]any{ "body": "Please keep the reply in the task discussion.", "createdBy": map[string]any{ "name": "alice", }, }, }, }, }, nil) require.Contains(t, formatted, "task discussion: #t9 \"Fix reply routing\"") require.Contains(t, formatted, "channel: #agentteam") require.Contains(t, formatted, "comment by @alice:\nPlease keep the reply in the task discussion.") require.Contains(t, formatted, `reply: post a reply (target="#tsk_...", content="...") with the task scope target to continue the task discussion`) require.Contains(t, formatted, "active task target: task discussion replies are scope-native; post a reply with the exact task scope target when present, task_checkpoint for high-signal checkpoints, and an ordinary reply only when deliberately addressing a channel/DM/thread audience") } func TestFormatReceiveMessagesIncludesTaskDispatchAttributesWithoutDiscussion(t *testing.T) { formatted := mcpformat.FormatReceiveMessages([]types.InboxMessage{ { MessageID: "tsk_223456789", Target: "#agentteam", ChannelType: "channel", ChannelName: "agentteam", SenderType: "human", SenderName: "alice", Content: "Assigned #t10 Fill task SPEC", Timestamp: "2026-04-20T10:00:00Z", MessageType: "task_dispatch", Metadata: map[string]any{ "task": map[string]any{ "taskNumber": 10, "title": "Fill task SPEC", "channelRef": "#agentteam", "status": "todo", "kind": "feature", "priority": "high", "assigneeProfile": map[string]any{ "actorType": "human", "principalId": "user-1", "memberId": "member-1", "displayName": "Alice Owner", }, "suggestedToYou": true, "reviewer": map[string]any{"name": "legacy-reviewer"}, "description": "Update task structured attributes before implementation.", "acceptanceCriteria": []any{"Task attributes are visible in the delivery"}, "reviewChecklist": []any{"No full task discussion fetch is required"}, "workspaceRefs": []any{"specs/task-attributes.md"}, }, }, }, }, nil) require.Contains(t, formatted, "task dispatch: #t10 \"Fill task SPEC\"") require.Contains(t, formatted, "channel: #agentteam") require.Contains(t, formatted, "task attributes:\nstatus: todo\nkind: feature\npriority: high\nassignee: @Alice Owner\ndescription:\nUpdate task structured attributes before implementation.") require.NotContains(t, formatted, "suggestedToYou") require.NotContains(t, formatted, "legacy-reviewer") require.NotContains(t, formatted, "reviewer:") require.Contains(t, formatted, "acceptanceCriteria:\n- Task attributes are visible in the delivery") require.Contains(t, formatted, "reviewChecklist:\n- No full task discussion fetch is required") require.Contains(t, formatted, "workspaceRefs:\n- specs/task-attributes.md") require.Contains(t, formatted, "instruction: Treat these structured task attributes as requirements.") require.Contains(t, formatted, "active task target: task discussion replies are scope-native; post a reply with the exact task scope target when present, task_checkpoint for high-signal checkpoints, and an ordinary reply only when deliberately addressing a channel/DM/thread audience") require.NotContains(t, formatted, "comment by @") require.NotContains(t, formatted, "reply: use add_discussion_entry") } func TestFormatReceiveMessagesIncludesTaskScopeDiscussionReplyTarget(t *testing.T) { formatted := mcpformat.FormatReceiveMessages([]types.InboxMessage{ { MessageID: "msg_task_scope_123456789", Target: "#tsk_tk8msx2pmcjpw9ujlqn0", ChannelType: "task", ChannelName: "tsk_tk8msx2pmcjpw9ujlqn0", SenderType: "human", SenderName: "alice", Content: "test3", Timestamp: "2026-05-20T09:22:00Z", MessageType: "task_discussion", }, }, nil) require.Contains(t, formatted, "target: #tsk_tk8msx2pmcjpw9ujlqn0") require.Contains(t, formatted, "task discussion: scope-native message") require.Contains(t, formatted, "scope: #tsk_tk8msx2pmcjpw9ujlqn0") require.Contains(t, formatted, `reply: post a reply (target="#tsk_tk8msx2pmcjpw9ujlqn0", content="...") to continue the task discussion`) require.NotContains(t, formatted, "add_discussion_entry") } // --------------------------------------------------------------------------- // WebSocket send queue tests // --------------------------------------------------------------------------- func TestSendJSONReturnsErrorWhenQueueFull(t *testing.T) { ch := &Protocol{ id: "agentteam-test", sendCh: make(chan []byte, 1), } ch.sendCh <- []byte(`{"type":"occupied"}`) err := ch.sendJSON(map[string]string{"type": "ready"}) require.ErrorIs(t, err, errSendBufferFull) require.Equal(t, uint64(1), ch.metrics.Summary().SendDropped) } func TestTelemetryHeadroomPreservesWakeStatusCapacity(t *testing.T) { ch := &Protocol{ id: "agentteam-test", sendCh: make(chan []byte, 4), } for range telemetryQueueLimit(cap(ch.sendCh)) { require.NoError(t, ch.sendJSON(map[string]string{"type": "agent:status"})) } err := ch.sendJSONTelemetry(map[string]string{"type": "agent:trajectory"}) require.ErrorIs(t, err, errSendBufferFull) ch.Emit(types.WakeStatusEvent{ AgentID: "agt_1", MessageID: "msg_1", WakeToken: "wake_1", Status: consts.WakeStatusFailed, }) require.Len(t, ch.sendCh, 4) for range 3 { <-ch.sendCh } require.Equal(t, "agent:wake_status", readAgentTeamTestMessageType(t, <-ch.sendCh)) require.Equal(t, uint64(1), ch.metrics.Summary().SendDropped) } func TestSendJSONBlockingWaitsForQueueSpace(t *testing.T) { ch := &Protocol{ id: "agentteam-test", sendCh: make(chan []byte, 1), } ch.sendCh <- []byte(`{"type":"occupied"}`) ctx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel() done := make(chan error, 1) go func() { done <- ch.sendJSONBlocking(ctx, map[string]string{"type": "critical"}) }() require.Eventually(t, func() bool { return len(ch.sendCh) == cap(ch.sendCh) }, 100*time.Millisecond, 5*time.Millisecond) <-ch.sendCh require.NoError(t, <-done) raw := <-ch.sendCh require.JSONEq(t, `{"type":"critical"}`, string(raw)) } func TestSendJSONBlockingReturnsContextErrorWhenQueueStaysFull(t *testing.T) { ch := &Protocol{ id: "agentteam-test", sendCh: make(chan []byte, 1), } ch.sendCh <- []byte(`{"type":"occupied"}`) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) defer cancel() err := ch.sendJSONBlocking(ctx, map[string]string{"type": "critical"}) require.ErrorIs(t, err, context.DeadlineExceeded) require.Equal(t, uint64(1), ch.metrics.Summary().SendDropped) } func TestWorkspaceListReturnsSendErrorWhenQueueFull(t *testing.T) { ch := &Protocol{ id: "agentteam-test", sendCh: make(chan []byte, 1), conn: &websocket.Conn{}, workspaceListRequests: make(map[string]chan workspaceListResult), } ch.sendCh <- []byte(`{"type":"occupied"}`) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Millisecond) defer cancel() _, err := ch.WorkspaceList(ctx) require.ErrorIs(t, err, context.DeadlineExceeded) require.Contains(t, err.Error(), "request agentteam workspaces") } // --------------------------------------------------------------------------- // AgentTeamMeta MediaDecorator tests // --------------------------------------------------------------------------- func TestAgentTeamMeta_MediaAnnotation_NoAttachments(t *testing.T) { m := &types.AgentTeamMeta{} if got := m.MediaAnnotation(); got != "" { t.Errorf("expected empty MediaAnnotation, got %q", got) } } func TestAgentTeamMeta_MediaAnnotation_WithNames(t *testing.T) { m := &types.AgentTeamMeta{ AttachmentNames: "image.png, doc.pdf", AttachmentIDs: "id1, id2", } got := m.MediaAnnotation() if got == "" { t.Fatal("expected non-empty MediaAnnotation") } // Should contain the filenames. for _, name := range []string{"image.png", "doc.pdf"} { if !contains(got, name) { t.Errorf("MediaAnnotation should contain %q, got %q", name, got) } } // Should contain the file view hint. if !contains(got, "attachment-view tool") { t.Errorf("MediaAnnotation should mention file view command, got %q", got) } } func TestAgentTeamMeta_MediaAnnotation_OnlyIDs(t *testing.T) { m := &types.AgentTeamMeta{ AttachmentIDs: "att_abc123", } got := m.MediaAnnotation() if got == "" { t.Fatal("expected non-empty MediaAnnotation when only IDs set") } if !contains(got, "att_abc123") { t.Errorf("MediaAnnotation should contain attachment ID, got %q", got) } } func TestAgentTeamMeta_ImplementsMediaDecorator(t *testing.T) { var _ types.MediaDecorator = &types.AgentTeamMeta{} } func TestAgentTeamMeta_ImplementsSessionConfigurator(t *testing.T) { var _ types.SessionConfigurator = &types.AgentTeamMeta{} } func TestAgentTeamMeta_ImplementsMetadata(t *testing.T) { var _ types.Metadata = &types.AgentTeamMeta{} } func contains(s, substr string) bool { return len(s) > 0 && len(substr) > 0 && (s == substr || len(s) >= len(substr) && containsSubstring(s, substr)) } func containsSubstring(s, sub string) bool { for i := 0; i+len(sub) <= len(s); i++ { if s[i:i+len(sub)] == sub { return true } } return false } // --------------------------------------------------------------------------- // DebugToggler tests // --------------------------------------------------------------------------- func TestProtocol_ImplementsDebugToggler(t *testing.T) { var _ types.DebugToggler = &Protocol{} } func TestToggleDebug(t *testing.T) { ch := &Protocol{ agents: make(map[string]*agentState), dedupRing: newDedupRing(10), sendCh: make(chan []byte, 1), } if ch.debug { t.Fatal("expected debug=false initially") } result := ch.ToggleDebug() if !result { t.Fatal("expected ToggleDebug to return true") } if !ch.debug { t.Fatal("expected debug=true after toggle") } result = ch.ToggleDebug() if result { t.Fatal("expected ToggleDebug to return false on second toggle") } } type roundTripFunc func(*http.Request) (*http.Response, error) func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) { return f(req) } func jsonRoundTripResponse(req *http.Request, body string) *http.Response { return &http.Response{ StatusCode: http.StatusOK, Header: http.Header{"Content-Type": []string{"application/json"}}, Body: io.NopCloser(strings.NewReader(body)), Request: req, } } func TestChannelSendMediaUsesScopeIDForUpload(t *testing.T) { uploadScopeID := make(chan string, 2) var resolveCalls atomic.Int32 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") switch r.URL.Path { case "/api/v1/workspaces/wsp_1/agents/server-agent/runtime/resolve-channel": resolveCalls.Add(1) require.Equal(t, http.MethodPost, r.Method) fmt.Fprint(w, `{"scopeId":"scp_ops"}`) case "/api/v1/workspaces/wsp_1/agents/server-agent/runtime/upload": require.Equal(t, http.MethodPost, r.Method) require.NoError(t, r.ParseMultipartForm(1<<20)) uploadScopeID <- r.FormValue("scopeId") require.Empty(t, r.FormValue(deprecatedUploadScopeAliasFormKeyForNegativeTest())) file, _, err := r.FormFile("file") require.NoError(t, err) require.NoError(t, file.Close()) fmt.Fprint(w, `{"id":"att_1"}`) case "/api/v1/workspaces/wsp_1/agents/server-agent/runtime/send": var body struct { AttachmentIDs []string `json:"attachmentIds"` } require.NoError(t, json.NewDecoder(r.Body).Decode(&body)) require.Equal(t, []string{"att_1"}, body.AttachmentIDs) fmt.Fprint(w, `{"messageId":"msg_1"}`) default: t.Fatalf("unexpected path %s", r.URL.Path) } })) defer srv.Close() filePath := filepath.Join(t.TempDir(), "pixel.png") require.NoError(t, os.WriteFile(filePath, []byte("png"), 0o644)) channel := &Protocol{ serverURL: srv.URL, workspaceID: "wsp_1", agentID: "server-agent", authToken: "tok", httpClient: srv.Client(), } require.NoError(t, channel.SendMedia(context.Background(), "server-agent", "#ops", filePath, "caption")) require.NoError(t, channel.SendMedia(context.Background(), "server-agent", "#ops", filePath, "caption")) require.Equal(t, "scp_ops", <-uploadScopeID) require.Equal(t, "scp_ops", <-uploadScopeID) require.Equal(t, int32(1), resolveCalls.Load(), "temporary API clients must share the Protocol resolver") } func TestChannelSendMediaInvalidatesStaleScopeWithoutReplayingUpload(t *testing.T) { var resolveCalls atomic.Int32 var uploadCalls atomic.Int32 var sendCalls atomic.Int32 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") switch r.URL.Path { case "/api/v1/workspaces/wsp_1/agents/server-agent/runtime/resolve-channel": scopeID := fmt.Sprintf("scp_%d", resolveCalls.Add(1)) _, _ = fmt.Fprintf(w, `{"scopeId":%q}`, scopeID) case "/api/v1/workspaces/wsp_1/agents/server-agent/runtime/upload": call := uploadCalls.Add(1) require.NoError(t, r.ParseMultipartForm(1<<20)) require.Equal(t, fmt.Sprintf("scp_%d", call), r.FormValue("scopeId")) if call == 1 { w.WriteHeader(http.StatusNotFound) _, _ = io.WriteString(w, `{"error":"scope not found"}`) return } _, _ = io.WriteString(w, `{"id":"att_1"}`) case "/api/v1/workspaces/wsp_1/agents/server-agent/runtime/send": sendCalls.Add(1) _, _ = io.WriteString(w, `{"messageId":"msg_1"}`) default: t.Fatalf("unexpected path %s", r.URL.Path) } })) defer srv.Close() filePath := filepath.Join(t.TempDir(), "pixel.png") require.NoError(t, os.WriteFile(filePath, []byte("png"), 0o644)) channel := &Protocol{ serverURL: srv.URL, workspaceID: "wsp_1", agentID: "server-agent", authToken: "tok", httpClient: srv.Client(), } err := channel.SendMedia(context.Background(), "server-agent", "#ops", filePath, "caption") require.Error(t, err) require.Equal(t, int32(1), uploadCalls.Load(), "failed upload must not be replayed") require.Equal(t, int32(0), sendCalls.Load()) require.NoError(t, channel.SendMedia(context.Background(), "server-agent", "#ops", filePath, "caption")) require.Equal(t, int32(2), resolveCalls.Load(), "404 must evict the stale mapping") require.Equal(t, int32(2), uploadCalls.Load()) require.Equal(t, int32(1), sendCalls.Load()) } func TestOutboundAgentBearerPathsRefreshOnUnauthorized(t *testing.T) { var refreshCalls int32 var mu sync.Mutex pathCalls := make(map[string]int) srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { switch r.URL.Path { case "/api/v1/workspaces/wsp_1/daemon/agents/server-agent/auth-token": require.Equal(t, "Bearer machine-key", r.Header.Get("Authorization")) atomic.AddInt32(&refreshCalls, 1) w.Header().Set("Content-Type", "application/json") _, _ = io.WriteString(w, `{"authToken":"fresh-agent-token"}`) return } mu.Lock() pathCalls[r.URL.Path]++ attempt := pathCalls[r.URL.Path] mu.Unlock() if attempt == 1 { require.Equal(t, "Bearer stale-agent-token", r.Header.Get("Authorization")) w.WriteHeader(http.StatusUnauthorized) _, _ = io.WriteString(w, `{"error":"Unauthorized"}`) return } require.Equal(t, "Bearer fresh-agent-token", r.Header.Get("Authorization")) w.Header().Set("Content-Type", "application/json") switch r.URL.Path { case "/api/v1/workspaces/wsp_1/agents/server-agent/runtime/send": _, _ = io.WriteString(w, `{"messageId":"msg_1"}`) case "/api/v1/workspaces/wsp_1/agents/server-agent/runtime/resolve-channel": _, _ = io.WriteString(w, `{"scopeId":"scp_ops"}`) case "/api/v1/workspaces/wsp_1/agents/server-agent/runtime/upload": require.NoError(t, r.ParseMultipartForm(1<<20)) require.Equal(t, "scp_ops", r.FormValue("scopeId")) file, _, err := r.FormFile("file") require.NoError(t, err) require.NoError(t, file.Close()) _, _ = io.WriteString(w, `{"id":"att_1"}`) default: t.Fatalf("unexpected path %s", r.URL.Path) } })) defer srv.Close() channel := &Protocol{ serverURL: srv.URL, apiKey: "machine-key", workspaceID: "wsp_1", httpClient: srv.Client(), agents: map[string]*agentState{"server-agent": {Token: "stale-agent-token"}}, agentTokenRefresh: map[string]*agentTokenRefreshState{}, } resetStaleToken := func() { channel.mu.Lock() channel.agents["server-agent"].Token = "stale-agent-token" channel.mu.Unlock() } require.NoError(t, channel.doSend( context.Background(), "server-agent", "#ops", "hello", nil, )) resetStaleToken() resolved, err := channel.resolveChannel(context.Background(), "server-agent", "#ops") require.NoError(t, err) require.Equal(t, "scp_ops", resolved.ScopeID) resetStaleToken() filePath := filepath.Join(t.TempDir(), "pixel.png") require.NoError(t, os.WriteFile(filePath, []byte("png"), 0o644)) attachmentID, err := channel.uploadFile( context.Background(), "server-agent", filePath, "scp_ops", ) require.NoError(t, err) require.Equal(t, "att_1", attachmentID) require.Equal(t, int32(3), atomic.LoadInt32(&refreshCalls)) for _, path := range []string{ "/api/v1/workspaces/wsp_1/agents/server-agent/runtime/send", "/api/v1/workspaces/wsp_1/agents/server-agent/runtime/resolve-channel", "/api/v1/workspaces/wsp_1/agents/server-agent/runtime/upload", } { require.Equal(t, 2, pathCalls[path], path) } } func TestDetectRuntimesRequiresCodexAppServerCapability(t *testing.T) { tmpDir := t.TempDir() codexPath := filepath.Join(tmpDir, "codex") require.NoError(t, os.WriteFile(codexPath, []byte("#!/bin/sh\nexit 0\n"), 0755)) t.Setenv("PATH", tmpDir) oldProbe := runtimeconfig.CodexAppServerAvailable t.Cleanup(func() { runtimeconfig.CodexAppServerAvailable = oldProbe }) runtimeconfig.CodexAppServerAvailable = func(string) bool { return false } runtimes := detectRuntimes() require.NotContains(t, runtimes, "codex") require.NotContains(t, runtimes, "codex-app-server") } func TestDetectRuntimesReportsBuiltinAndExternalOpenCodeSeparately(t *testing.T) { managedDir := t.TempDir() managedPath := filepath.Join(managedDir, "opencode") require.NoError(t, os.WriteFile(managedPath, []byte("#!/bin/sh\nexit 0\n"), 0o755)) require.NoError(t, os.WriteFile(filepath.Join(managedDir, "models.json"), []byte(`{"local":{"id":"local","name":"Local","env":[],"models":{}}}`+"\n"), 0o644)) t.Setenv("AGENTTEAM_MANAGED_OPENCODE_PATH", managedPath) oldManagedVersion := runtimeconfig.ManagedOpenCodeVersion runtimeconfig.ManagedOpenCodeVersion = "" t.Cleanup(func() { runtimeconfig.ManagedOpenCodeVersion = oldManagedVersion }) externalDir := t.TempDir() externalPath := filepath.Join(externalDir, "opencode") require.NoError(t, os.WriteFile(externalPath, []byte("#!/bin/sh\nexit 0\n"), 0o755)) t.Setenv("PATH", externalDir) runtimes := detectRuntimes() require.Contains(t, runtimes, "builtin") require.Contains(t, runtimes, "opencode") } func TestReadyMessage(t *testing.T) { // The ready message should contain all supported capability strings. expected := []string{ "agent:start", "agent:stop", "agent:interrupt", "agent:deliver", "agent:sleep", "agent:reset-workspace", "agent:permission-mode", "agent:maas-direct", "agent:runtime-session-authority", "agent:local-session-resume", "workspace:files", "workspace:target-path", "recovery:restore", "recovery:gc", "wake:status", daemonTransportCapabilityV1, daemonTransportElasticRequestCapability, daemonEndpointsCapabilityV1, } t.Setenv("PATH", t.TempDir()) // Create a minimal channel and capture what onConnect sends. ch := &Protocol{ id: "test", agents: make(map[string]*agentState), dedupRing: newDedupRing(10), sendCh: make(chan []byte, 16), serverURL: "http://localhost", apiKey: "testkey", } // Call onConnect (it will try to send to sendCh). ctx, cancel := context.WithCancel(context.Background()) cancel() // cancel immediately so the ticker goroutine exits quickly ch.onConnect(ctx) // Read the first message from sendCh (should be the ready message). select { case data := <-ch.sendCh: var msg map[string]json.RawMessage if err := json.Unmarshal(data, &msg); err != nil { t.Fatalf("failed to parse ready message: %v", err) } var msgType string if err := json.Unmarshal(msg["type"], &msgType); err != nil { t.Fatalf("failed to parse ready message type: %v", err) } if msgType != "ready" { t.Fatalf("expected type=ready, got %q", msgType) } var capabilities []string if err := json.Unmarshal(msg["capabilities"], &capabilities); err != nil { t.Fatalf("failed to parse capabilities: %v", err) } var runtimeDefaults map[string]profile.RuntimeDefaultSettings if err := json.Unmarshal(msg["runtimeDefaults"], &runtimeDefaults); err != nil { t.Fatalf("failed to parse runtime defaults: %v", err) } var tokenRefreshProof string if err := json.Unmarshal(msg["tokenRefreshProof"], &tokenRefreshProof); err != nil { t.Fatalf("failed to parse token refresh proof: %v", err) } if tokenRefreshProof == "" { t.Fatal("expected token refresh proof") } for _, removedKey := range []string{"message" + "Protocol", "message" + "ProtocolVersion"} { if _, exists := msg[removedKey]; exists { t.Fatalf("ready payload must not include %s", removedKey) } } capSet := make(map[string]bool) for _, c := range capabilities { capSet[c] = true } for _, exp := range expected { if !capSet[exp] { t.Errorf("missing expected capability %q in %v", exp, capabilities) } } if runtimeDefaults == nil { t.Fatal("expected runtimeDefaults object, got null") } var transport daemonTransportReadyMsg require.NoError(t, json.Unmarshal(msg["daemonTransport"], &transport)) require.Equal(t, daemonTransportSchemaVersion, transport.SchemaVersion) require.NotEmpty(t, transport.SessionGeneration) require.Equal(t, defaultEndpointRegistryVersion, transport.EndpointRegistryVersion) require.Equal(t, []daemonTransportEndpointRef{ {EndpointID: remoteEndpointMachineDashboard, Version: remoteEndpointMachineDashboardVersion}, {EndpointID: remoteEndpointMachineDirectoryList, Version: 1}, {EndpointID: remoteEndpointMachineMCPServerDelete, Version: 1}, {EndpointID: remoteEndpointMachineMCPServerList, Version: 1}, {EndpointID: remoteEndpointMachineMCPServerTest, Version: 1}, {EndpointID: remoteEndpointMachineMCPServerUpsert, Version: 1}, {EndpointID: remoteEndpointMachineRuntimesBundle, Version: 1}, {EndpointID: remoteEndpointMachineRuntimeConfigRead, Version: 1}, {EndpointID: remoteEndpointMachineRuntimeConfigUpdate, Version: 1}, {EndpointID: remoteEndpointMachineRuntimeConfigValidate, Version: 1}, {EndpointID: remoteEndpointMachineRuntimesDetect, Version: 1}, {EndpointID: remoteEndpointMachineSettings, Version: 1}, {EndpointID: remoteEndpointMachineSettingsUpdate, Version: 1}, {EndpointID: remoteEndpointMachineWorkspaceDelete, Version: 1}, {EndpointID: remoteEndpointMachineWorkspaceScan, Version: 1}, {EndpointID: remoteEndpointOutputTargetValidate, Version: 1}, {EndpointID: remoteEndpointReviewDiff, Version: 1}, {EndpointID: remoteEndpointReviewDiffBatch, Version: 1}, {EndpointID: remoteEndpointReviewFiles, Version: 1}, {EndpointID: remoteEndpointReviewFingerprint, Version: 1}, {EndpointID: remoteEndpointReviewSnapshot, Version: 1}, {EndpointID: remoteEndpointReviewSummary, Version: 1}, {EndpointID: remoteEndpointRuntimeMemorySyncRetry, Version: 1}, {EndpointID: remoteEndpointRuntimeMemorySyncStatus, Version: 1}, {EndpointID: remoteEndpointRuntimeRestoreRecoveryPoint, Version: 1}, {EndpointID: remoteEndpointRuntimeTrajectoryHistory, Version: 1}, {EndpointID: remoteEndpointWorkspaceApplySeedFiles, Version: 1}, {EndpointID: remoteEndpointWorkspaceDelete, Version: 1}, {EndpointID: remoteEndpointWorkspaceList, Version: 1}, {EndpointID: remoteEndpointWorkspaceRead, Version: 1}, }, transport.SupportedEndpoints) require.Contains(t, transport.SupportedLinkModes, "session") require.Contains(t, transport.SupportedLinkModes, "elastic_request") default: t.Fatal("no message sent to sendCh") } } func TestReadyMessageReusesTokenRefreshProof(t *testing.T) { t.Setenv("PATH", t.TempDir()) ch := &Protocol{id: "test"} first, _ := ch.readyMessage() second, _ := ch.readyMessage() firstProof, ok := first["tokenRefreshProof"].(string) require.True(t, ok) require.NotEmpty(t, firstProof) require.Equal(t, firstProof, second["tokenRefreshProof"]) require.Equal(t, firstProof, ch.currentTokenRefreshProof()) } func TestOnConnectReemitsRunningAgentTeamSessionTelemetry(t *testing.T) { t.Setenv("PATH", t.TempDir()) disp := &agentteamTestDispatcher{ sessions: []types.SessionSnapshot{{ AgentID: "local-agent", ServerAgentID: "server-agent", ChannelID: "test", Status: "running", RuntimeSessionID: "runtime-session-1", Activity: "thinking", ActivityDetail: "Investigating", }}, } ch := &Protocol{ id: "test", agents: make(map[string]*agentState), dedupRing: newDedupRing(10), sendCh: make(chan []byte, 16), serverURL: "http://localhost", apiKey: "testkey", disp: disp, } ctx, cancel := context.WithCancel(context.Background()) cancel() ch.onConnect(ctx) var ready map[string]any require.NoError(t, json.Unmarshal(<-ch.sendCh, &ready)) require.Equal(t, "ready", ready["type"]) require.Equal(t, []any{"server-agent"}, ready["runningAgents"]) require.Equal(t, map[string]any{"server-agent": "runtime-session-1"}, ready["runningAgentSessions"]) require.Equal(t, "machine:agent:list", readAgentTeamTestMessageType(t, <-ch.sendCh)) var session map[string]any require.NoError(t, json.Unmarshal(<-ch.sendCh, &session)) require.Equal(t, "agent:session", session["type"]) require.Equal(t, "server-agent", session["agentId"]) require.Equal(t, "runtime-session-1", session["sessionId"]) var status map[string]any require.NoError(t, json.Unmarshal(<-ch.sendCh, &status)) require.Equal(t, "agent:status", status["type"]) require.Equal(t, "server-agent", status["agentId"]) require.Equal(t, "active", status["status"]) var activity map[string]any require.NoError(t, json.Unmarshal(<-ch.sendCh, &activity)) require.Equal(t, "agent:activity", activity["type"]) require.Equal(t, "server-agent", activity["agentId"]) require.Equal(t, "thinking", activity["activity"]) require.Equal(t, "Investigating", activity["detail"]) } func TestOnConnectDoesNotEmitRunningSessionBeforeRuntimeSessionKnown(t *testing.T) { t.Setenv("PATH", t.TempDir()) disp := &agentteamTestDispatcher{ sessions: []types.SessionSnapshot{{ AgentID: "local-agent", ServerAgentID: "server-agent", ChannelID: "test", Status: "running", Activity: "thinking", ActivityDetail: "Investigating", }}, } ch := &Protocol{ id: "test", agents: make(map[string]*agentState), dedupRing: newDedupRing(10), sendCh: make(chan []byte, 16), serverURL: "http://localhost", apiKey: "testkey", disp: disp, } ctx, cancel := context.WithCancel(context.Background()) cancel() ch.onConnect(ctx) var ready map[string]any require.NoError(t, json.Unmarshal(<-ch.sendCh, &ready)) require.Equal(t, "ready", ready["type"]) require.Equal(t, []any{"server-agent"}, ready["runningAgents"]) require.NotContains(t, ready, "runningAgentSessions") require.Equal(t, "machine:agent:list", readAgentTeamTestMessageType(t, <-ch.sendCh)) require.Equal(t, "agent:status", readAgentTeamTestMessageType(t, <-ch.sendCh)) require.Equal(t, "agent:activity", readAgentTeamTestMessageType(t, <-ch.sendCh)) select { case payload := <-ch.sendCh: t.Fatalf("unexpected extra reconnect telemetry before session publish: %s", string(payload)) default: } } func readAgentTeamTestMessageType(t *testing.T, data []byte) string { t.Helper() var msg map[string]json.RawMessage require.NoError(t, json.Unmarshal(data, &msg)) var msgType string require.NoError(t, json.Unmarshal(msg["type"], &msgType)) return msgType } func TestResendMachineCapabilitiesSendsReadyOnly(t *testing.T) { t.Setenv("PATH", t.TempDir()) ch := &Protocol{ id: "test", agents: make(map[string]*agentState), dedupRing: newDedupRing(10), sendCh: make(chan []byte, 16), serverURL: "http://localhost", apiKey: "testkey", } ch.ResendMachineCapabilities() select { case data := <-ch.sendCh: var msg map[string]json.RawMessage require.NoError(t, json.Unmarshal(data, &msg)) var msgType string require.NoError(t, json.Unmarshal(msg["type"], &msgType)) require.Equal(t, "ready", msgType) if _, exists := msg["runtimes"]; !exists { t.Fatal("ready payload missing runtimes") } case <-time.After(time.Second): t.Fatal("no ready message sent") } select { case data := <-ch.sendCh: t.Fatalf("unexpected extra message: %s", string(data)) default: } } func TestStartAgentSyncTickerCancelsPreviousTicker(t *testing.T) { ch := &Protocol{ id: "test", sendCh: make(chan []byte, 16), } ctx, cancel := context.WithCancel(context.Background()) defer cancel() ch.startAgentSyncTicker(ctx) ch.mu.Lock() first := ch.syncTicker ch.mu.Unlock() require.NotNil(t, first) ch.startAgentSyncTicker(ctx) ch.mu.Lock() second := ch.syncTicker ch.mu.Unlock() require.NotNil(t, second) require.NotSame(t, first, second) select { case <-first.done: case <-time.After(time.Second): t.Fatal("first agent sync ticker did not stop after reconnect") } cancel() select { case <-second.done: case <-time.After(time.Second): t.Fatal("second agent sync ticker did not stop after channel context cancellation") } } func TestRequestAgentListSyncCoalescesConcurrentWaiters(t *testing.T) { ch := &Protocol{ id: "agentteam-test", sendCh: make(chan []byte, 16), } const waiterCount = 5 start := make(chan struct{}) results := make(chan bool, waiterCount) for range waiterCount { go func() { <-start results <- ch.requestAgentListSync(time.Second) }() } close(start) require.Eventually(t, func() bool { ch.mu.Lock() defer ch.mu.Unlock() return ch.agentListSyncInFlight && len(ch.agentListSyncWaiters) == waiterCount }, time.Second, 5*time.Millisecond) require.Eventually(t, func() bool { return len(ch.sendCh) == 1 }, time.Second, 5*time.Millisecond) raw := <-ch.sendCh require.JSONEq(t, `{"type":"machine:agent:list"}`, string(raw)) ch.completeAgentListSync(true) for range waiterCount { require.True(t, <-results) } } func TestRequestAgentListSyncTimeoutClearsWaiter(t *testing.T) { ch := &Protocol{ id: "agentteam-test", sendCh: make(chan []byte, 1), } ok := ch.requestAgentListSync(10 * time.Millisecond) require.False(t, ok) ch.mu.Lock() defer ch.mu.Unlock() require.False(t, ch.agentListSyncInFlight) require.Empty(t, ch.agentListSyncWaiters) } // --------------------------------------------------------------------------- // Interface compliance tests // --------------------------------------------------------------------------- func TestProtocol_ImplementsChannel(t *testing.T) { var _ types.ServerConnection = &Protocol{} } func TestProtocol_ImplementsEventReporter(t *testing.T) { var _ types.EventReporter = &Protocol{} } // --------------------------------------------------------------------------- // Workspace scan tests // --------------------------------------------------------------------------- func TestScanDirOneLevel_IncludesHiddenFiles(t *testing.T) { dir := t.TempDir() // Create a hidden file and a regular file. require.NoError(t, os.WriteFile(filepath.Join(dir, ".hidden"), []byte("abc"), 0644)) require.NoError(t, os.WriteFile(filepath.Join(dir, "visible.txt"), []byte("12345"), 0644)) _, totalSize, fileCount := workspacefs.ScanDirOneLevel(dir) if fileCount != 2 { t.Errorf("expected fileCount=2 (hidden+visible), got %d", fileCount) } if totalSize != 8 { // 3 + 5 t.Errorf("expected totalSize=8, got %d", totalSize) } } func TestScanDirOneLevel_DirMtimeIncluded(t *testing.T) { dir := t.TempDir() // Create a regular file with an old mtime. filePath := filepath.Join(dir, "old.txt") require.NoError(t, os.WriteFile(filePath, []byte("x"), 0644)) oldTime := time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC) require.NoError(t, os.Chtimes(filePath, oldTime, oldTime)) // Create a subdirectory with a newer mtime. subDir := filepath.Join(dir, "subdir") require.NoError(t, os.Mkdir(subDir, 0755)) newTime := time.Date(2025, 6, 15, 0, 0, 0, 0, time.UTC) require.NoError(t, os.Chtimes(subDir, newTime, newTime)) lastMod, totalSize, fileCount := workspacefs.ScanDirOneLevel(dir) // fileCount and totalSize should exclude the directory. if fileCount != 1 { t.Errorf("expected fileCount=1 (only the file, not the dir), got %d", fileCount) } if totalSize != 1 { t.Errorf("expected totalSize=1, got %d", totalSize) } // lastMod should reflect the directory's mtime (the newer one). if !lastMod.After(oldTime) { t.Errorf("expected lastMod to be after oldTime; lastMod=%v, oldTime=%v", lastMod, oldTime) } } func TestScanDirOneLevel_EmptyDir(t *testing.T) { dir := t.TempDir() lastMod, totalSize, fileCount := workspacefs.ScanDirOneLevel(dir) if fileCount != 0 || totalSize != 0 { t.Errorf("expected empty stats, got fileCount=%d totalSize=%d", fileCount, totalSize) } if !lastMod.IsZero() { t.Errorf("expected zero lastMod for empty dir, got %v", lastMod) } } // --------------------------------------------------------------------------- // Metrics tests // --------------------------------------------------------------------------- func TestMetrics_Summary_ZeroValue(t *testing.T) { var m metricsCollector s := m.Summary() if s.Reconnects != 0 || s.DedupHits != 0 || s.SendSuccess != 0 || s.SendErrors != 0 || s.SendDropped != 0 || s.APIRetries != 0 || s.AgentStarts != 0 || s.AgentStops != 0 || s.AgentSleeps != 0 || s.AvgSendLatencyMs != 0 || s.ConnectedSince != 0 { t.Fatalf("expected all zeros in default summary, got %+v", s) } } func TestMetrics_Counters(t *testing.T) { var m metricsCollector m.reconnects.Add(3) m.dedupHits.Add(10) m.sendSuccess.Add(5) m.sendErrors.Add(2) m.sendDropped.Add(1) m.apiRetries.Add(4) m.agentStarts.Add(7) m.agentStops.Add(6) m.agentSleeps.Add(8) m.connectedAt.Store(1000) s := m.Summary() if s.Reconnects != 3 { t.Errorf("reconnects: got %d, want 3", s.Reconnects) } if s.DedupHits != 10 { t.Errorf("dedupHits: got %d, want 10", s.DedupHits) } if s.SendSuccess != 5 { t.Errorf("sendSuccess: got %d, want 5", s.SendSuccess) } if s.SendErrors != 2 { t.Errorf("sendErrors: got %d, want 2", s.SendErrors) } if s.SendDropped != 1 { t.Errorf("sendDropped: got %d, want 1", s.SendDropped) } if s.APIRetries != 4 { t.Errorf("apiRetries: got %d, want 4", s.APIRetries) } if s.AgentStarts != 7 { t.Errorf("agentStarts: got %d, want 7", s.AgentStarts) } if s.AgentStops != 6 { t.Errorf("agentStops: got %d, want 6", s.AgentStops) } if s.AgentSleeps != 8 { t.Errorf("agentSleeps: got %d, want 8", s.AgentSleeps) } if s.ConnectedSince != 1000 { t.Errorf("connectedSince: got %d, want 1000", s.ConnectedSince) } } func TestMetrics_SendLatency(t *testing.T) { var m metricsCollector m.recordSendLatency(100 * time.Millisecond) m.recordSendLatency(200 * time.Millisecond) s := m.Summary() // Average should be 150ms. if s.AvgSendLatencyMs < 149 || s.AvgSendLatencyMs > 151 { t.Errorf("avgSendLatencyMs: got %f, want ~150", s.AvgSendLatencyMs) } } func renderedPromptPayloadMarkdown(rendered promptcompiler.RenderedPromptPayload) string { parts := make([]string, 0, len(rendered.RenderOrder)) for _, slot := range rendered.RenderOrder { value, ok := rendered.Slots[slot] if !ok { continue } body := "" if text, ok := value.(string); ok { body = text } else { data, err := json.MarshalIndent(value, "", " ") if err != nil { continue } body = string(data) } parts = append(parts, "## "+slot+"\n\n"+strings.TrimRight(body, "\n")) } if len(parts) == 0 { return "" } return strings.Join(parts, "\n\n") + "\n" } func assertAgentTeamGoldenText(t *testing.T, name, got string) { t.Helper() path := filepath.Join("testdata", name) if os.Getenv("UPDATE_GOLDEN") != "" { require.NoError(t, os.MkdirAll(filepath.Dir(path), 0o755)) require.NoError(t, os.WriteFile(path, []byte(got), 0o644)) } want, err := os.ReadFile(path) require.NoError(t, err) require.Equal(t, string(want), got) } func TestWorkspaceListRequestsOverWebSocket(t *testing.T) { t.Parallel() upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { conn, err := upgrader.Upgrade(w, r, nil) require.NoError(t, err) defer conn.Close() var request struct { Type string `json:"type"` RequestID string `json:"requestId"` } require.NoError(t, conn.ReadJSON(&request)) require.Equal(t, "machine:workspaces:list", request.Type) require.NotEmpty(t, request.RequestID) require.NoError(t, conn.WriteJSON(map[string]any{ "type": "machine:workspaces:list:response", "requestId": request.RequestID, "success": true, "workspaces": []map[string]string{ {"id": "wsp_b", "name": "Beta", "slug": "beta"}, {"id": "wsp_a", "name": "Alpha", "slug": "alpha"}, }, })) })) t.Cleanup(server.Close) conn, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(server.URL, "http"), nil) require.NoError(t, err) ch := &Protocol{ id: "agentteam-test", sendCh: make(chan []byte, sendBufSize), agents: make(map[string]*agentState), dedupRing: newDedupRing(dedupSetCap), workspaceListRequests: make(map[string]chan workspaceListResult), conn: conn, disp: &agentteamTestDispatcher{}, } pumpCtx, cancelPumps := context.WithCancel(context.Background()) t.Cleanup(cancelPumps) go ch.runPumps(pumpCtx) requestCtx, cancelRequest := context.WithTimeout(context.Background(), time.Second) defer cancelRequest() workspaces, err := ch.WorkspaceList(requestCtx) require.NoError(t, err) require.Equal(t, []types.AgentTeamWorkspaceSummary{ {ID: "wsp_b", Name: "Beta", Slug: "beta"}, {ID: "wsp_a", Name: "Alpha", Slug: "alpha"}, }, workspaces) } func TestWorkspaceListWrapsTimeout(t *testing.T) { t.Parallel() upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }} server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { conn, err := upgrader.Upgrade(w, r, nil) require.NoError(t, err) defer conn.Close() var request struct { Type string `json:"type"` RequestID string `json:"requestId"` } require.NoError(t, conn.ReadJSON(&request)) require.Equal(t, "machine:workspaces:list", request.Type) require.NotEmpty(t, request.RequestID) <-r.Context().Done() })) t.Cleanup(server.Close) conn, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(server.URL, "http"), nil) require.NoError(t, err) ch := &Protocol{ id: "agentteam-test", sendCh: make(chan []byte, sendBufSize), agents: make(map[string]*agentState), dedupRing: newDedupRing(dedupSetCap), workspaceListRequests: make(map[string]chan workspaceListResult), conn: conn, disp: &agentteamTestDispatcher{}, } pumpCtx, cancelPumps := context.WithCancel(context.Background()) t.Cleanup(cancelPumps) go ch.runPumps(pumpCtx) requestCtx, cancelRequest := context.WithTimeout(context.Background(), 50*time.Millisecond) defer cancelRequest() _, err = ch.WorkspaceList(requestCtx) require.Error(t, err) require.ErrorIs(t, err, types.ErrAgentTeamWorkspaceListTimeout) require.ErrorIs(t, err, context.DeadlineExceeded) } func TestHandleWorkspaceListResponseReturnsServerError(t *testing.T) { t.Parallel() resultCh := make(chan workspaceListResult, 1) ch := &Protocol{ id: "agentteam-test", workspaceListRequests: map[string]chan workspaceListResult{ "req_1": resultCh, }, } ch.handleWorkspaceListResponse([]byte(`{ "type":"machine:workspaces:list:response", "requestId":"req_1", "success":false, "error":"workspace list failed" }`)) select { case result := <-resultCh: require.ErrorContains(t, result.err, "workspace list failed") case <-time.After(time.Second): require.FailNow(t, "expected workspace list result") } } func TestHandleWorkspaceListResponseClassifiesUnsupported(t *testing.T) { t.Parallel() resultCh := make(chan workspaceListResult, 1) ch := &Protocol{ id: "agentteam-test", workspaceListRequests: map[string]chan workspaceListResult{ "req_1": resultCh, }, } ch.handleWorkspaceListResponse([]byte(`{ "type":"machine:workspaces:list:response", "requestId":"req_1", "success":false, "error":"workspace list unsupported" }`)) select { case result := <-resultCh: require.Error(t, result.err) require.True(t, errors.Is(result.err, types.ErrAgentTeamWorkspaceListUnsupported), "err = %v", result.err) case <-time.After(time.Second): require.FailNow(t, "expected workspace list result") } } func TestProtocolRunStateClearIsGenerationConditional(t *testing.T) { channel := &Protocol{} oldDone := make(chan struct{}) newDone := make(chan struct{}) _, newCancel := context.WithCancel(context.Background()) defer newCancel() channel.mu.Lock() channel.generation = 2 channel.done = newDone channel.cancel = newCancel channel.mu.Unlock() channel.clearRunState(1, oldDone) channel.mu.Lock() require.Equal(t, uint64(2), channel.generation) require.Equal(t, newDone, channel.done) require.NotNil(t, channel.cancel) channel.mu.Unlock() } func TestProtocolConcurrentStartWaitsForExactOldStopBoundary(t *testing.T) { t.Setenv("PATH", t.TempDir()) channel := &Protocol{ id: "generation-fence", enabled: true, serverURL: "http://127.0.0.1:1", apiKey: "machine-key", sendCh: make(chan []byte, sendBufSize), agents: make(map[string]*agentState), dedupRing: newDedupRing(10), disp: &agentteamTestDispatcher{}, } oldDone := make(chan struct{}) cancelCalled := make(chan struct{}) var cancelOnce sync.Once channel.mu.Lock() channel.generation = 1 channel.done = oldDone channel.cancel = func() { cancelOnce.Do(func() { close(cancelCalled) }) } channel.mu.Unlock() stopReturned := make(chan error, 1) go func() { ctx, cancel := context.WithTimeout(context.Background(), time.Second) defer cancel() stopReturned <- channel.Stop(ctx) }() select { case <-cancelCalled: case <-time.After(time.Second): t.Fatal("Stop did not acquire the lifecycle and cancel the old run") } startCtx, cancelStart := context.WithCancel(context.Background()) defer cancelStart() startReturned := make(chan error, 1) go func() { startReturned <- channel.Start(startCtx) }() select { case err := <-startReturned: t.Fatalf("Start crossed the still-open old Stop boundary: %v", err) case <-time.After(30 * time.Millisecond): } close(oldDone) require.NoError(t, <-stopReturned) require.NoError(t, <-startReturned) channel.mu.Lock() newGeneration := channel.generation newDone := channel.done channel.mu.Unlock() require.Equal(t, uint64(2), newGeneration) require.NotNil(t, newDone) require.NotEqual(t, oldDone, newDone) // A delayed cleanup from the old goroutine cannot erase the new run. channel.clearRunState(1, oldDone) channel.mu.Lock() require.Equal(t, uint64(2), channel.generation) require.Equal(t, newDone, channel.done) channel.mu.Unlock() cancelStart() ctx, cancelStop := context.WithTimeout(context.Background(), time.Second) defer cancelStop() require.NoError(t, channel.Stop(ctx)) } func TestShouldLogReconnectAttemptBatchesLongFailureStreaks(t *testing.T) { // Early failures log individually so an operator sees the immediate cause. for attempt := 1; attempt <= 10; attempt++ { require.True(t, shouldLogReconnectAttempt(attempt), "attempt %d should log", attempt) } // Beyond the first ten, only one summary line per batch keeps a dead // server entry from flooding the log on every backoff tick. for attempt := 11; attempt < reconnectLogBatchSize; attempt++ { require.False(t, shouldLogReconnectAttempt(attempt), "attempt %d should be silent", attempt) } require.True(t, shouldLogReconnectAttempt(reconnectLogBatchSize)) require.False(t, shouldLogReconnectAttempt(reconnectLogBatchSize+1)) require.True(t, shouldLogReconnectAttempt(2*reconnectLogBatchSize)) }