admin/isphere-ai-bridge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: harden send preview idempotency

Browse Source
This commit is contained in:
zhaoyilun
2026-07-10 19:24:53 +08:00
parent 138615a29c
commit 93112d6d99
8 changed files with 482 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

212
internal/tools/isphere_send_message.go
View File

@@ -1,6 +1,7 @@
package tools
import (
"bufio"
"context"
"crypto/sha256"
"encoding/hex"
@@ -9,18 +10,21 @@ import (
"os"
"path/filepath"
"strings"
"sync"
"time"
"github.com/modelcontextprotocol/go-sdk/mcp"
)
const (
ToolNameSendMessage = "isphere_send_message"
EnvSendMessageAuditPath = "ISPHERE_SEND_AUDIT_PATH"
sendMessageConnector = "implatform-sidecar"
sendMessagePreviewStage = "preview"
sendMessageBlockedStage = "blocked"
sendMessageProductionMode = "production"
ToolNameSendMessage = "isphere_send_message"
EnvSendMessageAuditPath = "ISPHERE_SEND_AUDIT_PATH"
EnvSendMessageIdempotencyPath = "ISPHERE_SEND_IDEMPOTENCY_PATH"
sendMessageConnector = "implatform-sidecar"
sendMessageConnectorMode = "sandbox-only-shell"
sendMessagePreviewStage = "preview"
sendMessageBlockedStage = "blocked"
sendMessageProductionMode = "production"
)
type SendMessageArgs struct {
@@ -36,6 +40,10 @@ type SendMessageAuditSink interface {
AppendSendMessageAudit(ctx context.Context, event SendMessageAuditEvent) error
}
type SendMessageIdempotencyStore interface {
ReserveSendMessageIdempotency(ctx context.Context, record SendMessageIdempotencyRecord) (SendMessageIdempotencyDecision, error)
}
type SendMessageAuditEvent struct {
Tool string `json:"tool"`
TargetType string `json:"target_type"`
@@ -57,14 +65,46 @@ type SendMessageAuditEvent struct {
IdempotencyKey string `json:"-"`
}
type SendMessageIdempotencyRecord struct {
Tool string `json:"tool"`
TargetRef string `json:"target_ref"`
ExecutionMode string `json:"execution_mode"`
Connector string `json:"connector"`
ConnectorStage string `json:"connector_stage"`
ContentSHA256 string `json:"content_sha256"`
IdempotencyKeySHA256 string `json:"idempotency_key_sha256"`
AuditRef string `json:"audit_ref"`
StartedAt string `json:"started_at"`
}
type SendMessageIdempotencyDecision struct {
Duplicate bool
Conflict bool
FirstAuditRef string
FirstStartedAt string
}
type fileSendMessageAuditSink struct {
path string
}
type fileSendMessageIdempotencyStore struct {
path string
}
var sendMessageIdempotencyFileMu sync.Mutex
func RegisterISphereSendMessageTool(server *mcp.Server, auditSink SendMessageAuditSink) {
RegisterISphereSendMessageToolWithState(server, auditSink, nil)
}
func RegisterISphereSendMessageToolWithState(server *mcp.Server, auditSink SendMessageAuditSink, idempotencyStore SendMessageIdempotencyStore) {
if auditSink == nil {
auditSink = defaultSendMessageAuditSink()
}
if idempotencyStore == nil {
idempotencyStore = defaultSendMessageIdempotencyStore()
}
mcp.AddTool[SendMessageArgs, map[string]any](server, &mcp.Tool{
Name: ToolNameSendMessage,
@@ -77,6 +117,11 @@ func RegisterISphereSendMessageTool(server *mcp.Server, auditSink SendMessageAud
}
finished := time.Now().UTC()
response, event := sendMessagePreviewResponse(normalized, started, finished)
decision, err := idempotencyStore.ReserveSendMessageIdempotency(ctx, sendMessageIdempotencyRecordFromAuditEvent(event))
if err != nil {
return nil, nil, fmt.Errorf("%s idempotency reserve failed: %w", ToolNameSendMessage, err)
}
applySendMessageIdempotencyDecision(response, &event, decision)
if err := auditSink.AppendSendMessageAudit(ctx, event); err != nil {
return nil, nil, fmt.Errorf("%s audit append failed: %w", ToolNameSendMessage, err)
}
@@ -91,6 +136,13 @@ func defaultSendMessageAuditSink() SendMessageAuditSink {
return fileSendMessageAuditSink{path: filepath.Join("runs", "send-audit", "send-message-preview.jsonl")}
}
func defaultSendMessageIdempotencyStore() SendMessageIdempotencyStore {
if path := strings.TrimSpace(os.Getenv(EnvSendMessageIdempotencyPath)); path != "" {
return fileSendMessageIdempotencyStore{path: path}
}
return fileSendMessageIdempotencyStore{path: filepath.Join("runs", "send-audit", "send-message-idempotency.jsonl")}
}
func (s fileSendMessageAuditSink) AppendSendMessageAudit(_ context.Context, event SendMessageAuditEvent) error {
if strings.TrimSpace(s.path) == "" {
return nil
@@ -113,6 +165,83 @@ func (s fileSendMessageAuditSink) AppendSendMessageAudit(_ context.Context, even
return nil
}
func (s fileSendMessageIdempotencyStore) ReserveSendMessageIdempotency(_ context.Context, record SendMessageIdempotencyRecord) (SendMessageIdempotencyDecision, error) {
if strings.TrimSpace(s.path) == "" {
return SendMessageIdempotencyDecision{}, nil
}
if record.IdempotencyKeySHA256 == "" {
return SendMessageIdempotencyDecision{}, fmt.Errorf("idempotency_key_sha256 is required")
}
sendMessageIdempotencyFileMu.Lock()
defer sendMessageIdempotencyFileMu.Unlock()
if err := os.MkdirAll(filepath.Dir(s.path), 0o755); err != nil {
return SendMessageIdempotencyDecision{}, err
}
existing, err := readSendMessageIdempotencyRecords(s.path)
if err != nil {
return SendMessageIdempotencyDecision{}, err
}
for _, candidate := range existing {
if candidate.IdempotencyKeySHA256 != record.IdempotencyKeySHA256 {
continue
}
decision := SendMessageIdempotencyDecision{
FirstAuditRef: candidate.AuditRef,
FirstStartedAt: candidate.StartedAt,
}
if candidate.TargetRef == record.TargetRef && candidate.ContentSHA256 == record.ContentSHA256 {
decision.Duplicate = true
return decision, nil
}
decision.Conflict = true
return decision, nil
}
payload, err := json.Marshal(record)
if err != nil {
return SendMessageIdempotencyDecision{}, err
}
file, err := os.OpenFile(s.path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
if err != nil {
return SendMessageIdempotencyDecision{}, err
}
defer file.Close()
if _, err := file.Write(append(payload, '\n')); err != nil {
return SendMessageIdempotencyDecision{}, err
}
return SendMessageIdempotencyDecision{}, nil
}
func readSendMessageIdempotencyRecords(path string) ([]SendMessageIdempotencyRecord, error) {
file, err := os.Open(path)
if err != nil {
if os.IsNotExist(err) {
return nil, nil
}
return nil, err
}
defer file.Close()
var records []SendMessageIdempotencyRecord
scanner := bufio.NewScanner(file)
for scanner.Scan() {
line := strings.TrimSpace(scanner.Text())
if line == "" {
continue
}
var record SendMessageIdempotencyRecord
if err := json.Unmarshal([]byte(line), &record); err != nil {
return nil, fmt.Errorf("decode idempotency record: %w", err)
}
records = append(records, record)
}
if err := scanner.Err(); err != nil {
return nil, err
}
return records, nil
}
type normalizedSendMessageArgs struct {
TargetType string
TargetID string
@@ -228,12 +357,14 @@ func sendMessagePreviewResponse(input normalizedSendMessageArgs, started time.Ti
SideEffects: sideEffects,
}
response := map[string]any{
"ok": ok,
"send_status": status,
"blocked_reason": blockedReason,
"execution_mode": input.ExecutionMode,
"connector": sendMessageConnector,
"connector_stage": stage,
"ok": ok,
"send_status": status,
"blocked_reason": blockedReason,
"execution_mode": input.ExecutionMode,
"connector": sendMessageConnector,
"connector_mode": sendMessageConnectorMode,
"connector_stage": stage,
"production_send_enabled": false,
"target": map[string]any{
"target_type": input.TargetType,
"target_id": input.TargetID,
@@ -245,6 +376,10 @@ func sendMessagePreviewResponse(input normalizedSendMessageArgs, started time.Ti
},
"idempotency": map[string]any{
"idempotency_key_sha256": input.IdempotencyKeySHA256,
"duplicate_detected": false,
"conflict_detected": false,
"first_audit_ref": nil,
"first_started_at": nil,
},
"side_effects": sideEffects,
"audit": map[string]any{
@@ -252,6 +387,7 @@ func sendMessagePreviewResponse(input normalizedSendMessageArgs, started time.Ti
"source": sendMessageConnector,
"execution_mode": input.ExecutionMode,
"connector": sendMessageConnector,
"connector_mode": sendMessageConnectorMode,
"connector_stage": stage,
"content_sha256": input.ContentSHA256,
"target_ref": input.TargetRef,
@@ -264,6 +400,58 @@ func sendMessagePreviewResponse(input normalizedSendMessageArgs, started time.Ti
return response, event
}
func sendMessageIdempotencyRecordFromAuditEvent(event SendMessageAuditEvent) SendMessageIdempotencyRecord {
return SendMessageIdempotencyRecord{
Tool: event.Tool,
TargetRef: event.TargetRef,
ExecutionMode: event.ExecutionMode,
Connector: event.Connector,
ConnectorStage: event.ConnectorStage,
ContentSHA256: event.ContentSHA256,
IdempotencyKeySHA256: event.IdempotencyKeySHA256,
AuditRef: event.AuditRef,
StartedAt: event.StartedAt,
}
}
func applySendMessageIdempotencyDecision(response map[string]any, event *SendMessageAuditEvent, decision SendMessageIdempotencyDecision) {
idempotency, _ := response["idempotency"].(map[string]any)
if idempotency == nil {
idempotency = map[string]any{}
response["idempotency"] = idempotency
}
idempotency["duplicate_detected"] = decision.Duplicate
idempotency["conflict_detected"] = decision.Conflict
if decision.FirstAuditRef != "" {
idempotency["first_audit_ref"] = decision.FirstAuditRef
}
if decision.FirstStartedAt != "" {
idempotency["first_started_at"] = decision.FirstStartedAt
}
if !decision.Conflict {
return
}
blockedReason := "idempotency key was already used for a different target or content"
response["ok"] = false
response["send_status"] = "blocked"
response["blocked_reason"] = blockedReason
response["connector_stage"] = sendMessageBlockedStage
audit, _ := response["audit"].(map[string]any)
if audit != nil {
audit["result"] = "blocked"
audit["connector_stage"] = sendMessageBlockedStage
audit["blocked_reason"] = blockedReason
}
if event != nil {
event.ConnectorStage = sendMessageBlockedStage
event.SendStatus = "blocked"
event.Result = "blocked"
}
}
func sendMessageNoSideEffects() map[string]any {
return map[string]any{
"sent_message": false,

202
internal/tools/isphere_send_message_test.go
View File

@@ -5,6 +5,7 @@ import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"strings"
"testing"
"github.com/modelcontextprotocol/go-sdk/mcp"
@@ -12,8 +13,9 @@ import (
func TestISphereSendMessagePreviewPlansAndAuditsWithoutRawContent(t *testing.T) {
audit := &fakeSendMessageAuditSink{}
idempotency := newFakeSendMessageIdempotencyStore()
session, cleanup := connectToolsTestSession(t, func(server *mcp.Server) {
RegisterISphereSendMessageTool(server, audit)
RegisterISphereSendMessageToolWithState(server, audit, idempotency)
})
defer cleanup()
@@ -90,8 +92,9 @@ func TestISphereSendMessagePreviewPlansAndAuditsWithoutRawContent(t *testing.T)
func TestISphereSendMessageProductionIsBlockedWithoutSideEffects(t *testing.T) {
audit := &fakeSendMessageAuditSink{}
idempotency := newFakeSendMessageIdempotencyStore()
session, cleanup := connectToolsTestSession(t, func(server *mcp.Server) {
RegisterISphereSendMessageTool(server, audit)
RegisterISphereSendMessageToolWithState(server, audit, idempotency)
})
defer cleanup()
@@ -145,7 +148,7 @@ func TestISphereSendMessageProductionIsBlockedWithoutSideEffects(t *testing.T) {
func TestISphereSendMessageRejectsContentHashMismatch(t *testing.T) {
session, cleanup := connectToolsTestSession(t, func(server *mcp.Server) {
RegisterISphereSendMessageTool(server, &fakeSendMessageAuditSink{})
RegisterISphereSendMessageToolWithState(server, &fakeSendMessageAuditSink{}, newFakeSendMessageIdempotencyStore())
})
defer cleanup()
@@ -165,6 +168,171 @@ func TestISphereSendMessageRejectsContentHashMismatch(t *testing.T) {
}
}
func TestISphereSendMessageDetectsDuplicateIdempotencyKey(t *testing.T) {
audit := &fakeSendMessageAuditSink{}
idempotency := newFakeSendMessageIdempotencyStore()
session, cleanup := connectToolsTestSession(t, func(server *mcp.Server) {
RegisterISphereSendMessageToolWithState(server, audit, idempotency)
})
defer cleanup()
content := "duplicate preview"
contentHash := sha256HexForSendMessageTest(content)
args := map[string]any{
"target_type": "direct",
"target_id": "alice@imopenfire1-lanzhou",
"content_text": content,
"content_sha256": contentHash,
"idempotency_key": "idem-duplicate-1",
"execution_mode": "preview",
}
if _, err := session.CallTool(context.Background(), &mcp.CallToolParams{Name: ToolNameSendMessage, Arguments: args}); err != nil {
t.Fatalf("first call: %v", err)
}
second, err := session.CallTool(context.Background(), &mcp.CallToolParams{Name: ToolNameSendMessage, Arguments: args})
if err != nil {
t.Fatalf("second call: %v", err)
}
if second.IsError {
t.Fatalf("duplicate should be structured preview metadata, got error: %+v", second)
}
var decoded struct {
OK bool `json:"ok"`
SendStatus string `json:"send_status"`
Idempotency map[string]any `json:"idempotency"`
SideEffects map[string]any `json:"side_effects"`
}
payload, _ := json.Marshal(second.StructuredContent)
if err := json.Unmarshal(payload, &decoded); err != nil {
t.Fatalf("decode duplicate payload %s: %v", payload, err)
}
if !decoded.OK || decoded.SendStatus != "planned" {
t.Fatalf("duplicate preview should remain planned/no-send: %s", payload)
}
if decoded.Idempotency["duplicate_detected"] != true || decoded.Idempotency["conflict_detected"] != false {
t.Fatalf("duplicate idempotency fields mismatch: %#v", decoded.Idempotency)
}
if decoded.Idempotency["first_audit_ref"] == "" {
t.Fatalf("duplicate response missing first_audit_ref: %#v", decoded.Idempotency)
}
if decoded.SideEffects["sent_message"] != false || decoded.SideEffects["clicked_ui"] != false {
t.Fatalf("duplicate preview should not have side effects: %#v", decoded.SideEffects)
}
if len(audit.events) != 2 || audit.events[1].Result != "planned" {
t.Fatalf("audit events = %+v, want two planned events", audit.events)
}
}
func TestISphereSendMessageBlocksIdempotencyKeyConflict(t *testing.T) {
audit := &fakeSendMessageAuditSink{}
idempotency := newFakeSendMessageIdempotencyStore()
session, cleanup := connectToolsTestSession(t, func(server *mcp.Server) {
RegisterISphereSendMessageToolWithState(server, audit, idempotency)
})
defer cleanup()
firstContent := "first body"
firstHash := sha256HexForSendMessageTest(firstContent)
_, err := session.CallTool(context.Background(), &mcp.CallToolParams{Name: ToolNameSendMessage, Arguments: map[string]any{
"target_type": "direct",
"target_id": "alice@imopenfire1-lanzhou",
"content_text": firstContent,
"content_sha256": firstHash,
"idempotency_key": "idem-conflict-1",
"execution_mode": "preview",
}})
if err != nil {
t.Fatalf("first call: %v", err)
}
secondContent := "different body"
secondHash := sha256HexForSendMessageTest(secondContent)
second, err := session.CallTool(context.Background(), &mcp.CallToolParams{Name: ToolNameSendMessage, Arguments: map[string]any{
"target_type": "direct",
"target_id": "alice@imopenfire1-lanzhou",
"content_text": secondContent,
"content_sha256": secondHash,
"idempotency_key": "idem-conflict-1",
"execution_mode": "preview",
}})
if err != nil {
t.Fatalf("conflict call: %v", err)
}
if second.IsError {
t.Fatalf("idempotency conflict should be structured blocked metadata, got error: %+v", second)
}
var decoded struct {
OK bool `json:"ok"`
SendStatus string `json:"send_status"`
BlockedReason string `json:"blocked_reason"`
ConnectorStage string `json:"connector_stage"`
Idempotency map[string]any `json:"idempotency"`
SideEffects map[string]any `json:"side_effects"`
}
payload, _ := json.Marshal(second.StructuredContent)
if err := json.Unmarshal(payload, &decoded); err != nil {
t.Fatalf("decode conflict payload %s: %v", payload, err)
}
if decoded.OK || decoded.SendStatus != "blocked" || decoded.ConnectorStage != "blocked" {
t.Fatalf("conflict should be blocked: %s", payload)
}
if !strings.Contains(strings.ToLower(decoded.BlockedReason), "idempotency") {
t.Fatalf("blocked reason should explain idempotency conflict: %q", decoded.BlockedReason)
}
if decoded.Idempotency["duplicate_detected"] != false || decoded.Idempotency["conflict_detected"] != true {
t.Fatalf("conflict idempotency fields mismatch: %#v", decoded.Idempotency)
}
if decoded.SideEffects["sent_message"] != false || decoded.SideEffects["typed_text"] != false {
t.Fatalf("conflict should not have side effects: %#v", decoded.SideEffects)
}
if len(audit.events) != 2 || audit.events[1].Result != "blocked" {
t.Fatalf("audit events = %+v, want second blocked event", audit.events)
}
}
func TestFileSendMessageIdempotencyStoreDetectsDuplicateAndConflict(t *testing.T) {
store := fileSendMessageIdempotencyStore{path: t.TempDir() + "\\send-idempotency.jsonl"}
first := SendMessageIdempotencyRecord{
Tool: ToolNameSendMessage,
TargetRef: "contact:alice@imopenfire1-lanzhou",
ContentSHA256: sha256HexForSendMessageTest("first body"),
IdempotencyKeySHA256: sha256HexForSendMessageTest("idem-file-store"),
AuditRef: "send-message:first",
StartedAt: "2026-07-10T00:00:00Z",
}
firstDecision, err := store.ReserveSendMessageIdempotency(context.Background(), first)
if err != nil {
t.Fatalf("reserve first: %v", err)
}
if firstDecision.Duplicate || firstDecision.Conflict {
t.Fatalf("first decision = %+v, want fresh", firstDecision)
}
duplicateDecision, err := store.ReserveSendMessageIdempotency(context.Background(), first)
if err != nil {
t.Fatalf("reserve duplicate: %v", err)
}
if !duplicateDecision.Duplicate || duplicateDecision.Conflict || duplicateDecision.FirstAuditRef != first.AuditRef {
t.Fatalf("duplicate decision = %+v", duplicateDecision)
}
conflict := first
conflict.ContentSHA256 = sha256HexForSendMessageTest("changed body")
conflict.AuditRef = "send-message:conflict"
conflictDecision, err := store.ReserveSendMessageIdempotency(context.Background(), conflict)
if err != nil {
t.Fatalf("reserve conflict: %v", err)
}
if !conflictDecision.Conflict || conflictDecision.Duplicate {
t.Fatalf("conflict decision = %+v", conflictDecision)
}
if conflictDecision.FirstAuditRef != first.AuditRef {
t.Fatalf("conflict first audit ref = %q, want %q", conflictDecision.FirstAuditRef, first.AuditRef)
}
}
type fakeSendMessageAuditSink struct {
events []SendMessageAuditEvent
}
@@ -174,6 +342,34 @@ func (f *fakeSendMessageAuditSink) AppendSendMessageAudit(_ context.Context, eve
return nil
}
type fakeSendMessageIdempotencyStore struct {
records []SendMessageIdempotencyRecord
}
func newFakeSendMessageIdempotencyStore() *fakeSendMessageIdempotencyStore {
return &fakeSendMessageIdempotencyStore{}
}
func (f *fakeSendMessageIdempotencyStore) ReserveSendMessageIdempotency(_ context.Context, record SendMessageIdempotencyRecord) (SendMessageIdempotencyDecision, error) {
for _, existing := range f.records {
if existing.IdempotencyKeySHA256 != record.IdempotencyKeySHA256 {
continue
}
decision := SendMessageIdempotencyDecision{
FirstAuditRef: existing.AuditRef,
FirstStartedAt: existing.StartedAt,
}
if existing.TargetRef == record.TargetRef && existing.ContentSHA256 == record.ContentSHA256 {
decision.Duplicate = true
return decision, nil
}
decision.Conflict = true
return decision, nil
}
f.records = append(f.records, record)
return SendMessageIdempotencyDecision{}, nil
}
func sha256HexForSendMessageTest(value string) string {
sum := sha256.Sum256([]byte(value))
return hex.EncodeToString(sum[:])