feat: add optional final native-frame admission hook

docs/native-send-admission.md

@@ -0,0 +1,28 @@
+# Optional native send admission (dark integration seam)
+
+`SendRequestExtra.NativeSend` is process-local and nil by default. The existing
+send path is unchanged when nil. Initial support is one ordinary text DM with
+an explicitly preallocated message ID; group, peer, bot, media and extra-node
+paths are refused before preparation. This is not a renderer or a grant.
+
+`Admit(ctx, id, write)` runs after native preparation and after the Noise socket
+write lock. It must synchronously invoke `write` at most once inside the final
+local/authority fence. Returning without invocation refuses the send. A retained
+callback cannot invoke after Admit returns. A refusal consumes no Noise nonce
+or first-frame header. `Entered()` is a non-blocking notification immediately
+before the actual WebSocket write call; it is evidence of entry, not provider
+acceptance. It cannot prove whether bytes reached a provider.
+
+The hook returns before the provider ACK wait. Callers must not hold Campaign
+locks through that wait. Persist an invocation CAS marker and preallocated ID
+before writing; an existing marker with no durable outcome is unknown and must
+never be reinvoked. No goroutine-start or AMQP-publish callback counts as entry.
+A frame-write failure or post-entry authority loss is not proof of non-delivery.
+
+Guarded sends do not populate HyperMeow's retry content cache/store and do not
+use the disconnect frame resend path. Integrators must also keep their
+`GetMessageForRetry` fallback from returning guarded Campaign content and must
+not implement an automatic retry around this API. Durable retry/receipt policy
+and the regional invocation/result outbox belong to the owning application.
+No Campaign consumer, positive authority adapter, database schema or runtime
+mount is supplied here. This seam alone does not establish native acceptance.

native_send.go

@@ -0,0 +1,63 @@
+package whatsmeow
+
+import (
+	"context"
+	"errors"
+
+	"google.golang.org/protobuf/reflect/protoreflect"
+
+	waBinary "github.com/polymorfa/hypermeow/binary"
+	"github.com/polymorfa/hypermeow/proto/waE2E"
+	"github.com/polymorfa/hypermeow/types"
+)
+
+var ErrNativeSendUnsupported = errors.New("native admission requires a preallocated ordinary text DM")
+
+// NativeSendHooks are process-local. Admit must synchronously invoke write at
+// most once under the caller's final local/authority fence. Entered is a short
+// notification immediately before the actual socket write; it must not block.
+// The hooks finish before SendMessage waits for the provider ACK.
+type NativeSendHooks struct {
+	Admit   func(context.Context, types.MessageID, func() error) error
+	Entered func()
+}
+
+func (cli *Client) sendNativeMessageNode(ctx context.Context, node waBinary.Node, id types.MessageID, hooks *NativeSendHooks) ([]byte, error) {
+	if hooks == nil {
+		return cli.sendNodeAndGetData(ctx, node)
+	}
+	if hooks.Admit == nil || hooks.Entered == nil || id == "" {
+		return nil, ErrNativeSendUnsupported
+	}
+	cli.socketLock.RLock()
+	sock := cli.socket
+	cli.socketLock.RUnlock()
+	// A shadow relay is a separate transport, not proof of native socket entry.
+	if sock == nil {
+		return nil, ErrNotConnected
+	}
+	payload, err := waBinary.Marshal(node)
+	if err != nil {
+		return nil, err
+	}
+	err = sock.SendFrameAdmitted(ctx, payload, func(ctx context.Context, write func() error) error {
+		return hooks.Admit(ctx, id, write)
+	}, hooks.Entered)
+	return payload, err
+}
+
+// Restrict the initial hook to one ordinary text payload. Unknown fields and
+// sibling message variants must not smuggle additional native effects.
+func validNativeText(message *waE2E.Message) bool {
+	if message == nil || message.Conversation == nil || len(message.ProtoReflect().GetUnknown()) != 0 {
+		return false
+	}
+	valid := true
+	message.ProtoReflect().Range(func(field protoreflect.FieldDescriptor, _ protoreflect.Value) bool {
+		if field.Name() != "conversation" {
+			valid = false
+		}
+		return valid
+	})
+	return valid
+}

native_send_test.go

@@ -0,0 +1,39 @@
+package whatsmeow
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"google.golang.org/protobuf/proto"
+
+	"github.com/polymorfa/hypermeow/proto/waE2E"
+	"github.com/polymorfa/hypermeow/types"
+)
+
+func TestNativeSendRejectsUnsupportedBeforePreparation(t *testing.T) {
+	cli := &Client{}
+	to := types.NewJID("15550000000", types.DefaultUserServer)
+	hooks := &NativeSendHooks{Admit: func(context.Context, types.MessageID, func() error) error {
+		t.Fatal("unexpected admission")
+		return nil
+	}, Entered: func() { t.Fatal("unexpected entry") }}
+	for _, tc := range []struct {
+		name    string
+		message *waE2E.Message
+		extra   SendRequestExtra
+	}{
+		{"missing ID", &waE2E.Message{Conversation: proto.String("text")}, SendRequestExtra{NativeSend: hooks}},
+		{"media", &waE2E.Message{ImageMessage: &waE2E.ImageMessage{}}, SendRequestExtra{ID: "original", NativeSend: hooks}},
+		{"sibling", &waE2E.Message{Conversation: proto.String("text"), ImageMessage: &waE2E.ImageMessage{}}, SendRequestExtra{ID: "original", NativeSend: hooks}},
+		{"peer", &waE2E.Message{Conversation: proto.String("text")}, SendRequestExtra{ID: "original", NativeSend: hooks, Peer: true}},
+		{"missing callback", &waE2E.Message{Conversation: proto.String("text")}, SendRequestExtra{ID: "original", NativeSend: &NativeSendHooks{}}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := cli.SendMessage(t.Context(), to, tc.message, tc.extra)
+			if !errors.Is(err, ErrNativeSendUnsupported) {
+				t.Fatalf("wrong refusal: %v", err)
+			}
+		})
+	}
+}

send.go

@@ -148,6 +148,10 @@ func setParticipantHashMismatch(resp *SendResponse, sent, acknowledged string) b
 //
 // Trying to add multiple extra parameters will return an error.
 type SendRequestExtra struct {
+	// NativeSend is an optional process-local final-frame admission. Supported
+	// only for a preallocated-ID ordinary one-to-one text send. Nil preserves
+	// the existing send path. It never crosses a serialized request.
+	NativeSend *NativeSendHooks
 	// The message ID to use when sending. If this is not provided, a random message ID will be generated
 	ID types.MessageID
 	// JID of the bot to be invoked (optional)
@@ -202,6 +206,15 @@ func (cli *Client) SendMessage(ctx context.Context, to types.JID, message *waE2E
 	} else if len(extra) == 1 {
 		req = extra[0]
 	}
+	if req.NativeSend != nil {
+		if req.ID == "" || req.NativeSend.Admit == nil || req.NativeSend.Entered == nil || req.Peer || !req.InlineBotJID.IsEmpty() ||
+			(to.Server != types.DefaultUserServer && to.Server != types.HiddenUserServer) || !validNativeText(message) || req.Meta != nil || req.AdditionalNodes != nil || req.MediaHandle != "" || to.IsBot() {
+			err = ErrNativeSendUnsupported
+			return
+		}
+		hooks := *req.NativeSend
+		req.NativeSend = &hooks
+	}
 	if to.Device > 0 && !req.Peer {
 		err = ErrRecipientADJID
 		return
@@ -300,6 +313,8 @@ func (cli *Client) SendMessage(ctx context.Context, to types.JID, message *waE2E
 		}
 	}
 
+	extraParams.nativeSend = req.NativeSend
+
 	var groupParticipants []types.JID
 	if to.Server == types.GroupServer || to.Server == types.BroadcastServer {
 		start := time.Now()
@@ -377,7 +392,7 @@ func (cli *Client) SendMessage(ctx context.Context, to types.JID, message *waE2E
 	defer cli.messageSendLock.Unlock()
 
 	// Peer message retries aren't implemented yet
-	if !req.Peer {
+	if !req.Peer && req.NativeSend == nil {
 		err = cli.addRecentMessage(ctx, to, req.ID, message, nil)
 		if err != nil {
 			return
@@ -434,6 +449,10 @@ func (cli *Client) SendMessage(ctx context.Context, to types.JID, message *waE2E
 		return
 	}
 	resp.DebugTimings.Resp = time.Since(start)
+	if isDisconnectNode(respNode) && req.NativeSend != nil {
+		err = &DisconnectedError{Action: "guarded message send"}
+		return
+	}
 	if isDisconnectNode(respNode) {
 		start = time.Now()
 		respNode, err = cli.retryFrame(ctx, "message send", req.ID, data, respNode, 0)
@@ -741,6 +760,7 @@ func (cli *Client) sendNewsletter(
 }
 
 type nodeExtraParams struct {
+	nativeSend      *NativeSendHooks
 	botNode         *waBinary.Node
 	metaNode        *waBinary.Node
 	additionalNodes *[]waBinary.Node
@@ -892,7 +912,7 @@ func (cli *Client) sendDM(
 	}
 
 	start = time.Now()
-	data, err := cli.sendNodeAndGetData(ctx, *node)
+	data, err := cli.sendNativeMessageNode(ctx, *node, id, extraParams.nativeSend)
 	timings.Send = time.Since(start)
 	if err != nil {
 		return "", nil, fmt.Errorf("failed to send message node: %w", err)

socket/framesocket.go

@@ -123,6 +123,10 @@ func (fs *FrameSocket) Context() context.Context {
 }
 
 func (fs *FrameSocket) SendFrame(data []byte) error {
+	return fs.sendFrame(data, nil, nil)
+}
+
+func (fs *FrameSocket) sendFrame(data []byte, entered func(), writeCtx context.Context) error {
 	conn := fs.conn.Load()
 	if conn == nil {
 		return ErrSocketClosed
@@ -151,7 +155,13 @@ func (fs *FrameSocket) SendFrame(data []byte) error {
 	// Copy actual frame data
 	copy(wholeFrame[headerLength+FrameLengthSize:], data)
 
-	return conn.Write(fs.cancelCtx, websocket.MessageBinary, wholeFrame)
+	if entered != nil {
+		entered()
+	}
+	if writeCtx == nil {
+		writeCtx = fs.cancelCtx
+	}
+	return conn.Write(writeCtx, websocket.MessageBinary, wholeFrame)
 }
 
 func (fs *FrameSocket) frameComplete() {

socket/native_admission.go

@@ -0,0 +1,50 @@
+package socket
+
+import (
+	"context"
+	"errors"
+	"sync"
+)
+
+var ErrNativeAdmissionClosed = errors.New("native send admission closed")
+
+// SendFrameAdmitted joins a process-local admission to an actual frame write.
+// It runs after the Noise write lock, never around goroutine scheduling or ACK
+// waiting. A refusal consumes neither the Noise nonce nor the first-frame header.
+func (ns *NoiseSocket) SendFrameAdmitted(ctx context.Context, plaintext []byte, admit func(context.Context, func() error) error, entered func()) error {
+	if admit == nil || entered == nil {
+		return ErrNativeAdmissionClosed
+	}
+	ns.writeLock.Lock()
+	defer ns.writeLock.Unlock()
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	ciphertext := ns.writeKey.Seal(nil, generateIV(ns.writeCounter), plaintext, nil)
+	var mu sync.Mutex
+	active, invoked := true, false
+	var writeErr error
+	err := admit(ctx, func() error {
+		mu.Lock()
+		defer mu.Unlock()
+		if !active || invoked {
+			return ErrNativeAdmissionClosed
+		}
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+		invoked = true
+		ns.writeCounter++
+		writeErr = ns.fs.sendFrame(ciphertext, entered, ctx)
+		return writeErr
+	})
+	mu.Lock()
+	active = false
+	didInvoke := invoked
+	result := writeErr
+	mu.Unlock()
+	if !didInvoke && err == nil {
+		return ErrNativeAdmissionClosed
+	}
+	return errors.Join(err, result)
+}