From 1f6dd97cd6f6f1de0fff6a767d7357541fc578e4 Mon Sep 17 00:00:00 2001
From: Pavel <177363085+pkcll@users.noreply.github.com>
Date: Fri, 16 Oct 2026 00:46:38 -0400
Subject: [PATCH 1/5] fix(chip-router): match atlas write-service
 Publish/PublishBatch semantics
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Aligns the router with the atlas chip-ingress write-service, addressing
the review findings on #2861:

- Publish (single event) now follows the write-service's all-or-nothing
  contract (atlas chip-ingress internal/server: Publish delegates to
  PublishCloudEvents with transaction_enabled=true): it fails the RPC
  when no subscribers are registered, when every forward fails, or when
  no subscriber accepts the event, and returns exactly one PublishResult
  on success. Previously it returned an empty PublishResponse with a nil
  error unconditionally, acking events that went nowhere.
- Failure RPCs use codes.Internal, the write-service's failure code, in
  place of codes.Unavailable.
- An empty batch short-circuits to a no-op success before the subscriber
  check, matching the write-service's empty-batch path.
- The per-event detail merge is now order-independent and sized to the
  batch: a later subscriber's results can extend detail beyond the first
  provider's length (previously trailing events were acked with no
  detail), and a detail-less success (empty results, nil error) now
  accepts the whole batch, clearing other subscribers' rejections —
  the intent documented in #2861 but not enforced by its merge.

Tests: single-event success/failure paths, empty batch, detail-less
acceptance, merge extension past the first provider's detail, rejection
surviving another subscriber's RPC failure; stub simplified to an
immutable struct.
---
 .../chiprouter/cmd/chip-router/main.go        |  97 +++++--
 .../chiprouter/cmd/chip-router/main_test.go   | 272 ++++++++++++++----
 2 files changed, 279 insertions(+), 90 deletions(-)

diff --git a/framework/components/chiprouter/cmd/chip-router/main.go b/framework/components/chiprouter/cmd/chip-router/main.go
index e680b6a86..e4f1e91c6 100644
--- a/framework/components/chiprouter/cmd/chip-router/main.go
+++ b/framework/components/chiprouter/cmd/chip-router/main.go
@@ -138,12 +138,26 @@ func run() error {
 	}
 }
 
+// Publish matches the write-service's single-event contract (atlas
+// chip-ingress internal/server): the RPC is all-or-nothing — it fails unless
+// at least one subscriber accepted the event, and returns exactly one
+// PublishResult on success.
 func (r *router) Publish(ctx context.Context, event *cepb.CloudEvent) (*chippb.PublishResponse, error) {
 	snapshot := r.snapshotSubscribers()
 	if len(snapshot) == 0 {
-		return &chippb.PublishResponse{}, nil
+		// Acking an event nothing carried would let a durable store delete it undelivered.
+		return nil, status.Error(codes.Internal, "no subscribers registered")
 	}
 
+	// anyForwarded: at least one subscriber's RPC succeeded. accepted: some
+	// subscriber accepted the event (an accepted result or a detail-less
+	// success) — sticky, so a later rejection cannot unset it. rejection:
+	// recorded only while no subscriber has accepted.
+	var mu sync.Mutex
+	anyForwarded := false
+	accepted := false
+	var rejection *chippb.PublishError
+
 	var group errgroup.Group
 	group.SetLimit(forwardParallel)
 	for _, sub := range snapshot {
@@ -151,32 +165,58 @@ func (r *router) Publish(ctx context.Context, event *cepb.CloudEvent) (*chippb.P
 			framework.L.Debug().Msgf("chip router forwarding event to subscriber id=%s name=%s endpoint=%s", sub.id, sub.name, sub.endpoint)
 			forwardCtx, cancel := context.WithTimeout(ctx, forwardTimeout)
 			defer cancel()
-			_, err := sub.client.Publish(forwardCtx, event)
+			resp, err := sub.client.Publish(forwardCtx, event)
 			if err != nil {
 				framework.L.Error().Msgf("chip router failed to forward event to subscriber id=%s name=%s endpoint=%s err=%v", sub.id, sub.name, sub.endpoint, err)
+				return nil
 			}
 			framework.L.Debug().Msgf("chip router forwarded event to subscriber id=%s", sub.id)
+			mu.Lock()
+			defer mu.Unlock()
+			anyForwarded = true
+			if results := resp.GetResults(); len(results) > 0 && results[0].GetError() != nil {
+				if !accepted && rejection == nil {
+					rejection = results[0].GetError()
+				}
+				return nil
+			}
+			accepted = true
 			return nil
 		})
 	}
 	_ = group.Wait()
-	return &chippb.PublishResponse{}, nil
+
+	if !anyForwarded {
+		return nil, status.Error(codes.Internal, fmt.Sprintf("all %d subscribers failed to accept the event", len(snapshot)))
+	}
+	if !accepted {
+		return nil, status.Error(codes.Internal, "event was not accepted by any subscriber")
+	}
+	return &chippb.PublishResponse{Results: []*chippb.PublishResult{{EventId: event.GetId()}}}, nil
 }
 
 func (r *router) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch) (*chippb.PublishResponse, error) {
+	// Matches the write-service: an empty batch is a no-op success and never
+	// reaches the subscribers.
+	if len(batch.GetEvents()) == 0 {
+		return &chippb.PublishResponse{Results: []*chippb.PublishResult{}}, nil
+	}
+
 	snapshot := r.snapshotSubscribers()
 	if len(snapshot) == 0 {
 		// Acking events nothing carried would let a durable store delete them undelivered.
-		return nil, status.Error(codes.Unavailable, "no subscribers registered")
+		return nil, status.Error(codes.Internal, "no subscribers registered")
 	}
 
-	// anyForwarded: at least one subscriber's RPC succeeded. detail: merged
-	// per-event results — an event is rejected only if every detail-providing
-	// subscriber rejected it; a subscriber accepting without per-event detail
-	// counts as accepting the whole batch.
+	// anyForwarded: at least one subscriber's RPC succeeded. anyBatchAccepted:
+	// a subscriber accepted the whole batch without per-event detail. detail:
+	// merged per-event results, index-aligned with the batch — an acceptance
+	// always wins over a rejection, and a rejection only fills an empty slot,
+	// so the outcome does not depend on subscriber completion order.
 	var mu sync.Mutex
 	anyForwarded := false
-	var detail []*chippb.PublishResult
+	anyBatchAccepted := false
+	detail := make([]*chippb.PublishResult, len(batch.GetEvents()))
 
 	var group errgroup.Group
 	group.SetLimit(forwardParallel)
@@ -194,19 +234,20 @@ func (r *router) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch
 			mu.Lock()
 			defer mu.Unlock()
 			anyForwarded = true
-			if results := resp.GetResults(); len(results) > 0 {
-				if detail == nil {
-					detail = results
-					return nil
+			results := resp.GetResults()
+			if len(results) == 0 {
+				// A detail-less success accepts the whole batch.
+				anyBatchAccepted = true
+				return nil
+			}
+			for i, res := range results {
+				if i >= len(detail) {
+					break
 				}
-				// An accepting result overrides a rejection for the same event.
-				for i := range detail {
-					if i >= len(results) {
-						break
-					}
-					if detail[i].GetError() != nil && results[i].GetError() == nil {
-						detail[i] = results[i]
-					}
+				if res.GetError() == nil {
+					detail[i] = res
+				} else if detail[i] == nil {
+					detail[i] = res
 				}
 			}
 			return nil
@@ -217,26 +258,24 @@ func (r *router) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch
 	// No subscriber accepted the batch: fail the RPC so every caller mode
 	// retains and retries.
 	if !anyForwarded {
-		return nil, status.Error(codes.Unavailable, fmt.Sprintf("all %d subscribers failed to accept the batch", len(snapshot)))
+		return nil, status.Error(codes.Internal, fmt.Sprintf("all %d subscribers failed to accept the batch", len(snapshot)))
 	}
 
+	anyRejected := false
 	results := make([]*chippb.PublishResult, 0, len(batch.GetEvents()))
 	for i, ev := range batch.GetEvents() {
 		result := &chippb.PublishResult{EventId: ev.GetId()}
-		if i < len(detail) && detail[i].GetError() != nil {
+		if !anyBatchAccepted && detail[i] != nil && detail[i].GetError() != nil {
 			result.Error = detail[i].GetError()
+			anyRejected = true
 		}
 		results = append(results, result)
 	}
 
 	// Transactional callers resolve the whole batch from the RPC outcome
 	// alone, so any unaccepted event must fail the RPC (all-or-nothing).
-	if batch.GetOptions().GetTransactionEnabled() {
-		for _, result := range results {
-			if result.Error != nil {
-				return nil, status.Error(codes.Unavailable, "not all events were accepted by a subscriber")
-			}
-		}
+	if batch.GetOptions().GetTransactionEnabled() && anyRejected {
+		return nil, status.Error(codes.Internal, "not all events were accepted by a subscriber")
 	}
 	return &chippb.PublishResponse{Results: results}, nil
 }
diff --git a/framework/components/chiprouter/cmd/chip-router/main_test.go b/framework/components/chiprouter/cmd/chip-router/main_test.go
index 3ce630356..1a14dfd83 100644
--- a/framework/components/chiprouter/cmd/chip-router/main_test.go
+++ b/framework/components/chiprouter/cmd/chip-router/main_test.go
@@ -2,7 +2,6 @@ package main
 
 import (
 	"context"
-	"sync"
 	"testing"
 
 	cepb "github.com/cloudevents/sdk-go/binding/format/protobuf/v2/pb"
@@ -12,21 +11,26 @@ import (
 	"google.golang.org/grpc/status"
 )
 
+// stubPublishClient is immutable after construction, so it is safe for the
+// router's concurrent forwards. A nil response makes Publish return the
+// write-service default for a single event (one accepted result) and
+// PublishBatch return a detail-less success (empty results).
 type stubPublishClient struct {
-	mu       sync.Mutex
-	batches  int
 	err      error
 	response *chippb.PublishResponse
 }
 
-func (s *stubPublishClient) Publish(context.Context, *cepb.CloudEvent, ...grpc.CallOption) (*chippb.PublishResponse, error) {
-	return &chippb.PublishResponse{}, nil
+func (s *stubPublishClient) Publish(_ context.Context, in *cepb.CloudEvent, _ ...grpc.CallOption) (*chippb.PublishResponse, error) {
+	if s.err != nil {
+		return nil, s.err
+	}
+	if s.response != nil {
+		return s.response, nil
+	}
+	return &chippb.PublishResponse{Results: []*chippb.PublishResult{{EventId: in.GetId()}}}, nil
 }
 
 func (s *stubPublishClient) PublishBatch(context.Context, *chippb.CloudEventBatch, ...grpc.CallOption) (*chippb.PublishResponse, error) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	s.batches++
 	if s.err != nil {
 		return nil, s.err
 	}
@@ -36,22 +40,34 @@ func (s *stubPublishClient) PublishBatch(context.Context, *chippb.CloudEventBatc
 	return &chippb.PublishResponse{}, nil
 }
 
-func (s *stubPublishClient) batchCount() int {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	return s.batches
-}
-
 func ptr(b bool) *bool { return &b }
 
+func eventOf(id string) *cepb.CloudEvent {
+	return &cepb.CloudEvent{Id: id, Source: "s", SpecVersion: "1.0", Type: "t"}
+}
+
 func batchOf(ids ...string) *chippb.CloudEventBatch {
 	events := make([]*cepb.CloudEvent, 0, len(ids))
 	for _, id := range ids {
-		events = append(events, &cepb.CloudEvent{Id: id, Source: "s", SpecVersion: "1.0", Type: "t"})
+		events = append(events, eventOf(id))
 	}
 	return &chippb.CloudEventBatch{Events: events}
 }
 
+func rejected(id, reason string) *chippb.PublishResult {
+	return &chippb.PublishResult{EventId: id, Error: &chippb.PublishError{Reason: reason}}
+}
+
+func requireCode(t *testing.T, err error, want codes.Code) {
+	t.Helper()
+	if err == nil {
+		t.Fatalf("want %s error, got nil", want)
+	}
+	if status.Code(err) != want {
+		t.Errorf("want %s, got %v", want, status.Code(err))
+	}
+}
+
 func TestPublishBatchPopulatesPerEventResults(t *testing.T) {
 	r := &router{subscribers: map[string]*subscriber{
 		"sink": {id: "sink", client: &stubPublishClient{}},
@@ -74,31 +90,34 @@ func TestPublishBatchPopulatesPerEventResults(t *testing.T) {
 	}
 }
 
-func TestPublishBatchNoSubscribersIsUnavailable(t *testing.T) {
+func TestPublishBatchEmptyBatchSucceedsWithoutSubscribers(t *testing.T) {
+	// The write-service treats an empty batch as a no-op success before any
+	// delivery attempt.
 	r := &router{subscribers: map[string]*subscriber{}}
 
-	_, err := r.PublishBatch(t.Context(), batchOf("e1"))
-	if err == nil {
-		t.Fatal("want error with no subscribers, got nil")
+	resp, err := r.PublishBatch(t.Context(), batchOf())
+	if err != nil {
+		t.Fatalf("PublishBatch: %v", err)
 	}
-	if status.Code(err) != codes.Unavailable {
-		t.Errorf("want Unavailable, got %v", status.Code(err))
+	if len(resp.Results) != 0 {
+		t.Errorf("want 0 results, got %d", len(resp.Results))
 	}
 }
 
-func TestPublishBatchAllForwardsFailedIsUnavailable(t *testing.T) {
+func TestPublishBatchNoSubscribersFailsWithInternal(t *testing.T) {
+	r := &router{subscribers: map[string]*subscriber{}}
+
+	_, err := r.PublishBatch(t.Context(), batchOf("e1"))
+	requireCode(t, err, codes.Internal)
+}
+
+func TestPublishBatchAllForwardsFailedFailsWithInternal(t *testing.T) {
 	r := &router{subscribers: map[string]*subscriber{
 		"sink": {id: "sink", client: &stubPublishClient{err: context.DeadlineExceeded}},
 	}}
 
-	// Transactional callers resolve the whole batch from the RPC outcome alone.
 	_, err := r.PublishBatch(t.Context(), batchOf("e1", "e2"))
-	if err == nil {
-		t.Fatal("want error when every forward fails, got nil")
-	}
-	if status.Code(err) != codes.Unavailable {
-		t.Errorf("want Unavailable, got %v", status.Code(err))
-	}
+	requireCode(t, err, codes.Internal)
 }
 
 func TestPublishBatchAggregatesDownstreamRejections(t *testing.T) {
@@ -108,10 +127,7 @@ func TestPublishBatchAggregatesDownstreamRejections(t *testing.T) {
 		"sink": {id: "sink", client: &stubPublishClient{response: &chippb.PublishResponse{
 			Results: []*chippb.PublishResult{
 				{EventId: "e1"},
-				{EventId: "e2", Error: &chippb.PublishError{
-					ErrorCode: chippb.PublishErrorCode_PUBLISH_ERROR_CODE_VALIDATION_FAILED,
-					Reason:    "invalid event",
-				}},
+				rejected("e2", "invalid event"),
 			},
 		}}},
 	}}
@@ -131,6 +147,105 @@ func TestPublishBatchAggregatesDownstreamRejections(t *testing.T) {
 	}
 }
 
+func TestPublishBatchRejectionSurvivesAnotherSubscribersRPCFailure(t *testing.T) {
+	// A rejects e1 per-event; B's forward fails. e1 is still reported
+	// rejected in partial-delivery mode — it was accepted nowhere.
+	r := &router{subscribers: map[string]*subscriber{
+		"a": {id: "a", client: &stubPublishClient{response: &chippb.PublishResponse{
+			Results: []*chippb.PublishResult{rejected("e1", "rejected by a")},
+		}}},
+		"b": {id: "b", client: &stubPublishClient{err: context.DeadlineExceeded}},
+	}}
+
+	resp, err := r.PublishBatch(t.Context(), batchOf("e1"))
+	if err != nil {
+		t.Fatalf("PublishBatch: %v", err)
+	}
+	if len(resp.Results) != 1 || resp.Results[0].Error == nil {
+		t.Errorf("want e1 rejected, got results=%+v", resp.Results)
+	}
+}
+
+func TestPublishBatchAnAcceptingSubscriberOverridesARejection(t *testing.T) {
+	// Two detail-providing subscribers disagree: the event is rejected by one
+	// and accepted by the other — acceptance wins.
+	r := &router{subscribers: map[string]*subscriber{
+		"a": {id: "a", client: &stubPublishClient{response: &chippb.PublishResponse{
+			Results: []*chippb.PublishResult{rejected("e1", "rejected by a")},
+		}}},
+		"b": {id: "b", client: &stubPublishClient{response: &chippb.PublishResponse{
+			Results: []*chippb.PublishResult{{EventId: "e1"}},
+		}}},
+	}}
+
+	resp, err := r.PublishBatch(t.Context(), batchOf("e1"))
+	if err != nil {
+		t.Fatalf("PublishBatch: %v", err)
+	}
+	if len(resp.Results) != 1 || resp.Results[0].Error != nil {
+		t.Errorf("want e1 accepted, got results=%+v", resp.Results)
+	}
+}
+
+func TestPublishBatchDetaillessAcceptanceOverridesRejections(t *testing.T) {
+	// A rejects both events per-event; B accepts the whole batch without
+	// per-event detail. The batch was delivered to B, so both events are
+	// acknowledged regardless of completion order.
+	r := &router{subscribers: map[string]*subscriber{
+		"a": {id: "a", client: &stubPublishClient{response: &chippb.PublishResponse{
+			Results: []*chippb.PublishResult{
+				rejected("e1", "rejected by a"),
+				rejected("e2", "rejected by a"),
+			},
+		}}},
+		"b": {id: "b", client: &stubPublishClient{}},
+	}}
+
+	resp, err := r.PublishBatch(t.Context(), batchOf("e1", "e2"))
+	if err != nil {
+		t.Fatalf("PublishBatch: %v", err)
+	}
+	if len(resp.Results) != 2 {
+		t.Fatalf("want 2 results, got %d", len(resp.Results))
+	}
+	for i, res := range resp.Results {
+		if res.Error != nil {
+			t.Errorf("results[%d].Error = %v, want nil (batch accepted by b)", i, res.Error)
+		}
+	}
+}
+
+func TestPublishBatchMergeExtendsBeyondFirstSubscriberDetail(t *testing.T) {
+	// A returns detail for only one of two events (contract violation); B
+	// returns full detail rejecting e2. e2's rejection must survive: detail
+	// is sized to the batch, not to the first subscriber's response.
+	r := &router{subscribers: map[string]*subscriber{
+		"a": {id: "a", client: &stubPublishClient{response: &chippb.PublishResponse{
+			Results: []*chippb.PublishResult{{EventId: "e1"}},
+		}}},
+		"b": {id: "b", client: &stubPublishClient{response: &chippb.PublishResponse{
+			Results: []*chippb.PublishResult{
+				{EventId: "e1"},
+				rejected("e2", "rejected by b"),
+			},
+		}}},
+	}}
+
+	resp, err := r.PublishBatch(t.Context(), batchOf("e1", "e2"))
+	if err != nil {
+		t.Fatalf("PublishBatch: %v", err)
+	}
+	if len(resp.Results) != 2 {
+		t.Fatalf("want 2 results, got %d", len(resp.Results))
+	}
+	if resp.Results[0].Error != nil {
+		t.Errorf("results[0].Error = %v, want nil", resp.Results[0].Error)
+	}
+	if resp.Results[1].Error == nil {
+		t.Error("results[1].Error = nil, want the rejection from b passed through")
+	}
+}
+
 func TestPublishBatchTransactionalPartialRejectionFailsRPC(t *testing.T) {
 	// Transactional callers resolve the whole batch from the RPC outcome
 	// alone, so a partial rejection must fail the RPC, not ack everything.
@@ -138,7 +253,7 @@ func TestPublishBatchTransactionalPartialRejectionFailsRPC(t *testing.T) {
 		"sink": {id: "sink", client: &stubPublishClient{response: &chippb.PublishResponse{
 			Results: []*chippb.PublishResult{
 				{EventId: "e1"},
-				{EventId: "e2", Error: &chippb.PublishError{Reason: "rejected"}},
+				rejected("e2", "rejected"),
 			},
 		}}},
 	}}
@@ -146,12 +261,7 @@ func TestPublishBatchTransactionalPartialRejectionFailsRPC(t *testing.T) {
 	batch := batchOf("e1", "e2")
 	batch.Options = &chippb.PublishOptions{TransactionEnabled: ptr(true)}
 	_, err := r.PublishBatch(t.Context(), batch)
-	if err == nil {
-		t.Fatal("want error for a partial rejection under transaction_enabled, got nil")
-	}
-	if status.Code(err) != codes.Unavailable {
-		t.Errorf("want Unavailable, got %v", status.Code(err))
-	}
+	requireCode(t, err, codes.Internal)
 }
 
 func TestPublishBatchTransactionalAllRejectedByDownstreamFailsRPC(t *testing.T) {
@@ -160,8 +270,8 @@ func TestPublishBatchTransactionalAllRejectedByDownstreamFailsRPC(t *testing.T)
 	r := &router{subscribers: map[string]*subscriber{
 		"sink": {id: "sink", client: &stubPublishClient{response: &chippb.PublishResponse{
 			Results: []*chippb.PublishResult{
-				{EventId: "e1", Error: &chippb.PublishError{Reason: "rejected"}},
-				{EventId: "e2", Error: &chippb.PublishError{Reason: "rejected"}},
+				rejected("e1", "rejected"),
+				rejected("e2", "rejected"),
 			},
 		}}},
 	}}
@@ -169,12 +279,7 @@ func TestPublishBatchTransactionalAllRejectedByDownstreamFailsRPC(t *testing.T)
 	batch := batchOf("e1", "e2")
 	batch.Options = &chippb.PublishOptions{TransactionEnabled: ptr(true)}
 	_, err := r.PublishBatch(t.Context(), batch)
-	if err == nil {
-		t.Fatal("want error when the subscriber rejected every event, got nil")
-	}
-	if status.Code(err) != codes.Unavailable {
-		t.Errorf("want Unavailable, got %v", status.Code(err))
-	}
+	requireCode(t, err, codes.Internal)
 }
 
 func TestPublishBatchTransactionalAllAcceptedPasses(t *testing.T) {
@@ -200,25 +305,70 @@ func TestPublishBatchTransactionalAllAcceptedPasses(t *testing.T) {
 	}
 }
 
-func TestPublishBatchAnAcceptingSubscriberOverridesADetaillessRejection(t *testing.T) {
-	// Two detail-providing subscribers disagree: the event is rejected by one
-	// and accepted by the other — acceptance wins.
+func TestPublishReturnsOneResult(t *testing.T) {
+	// The write-service returns exactly one result for a single event.
 	r := &router{subscribers: map[string]*subscriber{
-		"a": {id: "a", client: &stubPublishClient{response: &chippb.PublishResponse{
-			Results: []*chippb.PublishResult{
-				{EventId: "e1", Error: &chippb.PublishError{Reason: "rejected by a"}},
-			},
+		"sink": {id: "sink", client: &stubPublishClient{}},
+	}}
+
+	resp, err := r.Publish(t.Context(), eventOf("e1"))
+	if err != nil {
+		t.Fatalf("Publish: %v", err)
+	}
+	if len(resp.Results) != 1 {
+		t.Fatalf("want 1 result, got %d", len(resp.Results))
+	}
+	if resp.Results[0].EventId != "e1" {
+		t.Errorf("results[0].EventId = %q, want %q", resp.Results[0].EventId, "e1")
+	}
+	if resp.Results[0].Error != nil {
+		t.Errorf("results[0].Error = %v, want nil", resp.Results[0].Error)
+	}
+}
+
+func TestPublishNoSubscribersFailsWithInternal(t *testing.T) {
+	// Acks without delivery would let a durable store delete the event undelivered.
+	r := &router{subscribers: map[string]*subscriber{}}
+
+	_, err := r.Publish(t.Context(), eventOf("e1"))
+	requireCode(t, err, codes.Internal)
+}
+
+func TestPublishAllForwardsFailedFailsWithInternal(t *testing.T) {
+	r := &router{subscribers: map[string]*subscriber{
+		"sink": {id: "sink", client: &stubPublishClient{err: context.DeadlineExceeded}},
+	}}
+
+	_, err := r.Publish(t.Context(), eventOf("e1"))
+	requireCode(t, err, codes.Internal)
+}
+
+func TestPublishAllSubscribersRejectedFailsWithInternal(t *testing.T) {
+	// Single-event Publish is all-or-nothing (the write-service treats it as
+	// transactional): a per-event rejection from every subscriber fails the RPC.
+	r := &router{subscribers: map[string]*subscriber{
+		"sink": {id: "sink", client: &stubPublishClient{response: &chippb.PublishResponse{
+			Results: []*chippb.PublishResult{rejected("e1", "invalid event")},
 		}}},
-		"b": {id: "b", client: &stubPublishClient{response: &chippb.PublishResponse{
-			Results: []*chippb.PublishResult{
-				{EventId: "e1"},
-			},
+	}}
+
+	_, err := r.Publish(t.Context(), eventOf("e1"))
+	requireCode(t, err, codes.Internal)
+}
+
+func TestPublishDetaillessAcceptanceOverridesARejection(t *testing.T) {
+	// A rejects the event per-event; B accepts without per-event detail —
+	// the event was delivered to B, so Publish acknowledges it.
+	r := &router{subscribers: map[string]*subscriber{
+		"a": {id: "a", client: &stubPublishClient{response: &chippb.PublishResponse{
+			Results: []*chippb.PublishResult{rejected("e1", "rejected by a")},
 		}}},
+		"b": {id: "b", client: &stubPublishClient{}},
 	}}
 
-	resp, err := r.PublishBatch(t.Context(), batchOf("e1"))
+	resp, err := r.Publish(t.Context(), eventOf("e1"))
 	if err != nil {
-		t.Fatalf("PublishBatch: %v", err)
+		t.Fatalf("Publish: %v", err)
 	}
 	if len(resp.Results) != 1 || resp.Results[0].Error != nil {
 		t.Errorf("want e1 accepted, got results=%+v", resp.Results)

From 631a2b00a65a618ad4cb190dd6c75d9979f743f1 Mon Sep 17 00:00:00 2001
From: Pavel <177363085+pkcll@users.noreply.github.com>
Date: Fri, 16 Oct 2026 22:18:53 -0400
Subject: [PATCH 2/5] fix(chip-router): bump builder image to golang:1.26.6 to
 match go.mod

go.mod requires go >= 1.26.6 (bumped alongside the pkg/chipingress pin),
so the pinned golang:1.25.3 builder fails at 'go mod download' with
'go: go.mod requires go >= 1.26.6 (running go 1.25.3; GOTOOLCHAIN=local)'.
---
 framework/components/chiprouter/Dockerfile | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/framework/components/chiprouter/Dockerfile b/framework/components/chiprouter/Dockerfile
index 8d998ae0a..401868010 100644
--- a/framework/components/chiprouter/Dockerfile
+++ b/framework/components/chiprouter/Dockerfile
@@ -1,4 +1,4 @@
-FROM golang:1.25.3 AS builder
+FROM golang:1.26.6 AS builder
 
 WORKDIR /src
 

From 1edfbd6d0e4dde7601c0e3b955d92c2951c158a7 Mon Sep 17 00:00:00 2001
From: Pavel <177363085+pkcll@users.noreply.github.com>
Date: Fri, 16 Oct 2026 22:52:37 -0400
Subject: [PATCH 3/5] fix(chip-router): implement Ping for write-service
 surface parity

The write-service answers Ping with a static pong; the router inherited
UnimplementedChipIngressServer's Unimplemented for it, which the new
atlas parity suite's P0 surface probe flags as drift. Ping never needs
to reach subscribers, so answer directly like the write-service.
---
 .../components/chiprouter/cmd/chip-router/main.go    |  6 ++++++
 .../chiprouter/cmd/chip-router/main_test.go          | 12 ++++++++++++
 2 files changed, 18 insertions(+)

diff --git a/framework/components/chiprouter/cmd/chip-router/main.go b/framework/components/chiprouter/cmd/chip-router/main.go
index e4f1e91c6..f8094cf47 100644
--- a/framework/components/chiprouter/cmd/chip-router/main.go
+++ b/framework/components/chiprouter/cmd/chip-router/main.go
@@ -138,6 +138,12 @@ func run() error {
 	}
 }
 
+// Ping matches the write-service's Ping: a static health response that never
+// reaches the subscribers.
+func (r *router) Ping(context.Context, *chippb.EmptyRequest) (*chippb.PingResponse, error) {
+	return &chippb.PingResponse{Message: "pong"}, nil
+}
+
 // Publish matches the write-service's single-event contract (atlas
 // chip-ingress internal/server): the RPC is all-or-nothing — it fails unless
 // at least one subscriber accepted the event, and returns exactly one
diff --git a/framework/components/chiprouter/cmd/chip-router/main_test.go b/framework/components/chiprouter/cmd/chip-router/main_test.go
index 1a14dfd83..b8b2fa1a9 100644
--- a/framework/components/chiprouter/cmd/chip-router/main_test.go
+++ b/framework/components/chiprouter/cmd/chip-router/main_test.go
@@ -374,3 +374,15 @@ func TestPublishDetaillessAcceptanceOverridesARejection(t *testing.T) {
 		t.Errorf("want e1 accepted, got results=%+v", resp.Results)
 	}
 }
+
+func TestPingMatchesWriteService(t *testing.T) {
+	r := &router{subscribers: map[string]*subscriber{}}
+
+	resp, err := r.Ping(context.Background(), &chippb.EmptyRequest{})
+	if err != nil {
+		t.Fatalf("Ping: %v", err)
+	}
+	if resp.GetMessage() != "pong" {
+		t.Errorf("Ping message = %q, want %q", resp.GetMessage(), "pong")
+	}
+}

From b4e6931abc87180531c1e377c8d3a209631a3cfa Mon Sep 17 00:00:00 2001
From: Pavel <177363085+pkcll@users.noreply.github.com>
Date: Fri, 16 Oct 2026 23:25:13 -0400
Subject: [PATCH 4/5] fix(chip-router): make the image build self-contained (no
 framework dep in the binary)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The image build recipe (chainlink cre environment: docker_ctx =
framework/components/chiprouter) passes the module dir as the build
context, so go.mod's `replace framework => ../../` (a monorepo-dev
convenience inherited from main) cannot resolve inside the container and
'go mod download' fails. Published framework releases also do not
compile against this module's moby pins, so the replace cannot simply be
dropped for the module as a whole.

The router binary only used framework for its global logger, so:
- cmd/chip-router now carries its own zerolog logger, mirroring
  framework.L's setup (CTF_LOG_LEVEL env var, console writer to stderr);
  output format is unchanged.
- the Dockerfile drops the replace for the isolated build and runs
  'go mod tidy' so the published framework — needed only by the
  deploy-side library, which is not compiled into the image — resolves
  from the proxy for module-graph completeness.

The deploy-side library (chiprouter.go) keeps its framework dependency
and the monorepo replace.
---
 framework/components/chiprouter/Dockerfile    | 12 +++--
 .../chiprouter/cmd/chip-router/main.go        | 49 +++++++++++++------
 framework/components/chiprouter/go.mod        |  2 +-
 3 files changed, 44 insertions(+), 19 deletions(-)

diff --git a/framework/components/chiprouter/Dockerfile b/framework/components/chiprouter/Dockerfile
index 401868010..af44ef24b 100644
--- a/framework/components/chiprouter/Dockerfile
+++ b/framework/components/chiprouter/Dockerfile
@@ -2,11 +2,17 @@ FROM golang:1.26.6 AS builder
 
 WORKDIR /src
 
-COPY go.mod go.sum ./
-RUN go mod download
-
 COPY . .
 
+# The module's go.mod replaces the parent framework module with ../../ (a
+# monorepo-dev convenience) which cannot resolve inside this build's
+# module-dir context, and the published framework does not compile against
+# this module's moby pins. The router binary no longer imports framework
+# (only the deploy-side library does, and it is not compiled into the
+# image), so drop the replace and fetch the published framework module from
+# the proxy for module-graph completeness only.
+RUN sed -i '/^replace /d' go.mod && go mod tidy && go mod download
+
 ARG TARGETOS=linux
 ARG TARGETARCH=amd64
 ARG CTF_LOG_LEVEL=info
diff --git a/framework/components/chiprouter/cmd/chip-router/main.go b/framework/components/chiprouter/cmd/chip-router/main.go
index f8094cf47..96037e15f 100644
--- a/framework/components/chiprouter/cmd/chip-router/main.go
+++ b/framework/components/chiprouter/cmd/chip-router/main.go
@@ -17,15 +17,34 @@ import (
 	cepb "github.com/cloudevents/sdk-go/binding/format/protobuf/v2/pb"
 	"github.com/google/uuid"
 	chippb "github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb"
+	"github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/status"
-
-	"github.com/smartcontractkit/chainlink-testing-framework/framework"
 )
 
+// L is the router's logger. It mirrors framework.L's setup (CTF_LOG_LEVEL
+// env var, console writer to stderr) so output is unchanged now that the
+// binary no longer imports the framework module — the image build must stay
+// self-contained (see the Dockerfile: the monorepo-local `replace` cannot
+// resolve in its module-dir context).
+var L zerolog.Logger
+
+func init() {
+	lvlStr := os.Getenv("CTF_LOG_LEVEL")
+	if lvlStr == "" {
+		lvlStr = "info"
+	}
+	lvl, err := zerolog.ParseLevel(lvlStr)
+	if err != nil {
+		panic(err)
+	}
+	L = log.Output(zerolog.ConsoleWriter{Out: os.Stderr}).Level(lvl)
+}
+
 const (
 	defaultGRPCAddr  = "0.0.0.0:50051"
 	defaultAdminAddr = "0.0.0.0:50050"
@@ -67,7 +86,7 @@ type healthResponse struct {
 
 func main() {
 	if err := run(); err != nil {
-		framework.L.Error().Msgf("chip router failed: %v", err)
+		L.Error().Msgf("chip router failed: %v", err)
 		os.Exit(1)
 	}
 }
@@ -110,7 +129,7 @@ func run() error {
 	go func() { errCh <- grpcServer.Serve(grpcLis) }()
 	go func() { errCh <- adminServer.Serve(adminLis) }()
 
-	framework.L.Info().Msgf("chip router started: grpc=%s admin=%s", grpcAddr, adminAddr)
+	L.Info().Msgf("chip router started: grpc=%s admin=%s", grpcAddr, adminAddr)
 
 	select {
 	case <-ctx.Done():
@@ -133,7 +152,7 @@ func run() error {
 		if err == nil || errors.Is(err, http.ErrServerClosed) {
 			return nil
 		}
-		framework.L.Error().Msgf("chip router server error: %v", err)
+		L.Error().Msgf("chip router server error: %v", err)
 		return err
 	}
 }
@@ -168,15 +187,15 @@ func (r *router) Publish(ctx context.Context, event *cepb.CloudEvent) (*chippb.P
 	group.SetLimit(forwardParallel)
 	for _, sub := range snapshot {
 		group.Go(func() error {
-			framework.L.Debug().Msgf("chip router forwarding event to subscriber id=%s name=%s endpoint=%s", sub.id, sub.name, sub.endpoint)
+			L.Debug().Msgf("chip router forwarding event to subscriber id=%s name=%s endpoint=%s", sub.id, sub.name, sub.endpoint)
 			forwardCtx, cancel := context.WithTimeout(ctx, forwardTimeout)
 			defer cancel()
 			resp, err := sub.client.Publish(forwardCtx, event)
 			if err != nil {
-				framework.L.Error().Msgf("chip router failed to forward event to subscriber id=%s name=%s endpoint=%s err=%v", sub.id, sub.name, sub.endpoint, err)
+				L.Error().Msgf("chip router failed to forward event to subscriber id=%s name=%s endpoint=%s err=%v", sub.id, sub.name, sub.endpoint, err)
 				return nil
 			}
-			framework.L.Debug().Msgf("chip router forwarded event to subscriber id=%s", sub.id)
+			L.Debug().Msgf("chip router forwarded event to subscriber id=%s", sub.id)
 			mu.Lock()
 			defer mu.Unlock()
 			anyForwarded = true
@@ -228,15 +247,15 @@ func (r *router) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch
 	group.SetLimit(forwardParallel)
 	for _, sub := range snapshot {
 		group.Go(func() error {
-			framework.L.Debug().Msgf("chip router forwarding batch to subscriber id=%s name=%s endpoint=%s", sub.id, sub.name, sub.endpoint)
+			L.Debug().Msgf("chip router forwarding batch to subscriber id=%s name=%s endpoint=%s", sub.id, sub.name, sub.endpoint)
 			forwardCtx, cancel := context.WithTimeout(ctx, forwardTimeout)
 			defer cancel()
 			resp, err := sub.client.PublishBatch(forwardCtx, batch)
 			if err != nil {
-				framework.L.Error().Msgf("chip router failed to forward batch to subscriber id=%s name=%s endpoint=%s err=%v", sub.id, sub.name, sub.endpoint, err)
+				L.Error().Msgf("chip router failed to forward batch to subscriber id=%s name=%s endpoint=%s err=%v", sub.id, sub.name, sub.endpoint, err)
 				return nil
 			}
-			framework.L.Debug().Msgf("chip router forwarded batch to subscriber id=%s", sub.id)
+			L.Debug().Msgf("chip router forwarded batch to subscriber id=%s", sub.id)
 			mu.Lock()
 			defer mu.Unlock()
 			anyForwarded = true
@@ -315,13 +334,13 @@ func (r *router) handleSubscribers(w http.ResponseWriter, req *http.Request) {
 	r.mu.Lock()
 	if existing := r.findSubscriberLocked(body.Name, body.Endpoint); existing != nil {
 		r.mu.Unlock()
-		framework.L.Info().Msgf("chip router reused subscriber id=%s name=%s endpoint=%s", existing.id, existing.name, existing.endpoint)
+		L.Info().Msgf("chip router reused subscriber id=%s name=%s endpoint=%s", existing.id, existing.name, existing.endpoint)
 		w.Header().Set("Content-Type", "application/json")
 		_ = json.NewEncoder(w).Encode(registerSubscriberResponse{ID: existing.id})
 		return
 	}
 
-	framework.L.Info().Msgf("chip router attempting to register subscriber name=%s endpoint=%s", body.Name, body.Endpoint)
+	L.Info().Msgf("chip router attempting to register subscriber name=%s endpoint=%s", body.Name, body.Endpoint)
 	conn, err := grpc.NewClient(body.Endpoint, grpc.WithTransportCredentials(insecure.NewCredentials()))
 	if err != nil {
 		r.mu.Unlock()
@@ -337,7 +356,7 @@ func (r *router) handleSubscribers(w http.ResponseWriter, req *http.Request) {
 		client:   chippb.NewChipIngressClient(conn),
 	}
 	r.mu.Unlock()
-	framework.L.Info().Msgf("chip router registered subscriber id=%s name=%s endpoint=%s", id, body.Name, body.Endpoint)
+	L.Info().Msgf("chip router registered subscriber id=%s name=%s endpoint=%s", id, body.Name, body.Endpoint)
 	w.Header().Set("Content-Type", "application/json")
 	_ = json.NewEncoder(w).Encode(registerSubscriberResponse{ID: id})
 }
@@ -369,7 +388,7 @@ func (r *router) handleSubscriberByID(w http.ResponseWriter, req *http.Request)
 	r.mu.Unlock()
 	if ok && sub.conn != nil {
 		_ = sub.conn.Close()
-		framework.L.Info().Msgf("chip router unregistered subscriber id=%s name=%s endpoint=%s", id, sub.name, sub.endpoint)
+		L.Info().Msgf("chip router unregistered subscriber id=%s name=%s endpoint=%s", id, sub.name, sub.endpoint)
 	}
 	w.WriteHeader(http.StatusNoContent)
 }
diff --git a/framework/components/chiprouter/go.mod b/framework/components/chiprouter/go.mod
index 18347708e..90c5dd5a6 100644
--- a/framework/components/chiprouter/go.mod
+++ b/framework/components/chiprouter/go.mod
@@ -9,6 +9,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/moby/moby/api v1.54.1
 	github.com/pkg/errors v0.9.1
+	github.com/rs/zerolog v1.33.0
 	github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20260915184316-2730f1867c92
 	github.com/smartcontractkit/chainlink-testing-framework/framework v0.15.8
 	github.com/testcontainers/testcontainers-go v0.42.0
@@ -60,7 +61,6 @@ require (
 	github.com/opencontainers/image-spec v1.1.1 // indirect
 	github.com/pelletier/go-toml/v2 v2.2.4 // indirect
 	github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
-	github.com/rs/zerolog v1.33.0 // indirect
 	github.com/shirou/gopsutil/v4 v4.26.3 // indirect
 	github.com/sirupsen/logrus v1.9.4 // indirect
 	github.com/stretchr/testify v1.12.1 // indirect

From 0748e96a25068aa78be4b42d4de42d7a09400fe2 Mon Sep 17 00:00:00 2001
From: Pavel <177363085+pkcll@users.noreply.github.com>
Date: Fri, 23 Oct 2026 01:19:16 -0400
Subject: [PATCH 5/5] fix(chip-router): reject mismatched batch results and
 tighten Publish failure path

PublishBatch now treats a non-empty subscriber response whose result count
differs from the batch size as a contract violation (the write-service returns
exactly one result per event in input order): it is logged at Warn and counted
as a failed forward instead of silently acking the events it does not cover.
Empty responses stay a detail-less success, which chainlink's chip-testsink
subscriber relies on.

Publish drops the redundant anyForwarded flag and surfaces the subscriber's
rejection reason in the Internal error when no subscriber accepted the event.

Tests: rewrite the merge test into mismatch cases, make the detail-less
acceptance test genuinely detail-less, and cover the single-subscriber
empty-response Publish path. Also fix gofmt import order and make the
package logger unexported.
---
 .../chiprouter/cmd/chip-router/main.go        | 63 ++++++++++---------
 .../chiprouter/cmd/chip-router/main_test.go   | 53 +++++++++++++---
 2 files changed, 78 insertions(+), 38 deletions(-)

diff --git a/framework/components/chiprouter/cmd/chip-router/main.go b/framework/components/chiprouter/cmd/chip-router/main.go
index 96037e15f..de6a7df73 100644
--- a/framework/components/chiprouter/cmd/chip-router/main.go
+++ b/framework/components/chiprouter/cmd/chip-router/main.go
@@ -16,9 +16,9 @@ import (
 
 	cepb "github.com/cloudevents/sdk-go/binding/format/protobuf/v2/pb"
 	"github.com/google/uuid"
-	chippb "github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
+	chippb "github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -26,12 +26,12 @@ import (
 	"google.golang.org/grpc/status"
 )
 
-// L is the router's logger. It mirrors framework.L's setup (CTF_LOG_LEVEL
+// logger is the router's logger. It mirrors framework.L's setup (CTF_LOG_LEVEL
 // env var, console writer to stderr) so output is unchanged now that the
 // binary no longer imports the framework module — the image build must stay
 // self-contained (see the Dockerfile: the monorepo-local `replace` cannot
 // resolve in its module-dir context).
-var L zerolog.Logger
+var logger zerolog.Logger
 
 func init() {
 	lvlStr := os.Getenv("CTF_LOG_LEVEL")
@@ -42,7 +42,7 @@ func init() {
 	if err != nil {
 		panic(err)
 	}
-	L = log.Output(zerolog.ConsoleWriter{Out: os.Stderr}).Level(lvl)
+	logger = log.Output(zerolog.ConsoleWriter{Out: os.Stderr}).Level(lvl)
 }
 
 const (
@@ -86,7 +86,7 @@ type healthResponse struct {
 
 func main() {
 	if err := run(); err != nil {
-		L.Error().Msgf("chip router failed: %v", err)
+		logger.Error().Msgf("chip router failed: %v", err)
 		os.Exit(1)
 	}
 }
@@ -129,7 +129,7 @@ func run() error {
 	go func() { errCh <- grpcServer.Serve(grpcLis) }()
 	go func() { errCh <- adminServer.Serve(adminLis) }()
 
-	L.Info().Msgf("chip router started: grpc=%s admin=%s", grpcAddr, adminAddr)
+	logger.Info().Msgf("chip router started: grpc=%s admin=%s", grpcAddr, adminAddr)
 
 	select {
 	case <-ctx.Done():
@@ -152,7 +152,7 @@ func run() error {
 		if err == nil || errors.Is(err, http.ErrServerClosed) {
 			return nil
 		}
-		L.Error().Msgf("chip router server error: %v", err)
+		logger.Error().Msgf("chip router server error: %v", err)
 		return err
 	}
 }
@@ -174,12 +174,10 @@ func (r *router) Publish(ctx context.Context, event *cepb.CloudEvent) (*chippb.P
 		return nil, status.Error(codes.Internal, "no subscribers registered")
 	}
 
-	// anyForwarded: at least one subscriber's RPC succeeded. accepted: some
-	// subscriber accepted the event (an accepted result or a detail-less
-	// success) — sticky, so a later rejection cannot unset it. rejection:
-	// recorded only while no subscriber has accepted.
+	// accepted: some subscriber accepted the event (an accepted result or a
+	// detail-less success) — sticky, so a later rejection cannot unset it.
+	// rejection: recorded only while no subscriber has accepted.
 	var mu sync.Mutex
-	anyForwarded := false
 	accepted := false
 	var rejection *chippb.PublishError
 
@@ -187,18 +185,17 @@ func (r *router) Publish(ctx context.Context, event *cepb.CloudEvent) (*chippb.P
 	group.SetLimit(forwardParallel)
 	for _, sub := range snapshot {
 		group.Go(func() error {
-			L.Debug().Msgf("chip router forwarding event to subscriber id=%s name=%s endpoint=%s", sub.id, sub.name, sub.endpoint)
+			logger.Debug().Msgf("chip router forwarding event to subscriber id=%s name=%s endpoint=%s", sub.id, sub.name, sub.endpoint)
 			forwardCtx, cancel := context.WithTimeout(ctx, forwardTimeout)
 			defer cancel()
 			resp, err := sub.client.Publish(forwardCtx, event)
 			if err != nil {
-				L.Error().Msgf("chip router failed to forward event to subscriber id=%s name=%s endpoint=%s err=%v", sub.id, sub.name, sub.endpoint, err)
+				logger.Error().Msgf("chip router failed to forward event to subscriber id=%s name=%s endpoint=%s err=%v", sub.id, sub.name, sub.endpoint, err)
 				return nil
 			}
-			L.Debug().Msgf("chip router forwarded event to subscriber id=%s", sub.id)
+			logger.Debug().Msgf("chip router forwarded event to subscriber id=%s", sub.id)
 			mu.Lock()
 			defer mu.Unlock()
-			anyForwarded = true
 			if results := resp.GetResults(); len(results) > 0 && results[0].GetError() != nil {
 				if !accepted && rejection == nil {
 					rejection = results[0].GetError()
@@ -211,11 +208,11 @@ func (r *router) Publish(ctx context.Context, event *cepb.CloudEvent) (*chippb.P
 	}
 	_ = group.Wait()
 
-	if !anyForwarded {
-		return nil, status.Error(codes.Internal, fmt.Sprintf("all %d subscribers failed to accept the event", len(snapshot)))
-	}
 	if !accepted {
-		return nil, status.Error(codes.Internal, "event was not accepted by any subscriber")
+		if rejection != nil {
+			return nil, status.Error(codes.Internal, rejection.GetReason())
+		}
+		return nil, status.Error(codes.Internal, fmt.Sprintf("all %d subscribers failed to accept the event", len(snapshot)))
 	}
 	return &chippb.PublishResponse{Results: []*chippb.PublishResult{{EventId: event.GetId()}}}, nil
 }
@@ -247,28 +244,32 @@ func (r *router) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch
 	group.SetLimit(forwardParallel)
 	for _, sub := range snapshot {
 		group.Go(func() error {
-			L.Debug().Msgf("chip router forwarding batch to subscriber id=%s name=%s endpoint=%s", sub.id, sub.name, sub.endpoint)
+			logger.Debug().Msgf("chip router forwarding batch to subscriber id=%s name=%s endpoint=%s", sub.id, sub.name, sub.endpoint)
 			forwardCtx, cancel := context.WithTimeout(ctx, forwardTimeout)
 			defer cancel()
 			resp, err := sub.client.PublishBatch(forwardCtx, batch)
 			if err != nil {
-				L.Error().Msgf("chip router failed to forward batch to subscriber id=%s name=%s endpoint=%s err=%v", sub.id, sub.name, sub.endpoint, err)
+				logger.Error().Msgf("chip router failed to forward batch to subscriber id=%s name=%s endpoint=%s err=%v", sub.id, sub.name, sub.endpoint, err)
 				return nil
 			}
-			L.Debug().Msgf("chip router forwarded batch to subscriber id=%s", sub.id)
+			results := resp.GetResults()
+			if len(results) > 0 && len(results) != len(batch.GetEvents()) {
+				// The write-service returns one result per event in input order, so
+				// a short or long list cannot be aligned with the batch. Treat it
+				// as a failed forward rather than guess which events it covers.
+				logger.Warn().Msgf("chip router subscriber returned %d results for %d events, ignoring response id=%s name=%s endpoint=%s", len(results), len(batch.GetEvents()), sub.id, sub.name, sub.endpoint)
+				return nil
+			}
+			logger.Debug().Msgf("chip router forwarded batch to subscriber id=%s", sub.id)
 			mu.Lock()
 			defer mu.Unlock()
 			anyForwarded = true
-			results := resp.GetResults()
 			if len(results) == 0 {
 				// A detail-less success accepts the whole batch.
 				anyBatchAccepted = true
 				return nil
 			}
 			for i, res := range results {
-				if i >= len(detail) {
-					break
-				}
 				if res.GetError() == nil {
 					detail[i] = res
 				} else if detail[i] == nil {
@@ -334,13 +335,13 @@ func (r *router) handleSubscribers(w http.ResponseWriter, req *http.Request) {
 	r.mu.Lock()
 	if existing := r.findSubscriberLocked(body.Name, body.Endpoint); existing != nil {
 		r.mu.Unlock()
-		L.Info().Msgf("chip router reused subscriber id=%s name=%s endpoint=%s", existing.id, existing.name, existing.endpoint)
+		logger.Info().Msgf("chip router reused subscriber id=%s name=%s endpoint=%s", existing.id, existing.name, existing.endpoint)
 		w.Header().Set("Content-Type", "application/json")
 		_ = json.NewEncoder(w).Encode(registerSubscriberResponse{ID: existing.id})
 		return
 	}
 
-	L.Info().Msgf("chip router attempting to register subscriber name=%s endpoint=%s", body.Name, body.Endpoint)
+	logger.Info().Msgf("chip router attempting to register subscriber name=%s endpoint=%s", body.Name, body.Endpoint)
 	conn, err := grpc.NewClient(body.Endpoint, grpc.WithTransportCredentials(insecure.NewCredentials()))
 	if err != nil {
 		r.mu.Unlock()
@@ -356,7 +357,7 @@ func (r *router) handleSubscribers(w http.ResponseWriter, req *http.Request) {
 		client:   chippb.NewChipIngressClient(conn),
 	}
 	r.mu.Unlock()
-	L.Info().Msgf("chip router registered subscriber id=%s name=%s endpoint=%s", id, body.Name, body.Endpoint)
+	logger.Info().Msgf("chip router registered subscriber id=%s name=%s endpoint=%s", id, body.Name, body.Endpoint)
 	w.Header().Set("Content-Type", "application/json")
 	_ = json.NewEncoder(w).Encode(registerSubscriberResponse{ID: id})
 }
@@ -388,7 +389,7 @@ func (r *router) handleSubscriberByID(w http.ResponseWriter, req *http.Request)
 	r.mu.Unlock()
 	if ok && sub.conn != nil {
 		_ = sub.conn.Close()
-		L.Info().Msgf("chip router unregistered subscriber id=%s name=%s endpoint=%s", id, sub.name, sub.endpoint)
+		logger.Info().Msgf("chip router unregistered subscriber id=%s name=%s endpoint=%s", id, sub.name, sub.endpoint)
 	}
 	w.WriteHeader(http.StatusNoContent)
 }
diff --git a/framework/components/chiprouter/cmd/chip-router/main_test.go b/framework/components/chiprouter/cmd/chip-router/main_test.go
index b8b2fa1a9..10fd1e04b 100644
--- a/framework/components/chiprouter/cmd/chip-router/main_test.go
+++ b/framework/components/chiprouter/cmd/chip-router/main_test.go
@@ -215,10 +215,23 @@ func TestPublishBatchDetaillessAcceptanceOverridesRejections(t *testing.T) {
 	}
 }
 
-func TestPublishBatchMergeExtendsBeyondFirstSubscriberDetail(t *testing.T) {
-	// A returns detail for only one of two events (contract violation); B
-	// returns full detail rejecting e2. e2's rejection must survive: detail
-	// is sized to the batch, not to the first subscriber's response.
+func TestPublishBatchResultsMismatchFromOnlySubscriberFailsWithInternal(t *testing.T) {
+	// The write-service returns one result per event, so a short list cannot
+	// be aligned with the batch: it counts as a failed forward, not as an
+	// acknowledgement of the events it does not cover.
+	r := &router{subscribers: map[string]*subscriber{
+		"sink": {id: "sink", client: &stubPublishClient{response: &chippb.PublishResponse{
+			Results: []*chippb.PublishResult{{EventId: "e1"}},
+		}}},
+	}}
+
+	_, err := r.PublishBatch(t.Context(), batchOf("e1", "e2"))
+	requireCode(t, err, codes.Internal)
+}
+
+func TestPublishBatchResultsMismatchIsIgnoredInFavorOfFullDetail(t *testing.T) {
+	// A returns a short list (contract violation) and is dropped; B returns
+	// full detail rejecting e2, which must be the outcome.
 	r := &router{subscribers: map[string]*subscriber{
 		"a": {id: "a", client: &stubPublishClient{response: &chippb.PublishResponse{
 			Results: []*chippb.PublishResult{{EventId: "e1"}},
@@ -326,6 +339,28 @@ func TestPublishReturnsOneResult(t *testing.T) {
 	}
 }
 
+func TestPublishDetaillessSuccessAcceptsEvent(t *testing.T) {
+	// chainlink's chip-testsink subscriber returns an empty PublishResponse on
+	// success; the router must still acknowledge the event and echo its ID.
+	r := &router{subscribers: map[string]*subscriber{
+		"sink": {id: "sink", client: &stubPublishClient{response: &chippb.PublishResponse{}}},
+	}}
+
+	resp, err := r.Publish(t.Context(), eventOf("e1"))
+	if err != nil {
+		t.Fatalf("Publish: %v", err)
+	}
+	if len(resp.Results) != 1 {
+		t.Fatalf("want 1 result, got %d", len(resp.Results))
+	}
+	if resp.Results[0].EventId != "e1" {
+		t.Errorf("results[0].EventId = %q, want %q", resp.Results[0].EventId, "e1")
+	}
+	if resp.Results[0].Error != nil {
+		t.Errorf("results[0].Error = %v, want nil", resp.Results[0].Error)
+	}
+}
+
 func TestPublishNoSubscribersFailsWithInternal(t *testing.T) {
 	// Acks without delivery would let a durable store delete the event undelivered.
 	r := &router{subscribers: map[string]*subscriber{}}
@@ -354,16 +389,20 @@ func TestPublishAllSubscribersRejectedFailsWithInternal(t *testing.T) {
 
 	_, err := r.Publish(t.Context(), eventOf("e1"))
 	requireCode(t, err, codes.Internal)
+	if got := status.Convert(err).Message(); got != "invalid event" {
+		t.Errorf("error message = %q, want the subscriber's rejection reason %q", got, "invalid event")
+	}
 }
 
 func TestPublishDetaillessAcceptanceOverridesARejection(t *testing.T) {
-	// A rejects the event per-event; B accepts without per-event detail —
-	// the event was delivered to B, so Publish acknowledges it.
+	// A rejects the event per-event; B accepts with an empty response (no
+	// per-event detail) — the event was delivered to B, so Publish
+	// acknowledges it.
 	r := &router{subscribers: map[string]*subscriber{
 		"a": {id: "a", client: &stubPublishClient{response: &chippb.PublishResponse{
 			Results: []*chippb.PublishResult{rejected("e1", "rejected by a")},
 		}}},
-		"b": {id: "b", client: &stubPublishClient{}},
+		"b": {id: "b", client: &stubPublishClient{response: &chippb.PublishResponse{}}},
 	}}
 
 	resp, err := r.Publish(t.Context(), eventOf("e1"))