diff --git a/README.md b/README.md index 5ce10e5..1d3c468 100644 --- a/README.md +++ b/README.md @@ -15,6 +15,7 @@ 灵隙，英文名 **Synapse**，是一套专为 AI 智能体（Agent）设计的**协作协议栈**。 "灵隙"的含义： + - **灵** — 智能体的智慧与能力 - **隙** — 智能体之间的空间与连接 @@ -28,13 +29,13 @@ ### 典型场景 -| 场景 | 说明 | -|------|------| +| 场景 | 说明 | +|-------------------|----------------------------------------| | 🤖 **多 Agent 协作** | 一个 Agent 负责分析，另一个负责生成，第三个负责审核，协同完成一份报告 | -| 🔗 **系统集成** | 后端服务通过 REST API 调用 Agent 能力，无需实现完整协议 | -| 🧩 **任务分解** | 复杂任务拆分为多个子任务，分配给不同 Agent 并行执行 | -| 👥 **群体决策** | 多个 Agent 对同一问题进行投票，达成共识后再执行 | -| 🛡️ **人工兜底** | Agent 遇到无法决策的情况时，升级给人类做最终决定 | +| 🔗 **系统集成** | 后端服务通过 REST API 调用 Agent 能力，无需实现完整协议 | +| 🧩 **任务分解** | 复杂任务拆分为多个子任务，分配给不同 Agent 并行执行 | +| 👥 **群体决策** | 多个 Agent 对同一问题进行投票，达成共识后再执行 | +| 🛡️ **人工兜底** | Agent 遇到无法决策的情况时，升级给人类做最终决定 | --- @@ -50,8 +51,8 @@ import "synapse/sdk" client := sdk.NewClient(sdk.Config{ - Protocol: sdk.ProtoA2A, - AgentEndpoint: myEndpoint, +Protocol: sdk.ProtoA2A, +AgentEndpoint: myEndpoint, }) taskID, _ := client.SendTaskRequest(ctx, targetAgent, "text_generation", payload) @@ -64,8 +65,8 @@ result, _ := client.WaitForTaskA2A(ctx, taskID, 5*time.Minute) ```go client := sdk.NewClient(sdk.Config{ - Protocol: sdk.ProtoA2N, - AgentEndpoint: myEndpoint, +Protocol: sdk.ProtoA2N, +AgentEndpoint: myEndpoint, }) group, _ := client.CreateGroup(ctx, "分析小组", topology, nil, members) @@ -78,9 +79,9 @@ subtasks, _ := client.SubmitGroupTask(ctx, group.GroupID, "analysis", req, "majo ```go client := sdk.NewClient(sdk.Config{ - Protocol: sdk.ProtoP2A, - BaseURL: "http://agent-service:8080", - APIKey: "sk-xxx", +Protocol: sdk.ProtoP2A, +BaseURL: "http://agent-service:8080", +APIKey: "sk-xxx", }) result, _ := client.CreateAndWait(ctx, "text_generation", payload, 5*time.Second) @@ -94,28 +95,28 @@ result, _ := client.CreateAndWait(ctx, "text_generation", payload, 5*time.Second ## 七大组件层 -| 组件 | 名称 | 职责 | -|------|------|------| -| 灵脉 | Meridian | 寻址与路由 | -| 灵玺 | Sigil | 身份与凭证 | -| 灵鉴 | Speculum | 共识与裁决 | -| 灵泊 | Harbor | 异步队列与暂存 | -| 灵瞰 | Panorama | 可观测与度量 | -| 灵衡 | Libra | 价值结算与激励 | -| 灵枢 | Pivot | 紧急制动与治理 | +| 组件 | 名称 | 职责 | +|----|----------|---------| +| 灵脉 | Meridian | 寻址与路由 | +| 灵玺 | Sigil | 身份与凭证 | +| 灵鉴 | Speculum | 共识与裁决 | +| 灵泊 | Harbor | 异步队列与暂存 | +| 灵瞰 | Panorama | 可观测与度量 | +| 灵衡 | Libra | 价值结算与激励 | +| 灵枢 | Pivot | 紧急制动与治理 | --- ## 特色能力 -| 能力 | 说明 | -|------|------| -| 🔐 **去中心化身份** | 公钥派生 AgentID，无需中心化注册 | -| 🧠 **三级 HiL 升级** | Agent 建议 → 人工决策 → 专家仲裁 | -| 📦 **内容类型系统** | 文本 / JSON / 代码 / 文件 / 内容包 | -| 🔢 **数字分段错误码** | 9 大类 35+ 个错误码，便于国际化 | -| 🛤️ **双传输模式** | gRPC + HTTP REST，gRPC-Gateway 桥接 | -| ⚡ **熔断与治理** | 熔断器 + 5 级治理级别，保障系统稳定 | +| 能力 | 说明 | +|------------------|----------------------------------| +| 🔐 **去中心化身份** | 公钥派生 AgentID，无需中心化注册 | +| 🧠 **三级 HiL 升级** | Agent 建议 → 人工决策 → 专家仲裁 | +| 📦 **内容类型系统** | 文本 / JSON / 代码 / 文件 / 内容包 | +| 🔢 **数字分段错误码** | 9 大类 35+ 个错误码，便于国际化 | +| 🛤️ **双传输模式** | gRPC + HTTP REST，gRPC-Gateway 桥接 | +| ⚡ **熔断与治理** | 熔断器 + 5 级治理级别，保障系统稳定 | --- @@ -125,44 +126,93 @@ result, _ := client.CreateAndWait(ctx, "text_generation", payload, 5*time.Second # 初始化项目 go mod init my-agent -# 引入灵隙 SDK -go get github.com/synapse/sdk - -# 编写 Agent +# 引入灵隙 SDK（在项目中） +go mod edit -replace synapse=./path/to/synapse +go get synapse/pkg/sdk +go get synapse/pkg/types ``` +### P2A — 外部调用 Agent + ```go package main import ( - "context" - "synapse/sdk" - "synapse/pkg/types" + "context" + "fmt" + "time" + + "synapse/pkg/sdk" ) func main() { - cfg := sdk.Config{ - Protocol: sdk.ProtoA2A, - AgentEndpoint: types.AgentEndpoint{ - AgentID: "agent-001", - Endpoint: "http://localhost:8080", - }, - AuthSecret: "my-secret-key", - } + client := sdk.NewClient(sdk.Config{ + Protocol: sdk.ProtoP2A, + BaseURL: "http://agent-service:8080", + APIKey: "sk-xxx", + }) - client := sdk.NewClient(cfg) + result, err := client.CreateAndWait( + context.Background(), + "text_generation", + map[string]any{"prompt": "写一首关于AI的诗"}, + 5*time.Second, + ) + if err != nil { + panic(err) + } + fmt.Printf("任务完成: %+v\n", result.Result) +} +``` - // 发送任务 - taskID, _ := client.SendTaskRequest( - context.Background(), - types.AgentEndpoint{AgentID: "agent-002", Endpoint: "http://agent-002:8080"}, - "text_generation", - map[string]any{"prompt": "写一首诗"}, - ) +### A2A — Agent 间点对点协作 - // 等待结果 - result, _ := client.WaitForTaskA2A(context.Background(), taskID, 5*time.Minute) - println(result.Message) +```go +package main + +import ( + "context" + "fmt" + "time" + + "synapse/pkg/sdk" + "synapse/pkg/types" +) + +func main() { + cfg := sdk.Config{ + Protocol: sdk.ProtoA2A, + AgentEndpoint: types.AgentEndpoint{ + AgentID: "agent-001", + Endpoint: "http://localhost:8080", + }, + AuthSecret: "my-secret-key", + } + + client := sdk.NewClient(cfg) + + target := types.AgentEndpoint{ + AgentID: "agent-002", + Endpoint: "http://agent-002:8080", + } + + // 发送任务 + taskID, err := client.SendTaskRequest( + context.Background(), + target, + "text_generation", + map[string]any{"prompt": "写一首诗"}, + ) + if err != nil { + panic(err) + } + + // 等待结果 + state, err := client.WaitForTaskA2A(context.Background(), taskID, 5*time.Minute) + if err != nil { + panic(err) + } + fmt.Printf("任务状态: %s, 结果: %v\n", state.Status, state.Result) } ``` @@ -170,16 +220,103 @@ func main() { ## 协议文档 -| 文档 | 说明 | -|------|------| -| [docs/v1/index.md](docs/v1/index.md) | 协议索引与概览 | -| [docs/v1/raw-protocol.md](docs/v1/raw-protocol.md) | 原始协议（公共基础） | -| [docs/v1/a2a-protocol.md](docs/v1/a2a-protocol.md) | A2A 协议（点对点） | -| [docs/v1/a2n-protocol.md](docs/v1/a2n-protocol.md) | A2N 协议（群组协作） | -| [docs/v1/p2a-protocol.md](docs/v1/p2a-protocol.md) | P2A 协议（外部调用） | -| [docs/v1/identity-protocol-draft.md](docs/v1/identity-protocol-draft.md) | 去中心化身份与寻址（草案） | -| [docs/v1/components.md](docs/v1/components.md) | 七大组件层规范 | -| [docs/v1/error-codes.md](docs/v1/error-codes.md) | 数字分段错误码 | +| 文档 | 说明 | +|--------------------------------------------------------------------------------|---------------| +| [docs/v1/00_index.md](docs/v1/00_index.md) | 协议索引与概览 | +| [docs/v1/01_raw-protocol.md](docs/v1/01_raw-protocol.md) | 原始协议（公共基础） | +| [docs/v1/02_a2a-protocol.md](docs/v1/02_a2a-protocol.md) | A2A 协议（点对点） | +| [docs/v1/03_a2n-protocol.md](docs/v1/03_a2n-protocol.md) | A2N 协议（群组协作） | +| [docs/v1/04_p2a-protocol.md](docs/v1/04_p2a-protocol.md) | P2A 协议（外部调用） | +| [docs/v1/05_identity-protocol-draft.md](docs/v1/05_identity-protocol-draft.md) | 去中心化身份与寻址（草案） | +| [docs/v1/components.md](docs/v1/components.md) | 七大组件层规范 | +| [docs/v1/content-system.md](docs/v1/content-system.md) | 内容类型系统 | +| [docs/v1/capabilities-system.md](docs/v1/capabilities-system.md) | 能力发现系统 | +| [docs/v1/error-codes.md](docs/v1/error-codes.md) | 数字分段错误码 | +| [docs/v1/hil-mechanism.md](docs/v1/hil-mechanism.md) | 人工干预机制 (HiL) | +| [docs/v1/grpc-transport.md](docs/v1/grpc-transport.md) | gRPC 传输规范 | +| [docs/v1/transport-modes.md](docs/v1/transport-modes.md) | 传输模式说明 | +| [docs/v1/message-extensions.md](docs/v1/message-extensions.md) | 消息扩展机制 | + +--- + +## 项目结构 + +``` +synapse/ +├── cmd/ # 命令行入口（待实现） +├── configs/ # 配置文件目录 +├── docs/ # 文档 +│ ├── v0/ # v0 版本协议文档 +│ ├── v1/ # v1 版本协议文档 +│ └── poster*.svg # 宣传海报（多种配色） +├── internal/ # 内部实现代码 +│ ├── agent/ # Agent 协议实现 +│ │ ├── group/ # A2N 群组协议 +│ │ └── p2p/ # A2A 点对点协议 +│ ├── api/ # API 层 +│ │ ├── externalapi/ # P2A 外部 HTTP API +│ │ └── grpcapi/ # gRPC API +│ ├── capabilities/ # 能力发现 +│ ├── components/ # 七大核心组件 +│ │ ├── harbor/ # 灵泊 - 异步队列 +│ │ ├── libra/ # 灵衡 - 价值结算 +│ │ ├── meridian/ # 灵脉 - 寻址路由 +│ │ ├── panorama/ # 灵瞰 - 可观测性 +│ │ ├── pivot/ # 灵枢 - 熔断治理 +│ │ ├── sigil/ # 灵玺 - 身份凭证 +│ │ └── speculum/ # 灵鉴 - 共识裁决 +│ ├── hil/ # 人工干预机制 +│ ├── routing/ # 路由层 +│ └── transport/ # 传输层 +├── pkg/ # 公共库 +│ ├── sdk/ # 统一 SDK +│ └── types/ # 核心类型定义 +├── proto/ # Protocol Buffers 定义 +├── test/ # 测试数据和集成测试 +├── go.mod +├── go.sum +├── CHANGELOG.md +└── README.md +``` + +--- + +## 开发 + +### 编译 + +```bash +go build ./... +``` + +### 运行测试 + +```bash +# 运行所有测试（跳过自动生成代码的 vet 检查） +go test -vet=off ./... + +# 运行特定包测试 +go test -v ./pkg/types/... +go test -v ./internal/agent/p2p/... +``` + +### 重新生成 gRPC 代码 + +如果修改了 `proto/synapse.proto`，需要重新生成 Go 代码： + +```bash +# 安装必要工具 +go install google.golang.org/protobuf/cmd/protoc-gen-go@latest +go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@latest +go install github.com/grpc-ecosystem/grpc-gateway/v2/protoc-gen-grpc-gateway@latest +go install github.com/grpc-ecosystem/grpc-gateway/v2/protoc-gen-openapiv2@latest + +# 生成代码 +protoc -I proto/ \ + --go_out=. --go_opt=module=synapse \ + --go-grpc_out=. --go-grpc_opt=module=synapse \ + proto/synapse.proto +``` --- diff --git a/internal/agent/p2p/protocol.go b/internal/agent/p2p/protocol.go index b3dc85f..c181570 100644 --- a/internal/agent/p2p/protocol.go +++ b/internal/agent/p2p/protocol.go @@ -224,6 +224,10 @@ func (p *P2PProtocol) handleTaskRequest(ctx context.Context, msg *types.UnifiedS taskCtx, cancel := context.WithTimeout(ctx, p.config.TaskTimeout) defer cancel() + // Store task status + status := types.P2PStatusPending + p.tasks[taskID] = &status + // Execute task asynchronously go p.executeTask(taskCtx, msg, taskID, payload.Task.Type, payload.Task.Payload) @@ -233,6 +237,12 @@ func (p *P2PProtocol) handleTaskRequest(ctx context.Context, msg *types.UnifiedS // executeTask executes a task and handles callbacks func (p *P2PProtocol) executeTask(ctx context.Context, origMsg *types.UnifiedSynapseMessage, taskID, taskType string, payload map[string]interface{}) { + // Update status to executing + p.mu.Lock() + execStatus := types.P2PStatusExecuting + p.tasks[taskID] = &execStatus + p.mu.Unlock() + // Send initial status update p.sendStatusUpdate(ctx, taskID, types.P2PStatusExecuting, 0, "Task started") @@ -241,11 +251,19 @@ func (p *P2PProtocol) executeTask(ctx context.Context, origMsg *types.UnifiedSyn if err != nil { // Task failed + p.mu.Lock() + failStatus := types.P2PStatusFailed + p.tasks[taskID] = &failStatus + p.mu.Unlock() p.sendTaskFailed(ctx, taskID, "execution_failed", err.Error(), true, nil) return } // Task completed + p.mu.Lock() + compStatus := types.P2PStatusCompleted + p.tasks[taskID] = &compStatus + p.mu.Unlock() p.sendTaskComplete(ctx, taskID, result) } @@ -498,7 +516,18 @@ func (p *P2PProtocol) newAck(msg *types.UnifiedSynapseMessage, taskID, note stri // newNACK creates a NACK response func (p *P2PProtocol) newNACK(msg *types.UnifiedSynapseMessage, err *types.SynapseError) (*types.UnifiedSynapseMessage, error) { - return nil, err + errBytes, _ := json.Marshal(err) + return &types.UnifiedSynapseMessage{ + ID: types.GenerateMessageID(), + Version: "1.0.0", + Type: string(types.P2PNack), + Sender: p.agentEndpoint, + Receiver: &msg.Sender, + CorrelationID: msg.CorrelationID, + ReplyTo: msg.ID, + Payload: errBytes, + Timestamp: types.NowTimestamp(), + }, nil } // newReject creates a task rejection response diff --git a/internal/agent/p2p/protocol_test.go b/internal/agent/p2p/protocol_test.go index 7db486d..5826b1d 100644 --- a/internal/agent/p2p/protocol_test.go +++ b/internal/agent/p2p/protocol_test.go @@ -191,7 +191,7 @@ func TestHandleHeartbeat(t *testing.T) { ID: types.GenerateMessageID(), Version: "1.0.0", Type: string(types.P2PHeartbeat), - Sender: types.AgentEndpoint{AgentID: "agent-002"}, + Sender: types.AgentEndpoint{AgentID: "agent-002", Endpoint: "http://localhost:8081"}, Payload: json.RawMessage("{}"), Timestamp: types.NowTimestamp(), } @@ -217,7 +217,7 @@ func TestHandleAck(t *testing.T) { ID: types.GenerateMessageID(), Version: "1.0.0", Type: string(types.P2PAck), - Sender: types.AgentEndpoint{AgentID: "agent-002"}, + Sender: types.AgentEndpoint{AgentID: "agent-002", Endpoint: "http://localhost:8081"}, ReplyTo: "msg-001", Payload: json.RawMessage("{}"), Timestamp: types.NowTimestamp(), @@ -295,6 +295,10 @@ func TestGetTaskStatus(t *testing.T) { } proto := NewP2PProtocol(ep, nil) + proto.SetTaskHandler(&mockTaskHandler{ + canHandle: true, + result: "test result", + }) // 查询不存在的任务 status, ok := proto.GetTaskStatus("nonexistent") @@ -327,9 +331,9 @@ func TestGetTaskStatus(t *testing.T) { if !ok { t.Error("Task should be found") } - // 状态应该是 pending 或 confirmed - if status != types.P2PStatusPending && status != types.P2PStatusConfirmed { - t.Errorf("Status should be pending or confirmed: got %s", status) + // 状态应该是 pending 或 executing（异步执行可能已经开始） + if status != types.P2PStatusPending && status != types.P2PStatusExecuting && status != types.P2PStatusCompleted { + t.Errorf("Status should be pending/executing/completed: got %s", status) } } @@ -340,6 +344,10 @@ func TestListTasks(t *testing.T) { } proto := NewP2PProtocol(ep, nil) + proto.SetTaskHandler(&mockTaskHandler{ + canHandle: true, + result: "test result", + }) // 初始应该为空 tasks := proto.ListTasks() @@ -487,8 +495,14 @@ func TestHandleUnknownMessageType(t *testing.T) { Timestamp: types.NowTimestamp(), } - _, err := proto.HandleMessage(context.Background(), msg) - if err == nil { - t.Error("Unknown message type should return error") + resp, err := proto.HandleMessage(context.Background(), msg) + if err != nil { + t.Errorf("Unknown message type should return NACK response, not error: %v", err) + } + if resp == nil { + t.Error("Unknown message type should return NACK response") + } + if resp.Type != string(types.P2PNack) { + t.Errorf("Expected NACK response, got: %s", resp.Type) } } diff --git a/internal/api/grpcapi/server.go b/internal/api/grpcapi/server.go index 981dc47..50051cf 100644 --- a/internal/api/grpcapi/server.go +++ b/internal/api/grpcapi/server.go @@ -96,15 +96,24 @@ type VoteStatus struct { // NewGRPCServer creates a new gRPC server func NewGRPCServer(taskExec TaskExecutor, msgRouter MessageRouter, capsProvider CapabilitiesProvider, groupManager GroupManager) *GRPCServer { return &GRPCServer{ - taskExecutor: taskExec, - msgRouter: msgRouter, - capsProvider: capsProvider, - groupManager: groupManager, - taskService: &TaskServiceImpl{}, - messageService: &MessageServiceImpl{}, - capabilitiesService: &CapabilitiesServiceImpl{}, - groupService: &GroupServiceImpl{}, - healthService: &HealthServiceImpl{}, + taskExecutor: taskExec, + msgRouter: msgRouter, + capsProvider: capsProvider, + groupManager: groupManager, + taskService: &TaskServiceImpl{ + taskExecutor: taskExec, + streamClients: make(map[string]chan *TaskStatusUpdate), + }, + messageService: &MessageServiceImpl{ + msgRouter: msgRouter, + }, + capabilitiesService: &CapabilitiesServiceImpl{ + provider: capsProvider, + }, + groupService: &GroupServiceImpl{ + groupManager: groupManager, + }, + healthService: &HealthServiceImpl{}, } } @@ -124,11 +133,12 @@ func (s *GRPCServer) Start(grpcPort, gatewayPort int) error { } s.grpcServer = grpc.NewServer(opts...) - // Register services - // Note: These would be generated from proto - // RegisterTaskServiceServer(s.grpcServer, s.taskService) - // RegisterA2AMessageServiceServer(s.grpcServer, s.messageService) - // ... etc + // Register services using ServiceDesc + s.grpcServer.RegisterService(&TaskService_ServiceDesc, s.taskService) + s.grpcServer.RegisterService(&A2AMessageService_ServiceDesc, s.messageService) + s.grpcServer.RegisterService(&CapabilitiesService_ServiceDesc, s.capabilitiesService) + s.grpcServer.RegisterService(&GroupService_ServiceDesc, s.groupService) + s.grpcServer.RegisterService(&HealthService_ServiceDesc, s.healthService) log.Printf("[gRPC] Server started on port %d", grpcPort) if err := s.grpcServer.Serve(lis); err != nil { diff --git a/internal/components/pivot/pivot.go b/internal/components/pivot/pivot.go index a093efd..efece5f 100644 --- a/internal/components/pivot/pivot.go +++ b/internal/components/pivot/pivot.go @@ -73,8 +73,8 @@ func NewCircuitBreaker(name string, config *CircuitBreakerConfig) *CircuitBreake // Allow 是否允许请求通过 func (cb *CircuitBreaker) Allow() bool { - cb.mu.RLock() - defer cb.mu.RUnlock() + cb.mu.Lock() + defer cb.mu.Unlock() switch cb.state { case CircuitClosed: @@ -82,16 +82,10 @@ func (cb *CircuitBreaker) Allow() bool { case CircuitOpen: // 检查是否超时，可以进入半开状态 if time.Since(cb.lastFailure) > cb.timeout { - cb.mu.RUnlock() - cb.mu.Lock() - if cb.state == CircuitOpen && time.Since(cb.lastFailure) > cb.timeout { - cb.state = CircuitHalfOpen - cb.successCount = 0 - log.Printf("[灵枢] 熔断器 %s 进入半开状态", cb.name) - } - cb.mu.Unlock() - cb.mu.RLock() - return cb.state == CircuitHalfOpen + cb.state = CircuitHalfOpen + cb.successCount = 0 + log.Printf("[灵枢] 熔断器 %s 进入半开状态", cb.name) + return true } return false case CircuitHalfOpen: diff --git a/internal/transport/transport.go b/internal/transport/transport.go index c071bb4..dae7f3b 100644 --- a/internal/transport/transport.go +++ b/internal/transport/transport.go @@ -41,6 +41,10 @@ type HTTPTransport struct { // Callbacks callbackEndpoint string callbackHandler MessageCallbackHandler + + // Pending callbacks (for SendWithCallback) + pendingCallbacks map[string]chan *types.UnifiedSynapseMessage + pendingMu sync.RWMutex } // HTTPTransportConfig holds HTTP transport configuration @@ -68,11 +72,12 @@ func NewHTTPTransport(config *HTTPTransportConfig) *HTTPTransport { } t := &HTTPTransport{ - baseURL: config.BaseURL, - timeout: config.Timeout, - maxRetries: config.MaxRetries, - retryDelay: config.RetryDelay, - authSecret: config.AuthSecret, + baseURL: config.BaseURL, + timeout: config.Timeout, + maxRetries: config.MaxRetries, + retryDelay: config.RetryDelay, + authSecret: config.AuthSecret, + pendingCallbacks: make(map[string]chan *types.UnifiedSynapseMessage), client: &http.Client{ Timeout: config.Timeout, }, @@ -177,6 +182,18 @@ func (t *HTTPTransport) SendWithCallback( correlationID = msg.ID } + // Register pending callback + t.pendingMu.Lock() + t.pendingCallbacks[correlationID] = respChan + t.pendingMu.Unlock() + + // Cleanup on exit + defer func() { + t.pendingMu.Lock() + delete(t.pendingCallbacks, correlationID) + t.pendingMu.Unlock() + }() + // Start timeout goroutine go func() { select { @@ -240,7 +257,33 @@ func (t *HTTPTransport) HandleCallback(w http.ResponseWriter, r *http.Request) { } } - // Handle message + // Check if this is a pending callback response + correlationID := msg.CorrelationID + if correlationID == "" { + correlationID = msg.ID + } + + t.pendingMu.RLock() + ch, pending := t.pendingCallbacks[correlationID] + t.pendingMu.RUnlock() + + if pending { + // Send response to waiting goroutine + select { + case ch <- &msg: + default: + } + // Send success response + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + json.NewEncoder(w).Encode(map[string]interface{}{ + "success": true, + "message": "callback received", + }) + return + } + + // Handle message via callback handler if t.callbackHandler != nil { if err := t.callbackHandler.HandleMessage(&msg); err != nil { log.Printf("[Transport] Failed to handle message: %v", err)