feat: add fake send connector contract
This commit is contained in:
@@ -51,12 +51,16 @@ type SendMessageAuditEvent struct {
|
||||
TargetRef string `json:"target_ref"`
|
||||
ExecutionMode string `json:"execution_mode"`
|
||||
Connector string `json:"connector"`
|
||||
ConnectorMode string `json:"connector_mode"`
|
||||
ConnectorStage string `json:"connector_stage"`
|
||||
ContentSHA256 string `json:"content_sha256"`
|
||||
ContentLength int `json:"content_length"`
|
||||
IdempotencyKeySHA256 string `json:"idempotency_key_sha256"`
|
||||
SendStatus string `json:"send_status"`
|
||||
Result string `json:"result"`
|
||||
AckRef string `json:"ack_ref,omitempty"`
|
||||
ErrorCode string `json:"error_code,omitempty"`
|
||||
ErrorMessage string `json:"error_message,omitempty"`
|
||||
AuditRef string `json:"audit_ref"`
|
||||
StartedAt string `json:"started_at"`
|
||||
FinishedAt string `json:"finished_at"`
|
||||
@@ -105,6 +109,16 @@ func RegisterISphereSendMessageToolWithState(server *mcp.Server, auditSink SendM
|
||||
if idempotencyStore == nil {
|
||||
idempotencyStore = defaultSendMessageIdempotencyStore()
|
||||
}
|
||||
RegisterISphereSendMessageToolWithStateAndConnector(server, auditSink, idempotencyStore, nil)
|
||||
}
|
||||
|
||||
func RegisterISphereSendMessageToolWithStateAndConnector(server *mcp.Server, auditSink SendMessageAuditSink, idempotencyStore SendMessageIdempotencyStore, connector SendMessageConnector) {
|
||||
if auditSink == nil {
|
||||
auditSink = defaultSendMessageAuditSink()
|
||||
}
|
||||
if idempotencyStore == nil {
|
||||
idempotencyStore = defaultSendMessageIdempotencyStore()
|
||||
}
|
||||
|
||||
mcp.AddTool[SendMessageArgs, map[string]any](server, &mcp.Tool{
|
||||
Name: ToolNameSendMessage,
|
||||
@@ -116,12 +130,16 @@ func RegisterISphereSendMessageToolWithState(server *mcp.Server, auditSink SendM
|
||||
return nil, nil, err
|
||||
}
|
||||
finished := time.Now().UTC()
|
||||
response, event := sendMessagePreviewResponse(normalized, started, finished)
|
||||
response, event := sendMessagePreviewResponse(normalized, started, finished, connector != nil)
|
||||
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 normalized.ExecutionMode == sendMessageProductionMode && connector != nil && !decision.Duplicate && !decision.Conflict {
|
||||
result, connectorErr := connector.ExecuteSendMessage(ctx, sendMessageConnectorRequestFromNormalized(normalized))
|
||||
applySendMessageConnectorResult(response, &event, result, connectorErr)
|
||||
}
|
||||
if err := auditSink.AppendSendMessageAudit(ctx, event); err != nil {
|
||||
return nil, nil, fmt.Errorf("%s audit append failed: %w", ToolNameSendMessage, err)
|
||||
}
|
||||
@@ -325,17 +343,21 @@ func normalizeSendMessageExecutionMode(value string) (string, error) {
|
||||
}
|
||||
}
|
||||
|
||||
func sendMessagePreviewResponse(input normalizedSendMessageArgs, started time.Time, finished time.Time) (map[string]any, SendMessageAuditEvent) {
|
||||
func sendMessagePreviewResponse(input normalizedSendMessageArgs, started time.Time, finished time.Time, connectorAvailable bool) (map[string]any, SendMessageAuditEvent) {
|
||||
sideEffects := sendMessageNoSideEffects()
|
||||
status := "planned"
|
||||
stage := sendMessagePreviewStage
|
||||
ok := true
|
||||
blockedReason := any(nil)
|
||||
if input.ExecutionMode == sendMessageProductionMode {
|
||||
ok = false
|
||||
status = "blocked"
|
||||
stage = sendMessageBlockedStage
|
||||
blockedReason = "production send is blocked until dynamic observation, idempotency audit, and one approved sandbox send gate pass"
|
||||
if connectorAvailable {
|
||||
stage = sendMessagePreviewStage
|
||||
} else {
|
||||
ok = false
|
||||
status = "blocked"
|
||||
stage = sendMessageBlockedStage
|
||||
blockedReason = "production send is blocked until dynamic observation, idempotency audit, and one approved sandbox send gate pass"
|
||||
}
|
||||
}
|
||||
auditRef := sendMessageAuditRef(input.ContentSHA256, input.IdempotencyKeySHA256)
|
||||
event := SendMessageAuditEvent{
|
||||
@@ -345,6 +367,7 @@ func sendMessagePreviewResponse(input normalizedSendMessageArgs, started time.Ti
|
||||
TargetRef: input.TargetRef,
|
||||
ExecutionMode: input.ExecutionMode,
|
||||
Connector: sendMessageConnector,
|
||||
ConnectorMode: sendMessageConnectorMode,
|
||||
ConnectorStage: stage,
|
||||
ContentSHA256: input.ContentSHA256,
|
||||
ContentLength: input.ContentLength,
|
||||
@@ -400,6 +423,54 @@ func sendMessagePreviewResponse(input normalizedSendMessageArgs, started time.Ti
|
||||
return response, event
|
||||
}
|
||||
|
||||
func applySendMessageConnectorResult(response map[string]any, event *SendMessageAuditEvent, result SendMessageConnectorResult, err error) {
|
||||
normalized := normalizeSendMessageConnectorResult(result, err)
|
||||
response["ok"] = normalized.Accepted
|
||||
response["send_status"] = normalized.Status
|
||||
response["connector_mode"] = normalized.ConnectorMode
|
||||
response["connector_stage"] = normalized.Status
|
||||
if normalized.Accepted {
|
||||
response["blocked_reason"] = nil
|
||||
} else if normalized.ErrorMessage != "" {
|
||||
response["blocked_reason"] = normalized.ErrorMessage
|
||||
}
|
||||
if normalized.AckRef != "" {
|
||||
response["ack_ref"] = normalized.AckRef
|
||||
}
|
||||
if normalized.ErrorCode != "" {
|
||||
response["error_code"] = normalized.ErrorCode
|
||||
}
|
||||
if normalized.ErrorMessage != "" {
|
||||
response["error_message"] = normalized.ErrorMessage
|
||||
}
|
||||
|
||||
audit, _ := response["audit"].(map[string]any)
|
||||
if audit != nil {
|
||||
audit["result"] = normalized.Status
|
||||
audit["connector_mode"] = normalized.ConnectorMode
|
||||
audit["connector_stage"] = normalized.Status
|
||||
if normalized.AckRef != "" {
|
||||
audit["ack_ref"] = normalized.AckRef
|
||||
}
|
||||
if normalized.ErrorCode != "" {
|
||||
audit["error_code"] = normalized.ErrorCode
|
||||
}
|
||||
if normalized.ErrorMessage != "" {
|
||||
audit["error_message"] = normalized.ErrorMessage
|
||||
}
|
||||
}
|
||||
|
||||
if event != nil {
|
||||
event.ConnectorMode = normalized.ConnectorMode
|
||||
event.ConnectorStage = normalized.Status
|
||||
event.SendStatus = normalized.Status
|
||||
event.Result = normalized.Status
|
||||
event.AckRef = normalized.AckRef
|
||||
event.ErrorCode = normalized.ErrorCode
|
||||
event.ErrorMessage = normalized.ErrorMessage
|
||||
}
|
||||
}
|
||||
|
||||
func sendMessageIdempotencyRecordFromAuditEvent(event SendMessageAuditEvent) SendMessageIdempotencyRecord {
|
||||
return SendMessageIdempotencyRecord{
|
||||
Tool: event.Tool,
|
||||
|
||||
@@ -146,6 +146,153 @@ func TestISphereSendMessageProductionIsBlockedWithoutSideEffects(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestISphereSendMessageFakeConnectorAcceptedAudit(t *testing.T) {
|
||||
audit := &fakeSendMessageAuditSink{}
|
||||
idempotency := newFakeSendMessageIdempotencyStore()
|
||||
connector := &fakeSendMessageConnector{
|
||||
result: SendMessageConnectorResult{
|
||||
Accepted: true,
|
||||
Status: "accepted",
|
||||
AckRef: "fake-ack-1",
|
||||
ConnectorMode: "fake",
|
||||
},
|
||||
}
|
||||
session, cleanup := connectToolsTestSession(t, func(server *mcp.Server) {
|
||||
RegisterISphereSendMessageToolWithStateAndConnector(server, audit, idempotency, connector)
|
||||
})
|
||||
defer cleanup()
|
||||
|
||||
content := "fake connector accepted"
|
||||
contentHash := sha256HexForSendMessageTest(content)
|
||||
callResult, err := session.CallTool(context.Background(), &mcp.CallToolParams{
|
||||
Name: ToolNameSendMessage,
|
||||
Arguments: map[string]any{
|
||||
"target_type": "direct",
|
||||
"target_id": "alice@imopenfire1-lanzhou",
|
||||
"content_text": content,
|
||||
"content_sha256": contentHash,
|
||||
"idempotency_key": "idem-fake-accepted-1",
|
||||
"execution_mode": "production",
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("call %s fake connector accepted: %v", ToolNameSendMessage, err)
|
||||
}
|
||||
if callResult.IsError {
|
||||
t.Fatalf("fake connector accepted result should be structured, got error: %+v", callResult)
|
||||
}
|
||||
|
||||
var decoded struct {
|
||||
OK bool `json:"ok"`
|
||||
SendStatus string `json:"send_status"`
|
||||
ConnectorMode string `json:"connector_mode"`
|
||||
ProductionSendEnabled bool `json:"production_send_enabled"`
|
||||
SideEffects map[string]any `json:"side_effects"`
|
||||
Audit map[string]any `json:"audit"`
|
||||
}
|
||||
payload, _ := json.Marshal(callResult.StructuredContent)
|
||||
if err := json.Unmarshal(payload, &decoded); err != nil {
|
||||
t.Fatalf("decode fake accepted payload %s: %v", payload, err)
|
||||
}
|
||||
if !decoded.OK || decoded.SendStatus != "accepted" || decoded.ConnectorMode != "fake" {
|
||||
t.Fatalf("unexpected fake accepted response: %s", payload)
|
||||
}
|
||||
if decoded.ProductionSendEnabled {
|
||||
t.Fatalf("fake connector must not flip default production_send_enabled: %s", payload)
|
||||
}
|
||||
if decoded.SideEffects["sent_message"] != false || decoded.SideEffects["clicked_ui"] != false || decoded.SideEffects["captured_network"] != false {
|
||||
t.Fatalf("fake connector should not report real side effects: %#v", decoded.SideEffects)
|
||||
}
|
||||
if decoded.Audit["ack_ref"] != "fake-ack-1" || decoded.Audit["result"] != "accepted" {
|
||||
t.Fatalf("unexpected fake accepted audit: %#v", decoded.Audit)
|
||||
}
|
||||
if strings.Contains(string(payload), content) || strings.Contains(string(payload), "idem-fake-accepted-1") {
|
||||
t.Fatalf("fake accepted payload leaked raw content or idempotency key: %s", payload)
|
||||
}
|
||||
if connector.calls != 1 {
|
||||
t.Fatalf("fake connector calls = %d, want 1", connector.calls)
|
||||
}
|
||||
if len(audit.events) != 1 {
|
||||
t.Fatalf("audit events = %+v, want one", audit.events)
|
||||
}
|
||||
event := audit.events[0]
|
||||
if event.Result != "accepted" || event.AckRef != "fake-ack-1" || event.ConnectorMode != "fake" {
|
||||
t.Fatalf("unexpected audit event: %+v", event)
|
||||
}
|
||||
if event.ContentText != "" || event.IdempotencyKey != "" {
|
||||
t.Fatalf("audit event leaked raw content or idempotency key: %+v", event)
|
||||
}
|
||||
}
|
||||
|
||||
func TestISphereSendMessageFakeConnectorFailureAudit(t *testing.T) {
|
||||
audit := &fakeSendMessageAuditSink{}
|
||||
idempotency := newFakeSendMessageIdempotencyStore()
|
||||
connector := &fakeSendMessageConnector{
|
||||
result: SendMessageConnectorResult{
|
||||
Accepted: false,
|
||||
Status: "failed",
|
||||
ErrorCode: "fake_rejected",
|
||||
ErrorMessage: "fake connector rejected the request",
|
||||
ConnectorMode: "fake",
|
||||
},
|
||||
}
|
||||
session, cleanup := connectToolsTestSession(t, func(server *mcp.Server) {
|
||||
RegisterISphereSendMessageToolWithStateAndConnector(server, audit, idempotency, connector)
|
||||
})
|
||||
defer cleanup()
|
||||
|
||||
content := "fake connector failed"
|
||||
contentHash := sha256HexForSendMessageTest(content)
|
||||
callResult, err := session.CallTool(context.Background(), &mcp.CallToolParams{
|
||||
Name: ToolNameSendMessage,
|
||||
Arguments: map[string]any{
|
||||
"target_type": "group",
|
||||
"target_id": "project-room@conference.imopenfire1-lanzhou",
|
||||
"content_text": content,
|
||||
"content_sha256": contentHash,
|
||||
"idempotency_key": "idem-fake-failed-1",
|
||||
"execution_mode": "production",
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("call %s fake connector failure: %v", ToolNameSendMessage, err)
|
||||
}
|
||||
if callResult.IsError {
|
||||
t.Fatalf("fake connector failure should be structured, got error: %+v", callResult)
|
||||
}
|
||||
|
||||
var decoded struct {
|
||||
OK bool `json:"ok"`
|
||||
SendStatus string `json:"send_status"`
|
||||
ErrorCode string `json:"error_code"`
|
||||
SideEffects map[string]any `json:"side_effects"`
|
||||
Audit map[string]any `json:"audit"`
|
||||
}
|
||||
payload, _ := json.Marshal(callResult.StructuredContent)
|
||||
if err := json.Unmarshal(payload, &decoded); err != nil {
|
||||
t.Fatalf("decode fake failure payload %s: %v", payload, err)
|
||||
}
|
||||
if decoded.OK || decoded.SendStatus != "failed" || decoded.ErrorCode != "fake_rejected" {
|
||||
t.Fatalf("unexpected fake failure response: %s", payload)
|
||||
}
|
||||
if decoded.SideEffects["sent_message"] != false || decoded.SideEffects["typed_text"] != false {
|
||||
t.Fatalf("fake failure should not report real side effects: %#v", decoded.SideEffects)
|
||||
}
|
||||
if decoded.Audit["error_code"] != "fake_rejected" || decoded.Audit["result"] != "failed" {
|
||||
t.Fatalf("unexpected fake failure audit: %#v", decoded.Audit)
|
||||
}
|
||||
if strings.Contains(string(payload), content) || strings.Contains(string(payload), "idem-fake-failed-1") {
|
||||
t.Fatalf("fake failure payload leaked raw content or idempotency key: %s", payload)
|
||||
}
|
||||
if len(audit.events) != 1 {
|
||||
t.Fatalf("audit events = %+v, want one", audit.events)
|
||||
}
|
||||
event := audit.events[0]
|
||||
if event.Result != "failed" || event.ErrorCode != "fake_rejected" || event.ConnectorMode != "fake" {
|
||||
t.Fatalf("unexpected audit event: %+v", event)
|
||||
}
|
||||
}
|
||||
|
||||
func TestISphereSendMessageRejectsContentHashMismatch(t *testing.T) {
|
||||
session, cleanup := connectToolsTestSession(t, func(server *mcp.Server) {
|
||||
RegisterISphereSendMessageToolWithState(server, &fakeSendMessageAuditSink{}, newFakeSendMessageIdempotencyStore())
|
||||
@@ -370,6 +517,17 @@ func (f *fakeSendMessageIdempotencyStore) ReserveSendMessageIdempotency(_ contex
|
||||
return SendMessageIdempotencyDecision{}, nil
|
||||
}
|
||||
|
||||
type fakeSendMessageConnector struct {
|
||||
result SendMessageConnectorResult
|
||||
err error
|
||||
calls int
|
||||
}
|
||||
|
||||
func (f *fakeSendMessageConnector) ExecuteSendMessage(_ context.Context, _ SendMessageConnectorRequest) (SendMessageConnectorResult, error) {
|
||||
f.calls++
|
||||
return f.result, f.err
|
||||
}
|
||||
|
||||
func sha256HexForSendMessageTest(value string) string {
|
||||
sum := sha256.Sum256([]byte(value))
|
||||
return hex.EncodeToString(sum[:])
|
||||
|
||||
77
internal/tools/send_message_connector.go
Normal file
77
internal/tools/send_message_connector.go
Normal file
@@ -0,0 +1,77 @@
|
||||
package tools
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
)
|
||||
|
||||
type SendMessageConnector interface {
|
||||
ExecuteSendMessage(context.Context, SendMessageConnectorRequest) (SendMessageConnectorResult, error)
|
||||
}
|
||||
|
||||
type SendMessageConnectorRequest struct {
|
||||
TargetType string
|
||||
TargetID string
|
||||
TargetRef string
|
||||
ContentSHA256 string
|
||||
ContentLength int
|
||||
IdempotencyKeySHA256 string
|
||||
ExecutionMode string
|
||||
}
|
||||
|
||||
type SendMessageConnectorResult struct {
|
||||
Accepted bool
|
||||
Status string
|
||||
AckRef string
|
||||
ErrorCode string
|
||||
ErrorMessage string
|
||||
ConnectorMode string
|
||||
}
|
||||
|
||||
func sendMessageConnectorRequestFromNormalized(input normalizedSendMessageArgs) SendMessageConnectorRequest {
|
||||
return SendMessageConnectorRequest{
|
||||
TargetType: input.TargetType,
|
||||
TargetID: input.TargetID,
|
||||
TargetRef: input.TargetRef,
|
||||
ContentSHA256: input.ContentSHA256,
|
||||
ContentLength: input.ContentLength,
|
||||
IdempotencyKeySHA256: input.IdempotencyKeySHA256,
|
||||
ExecutionMode: input.ExecutionMode,
|
||||
}
|
||||
}
|
||||
|
||||
func normalizeSendMessageConnectorResult(result SendMessageConnectorResult, err error) SendMessageConnectorResult {
|
||||
normalized := result
|
||||
normalized.Status = strings.TrimSpace(normalized.Status)
|
||||
normalized.AckRef = strings.TrimSpace(normalized.AckRef)
|
||||
normalized.ErrorCode = strings.TrimSpace(normalized.ErrorCode)
|
||||
normalized.ErrorMessage = strings.TrimSpace(normalized.ErrorMessage)
|
||||
normalized.ConnectorMode = strings.TrimSpace(normalized.ConnectorMode)
|
||||
if normalized.ConnectorMode == "" {
|
||||
normalized.ConnectorMode = sendMessageConnectorMode
|
||||
}
|
||||
if err != nil {
|
||||
normalized.Accepted = false
|
||||
if normalized.Status == "" {
|
||||
normalized.Status = "failed"
|
||||
}
|
||||
if normalized.ErrorCode == "" {
|
||||
normalized.ErrorCode = "connector_error"
|
||||
}
|
||||
if normalized.ErrorMessage == "" {
|
||||
normalized.ErrorMessage = err.Error()
|
||||
}
|
||||
return normalized
|
||||
}
|
||||
if normalized.Status == "" {
|
||||
if normalized.Accepted {
|
||||
normalized.Status = "accepted"
|
||||
} else {
|
||||
normalized.Status = "failed"
|
||||
}
|
||||
}
|
||||
if !normalized.Accepted && normalized.ErrorCode == "" {
|
||||
normalized.ErrorCode = "connector_rejected"
|
||||
}
|
||||
return normalized
|
||||
}
|
||||
Reference in New Issue
Block a user