diff --git a/.changes/employee-chat-convergence.md b/.changes/employee-chat-convergence.md index bd9bb896c..643e8d696 100644 --- a/.changes/employee-chat-convergence.md +++ b/.changes/employee-chat-convergence.md @@ -5,4 +5,4 @@ category: Changed - **数字员工回复复用 chat** — 通用发送与引用回复增加 stdin 正文、投递回执和绑定上下文校验；本地数字员工 worker 使用通用 chat 回传，删除重复的 `channel reply` / `channel operator-private`，配套 DSH 要求 `chatDelivery=true`，缺失时提示升级；双方须配套升级。 - **A2UI 审批卡片兼容** — 发卡命令支持 `--summary` 自定义聊天预览，满足 DSH 数字员工的卡片能力探测；省略时继续使用原有默认摘要。 -local_agent 聊天权限统一读取 DEAP 已发布可见范围（成员及部门子树）；其他场景保留本地白名单。DWS/DSH 共用绑定查询判定，固定管理 Profile，查询失败不回退放行。 +local_agent 私聊权限读取 DEAP 已发布可见范围（成员及部门子树）；群聊以已入群为准，不叠加个人可见范围或本地群白名单，并修复空白名单漏订阅群内 @ 消息；群聊仅原生 @ 当前员工才触发，普通消息不进入模型。DSH 按 groupMembershipAccess 能力传递群上下文；其他场景保留本地白名单。DWS/DSH 共用绑定查询判定，固定管理 Profile，查询失败不回退放行。 diff --git a/docs/digital-employee-adapters.md b/docs/digital-employee-adapters.md index 162570cd5..3a00a0179 100644 --- a/docs/digital-employee-adapters.md +++ b/docs/digital-employee-adapters.md @@ -63,9 +63,9 @@ dws dingtalk-tag connect --agent-uuid --channel qoder --daemon --alw connect 使用主管取得一次性授权码，严格使用返回的 Client ID 换票，按发布详情校验员工身份后保存独立 Profile，不切换主管 Current。普通 Agent 的进程、会话及处理记录按员工 Profile 摘要和 Adapter 隔离，不用共享 OAuth Client ID 隔离员工。 -`manage create` 创建并通过 connect 绑定的 `local_agent`，聊天权限以 DEAP 已发布可见范围为准：`manage set-visibility` 保存草稿后仍需 publish。每条入站消息重新查询 published，成员范围使用员工身份精确转换开放 ID，ALL 必须确认本企业成员，部门范围包含子部门（遍历最多 256 个部门，超限或查询失败拒绝处理）。私聊与群聊均按发送者校验，不叠加本地用户/群白名单；operator 不绕过聊天权限，但敏感操作审批仍只接受绑定 operator。 +`manage create` 创建并通过 connect 绑定的 `local_agent`，聊天权限以 DEAP 已发布可见范围为准：`manage set-visibility` 保存草稿后仍需 publish。每条入站消息重新查询 published，成员范围使用员工身份精确转换开放 ID，ALL 必须确认本企业成员，部门范围包含子部门（遍历最多 256 个部门，超限或查询失败拒绝处理）。私聊按发送者校验；local_agent 已被拉入群即视为群内可见，群消息不再叠加个人可见范围或本地用户/群白名单。受管员工只订阅原生 @ 当前员工的消息，普通群消息不进入模型；入队前核实已发布身份、类型与在线状态；operator 不绕过聊天权限，但敏感操作审批仍只接受绑定 operator。 -绑定固定保存管理账号 Profile，不受全局 current 切换影响。旧服务端绑定使用原管理账号执行 connect restart 迁移；缺管理账号、身份冲突或查询失败时停止授权，不回退本地白名单。`channel binding --stdin` 可选接收 senderOpenDingTalkId、senderName，返回 accessPolicy 和 DEAP allowed 判定；仅有本地配置的旧通道、非 local_agent 和机器人继续采用本地白名单。名称仅作检索提示，必须精确匹配开放 ID 并确认企业成员身份，绝不按同名放行。 +绑定固定保存管理账号 Profile，不受全局 current 切换影响。旧服务端绑定使用原管理账号执行 connect restart 迁移；缺管理账号、身份冲突或查询失败时停止授权，不回退本地白名单。`channel binding --stdin` 可选接收 senderOpenDingTalkId、senderName、conversationType（direct/group，省略为 direct），返回 accessPolicy 和 DEAP allowed 判定；Adapter 仅在 capabilities.groupMembershipAccess=true 时发送 conversationType，以兼容旧 DWS；仅有本地配置的旧通道、非 local_agent 和机器人继续采用本地白名单。名称仅作检索提示，必须精确匹配开放 ID 并确认企业成员身份，绝不按同名放行。 不同 Adapter 的已有绑定返回冲突，不隐式替换 DSH。 diff --git a/internal/helpers/deap_agent_adapter.go b/internal/helpers/deap_agent_adapter.go index 6de5a0865..08b0df4c4 100644 --- a/internal/helpers/deap_agent_adapter.go +++ b/internal/helpers/deap_agent_adapter.go @@ -310,7 +310,7 @@ func digitalEmployeeResultSpec() *contract.ResultSpec { } func digitalEmployeeCapabilitiesResultSpec() *contract.ResultSpec { - return &contract.ResultSpec{Outcomes: []contract.ResultOutcome{"success"}, DataSchema: json.RawMessage(`{"type":"object","properties":{"schemaVersion":{"type":"integer","description":"输入 Schema 版本"},"protocolVersion":{"type":"integer","description":"机器协议版本"},"channel":{"type":"string","description":"Adapter 类型"},"auditMode":{"type":"string","description":"审计要求"},"capabilities":{"type":"object","description":"支持的协议操作","properties":{"eventConsume":{"type":"boolean","description":"支持 Event Consumer"},"chatDelivery":{"type":"boolean","description":"通用 chat 支持 stdin、员工上下文和统一发送回执"},"visibilityAccess":{"type":"boolean","description":"支持绑定查询的访问判定"}}}}}`)} + return &contract.ResultSpec{Outcomes: []contract.ResultOutcome{"success"}, DataSchema: json.RawMessage(`{"type":"object","properties":{"schemaVersion":{"type":"integer","description":"输入 Schema 版本"},"protocolVersion":{"type":"integer","description":"机器协议版本"},"channel":{"type":"string","description":"Adapter 类型"},"auditMode":{"type":"string","description":"审计要求"},"capabilities":{"type":"object","description":"支持的协议操作","properties":{"eventConsume":{"type":"boolean","description":"支持 Event Consumer"},"chatDelivery":{"type":"boolean","description":"通用 chat 支持 stdin、员工上下文和统一发送回执"},"visibilityAccess":{"type":"boolean","description":"支持绑定查询的访问判定"},"groupMembershipAccess":{"type":"boolean","description":"支持 conversationType 区分 local_agent 已入群授权与私聊可见范围"}}}}}`)} } func newDigitalEmployeeStatusCommand() *cobra.Command { diff --git a/internal/helpers/deap_agent_channel.go b/internal/helpers/deap_agent_channel.go index c46e61ec3..4b14edbc9 100644 --- a/internal/helpers/deap_agent_channel.go +++ b/internal/helpers/deap_agent_channel.go @@ -349,7 +349,7 @@ func newDeapChannelCapabilitiesCommand() *cobra.Command { RunE: func(cmd *cobra.Command, _ []string) error { return writeDWSMachineEnvelope(cmd, map[string]any{ "schemaVersion": 1, "protocolVersion": 1, "channel": devAppStringFlag(cmd, "channel"), "auditMode": "local_required", - "capabilities": map[string]any{"eventConsume": true, "chatDelivery": true, "visibilityAccess": true}, + "capabilities": map[string]any{"eventConsume": true, "chatDelivery": true, "visibilityAccess": true, "groupMembershipAccess": true}, }) }, Contract: c, diff --git a/internal/helpers/deap_agent_channel_test.go b/internal/helpers/deap_agent_channel_test.go index 5e2cbaec8..d2871a4cb 100644 --- a/internal/helpers/deap_agent_channel_test.go +++ b/internal/helpers/deap_agent_channel_test.go @@ -186,9 +186,10 @@ func TestCrossPlatformCoverageDingTalkTagChannelCapabilitiesUsesDWSMachineEnvelo ProtocolVersion int `json:"protocolVersion"` AuditMode string `json:"auditMode"` Capabilities struct { - EventConsume bool `json:"eventConsume"` - ChatDelivery bool `json:"chatDelivery"` - VisibilityAccess bool `json:"visibilityAccess"` + EventConsume bool `json:"eventConsume"` + ChatDelivery bool `json:"chatDelivery"` + VisibilityAccess bool `json:"visibilityAccess"` + GroupMembershipAccess bool `json:"groupMembershipAccess"` } `json:"capabilities"` } `json:"data"` } @@ -196,7 +197,7 @@ func TestCrossPlatformCoverageDingTalkTagChannelCapabilitiesUsesDWSMachineEnvelo t.Fatalf("decode output %q: %v", output.String(), err) } if !envelope.OK || envelope.Outcome != "success" || envelope.Data.ProtocolVersion != 1 || envelope.Data.AuditMode != "local_required" || - !envelope.Data.Capabilities.EventConsume || !envelope.Data.Capabilities.ChatDelivery || !envelope.Data.Capabilities.VisibilityAccess { + !envelope.Data.Capabilities.EventConsume || !envelope.Data.Capabilities.ChatDelivery || !envelope.Data.Capabilities.VisibilityAccess || !envelope.Data.Capabilities.GroupMembershipAccess { t.Fatalf("capability envelope = %#v", envelope) } } diff --git a/internal/helpers/deap_agent_lifecycle.go b/internal/helpers/deap_agent_lifecycle.go index 113af4c53..c15c2c3c6 100644 --- a/internal/helpers/deap_agent_lifecycle.go +++ b/internal/helpers/deap_agent_lifecycle.go @@ -258,14 +258,15 @@ func newEmployeeBindingCommand() *cobra.Command { c := digitalEmployeeChannelContract("channel_binding", "binding", "读取数字员工权威绑定；可选传入发送者，按已发布 DEAP 范围返回访问判定。", "DSH 启动员工之前核验绑定版本与运行期望") c.Result = &contract.ResultSpec{Outcomes: []contract.ResultOutcome{"success"}, DataSchema: json.RawMessage(`{"type":"object","properties":{"agentUuid":{"type":"string","description":"员工 ID"},"dwsProfile":{"type":"string","description":"精确员工 Profile"},"channel":{"type":"string","description":"绑定 Adapter"},"bindingRevision":{"type":"integer","description":"绑定代数"},"bindingState":{"type":"string","description":"绑定状态"},"desiredState":{"type":"string","description":"运行期望"},"accessPolicy":{"type":"string","description":"访问权威：deap_visibility 或 local_allowlist"},"allowed":{"type":"boolean","description":"DEAP 对指定发送者的访问判定；本地白名单策略不返回"}}}`)} return NewLeafCommand(LeafSpec{Use: "binding", Short: "读取员工绑定权威状态", PostMount: deapAgentNoArgs, OutputRollout: output.RolloutUnifiedActive, - Flags: []LeafFlag{{Name: "channel", Required: true, Enum: digitalEmployeeChannels(), Usage: "绑定 Adapter"}, {Name: "stdin", Kind: LeafBool, Required: true, Usage: "JSON：agentUuid、bindingRevision；可选 senderOpenDingTalkId、senderName 查询访问权限"}}, + Flags: []LeafFlag{{Name: "channel", Required: true, Enum: digitalEmployeeChannels(), Usage: "绑定 Adapter"}, {Name: "stdin", Kind: LeafBool, Required: true, Usage: "JSON：agentUuid、bindingRevision；可选 senderOpenDingTalkId、senderName、conversationType（direct/group，默认 direct）查询访问权限"}}, Safety: contract.SafetySpec{Effect: "read", Risk: "low", Confirmation: "not_required", Idempotency: "idempotent"}, Contract: c, RunE: func(cmd *cobra.Command, _ []string) error { var in struct { - AgentUUID string `json:"agentUuid"` - BindingRevision uint64 `json:"bindingRevision"` - SenderID string `json:"senderOpenDingTalkId,omitempty"` - SenderName string `json:"senderName,omitempty"` + AgentUUID string `json:"agentUuid"` + BindingRevision uint64 `json:"bindingRevision"` + SenderID string `json:"senderOpenDingTalkId,omitempty"` + SenderName string `json:"senderName,omitempty"` + ConversationType string `json:"conversationType,omitempty"` } if err := decodeBoundedDigitalEmployeeStdin(cmd, &in); err != nil { return err @@ -279,7 +280,11 @@ func newEmployeeBindingCommand() *cobra.Command { } result := map[string]any{"agentUuid": b.AgentUUID, "dwsProfile": b.DWSProfile, "channel": bindingChannel(b), "bindingRevision": b.BindingRevision, "bindingState": employeeBindingState(b), "desiredState": employeeDesiredState(b)} if in.SenderID != "" { - policy, allowed, err := employeeVisibilityAccess(cmd.Context(), b, in.SenderID, in.SenderName) + conversationType := in.ConversationType + if conversationType == "" { + conversationType = "direct" + } + policy, allowed, err := employeeConversationAccess(cmd.Context(), b, in.SenderID, in.SenderName, conversationType) if err != nil { return err } diff --git a/internal/helpers/deap_agent_lifecycle_test.go b/internal/helpers/deap_agent_lifecycle_test.go index a07a037db..7ec60dc5b 100644 --- a/internal/helpers/deap_agent_lifecycle_test.go +++ b/internal/helpers/deap_agent_lifecycle_test.go @@ -189,3 +189,43 @@ func TestCrossPlatformCoverageEmployeeBindingRevisionFencesOldReplies(t *testing t.Fatal("unbound reply accepted") } } + +func TestCrossPlatformCoverageEmployeeBindingGroupContext(t *testing.T) { + for _, typ := range []string{"group", "direct", "", "invalid"} { + t.Run("type_"+typ, func(t *testing.T) { + dir, b := lifecycleFixture(t) + b.SupervisorProfile = "corp:manager" + if err := saveDigitalEmployeeBinding(dir, b); err != nil { + t.Fatal(err) + } + auth.SetRuntimeProfile(b.DWSProfile) + t.Cleanup(func() { auth.SetRuntimeProfile("") }) + testseam.Swap(t, &employeeVisibilityCall, func(_ context.Context, _, _, tool string, _ map[string]any) (map[string]any, error) { + if tool != deapAgentDetailTool { + t.Fatalf("unexpected lookup %s", tool) + } + return map[string]any{"data": map[string]any{"agentUuid": b.AgentUUID, "type": "local_agent", "snapshot": "published", "status": "online", "visibility": "PARTIAL", "staffIds": []any{}, "profile": map[string]any{"corpId": "corp", "userId": "employee"}}}, nil + }) + cmd := newEmployeeBindingCommand() + cmd.SetContext(context.Background()) + _ = cmd.Flags().Set("channel", "dsh") + _ = cmd.Flags().Set("stdin", "true") + cmd.SetIn(strings.NewReader(fmt.Sprintf(`{"agentUuid":%q,"bindingRevision":7,"senderOpenDingTalkId":"outsider","conversationType":%q}`, b.AgentUUID, typ))) + var out bytes.Buffer + cmd.SetOut(&out) + err := cmd.RunE(cmd, nil) + if typ == "invalid" { + if err == nil { + t.Fatal("invalid context accepted") + } + return + } + if err != nil { + t.Fatal(err) + } + if strings.Contains(out.String(), `"allowed": true`) != (typ == "group") { + t.Fatalf("output=%s", out.String()) + } + }) + } +} diff --git a/internal/helpers/deap_agent_runtime.go b/internal/helpers/deap_agent_runtime.go index 5251704ad..e75637972 100644 --- a/internal/helpers/deap_agent_runtime.go +++ b/internal/helpers/deap_agent_runtime.go @@ -235,7 +235,7 @@ func (r *employeeRuntime) accept(e employeeEvent) bool { } if r.cfg.Binding.SupervisorProfile != "" || r.cfg.Binding.RuntimeBindingID != "" { switch e.Type { - case "user_im_message_receive_o2o_all", "user_im_message_receive_o2o", "user_im_message_receive_group_all", "user_im_message_receive_group", "user_im_message_receive_at": + case "user_im_message_receive_o2o_all", "user_im_message_receive_o2o", "user_im_message_receive_at": return true default: return false @@ -259,7 +259,11 @@ func (r *employeeRuntime) enqueue(e employeeEvent) error { return nil } if r.cfg.Binding.SupervisorProfile != "" || r.cfg.Binding.RuntimeBindingID != "" { - policy, allowed, err := employeeVisibilityAccess(r.ctx, r.cfg.Binding, e.SenderID, e.SenderName) + conversationType := "group" + if e.Type == "user_im_message_receive_o2o_all" || e.Type == "user_im_message_receive_o2o" { + conversationType = "direct" + } + policy, allowed, err := employeeConversationAccess(r.ctx, r.cfg.Binding, e.SenderID, e.SenderName, conversationType) if err != nil { return &employeeRunError{Code: "visibility_access_unavailable"} } @@ -505,7 +509,10 @@ func runEmployeeWorker(parent context.Context, cfg digitalEmployeeAdapterConfig, } defer func() { cancel(); r.wg.Wait() }() args := []string{"event", "consume", "user_im_message_receive_o2o_all"} - if len(cfg.Options.AllowedGroups) > 0 { + // 入群授权与响应触发独立：受管员工只订阅原生 @ 事件，普通群消息不交给模型。 + if cfg.Binding.SupervisorProfile != "" || cfg.Binding.RuntimeBindingID != "" { + args = append(args, "user_im_message_receive_at") + } else if len(cfg.Options.AllowedGroups) > 0 { args = append(args, "user_im_message_receive_group_all") } args = append(args, "--flatten", "--format", "ndjson") diff --git a/internal/helpers/deap_agent_visibility.go b/internal/helpers/deap_agent_visibility.go index 2d301ffa7..d20be8db6 100644 --- a/internal/helpers/deap_agent_visibility.go +++ b/internal/helpers/deap_agent_visibility.go @@ -52,6 +52,14 @@ func employeeVisibilityProfileCall(ctx context.Context, profile, server, tool st } func employeeVisibilityAccess(ctx context.Context, b digitalEmployeeBinding, sender, senderName string) (string, bool, error) { + return employeeConversationAccess(ctx, b, sender, senderName, "direct") +} + +// 群事件来自员工自身订阅；local_agent 已入群即有群内访问权，私聊仍查个人可见范围。 +func employeeConversationAccess(ctx context.Context, b digitalEmployeeBinding, sender, senderName, conversationType string) (string, bool, error) { + if conversationType != "direct" && conversationType != "group" { + return "deap_visibility", false, fmt.Errorf("visibility_conversation_type_invalid") + } if b.SupervisorProfile == "" && b.RuntimeBindingID == "" { return "local_allowlist", false, nil } @@ -80,6 +88,9 @@ func employeeVisibilityAccess(ctx context.Context, b digitalEmployeeBinding, sen if jsonScalar(data["status"]) != "online" || sender == identity.OpenDingTalkID { return policy, false, nil } + if conversationType == "group" { + return policy, true, nil + } scope := jsonScalar(data["visibility"]) if scope != "ALL" && scope != "PARTIAL" { return policy, false, fmt.Errorf("visibility_scope_unknown") diff --git a/internal/helpers/deap_agent_visibility_test.go b/internal/helpers/deap_agent_visibility_test.go index 4027d704e..4bd3cfdfd 100644 --- a/internal/helpers/deap_agent_visibility_test.go +++ b/internal/helpers/deap_agent_visibility_test.go @@ -171,3 +171,40 @@ func TestCrossPlatformCoverageEmployeeVisibilityMigration(t *testing.T) { }) } } + +func TestCrossPlatformCoverageEmployeeJoinedGroupAccess(t *testing.T) { + for _, tc := range []struct { + name, kind, eventType string + allow, fail bool + }{ + {"ordinary group does not trigger", "local_agent", "user_im_message_receive_group_all", false, false}, + {"local mention outside personal visibility", "local_agent", "user_im_message_receive_at", true, false}, + {"private remains restricted", "local_agent", "user_im_message_receive_o2o_all", false, false}, + {"other type keeps allowlist", "open_code", "user_im_message_receive_group_all", false, false}, + {"unknown type fails closed", "unknown", "user_im_message_receive_at", false, true}, + {"query failure fails closed", "error", "user_im_message_receive_at", false, true}, + } { + t.Run(tc.name, func(t *testing.T) { + r, e := employeeLedgerFixture(t) + r.cfg.Binding = digitalEmployeeBinding{AgentUUID: "agent", DWSProfile: "corp:employee", SupervisorProfile: "corp:manager"} + r.cfg.Options.AllowedUsers = nil + r.cfg.Options.AllowedGroups = nil + e.Type = tc.eventType + q := make(chan employeeEvent, 1) + r.queues[e.ConversationID] = q + testseam.Swap(t, &employeeVisibilityCall, func(_ context.Context, _, _, tool string, _ map[string]any) (map[string]any, error) { + if tc.kind == "error" { + return nil, fmt.Errorf("unavailable") + } + if tool != deapAgentDetailTool { + t.Fatal("group must not resolve personal visibility") + } + return map[string]any{"data": map[string]any{"agentUuid": "agent", "type": tc.kind, "snapshot": "published", "status": "online", "visibility": "PARTIAL", "staffIds": []any{}, "profile": map[string]any{"corpId": "corp", "userId": "employee"}}}, nil + }) + err := r.enqueue(e) + if (err != nil) != tc.fail || (len(q) > 0) != tc.allow { + t.Fatalf("queued=%d err=%v", len(q), err) + } + }) + } +} diff --git a/internal/helpers/deap_agent_worker_stream_test.go b/internal/helpers/deap_agent_worker_stream_test.go index 6187ccd25..055acaf79 100644 --- a/internal/helpers/deap_agent_worker_stream_test.go +++ b/internal/helpers/deap_agent_worker_stream_test.go @@ -80,11 +80,15 @@ func TestEmployeeStreamBoundaryFixture(t *testing.T) { } func TestCrossPlatformCoverageEmployeeWorkerStreamFailures(t *testing.T) { - for _, scenario := range []string{"prepare", "executor-state", "audit", "ledger", "command", "transport-state", "running-state", "exit-before-ready", "stderr-oversized", "stdout-oversized", "exit-after-ready", "unobserved", "transport-write", "enqueue", "fatal", "cancel-start", "cancel-running", "hang-exit", "flood-stdout", "flood-stderr"} { + for _, scenario := range []string{"group-subscription", "prepare", "executor-state", "audit", "ledger", "command", "transport-state", "running-state", "exit-before-ready", "stderr-oversized", "stdout-oversized", "exit-after-ready", "unobserved", "transport-write", "enqueue", "fatal", "cancel-start", "cancel-running", "hang-exit", "flood-stdout", "flood-stderr"} { t.Run(scenario, func(t *testing.T) { cfg := employeeBoundaryConfig(t) cfg.Options.AllowedUsers = []string{"owner"} cfg.Options.AllowedGroups = []string{"group"} + if scenario == "group-subscription" { + cfg.Options.AllowedGroups = nil + cfg.Binding.SupervisorProfile = "corp:manager" + } dir := digitalEmployeeRuntimeDir(cfg.Binding.DWSProfile) closed := false testseam.Swap(t, &digitalEmployeeNewForwarder, func(context.Context, digitalEmployeeAdapterConfig) (forwarder, error) { @@ -95,7 +99,10 @@ func TestCrossPlatformCoverageEmployeeWorkerStreamFailures(t *testing.T) { }) t.Setenv("DWS_STREAM_BOUNDARY", scenario) t.Setenv("DWS_STREAM_BOUNDARY_DIR", dir) - testseam.Swap(t, &employeeExecCommand, func(ctx context.Context, _ string, _ ...string) *exec.Cmd { + testseam.Swap(t, &employeeExecCommand, func(ctx context.Context, _ string, args ...string) *exec.Cmd { + if scenario == "group-subscription" && (!strings.Contains(strings.Join(args, " "), "user_im_message_receive_at") || strings.Contains(strings.Join(args, " "), "user_im_message_receive_group_all")) { + t.Error("managed employee must subscribe only to group mentions") + } cmd := exec.CommandContext(ctx, os.Args[0], "-test.run=^TestEmployeeStreamBoundaryFixture$", "--", "employee-stream-boundary") if scenario == "hang-exit" { cmd.Cancel = func() error { return nil } @@ -138,7 +145,7 @@ func TestCrossPlatformCoverageEmployeeWorkerStreamFailures(t *testing.T) { if scenario == "hang-exit" { testseam.Swap(t, &employeeConsumerStopTimeout, time.Millisecond) } - graceful := scenario == "cancel-start" || scenario == "cancel-running" || scenario == "hang-exit" || strings.HasPrefix(scenario, "flood-") + graceful := scenario == "group-subscription" || scenario == "cancel-start" || scenario == "cancel-running" || scenario == "hang-exit" || strings.HasPrefix(scenario, "flood-") if graceful { done := make(chan struct{}) go func() {