Align Go and Python prekey publish operation IDs with payload semantics

anp/direct_e2ee/client.py

@@ -65,7 +65,7 @@ def __init__(
 
     def publish_prekey_bundle(self) -> Dict[str, Any]:
         bundle = self._prekey_manager.ensure_fresh_prekey_bundle()
-        return self._prekey_manager.publish_prekey_bundle(bundle)
+        return self._prekey_manager.publish_prekey_bundle(bundle, f"op-publish-{bundle.bundle_id}")
 
     def ensure_fresh_prekey_bundle(self) -> PrekeyBundle:
         return self._prekey_manager.ensure_fresh_prekey_bundle()

anp/direct_e2ee/prekey_manager.py

@@ -104,7 +104,13 @@ def build_prekey_bundle(
             proof=signed_bundle["proof"],
         )
 
-    def publish_prekey_bundle(self, bundle: PrekeyBundle) -> Dict[str, Any]:
+    def publish_prekey_bundle(self, bundle: PrekeyBundle, operation_id: str) -> Dict[str, Any]:
+        """Publish a signed prekey bundle over the RPC boundary.
+
+        ``operation_id`` identifies the complete publish payload: reuse it
+        when retrying the same payload and provide a new one when the payload
+        changes.
+        """
         if self._rpc_client is None:
             raise DirectE2eeError("RPC client is not configured", "rpc_unavailable")
         return self._rpc_client(
@@ -115,7 +121,7 @@ def publish_prekey_bundle(self, bundle: PrekeyBundle) -> Dict[str, Any]:
                     "profile": "anp.direct.e2ee.v1",
                     "security_profile": "transport-protected",
                     "sender_did": self._local_did,
-                    "operation_id": f"op-publish-{bundle.bundle_id}",
+                    "operation_id": operation_id,
                 },
                 "body": {
                     "prekey_bundle": bundle.to_dict(),
@@ -132,12 +138,12 @@ def ensure_fresh_prekey_bundle(self) -> PrekeyBundle:
             )
             bundle = self.build_prekey_bundle(signed_prekey)
             if self._rpc_client is not None:
-                self.publish_prekey_bundle(bundle)
+                self.publish_prekey_bundle(bundle, f"op-publish-{bundle.bundle_id}")
             return bundle
         _, metadata = latest
         bundle = self.build_prekey_bundle(metadata)
         if self._rpc_client is not None:
-            self.publish_prekey_bundle(bundle)
+            self.publish_prekey_bundle(bundle, f"op-publish-{bundle.bundle_id}")
         return bundle
 
     @staticmethod

anp/unittest/direct_e2ee/test_direct_e2ee.py

@@ -95,6 +95,31 @@ def test_prekey_bundle_round_trip(tmp_path: Path) -> None:
     PrekeyManager.verify_prekey_bundle(bundle, bob_doc)
 
 
+def test_publish_prekey_bundle_uses_explicit_operation_id(tmp_path: Path) -> None:
+    bob_doc, bob_keys = _build_identity("b.example", ["agents", "bob"])
+    bob_did = bob_doc["id"]
+    bob_signing_key, _ = _load_identity_keys(bob_keys)
+    rpc = FakeRpcClient({})
+    manager = PrekeyManager(
+        local_did=bob_did,
+        static_key_agreement_id=f"{bob_did}#key-3",
+        signing_private_key=bob_signing_key,
+        signing_verification_method=f"{bob_did}#key-1",
+        signed_prekey_store=FileSignedPrekeyStore(tmp_path / "spk"),
+        rpc_client=rpc,
+    )
+    _, signed_prekey = manager.generate_signed_prekey("spk-bob-001", "2026-04-07T00:00:00Z")
+    bundle = manager.build_prekey_bundle(signed_prekey, bundle_id="bundle-bob-001")
+
+    manager.publish_prekey_bundle(bundle, "op-publish-custom-001")
+    publish_params = rpc.calls[-1][1]
+    assert publish_params["meta"]["operation_id"] == "op-publish-custom-001"
+
+    ensured = manager.ensure_fresh_prekey_bundle()
+    publish_params = rpc.calls[-1][1]
+    assert publish_params["meta"]["operation_id"] == f"op-publish-{ensured.bundle_id}"
+
+
 def test_session_init_and_follow_up_round_trip(tmp_path: Path) -> None:
     alice_doc, alice_keys = _build_identity("a.example", ["agents", "alice"])
     bob_doc, bob_keys = _build_identity("b.example", ["agents", "bob"])

golang/direct_e2ee/direct_e2ee_test.go

@@ -131,6 +131,76 @@ func TestPublishPrekeyBundleDoesNotStripOneTimePrekeysOnFailure(t *testing.T) {
 	}
 }
 
+func TestPublishPrekeyBundleOperationIDIdentifiesCompletePayload(t *testing.T) {
+	bobBundle, err := authentication.CreateDidWBADocument("b.example", didOptions("b.example", "bob"))
+	if err != nil {
+		t.Fatalf("CreateDidWBADocument failed: %v", err)
+	}
+	bobDID := stringValue(bobBundle.DidDocument["id"])
+	signingPrivateKey, err := anp.PrivateKeyFromPEM(bobBundle.Keys[authentication.VMKeyAuth].PrivateKeyPEM)
+	if err != nil {
+		t.Fatalf("PrivateKeyFromPEM failed: %v", err)
+	}
+	signedPrekeyStore, err := NewFileSignedPrekeyStore(filepath.Join(t.TempDir(), "spk"))
+	if err != nil {
+		t.Fatalf("NewFileSignedPrekeyStore failed: %v", err)
+	}
+	oneTimePrekeyStore, err := NewFileOneTimePrekeyStore(filepath.Join(t.TempDir(), "opk"))
+	if err != nil {
+		t.Fatalf("NewFileOneTimePrekeyStore failed: %v", err)
+	}
+	serviceDID, err := messageServiceDIDFromDocument(bobBundle.DidDocument)
+	if err != nil {
+		t.Fatalf("messageServiceDIDFromDocument failed: %v", err)
+	}
+	rpc := &fakeRPCClient{expectedServiceDID: serviceDID}
+	manager := NewPrekeyManager(bobDID, serviceDID, bobDID+"#"+authentication.VMKeyE2EEAgreement, signingPrivateKey, bobDID+"#"+authentication.VMKeyAuth, signedPrekeyStore, rpc.Call, oneTimePrekeyStore)
+	_, signedPrekey, err := manager.GenerateSignedPrekey("spk-bob-001", "2026-04-07T00:00:00Z")
+	if err != nil {
+		t.Fatalf("GenerateSignedPrekey failed: %v", err)
+	}
+	bundle, err := manager.BuildPrekeyBundle(signedPrekey, "bundle-bob-001", "")
+	if err != nil {
+		t.Fatalf("BuildPrekeyBundle failed: %v", err)
+	}
+
+	// A bundle-only payload derives its operation ID from the bundle ID alone.
+	if _, err := manager.PublishPrekeyBundle(bundle); err != nil {
+		t.Fatalf("PublishPrekeyBundle failed: %v", err)
+	}
+	if got := rpc.publishOperationIDs[0]; got != "op-publish-bundle-bob-001" {
+		t.Fatalf("operation_id without OPKs = %q, want op-publish-bundle-bob-001", got)
+	}
+
+	// With an OPK sidecar, exact retries of the same payload reuse the ID.
+	if _, err := manager.EnsureFreshOneTimePrekeys(2); err != nil {
+		t.Fatalf("EnsureFreshOneTimePrekeys failed: %v", err)
+	}
+	for i := 0; i < 2; i++ {
+		if _, err := manager.PublishPrekeyBundle(bundle); err != nil {
+			t.Fatalf("PublishPrekeyBundle with OPKs failed: %v", err)
+		}
+	}
+	withOPKs := rpc.publishOperationIDs[1]
+	if !strings.HasPrefix(withOPKs, "op-publish-bundle-bob-001-opks-") {
+		t.Fatalf("operation_id with OPKs = %q, want op-publish-bundle-bob-001-opks-*", withOPKs)
+	}
+	if rpc.publishOperationIDs[2] != withOPKs {
+		t.Fatalf("exact retry used %q, want reused operation ID %q", rpc.publishOperationIDs[2], withOPKs)
+	}
+
+	// A replenished OPK sidecar changes the payload, so a new ID is required.
+	if _, _, err := manager.GenerateOneTimePrekey("opk-extra-001"); err != nil {
+		t.Fatalf("GenerateOneTimePrekey failed: %v", err)
+	}
+	if _, err := manager.PublishPrekeyBundle(bundle); err != nil {
+		t.Fatalf("PublishPrekeyBundle with replenished OPKs failed: %v", err)
+	}
+	if got := rpc.publishOperationIDs[3]; got == withOPKs {
+		t.Fatalf("replenished OPK sidecar reused operation ID %q, want a new one", got)
+	}
+}
+
 func TestSessionInitAndFollowUpRoundTrip(t *testing.T) {
 	aliceDoc, bobDoc, aliceStatic, bobStatic, bobSPK, bundle := buildSessionFixtures(t)
 	aliceDID := stringValue(aliceDoc["id"])
@@ -440,6 +510,7 @@ type fakeRPCClient struct {
 	totalGetPrekeyCalls   int
 	failPublishWithOPK    bool
 	publishAttempts       int
+	publishOperationIDs   []string
 	getPrekeyOperationIDs []string
 	calls                 [][2]any
 }
@@ -450,6 +521,7 @@ func (f *fakeRPCClient) Call(method string, params map[string]any) (map[string]a
 	case "direct.e2ee.publish_prekey_bundle":
 		f.publishAttempts++
 		meta := params["meta"].(map[string]any)
+		f.publishOperationIDs = append(f.publishOperationIDs, stringValue(meta["operation_id"]))
 		target := meta["target"].(map[string]any)
 		if target["kind"] != "service" || target["did"] != f.expectedServiceDID {
 			return nil, invalidField("publish_prekey_bundle target")

golang/direct_e2ee/prekey_manager.go

@@ -2,8 +2,10 @@ package directe2ee
 
 import (
 	"crypto/ecdh"
+	"crypto/sha256"
 	"fmt"
 	"sort"
+	"strings"
 	"time"
 
 	anp "github.com/agent-network-protocol/anp/golang"
@@ -84,6 +86,10 @@ func (m *PrekeyManager) BuildPrekeyBundle(signedPrekey SignedPrekey, bundleID st
 }
 
 // PublishPrekeyBundle publishes a prekey bundle over the RPC boundary.
+// The publish operation ID identifies the complete publish payload: it is
+// derived from the bundle ID plus a digest of the OPK sidecar key IDs, so an
+// exact retry of the same payload reuses the ID while a replenished OPK
+// sidecar produces a new one.
 func (m *PrekeyManager) PublishPrekeyBundle(bundle PrekeyBundle) (map[string]any, error) {
 	if m.rpcClient == nil {
 		return nil, &Error{Code: "rpc_unavailable", Message: "RPC client is not configured"}
@@ -96,7 +102,22 @@ func (m *PrekeyManager) PublishPrekeyBundle(bundle PrekeyBundle) (map[string]any
 	if len(oneTimePrekeys) > 0 {
 		body["one_time_prekeys"] = oneTimePrekeysToAny(oneTimePrekeys)
 	}
-	return m.rpcClient("direct.e2ee.publish_prekey_bundle", map[string]any{"meta": map[string]any{"anp_version": "1.0", "profile": "anp.direct.e2ee.v1", "security_profile": "transport-protected", "sender_did": m.localDID, "target": map[string]any{"kind": "service", "did": m.localServiceDID}, "operation_id": "op-publish-" + bundle.BundleID}, "body": body})
+	return m.rpcClient("direct.e2ee.publish_prekey_bundle", map[string]any{"meta": map[string]any{"anp_version": "1.0", "profile": "anp.direct.e2ee.v1", "security_profile": "transport-protected", "sender_did": m.localDID, "target": map[string]any{"kind": "service", "did": m.localServiceDID}, "operation_id": publishOperationID(bundle.BundleID, oneTimePrekeys)}, "body": body})
+}
+
+// publishOperationID derives the publish operation ID from the complete
+// publish payload rather than the stable bundle ID alone. oneTimePrekeys must
+// be sorted by key ID (listOneTimePrekeys already guarantees this).
+func publishOperationID(bundleID string, oneTimePrekeys []OneTimePrekey) string {
+	if len(oneTimePrekeys) == 0 {
+		return "op-publish-" + bundleID
+	}
+	keyIDs := make([]string, 0, len(oneTimePrekeys))
+	for _, oneTimePrekey := range oneTimePrekeys {
+		keyIDs = append(keyIDs, oneTimePrekey.KeyID)
+	}
+	digest := sha256.Sum256([]byte(strings.Join(keyIDs, ",")))
+	return fmt.Sprintf("op-publish-%s-opks-%x", bundleID, digest[:4])
 }
 
 // EnsureFreshPrekeyBundle returns the latest prekey bundle or creates one.