diff --git a/docs/architecture-review-issue-217.md b/docs/architecture-review-issue-217.md new file mode 100644 index 00000000..ef1432de --- /dev/null +++ b/docs/architecture-review-issue-217.md @@ -0,0 +1,134 @@ +# NeoCode 架构评审记录（Issue #217） + +本文档用于沉淀 `Issue #217` 中对当前仓库架构的复核结论，重点关注主链路、职责边界、无效抽象与可维护性风险。 + +## 评审范围 + +- `internal/app` +- `internal/gateway` +- `internal/runtime` +- `internal/tools` +- `docs` +- 根目录 `AGENTS.md` + +## 总体结论 + +当前项目主链路 `TUI -> Runtime -> Provider / Tool Manager` 基本清晰，运行时编排、工具调度、配置管理与 TUI 展示的边界总体可读，适合作为 Agent CLI 的 MVP 基础。 + +> 2026-04-10 修复进展：`internal/gateway` 已按建议移除；待审批权限状态改为由 `runtime.Service` 实例持有；缺失 workspace sandbox 时会显式报错；关键交互事件开始在投递失败时返回错误。 + +但仓库中仍存在几类会直接拉低可维护性和演进效率的问题： + +- 存在未落地的抽象层，增加理解成本且没有产生真实隔离收益。 +- 个别运行时状态通过包级全局变量维护，削弱封装并引入并发与测试隔离风险。 +- 部分安全兜底行为过于宽松，容易在调用方漏配时悄悄退化为不安全路径。 +- 少量关键事件投递失败后没有被显式处理，可能造成 UI 视图与真实运行状态不一致。 + +## 主要问题 + +### 1. `internal/gateway` 当前属于未落地抽象 + +相关位置： + +- `internal/gateway/contracts.go` + +现状： + +- `Gateway` 与 `RuntimePort` 只定义了契约，没有看到对应实现或被主链路消费。 +- `PermissionResolutionDecision`、`CompactResult`、`RunInput` 等类型与 `internal/runtime` 中的运行时结构存在明显语义重叠。 +- 当前 TUI 直接消费 runtime 事件与能力，没有经过 gateway 这一层。 + +影响： + +- 形成“为了将来可能需要”而保留的预留层。 +- 新人阅读时会误判系统存在第二套接入边界。 +- 同类结构重复定义，后续演进时容易出现字段漂移。 + +建议： + +- 如果当前版本没有 HTTP / RPC 网关落地计划，先删除 `internal/gateway`，统一以 runtime 契约作为唯一编排入口。 +- 如果必须保留，则应尽快补齐真实实现，并让 gateway 直接复用 runtime 领域类型，避免重复建模。 + +### 2. 权限待决状态使用包级全局变量，封装性不足 + +相关位置： + +- `internal/runtime/permission.go` + +现状： + +- `runtimePendingPermissions` 通过包级全局变量保存 `*Service -> pendingPermissionRequest` 的映射。 +- 状态生命周期没有收敛在 `Service` 实例内部。 + +影响： + +- 多实例测试或未来并发运行时更容易产生隐式共享状态。 +- 状态释放路径不够直观，后续改动时容易引入泄漏或串扰。 +- 当前结构默认每个 `Service` 同时只有一个待审批请求，不利于并行工具调用扩展。 + +建议： + +- 将待审批状态下沉到 `Service` 字段中，由实例自己持有 `pending request` 与互斥锁。 +- 若后续要支持并行工具调用，建议改成按 `requestID` 建索引，而不是单值槽位。 + +### 3. `NoopWorkspaceSandbox` 的兜底语义过宽 + +相关位置： + +- `internal/tools/manager.go` +- `internal/app/bootstrap.go` + +现状： + +- `buildToolManager` 正常路径会注入 `security.NewWorkspaceSandbox()`，这条链路是合理的。 +- 但 `tools.NewManager` 在 `sandbox == nil` 时会回退到 `NoopWorkspaceSandbox`。 +- 该实现的 `Check` 直接返回 `nil, ctx.Err()`；在上下文正常时等价于“不做任何工作区限制”。 + +影响： + +- 调用方一旦漏传 sandbox，不会快速失败，而是静默退化为无隔离执行。 +- 类型名称中的 `Noop` 容易让人忽略它实际上是在跳过关键安全检查。 + +建议： + +- 优先改为构造期强校验，要求调用方显式传入 sandbox。 +- 如果保留兜底实现，也应使用更明确的命名，并在运行时输出清晰错误，避免静默放行。 + +### 4. 关键事件发射失败后未被统一处理 + +相关位置： + +- `internal/runtime/runtime.go` + +现状： + +- `s.emit(...)` 在运行主循环中多次调用，但大多数返回值没有被处理。 +- 一旦事件通道阻塞或上下文取消，事件丢失不会总是反映到上层控制流。 + +影响： + +- UI 可能漏掉 `EventToolStart`、`EventToolResult`、`EventAgentDone` 等关键状态。 +- 当展示层状态与真实运行进度不一致时，问题排查会比较困难。 + +建议： + +- 对关键事件建立统一的错误处理策略。 +- 至少对会影响交互一致性的事件进行检查，并决定是中止当前运行、写入诊断日志，还是降级告警。 + +## 优先级建议 + +建议按以下顺序治理： + +1. 先处理 `runtimePendingPermissions` 的实例内收敛，降低运行时共享状态风险。 +2. 再收紧 `NoopWorkspaceSandbox` 的默认行为，避免安全能力因误用而失效。 +3. 随后清理 `internal/gateway`，减少空抽象和重复类型。 +4. 最后补强事件投递失败时的处理策略，提高 TUI 与 runtime 的一致性。 + +## 收益预期 + +完成上述收敛后，项目会得到几项直接收益： + +- 主链路更单纯，仓库学习成本更低。 +- 运行时状态边界更清晰，测试隔离性更好。 +- 安全能力从“依赖正确接线”变成“默认不允许漏接”。 +- 事件驱动链路更可观测，后续排查 UI 异常更直接。 diff --git a/internal/gateway/contracts.go b/internal/gateway/contracts.go deleted file mode 100644 index 88eba546..00000000 --- a/internal/gateway/contracts.go +++ /dev/null @@ -1,174 +0,0 @@ -package gateway - -import ( - "context" - "time" -) - -// RuntimeEventType 表示运行事件类型。 -type RuntimeEventType string - -const ( - // RuntimeEventTypeRunProgress 表示运行过程事件。 - RuntimeEventTypeRunProgress RuntimeEventType = "run_progress" - // RuntimeEventTypeRunDone 表示运行完成事件。 - RuntimeEventTypeRunDone RuntimeEventType = "run_done" - // RuntimeEventTypeRunError 表示运行错误事件。 - RuntimeEventTypeRunError RuntimeEventType = "run_error" -) - -// PermissionResolutionDecision 表示权限审批最终决策。 -type PermissionResolutionDecision string - -const ( - // PermissionResolutionAllowOnce 表示仅本次允许。 - PermissionResolutionAllowOnce PermissionResolutionDecision = "allow_once" - // PermissionResolutionAllowSession 表示在当前会话中持续允许。 - PermissionResolutionAllowSession PermissionResolutionDecision = "allow_session" - // PermissionResolutionReject 表示拒绝本次审批。 - PermissionResolutionReject PermissionResolutionDecision = "reject" -) - -// PermissionResolutionInput 表示一次权限审批决策输入。 -type PermissionResolutionInput struct { - // RequestID 是待审批请求标识。 - RequestID string `json:"request_id"` - // Decision 是审批决策值。 - Decision PermissionResolutionDecision `json:"decision"` -} - -// RunInput 表示网关向下游运行端口发起 run 动作时的输入。 -type RunInput struct { - // RequestID 是客户端请求标识。 - RequestID string - // SessionID 是会话标识。 - SessionID string - // RunID 是运行标识。 - RunID string - // InputText 是文本输入。 - InputText string - // InputParts 是多模态输入分片。 - InputParts []InputPart - // Workdir 是请求级工作目录覆盖值。 - Workdir string -} - -// CompactInput 表示网关向下游运行端口发起 compact 动作时的输入。 -type CompactInput struct { - // RequestID 是客户端请求标识。 - RequestID string - // SessionID 是会话标识。 - SessionID string - // RunID 是运行标识。 - RunID string -} - -// CompactResult 表示 compact 动作完成后返回的结果。 -type CompactResult struct { - // Applied 表示是否实际应用压缩结果。 - Applied bool - // BeforeChars 是压缩前字符数。 - BeforeChars int - // AfterChars 是压缩后字符数。 - AfterChars int - // SavedRatio 是压缩节省比例。 - SavedRatio float64 - // TriggerMode 是触发模式标识。 - TriggerMode string - // TranscriptID 是压缩产物标识。 - TranscriptID string - // TranscriptPath 是压缩产物路径。 - TranscriptPath string -} - -// RuntimeEvent 表示运行端口推送给网关的统一事件。 -type RuntimeEvent struct { - // Type 是事件类型。 - Type RuntimeEventType `json:"type"` - // RunID 是运行标识。 - RunID string `json:"run_id,omitempty"` - // SessionID 是会话标识。 - SessionID string `json:"session_id,omitempty"` - // Payload 是事件扩展负载。 - Payload any `json:"payload,omitempty"` -} - -// ToolCall 表示助手消息中的工具调用元数据。 -type ToolCall struct { - // ID 是工具调用标识。 - ID string `json:"id"` - // Name 是工具名。 - Name string `json:"name"` - // Arguments 是工具参数 JSON 字符串。 - Arguments string `json:"arguments"` -} - -// SessionMessage 表示会话消息快照中的单条消息。 -type SessionMessage struct { - // Role 是消息角色。 - Role string `json:"role"` - // Content 是消息内容。 - Content string `json:"content"` - // ToolCalls 是 assistant 发起的工具调用元数据。 - ToolCalls []ToolCall `json:"tool_calls,omitempty"` - // ToolCallID 是工具消息关联的调用标识。 - ToolCallID string `json:"tool_call_id,omitempty"` - // IsError 表示该消息是否为错误结果。 - IsError bool `json:"is_error,omitempty"` -} - -// Session 表示网关视角的会话详情。 -type Session struct { - // ID 是会话标识。 - ID string `json:"id"` - // Title 是会话标题。 - Title string `json:"title"` - // CreatedAt 是会话创建时间。 - CreatedAt time.Time `json:"created_at"` - // UpdatedAt 是会话更新时间。 - UpdatedAt time.Time `json:"updated_at"` - // Workdir 是会话工作目录。 - Workdir string `json:"workdir,omitempty"` - // Messages 是会话消息快照。 - Messages []SessionMessage `json:"messages,omitempty"` -} - -// SessionSummary 表示会话列表项摘要。 -type SessionSummary struct { - // ID 是会话标识。 - ID string `json:"id"` - // Title 是会话标题。 - Title string `json:"title"` - // CreatedAt 是会话创建时间。 - CreatedAt time.Time `json:"created_at"` - // UpdatedAt 是会话更新时间。 - UpdatedAt time.Time `json:"updated_at"` -} - -// RuntimePort 定义网关访问运行时编排的下游端口契约。 -type RuntimePort interface { - // Run 启动一次运行编排。 - Run(ctx context.Context, input RunInput) error - // Compact 对指定会话触发一次手动压缩。 - Compact(ctx context.Context, input CompactInput) (CompactResult, error) - // ResolvePermission 向运行时提交一次权限审批决策。 - ResolvePermission(ctx context.Context, input PermissionResolutionInput) error - // CancelActiveRun 取消当前活跃运行。 - CancelActiveRun() bool - // Events 返回统一运行事件流。 - Events() <-chan RuntimeEvent - // ListSessions 返回会话摘要列表。 - ListSessions(ctx context.Context) ([]SessionSummary, error) - // LoadSession 加载指定会话详情。 - LoadSession(ctx context.Context, id string) (Session, error) - // SetSessionWorkdir 更新会话工作目录。 - SetSessionWorkdir(ctx context.Context, sessionID string, workdir string) (Session, error) -} - -// Gateway 定义网关主契约。 -type Gateway interface { - // Serve 启动网关服务并绑定运行端口。 - Serve(ctx context.Context, runtimePort RuntimePort) error - // Close 优雅关闭网关服务。 - Close(ctx context.Context) error -} diff --git a/internal/gateway/contracts_test.go b/internal/gateway/contracts_test.go deleted file mode 100644 index e4a4e50c..00000000 --- a/internal/gateway/contracts_test.go +++ /dev/null @@ -1,40 +0,0 @@ -package gateway - -import "context" - -// runtimePortCompileStub 用于编译期验证 RuntimePort 契约完整性。 -type runtimePortCompileStub struct{} - -func (s *runtimePortCompileStub) Run(_ context.Context, _ RunInput) error { - return nil -} - -func (s *runtimePortCompileStub) Compact(_ context.Context, _ CompactInput) (CompactResult, error) { - return CompactResult{}, nil -} - -func (s *runtimePortCompileStub) ResolvePermission(_ context.Context, _ PermissionResolutionInput) error { - return nil -} - -func (s *runtimePortCompileStub) CancelActiveRun() bool { - return false -} - -func (s *runtimePortCompileStub) Events() <-chan RuntimeEvent { - return nil -} - -func (s *runtimePortCompileStub) ListSessions(_ context.Context) ([]SessionSummary, error) { - return nil, nil -} - -func (s *runtimePortCompileStub) LoadSession(_ context.Context, _ string) (Session, error) { - return Session{}, nil -} - -func (s *runtimePortCompileStub) SetSessionWorkdir(_ context.Context, _, _ string) (Session, error) { - return Session{}, nil -} - -var _ RuntimePort = (*runtimePortCompileStub)(nil) diff --git a/internal/gateway/errors.go b/internal/gateway/errors.go deleted file mode 100644 index c5e72aba..00000000 --- a/internal/gateway/errors.go +++ /dev/null @@ -1,54 +0,0 @@ -package gateway - -import "fmt" - -// ErrorCode 表示网关协议层稳定错误码。 -type ErrorCode string - -const ( - // ErrorCodeInvalidFrame 表示帧结构或帧类型非法。 - ErrorCodeInvalidFrame ErrorCode = "invalid_frame" - // ErrorCodeInvalidAction 表示动作值非法。 - ErrorCodeInvalidAction ErrorCode = "invalid_action" - // ErrorCodeInvalidMultimodalPayload 表示多模态输入负载非法。 - ErrorCodeInvalidMultimodalPayload ErrorCode = "invalid_multimodal_payload" - // ErrorCodeMissingRequiredField 表示缺少必填字段。 - ErrorCodeMissingRequiredField ErrorCode = "missing_required_field" - // ErrorCodeUnsupportedAction 表示动作暂不支持。 - ErrorCodeUnsupportedAction ErrorCode = "unsupported_action" - // ErrorCodeInternalError 表示网关内部错误。 - ErrorCodeInternalError ErrorCode = "internal_error" -) - -var stableErrorCodes = map[string]struct{}{ - string(ErrorCodeInvalidFrame): {}, - string(ErrorCodeInvalidAction): {}, - string(ErrorCodeInvalidMultimodalPayload): {}, - string(ErrorCodeMissingRequiredField): {}, - string(ErrorCodeUnsupportedAction): {}, - string(ErrorCodeInternalError): {}, -} - -// String 返回错误码的字符串值。 -func (c ErrorCode) String() string { - return string(c) -} - -// NewFrameError 创建统一格式的协议错误对象。 -func NewFrameError(code ErrorCode, message string) *FrameError { - return &FrameError{ - Code: code.String(), - Message: message, - } -} - -// NewMissingRequiredFieldError 创建缺少必填字段的错误对象。 -func NewMissingRequiredFieldError(field string) *FrameError { - return NewFrameError(ErrorCodeMissingRequiredField, fmt.Sprintf("missing required field: %s", field)) -} - -// IsStableErrorCode 判断给定字符串是否为网关稳定错误码。 -func IsStableErrorCode(code string) bool { - _, exists := stableErrorCodes[code] - return exists -} diff --git a/internal/gateway/errors_test.go b/internal/gateway/errors_test.go deleted file mode 100644 index 1c07cc5d..00000000 --- a/internal/gateway/errors_test.go +++ /dev/null @@ -1,37 +0,0 @@ -package gateway - -import "testing" - -func TestStableErrorCodes(t *testing.T) { - codes := []ErrorCode{ - ErrorCodeInvalidFrame, - ErrorCodeInvalidAction, - ErrorCodeInvalidMultimodalPayload, - ErrorCodeMissingRequiredField, - ErrorCodeUnsupportedAction, - ErrorCodeInternalError, - } - - for _, code := range codes { - if !IsStableErrorCode(code.String()) { - t.Fatalf("expected code %q to be stable", code) - } - } - - if IsStableErrorCode("unknown_code") { - t.Fatalf("unknown code should not be stable") - } -} - -func TestNewMissingRequiredFieldError(t *testing.T) { - err := NewMissingRequiredFieldError("session_id") - if err == nil { - t.Fatalf("expected non-nil error") - } - if err.Code != ErrorCodeMissingRequiredField.String() { - t.Fatalf("error code mismatch: got %q", err.Code) - } - if err.Message == "" { - t.Fatalf("error message should not be empty") - } -} diff --git a/internal/gateway/types.go b/internal/gateway/types.go deleted file mode 100644 index 281c8132..00000000 --- a/internal/gateway/types.go +++ /dev/null @@ -1,97 +0,0 @@ -package gateway - -// FrameType 表示网关协议帧类型。 -type FrameType string - -const ( - // FrameTypeRequest 表示客户端发往网关的请求帧。 - FrameTypeRequest FrameType = "request" - // FrameTypeEvent 表示网关推送给客户端的事件帧。 - FrameTypeEvent FrameType = "event" - // FrameTypeError 表示网关推送给客户端的错误帧。 - FrameTypeError FrameType = "error" - // FrameTypeAck 表示网关对请求的接收确认帧。 - FrameTypeAck FrameType = "ack" -) - -// FrameAction 表示请求动作类型。 -type FrameAction string - -const ( - // FrameActionRun 表示发起一次运行。 - FrameActionRun FrameAction = "run" - // FrameActionCompact 表示触发一次手动压缩。 - FrameActionCompact FrameAction = "compact" - // FrameActionCancel 表示取消当前活跃运行。 - FrameActionCancel FrameAction = "cancel" - // FrameActionListSessions 表示获取会话摘要列表。 - FrameActionListSessions FrameAction = "list_sessions" - // FrameActionLoadSession 表示加载指定会话详情。 - FrameActionLoadSession FrameAction = "load_session" - // FrameActionSetSessionWorkdir 表示设置会话工作目录。 - FrameActionSetSessionWorkdir FrameAction = "set_session_workdir" - // FrameActionResolvePermission 表示提交一次权限审批决策。 - FrameActionResolvePermission FrameAction = "resolve_permission" -) - -// InputPartType 表示多模态输入分片类型。 -type InputPartType string - -const ( - // InputPartTypeText 表示文本分片。 - InputPartTypeText InputPartType = "text" - // InputPartTypeImage 表示图片分片。 - InputPartTypeImage InputPartType = "image" -) - -// Media 表示非文本输入的媒体描述。 -type Media struct { - // URI 是媒体资源地址。 - URI string `json:"uri"` - // MimeType 是媒体 MIME 类型。 - MimeType string `json:"mime_type"` - // FileName 是媒体文件名。 - FileName string `json:"file_name,omitempty"` -} - -// InputPart 表示网关协议中的多模态输入分片。 -type InputPart struct { - // Type 表示分片类型，如 text / image。 - Type InputPartType `json:"type"` - // Text 是文本分片内容，仅 text 类型使用。 - Text string `json:"text,omitempty"` - // Media 是非文本分片媒体信息，仅 image 等类型使用。 - Media *Media `json:"media,omitempty"` -} - -// FrameError 表示协议帧中的错误信息。 -type FrameError struct { - // Code 是稳定错误码，供客户端做分支判断。 - Code string `json:"code"` - // Message 是面向用户或日志的错误描述。 - Message string `json:"message"` -} - -// MessageFrame 是网关与客户端之间的统一通信帧。 -type MessageFrame struct { - // Type 是帧类型。 - Type FrameType `json:"type"` - // Action 是请求动作，非 request 帧可为空。 - Action FrameAction `json:"action,omitempty"` - // RequestID 是客户端请求幂等标识。 - RequestID string `json:"request_id,omitempty"` - // RunID 是运行标识。 - RunID string `json:"run_id,omitempty"` - // SessionID 是会话标识。 - SessionID string `json:"session_id,omitempty"` - // InputText 是文本输入内容。 - InputText string `json:"input_text,omitempty"` - // InputParts 是多模态输入分片。 - InputParts []InputPart `json:"input_parts,omitempty"` - // Workdir 是本次请求的工作目录覆盖值。 - Workdir string `json:"workdir,omitempty"` - // Payload 是动作扩展负载或事件负载。 - Payload any `json:"payload,omitempty"` - // Error 是错误帧负载。 - Error *FrameError `json:"error,omitempty"` -} diff --git a/internal/gateway/types_test.go b/internal/gateway/types_test.go deleted file mode 100644 index 8641b0f9..00000000 --- a/internal/gateway/types_test.go +++ /dev/null @@ -1,120 +0,0 @@ -package gateway - -import ( - "encoding/json" - "testing" - "time" -) - -func TestMessageFrameJSONRoundTrip(t *testing.T) { - original := MessageFrame{ - Type: FrameTypeRequest, - Action: FrameActionRun, - RequestID: "req_001", - RunID: "run_123", - SessionID: "sess_abc", - InputText: "请分析这张图", - InputParts: []InputPart{ - { - Type: InputPartTypeText, - Text: "请先读取图片中的文字", - }, - { - Type: InputPartTypeImage, - Media: &Media{ - URI: "file:///workspace/assets/screen.png", - MimeType: "image/png", - FileName: "screen.png", - }, - }, - }, - Workdir: "/workspace/project", - Error: &FrameError{ - Code: ErrorCodeInvalidFrame.String(), - Message: "invalid frame", - }, - } - - payload, err := json.Marshal(original) - if err != nil { - t.Fatalf("marshal failed: %v", err) - } - - var decoded MessageFrame - if err := json.Unmarshal(payload, &decoded); err != nil { - t.Fatalf("unmarshal failed: %v", err) - } - - if decoded.Type != original.Type { - t.Fatalf("type mismatch: got %q want %q", decoded.Type, original.Type) - } - if decoded.Action != original.Action { - t.Fatalf("action mismatch: got %q want %q", decoded.Action, original.Action) - } - if decoded.RequestID != original.RequestID { - t.Fatalf("request_id mismatch: got %q want %q", decoded.RequestID, original.RequestID) - } - if decoded.SessionID != original.SessionID { - t.Fatalf("session_id mismatch: got %q want %q", decoded.SessionID, original.SessionID) - } - if len(decoded.InputParts) != 2 { - t.Fatalf("input_parts mismatch: got %d want %d", len(decoded.InputParts), 2) - } - if decoded.InputParts[1].Media == nil || decoded.InputParts[1].Media.MimeType != "image/png" { - t.Fatalf("media mime_type mismatch in image part") - } - if decoded.Error == nil || decoded.Error.Code != original.Error.Code { - t.Fatalf("error code mismatch") - } -} - -func TestSessionMessageToolCallsRoundTrip(t *testing.T) { - now := time.Now().UTC().Truncate(time.Second) - original := Session{ - ID: "sess_1", - Title: "demo", - CreatedAt: now, - UpdatedAt: now, - Messages: []SessionMessage{ - { - Role: "assistant", - Content: "我准备调用工具", - ToolCalls: []ToolCall{ - { - ID: "call_1", - Name: "filesystem_read", - Arguments: `{"path":"README.md"}`, - }, - }, - }, - { - Role: "tool", - Content: "tool result", - ToolCallID: "call_1", - }, - }, - } - - encoded, err := json.Marshal(original) - if err != nil { - t.Fatalf("marshal session failed: %v", err) - } - - var decoded Session - if err := json.Unmarshal(encoded, &decoded); err != nil { - t.Fatalf("unmarshal session failed: %v", err) - } - - if len(decoded.Messages) != 2 { - t.Fatalf("message count mismatch: got %d want %d", len(decoded.Messages), 2) - } - if len(decoded.Messages[0].ToolCalls) != 1 { - t.Fatalf("assistant tool_calls count mismatch: got %d want %d", len(decoded.Messages[0].ToolCalls), 1) - } - if decoded.Messages[0].ToolCalls[0].Name != "filesystem_read" { - t.Fatalf("tool call name mismatch: got %q", decoded.Messages[0].ToolCalls[0].Name) - } - if decoded.Messages[0].ToolCalls[0].Arguments != `{"path":"README.md"}` { - t.Fatalf("tool call arguments mismatch: got %q", decoded.Messages[0].ToolCalls[0].Arguments) - } -} diff --git a/internal/gateway/validate.go b/internal/gateway/validate.go deleted file mode 100644 index 3be96857..00000000 --- a/internal/gateway/validate.go +++ /dev/null @@ -1,189 +0,0 @@ -package gateway - -import ( - "encoding/json" - "errors" - "strings" -) - -// ValidateFrame 校验网关协议帧是否满足基础契约约束。 -func ValidateFrame(frame MessageFrame) *FrameError { - if !isValidFrameType(frame.Type) { - return NewFrameError(ErrorCodeInvalidFrame, "invalid frame type") - } - - if strings.TrimSpace(string(frame.Action)) != "" && !isValidFrameAction(frame.Action) { - return NewFrameError(ErrorCodeInvalidAction, "invalid action") - } - - if frame.Type == FrameTypeRequest { - return validateRequestFrame(frame) - } - - return nil -} - -// validateRequestFrame 校验 request 帧的动作及动作所需字段。 -func validateRequestFrame(frame MessageFrame) *FrameError { - if strings.TrimSpace(string(frame.Action)) == "" { - return NewMissingRequiredFieldError("action") - } - - switch frame.Action { - case FrameActionRun: - return validateRunFrame(frame) - case FrameActionCompact, FrameActionLoadSession: - if strings.TrimSpace(frame.SessionID) == "" { - return NewMissingRequiredFieldError("session_id") - } - case FrameActionSetSessionWorkdir: - if strings.TrimSpace(frame.SessionID) == "" { - return NewMissingRequiredFieldError("session_id") - } - if strings.TrimSpace(frame.Workdir) == "" { - return NewMissingRequiredFieldError("workdir") - } - case FrameActionResolvePermission: - return validateResolvePermissionFrame(frame) - case FrameActionCancel, FrameActionListSessions: - return nil - default: - return NewFrameError(ErrorCodeInvalidAction, "invalid action") - } - - if len(frame.InputParts) > 0 { - return validateInputParts(frame.InputParts) - } - - return nil -} - -// validateRunFrame 校验 run 动作的输入字段是否完整且合法。 -func validateRunFrame(frame MessageFrame) *FrameError { - hasText := strings.TrimSpace(frame.InputText) != "" - hasParts := len(frame.InputParts) > 0 - if !hasText && !hasParts { - return NewMissingRequiredFieldError("input_text_or_input_parts") - } - - if hasParts { - return validateInputParts(frame.InputParts) - } - - return nil -} - -// validateResolvePermissionFrame 校验 resolve_permission 动作所需字段。 -func validateResolvePermissionFrame(frame MessageFrame) *FrameError { - if frame.Payload == nil { - return NewMissingRequiredFieldError("payload") - } - - input, err := decodePermissionResolutionInput(frame.Payload) - if err != nil { - return NewFrameError(ErrorCodeInvalidAction, "invalid resolve_permission payload") - } - if strings.TrimSpace(input.RequestID) == "" { - return NewMissingRequiredFieldError("payload.request_id") - } - if !isValidPermissionResolutionDecision(input.Decision) { - return NewFrameError(ErrorCodeInvalidAction, "invalid resolve_permission decision") - } - - return nil -} - -// decodePermissionResolutionInput 将 payload 解析为权限审批决策输入。 -func decodePermissionResolutionInput(payload any) (PermissionResolutionInput, error) { - if direct, ok := payload.(PermissionResolutionInput); ok { - return direct, nil - } - if ptr, ok := payload.(*PermissionResolutionInput); ok { - if ptr == nil { - return PermissionResolutionInput{}, errors.New("permission payload is nil") - } - return *ptr, nil - } - - raw, err := json.Marshal(payload) - if err != nil { - return PermissionResolutionInput{}, err - } - - var input PermissionResolutionInput - if err := json.Unmarshal(raw, &input); err != nil { - return PermissionResolutionInput{}, err - } - return input, nil -} - -// isValidPermissionResolutionDecision 判断审批决策是否属于受支持集合。 -func isValidPermissionResolutionDecision(decision PermissionResolutionDecision) bool { - switch decision { - case PermissionResolutionAllowOnce, PermissionResolutionAllowSession, PermissionResolutionReject: - return true - default: - return false - } -} - -// validateInputParts 校验多模态输入分片数组。 -func validateInputParts(parts []InputPart) *FrameError { - for index := range parts { - if err := validateInputPart(parts[index], index); err != nil { - return err - } - } - return nil -} - -// validateInputPart 校验单个多模态输入分片。 -func validateInputPart(part InputPart, index int) *FrameError { - switch part.Type { - case InputPartTypeText: - if strings.TrimSpace(part.Text) == "" { - return NewFrameError(ErrorCodeInvalidMultimodalPayload, "input_parts[text] requires non-empty text") - } - case InputPartTypeImage: - if part.Media == nil { - return NewFrameError(ErrorCodeInvalidMultimodalPayload, "input_parts[image] requires media") - } - if strings.TrimSpace(part.Media.URI) == "" { - return NewFrameError(ErrorCodeInvalidMultimodalPayload, "input_parts[image] requires media.uri") - } - if strings.TrimSpace(part.Media.MimeType) == "" { - return NewFrameError(ErrorCodeInvalidMultimodalPayload, "input_parts[image] requires media.mime_type") - } - default: - _ = index - return NewFrameError(ErrorCodeInvalidMultimodalPayload, "input_parts contains unsupported type") - } - - return nil -} - -// isValidFrameType 判断帧类型是否属于协议定义集合。 -func isValidFrameType(frameType FrameType) bool { - switch frameType { - case FrameTypeRequest, FrameTypeEvent, FrameTypeError, FrameTypeAck: - return true - default: - return false - } -} - -// isValidFrameAction 判断动作是否属于协议定义集合。 -func isValidFrameAction(action FrameAction) bool { - switch action { - case FrameActionRun, - FrameActionCompact, - FrameActionCancel, - FrameActionListSessions, - FrameActionLoadSession, - FrameActionSetSessionWorkdir, - FrameActionResolvePermission: - return true - default: - return false - } -} diff --git a/internal/gateway/validate_test.go b/internal/gateway/validate_test.go deleted file mode 100644 index d1e86c92..00000000 --- a/internal/gateway/validate_test.go +++ /dev/null @@ -1,287 +0,0 @@ -package gateway - -import ( - "strings" - "testing" -) - -func TestValidateFrame_BasicRules(t *testing.T) { - tests := []struct { - name string - frame MessageFrame - wantNil bool - wantCode string - wantField string - }{ - { - name: "valid run with input_text", - frame: MessageFrame{ - Type: FrameTypeRequest, - Action: FrameActionRun, - InputText: "hello", - }, - wantNil: true, - }, - { - name: "invalid frame type", - frame: MessageFrame{ - Type: FrameType("unknown"), - }, - wantCode: ErrorCodeInvalidFrame.String(), - }, - { - name: "request missing action", - frame: MessageFrame{ - Type: FrameTypeRequest, - }, - wantCode: ErrorCodeMissingRequiredField.String(), - wantField: "action", - }, - { - name: "request invalid action", - frame: MessageFrame{ - Type: FrameTypeRequest, - Action: FrameAction("foo"), - }, - wantCode: ErrorCodeInvalidAction.String(), - }, - { - name: "run missing both input_text and input_parts", - frame: MessageFrame{ - Type: FrameTypeRequest, - Action: FrameActionRun, - }, - wantCode: ErrorCodeMissingRequiredField.String(), - wantField: "input_text_or_input_parts", - }, - { - name: "compact missing session_id", - frame: MessageFrame{ - Type: FrameTypeRequest, - Action: FrameActionCompact, - }, - wantCode: ErrorCodeMissingRequiredField.String(), - wantField: "session_id", - }, - { - name: "load_session missing session_id", - frame: MessageFrame{ - Type: FrameTypeRequest, - Action: FrameActionLoadSession, - }, - wantCode: ErrorCodeMissingRequiredField.String(), - wantField: "session_id", - }, - { - name: "set_session_workdir missing workdir", - frame: MessageFrame{ - Type: FrameTypeRequest, - Action: FrameActionSetSessionWorkdir, - SessionID: "sess_1", - }, - wantCode: ErrorCodeMissingRequiredField.String(), - wantField: "workdir", - }, - { - name: "set_session_workdir valid", - frame: MessageFrame{ - Type: FrameTypeRequest, - Action: FrameActionSetSessionWorkdir, - SessionID: "sess_1", - Workdir: "/workspace", - }, - wantNil: true, - }, - { - name: "resolve_permission valid struct payload", - frame: MessageFrame{ - Type: FrameTypeRequest, - Action: FrameActionResolvePermission, - Payload: PermissionResolutionInput{ - RequestID: "perm-1", - Decision: PermissionResolutionAllowOnce, - }, - }, - wantNil: true, - }, - { - name: "resolve_permission missing payload", - frame: MessageFrame{ - Type: FrameTypeRequest, - Action: FrameActionResolvePermission, - }, - wantCode: ErrorCodeMissingRequiredField.String(), - wantField: "payload", - }, - { - name: "resolve_permission missing request_id", - frame: MessageFrame{ - Type: FrameTypeRequest, - Action: FrameActionResolvePermission, - Payload: map[string]any{ - "decision": "allow_session", - }, - }, - wantCode: ErrorCodeMissingRequiredField.String(), - wantField: "payload.request_id", - }, - { - name: "resolve_permission invalid decision", - frame: MessageFrame{ - Type: FrameTypeRequest, - Action: FrameActionResolvePermission, - Payload: map[string]any{ - "request_id": "perm-1", - "decision": "allow_forever", - }, - }, - wantCode: ErrorCodeInvalidAction.String(), - }, - { - name: "event frame allows empty action", - frame: MessageFrame{ - Type: FrameTypeEvent, - }, - wantNil: true, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - err := ValidateFrame(tt.frame) - if tt.wantNil { - if err != nil { - t.Fatalf("expected nil error, got: %#v", err) - } - return - } - - if err == nil { - t.Fatalf("expected validation error, got nil") - } - if err.Code != tt.wantCode { - t.Fatalf("error code mismatch: got %q want %q", err.Code, tt.wantCode) - } - if tt.wantField != "" && !strings.Contains(err.Message, tt.wantField) { - t.Fatalf("expected message to contain %q, got %q", tt.wantField, err.Message) - } - }) - } -} - -func TestValidateFrame_MultimodalPayloadRules(t *testing.T) { - tests := []struct { - name string - frame MessageFrame - wantNil bool - wantCode string - }{ - { - name: "valid text part", - frame: MessageFrame{ - Type: FrameTypeRequest, - Action: FrameActionRun, - InputParts: []InputPart{ - {Type: InputPartTypeText, Text: "hello"}, - }, - }, - wantNil: true, - }, - { - name: "valid image part", - frame: MessageFrame{ - Type: FrameTypeRequest, - Action: FrameActionRun, - InputParts: []InputPart{ - { - Type: InputPartTypeImage, - Media: &Media{ - URI: "file:///a.png", - MimeType: "image/png", - }, - }, - }, - }, - wantNil: true, - }, - { - name: "text part with empty text", - frame: MessageFrame{ - Type: FrameTypeRequest, - Action: FrameActionRun, - InputParts: []InputPart{ - {Type: InputPartTypeText, Text: " "}, - }, - }, - wantCode: ErrorCodeInvalidMultimodalPayload.String(), - }, - { - name: "image part missing media", - frame: MessageFrame{ - Type: FrameTypeRequest, - Action: FrameActionRun, - InputParts: []InputPart{ - {Type: InputPartTypeImage}, - }, - }, - wantCode: ErrorCodeInvalidMultimodalPayload.String(), - }, - { - name: "image part missing media.uri", - frame: MessageFrame{ - Type: FrameTypeRequest, - Action: FrameActionRun, - InputParts: []InputPart{ - { - Type: InputPartTypeImage, - Media: &Media{MimeType: "image/png"}, - }, - }, - }, - wantCode: ErrorCodeInvalidMultimodalPayload.String(), - }, - { - name: "image part missing media.mime_type", - frame: MessageFrame{ - Type: FrameTypeRequest, - Action: FrameActionRun, - InputParts: []InputPart{ - { - Type: InputPartTypeImage, - Media: &Media{URI: "file:///a.png"}, - }, - }, - }, - wantCode: ErrorCodeInvalidMultimodalPayload.String(), - }, - { - name: "unsupported part type", - frame: MessageFrame{ - Type: FrameTypeRequest, - Action: FrameActionRun, - InputParts: []InputPart{ - {Type: InputPartType("audio")}, - }, - }, - wantCode: ErrorCodeInvalidMultimodalPayload.String(), - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - err := ValidateFrame(tt.frame) - if tt.wantNil { - if err != nil { - t.Fatalf("expected nil error, got: %#v", err) - } - return - } - if err == nil { - t.Fatalf("expected validation error, got nil") - } - if err.Code != tt.wantCode { - t.Fatalf("error code mismatch: got %q want %q", err.Code, tt.wantCode) - } - }) - } -} diff --git a/internal/runtime/permission.go b/internal/runtime/permission.go index 00a0fb76..9c0354a0 100644 --- a/internal/runtime/permission.go +++ b/internal/runtime/permission.go @@ -5,7 +5,6 @@ import ( "errors" "fmt" "strings" - "sync" "time" providertypes "neo-code/internal/provider/types" @@ -46,14 +45,6 @@ type pendingPermissionRequest struct { Submitted bool } -var runtimePendingPermissions = struct { - mu sync.Mutex - nextID uint64 - byRun map[*Service]*pendingPermissionRequest -}{ - byRun: make(map[*Service]*pendingPermissionRequest), -} - // ResolvePermission 接收 UI 的审批决定，并唤醒 runtime 中等待的工具调用。 func (s *Service) ResolvePermission(ctx context.Context, input PermissionResolutionInput) error { requestID := strings.TrimSpace(input.RequestID) @@ -68,20 +59,20 @@ func (s *Service) ResolvePermission(ctx context.Context, input PermissionResolut return err } - runtimePendingPermissions.mu.Lock() - pending := runtimePendingPermissions.byRun[s] + s.pendingPermissionMu.Lock() + pending := s.pendingPermission if pending == nil || pending.RequestID != requestID { - runtimePendingPermissions.mu.Unlock() + s.pendingPermissionMu.Unlock() return fmt.Errorf("runtime: permission request %q not found", requestID) } // Submitted 标记用于避免重复提交同一个 request 导致阻塞。 if pending.Submitted { - runtimePendingPermissions.mu.Unlock() + s.pendingPermissionMu.Unlock() return nil } pending.Submitted = true resultCh := pending.ResultCh - runtimePendingPermissions.mu.Unlock() + s.pendingPermissionMu.Unlock() // 非阻塞提交，避免 UI 重复触发导致写满 channel 后长时间卡住。 select { @@ -134,7 +125,7 @@ func (s *Service) executeToolCallWithPermission(ctx context.Context, input permi return tools.ToolResult{}, rememberErr } } - s.emit(ctx, EventPermissionResolved, input.RunID, input.SessionID, PermissionResolvedPayload{ + if err := s.emitRequired(ctx, EventPermissionResolved, input.RunID, input.SessionID, PermissionResolvedPayload{ RequestID: requestID, ToolCallID: input.Call.ID, ToolName: input.Call.Name, @@ -148,7 +139,9 @@ func (s *Service) executeToolCallWithPermission(ctx context.Context, input permi RuleID: permissionErr.RuleID(), RememberScope: string(scope), ResolvedAs: "rejected", - }) + }); err != nil { + return tools.ToolResult{}, err + } return tools.ToolResult{ ToolCallID: input.Call.ID, Name: input.Call.Name, @@ -160,7 +153,7 @@ func (s *Service) executeToolCallWithPermission(ctx context.Context, input permi if rememberErr := s.toolManager.RememberSessionDecision(input.SessionID, permissionErr.Action(), scope); rememberErr != nil { return tools.ToolResult{}, rememberErr } - s.emit(ctx, EventPermissionResolved, input.RunID, input.SessionID, PermissionResolvedPayload{ + if err := s.emitRequired(ctx, EventPermissionResolved, input.RunID, input.SessionID, PermissionResolvedPayload{ RequestID: requestID, ToolCallID: input.Call.ID, ToolName: input.Call.Name, @@ -174,7 +167,9 @@ func (s *Service) executeToolCallWithPermission(ctx context.Context, input permi RuleID: permissionErr.RuleID(), RememberScope: string(scope), ResolvedAs: "approved", - }) + }); err != nil { + return tools.ToolResult{}, err + } retryCtx, retryCancel := context.WithTimeout(ctx, input.ToolTimeout) retryResult, retryErr := s.toolManager.Execute(retryCtx, callInput) @@ -191,7 +186,7 @@ func (s *Service) awaitPermissionDecision( request := registerPendingPermission(s, input, permissionErr.Action()) defer clearPendingPermission(s, request.RequestID) - s.emit(ctx, EventPermissionRequest, input.RunID, input.SessionID, PermissionRequestPayload{ + if err := s.emitRequired(ctx, EventPermissionRequest, input.RunID, input.SessionID, PermissionRequestPayload{ RequestID: request.RequestID, ToolCallID: input.Call.ID, ToolName: input.Call.Name, @@ -203,7 +198,9 @@ func (s *Service) awaitPermissionDecision( Decision: permissionErr.Decision(), Reason: permissionErr.Reason(), RuleID: permissionErr.RuleID(), - }) + }); err != nil { + return "", "", request.RequestID, err + } select { case <-ctx.Done(): @@ -219,30 +216,30 @@ func (s *Service) awaitPermissionDecision( // registerPendingPermission 注册待审批请求并返回 request id。 func registerPendingPermission(s *Service, input permissionExecutionInput, action security.Action) pendingPermissionRequest { - runtimePendingPermissions.mu.Lock() - defer runtimePendingPermissions.mu.Unlock() + s.pendingPermissionMu.Lock() + defer s.pendingPermissionMu.Unlock() - runtimePendingPermissions.nextID++ + s.nextPermissionID++ request := pendingPermissionRequest{ - RequestID: fmt.Sprintf("perm-%d", runtimePendingPermissions.nextID), + RequestID: fmt.Sprintf("perm-%d", s.nextPermissionID), RunID: input.RunID, SessionID: input.SessionID, Call: input.Call, Action: action, ResultCh: make(chan PermissionResolutionDecision, 1), } - runtimePendingPermissions.byRun[s] = &request + s.pendingPermission = &request return request } // clearPendingPermission 清理待审批请求，避免跨请求误用。 func clearPendingPermission(s *Service, requestID string) { - runtimePendingPermissions.mu.Lock() - defer runtimePendingPermissions.mu.Unlock() + s.pendingPermissionMu.Lock() + defer s.pendingPermissionMu.Unlock() - current := runtimePendingPermissions.byRun[s] + current := s.pendingPermission if current != nil && current.RequestID == requestID { - delete(runtimePendingPermissions.byRun, s) + s.pendingPermission = nil } } diff --git a/internal/runtime/permission_test.go b/internal/runtime/permission_test.go index 86513069..93e840d7 100644 --- a/internal/runtime/permission_test.go +++ b/internal/runtime/permission_test.go @@ -144,6 +144,72 @@ func TestResolvePermissionDuplicateSubmissionIsNonBlocking(t *testing.T) { } } +func TestPendingPermissionStateLivesOnService(t *testing.T) { + t.Parallel() + + service := NewWithFactory( + newRuntimeConfigManager(t), + &stubToolManager{}, + newMemoryStore(), + &scriptedProviderFactory{provider: &scriptedProvider{}}, + nil, + ) + + request := registerPendingPermission(service, permissionExecutionInput{ + RunID: "run-local", + SessionID: "session-local", + Call: providertypes.ToolCall{ + ID: "call-local", + Name: "webfetch", + }, + }, security.Action{ + Type: security.ActionTypeRead, + Payload: security.ActionPayload{ + ToolName: "webfetch", + Resource: "webfetch", + }, + }) + + if service.pendingPermission == nil || service.pendingPermission.RequestID != request.RequestID { + t.Fatalf("expected pending permission stored on service, got %#v", service.pendingPermission) + } + + clearPendingPermission(service, request.RequestID) + if service.pendingPermission != nil { + t.Fatalf("expected pending permission cleared from service, got %#v", service.pendingPermission) + } +} + +func TestAwaitPermissionDecisionPropagatesCriticalEventFailure(t *testing.T) { + t.Parallel() + + service := &Service{ + events: make(chan RuntimeEvent, 1), + } + service.events <- RuntimeEvent{Type: EventError} + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, _, requestID, err := service.awaitPermissionDecision(ctx, permissionExecutionInput{ + RunID: "run-event-fail", + SessionID: "session-event-fail", + Call: providertypes.ToolCall{ + ID: "call-event-fail", + Name: "webfetch", + }, + }, &tools.PermissionDecisionError{}) + if requestID == "" { + t.Fatalf("expected request id even on emit failure") + } + if err == nil || !containsError(err, "emit permission_request event") { + t.Fatalf("expected permission_request emit error, got %v", err) + } + if service.pendingPermission != nil { + t.Fatalf("expected deferred cleanup to clear pending permission, got %#v", service.pendingPermission) + } +} + func TestServiceRunPermissionRejectFlow(t *testing.T) { t.Parallel() @@ -342,7 +408,7 @@ func TestExecuteToolCallWithPermissionReturnsContextCanceledFromEmitChunk(t *tes registry := tools.NewRegistry() registry.Register(&stubTool{ - name: "filesystem_read_file", + name: "webfetch", executeFn: func(_ context.Context, input tools.ToolCallInput) (tools.ToolResult, error) { if input.EmitChunk == nil { t.Fatalf("expected EmitChunk callback") @@ -379,8 +445,8 @@ func TestExecuteToolCallWithPermissionReturnsContextCanceledFromEmitChunk(t *tes SessionID: "session-canceled", Call: providertypes.ToolCall{ ID: "call-canceled", - Name: "filesystem_read_file", - Arguments: `{"path":"README.md"}`, + Name: "webfetch", + Arguments: `{"url":"https://example.com"}`, }, ToolTimeout: time.Second, }) @@ -409,7 +475,7 @@ func TestExecuteToolCallWithPermissionDoesNotRecheckContextAfterSuccessfulEmit(t var cancel context.CancelFunc registry := tools.NewRegistry() registry.Register(&stubTool{ - name: "filesystem_read_file", + name: "webfetch", executeFn: func(_ context.Context, input tools.ToolCallInput) (tools.ToolResult, error) { if input.EmitChunk == nil { t.Fatalf("expected EmitChunk callback") @@ -447,8 +513,8 @@ func TestExecuteToolCallWithPermissionDoesNotRecheckContextAfterSuccessfulEmit(t SessionID: "session-successful-emit", Call: providertypes.ToolCall{ ID: "call-successful-emit", - Name: "filesystem_read_file", - Arguments: `{"path":"README.md"}`, + Name: "webfetch", + Arguments: `{"url":"https://example.com"}`, }, ToolTimeout: time.Second, }) @@ -465,7 +531,7 @@ func TestExecuteToolCallWithPermissionReturnsContextCanceledWhenChunkNotDelivere registry := tools.NewRegistry() registry.Register(&stubTool{ - name: "filesystem_read_file", + name: "webfetch", executeFn: func(_ context.Context, input tools.ToolCallInput) (tools.ToolResult, error) { if input.EmitChunk == nil { t.Fatalf("expected EmitChunk callback") @@ -511,8 +577,8 @@ func TestExecuteToolCallWithPermissionReturnsContextCanceledWhenChunkNotDelivere SessionID: "session-canceled-blocked", Call: providertypes.ToolCall{ ID: "call-canceled-blocked", - Name: "filesystem_read_file", - Arguments: `{"path":"README.md"}`, + Name: "webfetch", + Arguments: `{"url":"https://example.com"}`, }, ToolTimeout: time.Second, }) diff --git a/internal/runtime/runtime.go b/internal/runtime/runtime.go index 2a89957c..7cd247a6 100644 --- a/internal/runtime/runtime.go +++ b/internal/runtime/runtime.go @@ -148,6 +148,9 @@ type Service struct { activeRunCancel context.CancelFunc // 当前活跃 Run 的取消函数。 sessionInputTokens int // 当前会话累计输入 token。 sessionOutputTokens int // 当前会话累计输出 token。 + pendingPermissionMu sync.Mutex // 保护待审批请求状态，避免不同运行实例共享全局变量。 + pendingPermission *pendingPermissionRequest + nextPermissionID uint64 } func NewWithFactory( @@ -209,7 +212,9 @@ func (s *Service) Run(ctx context.Context, input UserInput) error { if err := s.sessionStore.Save(ctx, &session); err != nil { return s.handleRunError(ctx, input.RunID, session.ID, err) } - s.emit(ctx, EventUserMessage, input.RunID, session.ID, userMessage) + if err := s.emitRequired(ctx, EventUserMessage, input.RunID, session.ID, userMessage); err != nil { + return s.handleRunError(ctx, input.RunID, session.ID, err) + } autoCompacted := false reactiveRetried := false @@ -227,7 +232,7 @@ func (s *Service) Run(ctx context.Context, input UserInput) error { } if attempt >= maxLoops { err := errors.New("runtime: max loop reached") - s.emit(ctx, EventError, input.RunID, session.ID, err.Error()) + _ = s.emit(ctx, EventError, input.RunID, session.ID, err.Error()) return err } @@ -332,7 +337,9 @@ func (s *Service) Run(ctx context.Context, input UserInput) error { return s.handleRunError(ctx, input.RunID, session.ID, err) } if len(assistant.ToolCalls) == 0 { - s.emit(ctx, EventAgentDone, input.RunID, session.ID, assistant) + if err := s.emitRequired(ctx, EventAgentDone, input.RunID, session.ID, assistant); err != nil { + return s.handleRunError(ctx, input.RunID, session.ID, err) + } return nil } @@ -340,7 +347,9 @@ func (s *Service) Run(ctx context.Context, input UserInput) error { if err := ctx.Err(); err != nil { return s.handleRunError(ctx, input.RunID, session.ID, err) } - s.emit(ctx, EventToolStart, input.RunID, session.ID, call) + if err := s.emitRequired(ctx, EventToolStart, input.RunID, session.ID, call); err != nil { + return s.handleRunError(ctx, input.RunID, session.ID, err) + } result, execErr := s.executeToolCallWithPermission(ctx, permissionExecutionInput{ RunID: input.RunID, @@ -362,7 +371,15 @@ func (s *Service) Run(ctx context.Context, input UserInput) error { result.Content = execErr.Error() } if permissionEvent, ok := permissionEventFromError(execErr); ok { - s.emit(ctx, EventPermissionResolved, input.RunID, session.ID, permissionEvent.toResolvedPayload()) + if err := s.emitRequired( + ctx, + EventPermissionResolved, + input.RunID, + session.ID, + permissionEvent.toResolvedPayload(), + ); err != nil { + return s.handleRunError(ctx, input.RunID, session.ID, err) + } } toolMessage := providertypes.Message{ @@ -375,7 +392,7 @@ func (s *Service) Run(ctx context.Context, input UserInput) error { session.UpdatedAt = time.Now() if err := s.sessionStore.Save(ctx, &session); err != nil { if execErr != nil && errors.Is(err, context.Canceled) { - s.emit(ctx, EventToolResult, input.RunID, session.ID, result) + _ = s.emit(ctx, EventToolResult, input.RunID, session.ID, result) } return s.handleRunError(ctx, input.RunID, session.ID, err) } @@ -385,7 +402,9 @@ func (s *Service) Run(ctx context.Context, input UserInput) error { } } - s.emit(ctx, EventToolResult, input.RunID, session.ID, result) + if err := s.emitRequired(ctx, EventToolResult, input.RunID, session.ID, result); err != nil { + return s.handleRunError(ctx, input.RunID, session.ID, err) + } if execErr != nil { if err := ctx.Err(); err != nil { return s.handleRunError(ctx, input.RunID, session.ID, err) @@ -515,6 +534,14 @@ func (s *Service) emit(ctx context.Context, kind EventType, runID string, sessio } } +// emitRequired 用于投递影响交互一致性的关键事件；投递失败时返回带事件名的错误。 +func (s *Service) emitRequired(ctx context.Context, kind EventType, runID string, sessionID string, payload any) error { + if err := s.emit(ctx, kind, runID, sessionID, payload); err != nil { + return fmt.Errorf("runtime: emit %s event: %w", kind, err) + } + return nil +} + // handleProviderStreamEvent 解析并应用单条 provider 流式事件，缺失载荷或未知类型时返回错误。 func handleProviderStreamEvent( event providertypes.StreamEvent, diff --git a/internal/runtime/runtime_test.go b/internal/runtime/runtime_test.go index acb72e97..cc2b34f1 100644 --- a/internal/runtime/runtime_test.go +++ b/internal/runtime/runtime_test.go @@ -3085,6 +3085,23 @@ func TestEmitDropsWhenChannelFullAndContextCanceled(t *testing.T) { } } +func TestEmitRequiredWrapsCriticalEventError(t *testing.T) { + t.Parallel() + + service := &Service{ + events: make(chan RuntimeEvent, 1), + } + service.events <- RuntimeEvent{Type: EventAgentChunk} + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + err := service.emitRequired(ctx, EventToolStart, "run-id", "session-id", "payload") + if err == nil || !containsError(err, "emit tool_start event") { + t.Fatalf("expected wrapped tool_start emit error, got %v", err) + } +} + func TestCallProviderWithRetryReturnsCombinedForwardError(t *testing.T) { t.Parallel() diff --git a/internal/tools/manager.go b/internal/tools/manager.go index dab53833..5caaf090 100644 --- a/internal/tools/manager.go +++ b/internal/tools/manager.go @@ -40,13 +40,33 @@ type WorkspaceSandbox interface { Check(ctx context.Context, action security.Action) (*security.WorkspaceExecutionPlan, error) } -// NoopWorkspaceSandbox keeps the explicit sandbox stage in the execution chain -// without changing current behavior. -type NoopWorkspaceSandbox struct{} +type missingWorkspaceSandbox struct{} -// Check implements WorkspaceSandbox. -func (NoopWorkspaceSandbox) Check(ctx context.Context, action security.Action) (*security.WorkspaceExecutionPlan, error) { - return nil, ctx.Err() +// Check 在调用方漏传 sandbox 时明确返回错误，避免静默退化为无隔离执行。 +func (missingWorkspaceSandbox) Check(ctx context.Context, action security.Action) (*security.WorkspaceExecutionPlan, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + if !requiresWorkspaceSandbox(action) { + return nil, nil + } + return nil, errors.New("tools: workspace sandbox is required") +} + +// requiresWorkspaceSandbox 判断当前动作是否会触及本地工作区边界。 +func requiresWorkspaceSandbox(action security.Action) bool { + switch action.Type { + case security.ActionTypeBash: + return true + case security.ActionTypeRead, security.ActionTypeWrite: + targetType := action.Payload.SandboxTargetType + if targetType == "" { + targetType = action.Payload.TargetType + } + return targetType == security.TargetTypePath || targetType == security.TargetTypeDirectory + default: + return false + } } // PermissionDecisionError reports a non-allow permission decision. @@ -149,7 +169,7 @@ func NewManager(executor Executor, engine security.PermissionEngine, sandbox Wor engine = defaultEngine } if sandbox == nil { - sandbox = NoopWorkspaceSandbox{} + sandbox = missingWorkspaceSandbox{} } return &DefaultManager{ diff --git a/internal/tools/manager_test.go b/internal/tools/manager_test.go index 45e7de66..a2dcbb51 100644 --- a/internal/tools/manager_test.go +++ b/internal/tools/manager_test.go @@ -538,7 +538,7 @@ func TestDefaultManagerSessionPermissionMemory(t *testing.T) { if err != nil { t.Fatalf("new engine: %v", err) } - manager, err := NewManager(registry, engine, nil) + manager, err := NewManager(registry, engine, &stubSandbox{}) if err != nil { t.Fatalf("new manager: %v", err) } @@ -748,7 +748,7 @@ func TestDefaultManagerSessionPermissionMemory(t *testing.T) { if err != nil { t.Fatalf("new engine: %v", err) } - manager, err := NewManager(registry, engine, nil) + manager, err := NewManager(registry, engine, &stubSandbox{}) if err != nil { t.Fatalf("new manager: %v", err) } @@ -809,7 +809,7 @@ func TestDefaultManagerSessionPermissionMemory(t *testing.T) { if err != nil { t.Fatalf("new engine: %v", err) } - manager, err := NewManager(registry, engine, nil) + manager, err := NewManager(registry, engine, &stubSandbox{}) if err != nil { t.Fatalf("new manager: %v", err) } @@ -1060,16 +1060,34 @@ func TestPermissionMapperHelpers(t *testing.T) { } } -func TestNoopWorkspaceSandbox(t *testing.T) { +func TestMissingWorkspaceSandbox(t *testing.T) { t.Parallel() - sandbox := NoopWorkspaceSandbox{} - plan, err := sandbox.Check(context.Background(), security.Action{}) + sandbox := missingWorkspaceSandbox{} + plan, err := sandbox.Check(context.Background(), security.Action{ + Type: security.ActionTypeWrite, + Payload: security.ActionPayload{ + TargetType: security.TargetTypePath, + }, + }) + if plan != nil { + t.Fatalf("expected nil workspace plan, got %#v", plan) + } + if err == nil || !strings.Contains(err.Error(), "workspace sandbox is required") { + t.Fatalf("expected required sandbox error, got %v", err) + } + + plan, err = sandbox.Check(context.Background(), security.Action{ + Type: security.ActionTypeRead, + Payload: security.ActionPayload{ + TargetType: security.TargetTypeURL, + }, + }) if err != nil { - t.Fatalf("expected nil error, got %v", err) + t.Fatalf("expected non-workspace action to pass, got %v", err) } if plan != nil { - t.Fatalf("expected nil workspace plan, got %#v", plan) + t.Fatalf("expected nil plan for non-workspace action, got %#v", plan) } ctx, cancel := context.WithCancel(context.Background())