灵隙(synapse) v1.0.0 版本

2026-06-29 00:48:45 +08:00
parent 6d1234efff
commit db716f94ac
6 changed files with 334 additions and 107 deletions
--- 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
+```

 ---

--- 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
--- 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)
 	}
 }
--- 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 {
--- 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:
--- 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)