diff --git a/chain_capabilities/evm/actions/actions.go b/chain_capabilities/evm/actions/actions.go
index 2a1903f9e..2ab576fc6 100644
--- a/chain_capabilities/evm/actions/actions.go
+++ b/chain_capabilities/evm/actions/actions.go
@@ -37,6 +37,7 @@ import (
 	"github.com/smartcontractkit/capabilities/chain_capabilities/evm/internal/contracts"
 	"github.com/smartcontractkit/capabilities/chain_capabilities/evm/metering"
 	"github.com/smartcontractkit/capabilities/chain_capabilities/evm/monitoring"
+	"github.com/smartcontractkit/capabilities/chain_capabilities/evm/txevents"
 	ctypes "github.com/smartcontractkit/capabilities/libs/chainconsensus/types"
 	commonMon "github.com/smartcontractkit/capabilities/libs/monitoring"
 )
@@ -63,10 +64,22 @@ type EVM struct {
 	writeReportL1FeeActive                     limits.RangeLimiter[commoncfg.Timestamp]
 
 	transmissionScheduler ts.TransmissionScheduler
+
+	// txEvents publishes ChainTransactionObserved events for the Transactions Viewer.
+	txEvents txevents.Emitter
+}
+
+// Option configures optional behaviour of the EVM capability.
+type Option func(*EVM)
+
+// WithTxEvents sets the emitter for ChainTransactionObserved events. A nil or NopEmitter
+// disables emission.
+func WithTxEvents(emitter txevents.Emitter) Option {
+	return func(e *EVM) { e.txEvents = emitter }
 }
 
 func NewEVM(cfg config.Config, evmService types.EVMService, lggr logger.Logger, beholderProcessor beholder.ProtoProcessor,
-	messageBuilder *monitoring.MessageBuilder, handler chainconsensus.RequestHandler, chainSelector uint64, limitsFactory limits.Factory, transmissionScheduler ts.TransmissionScheduler) (*EVM, caperrors.Error) {
+	messageBuilder *monitoring.MessageBuilder, handler chainconsensus.RequestHandler, chainSelector uint64, limitsFactory limits.Factory, transmissionScheduler ts.TransmissionScheduler, opts ...Option) (*EVM, caperrors.Error) {
 	keystoneForwarderAddress := common.HexToAddress(cfg.CREForwarderAddress)
 	if keystoneForwarderAddress == (common.Address{}) {
 		return &EVM{}, caperrors.NewPublicSystemError(errors.New("keystone forwarder address is not set"), caperrors.FailedPrecondition)
@@ -88,6 +101,10 @@ func NewEVM(cfg config.Config, evmService types.EVMService, lggr logger.Logger,
 		consensusHandler:         handler,
 		chainSelector:            chainSelector,
 		transmissionScheduler:    transmissionScheduler,
+		txEvents:                 txevents.NopEmitter{},
+	}
+	for _, opt := range opts {
+		opt(e)
 	}
 	err = e.initLimiters(limitsFactory)
 	if err != nil {
diff --git a/chain_capabilities/evm/actions/tx_events.go b/chain_capabilities/evm/actions/tx_events.go
new file mode 100644
index 000000000..f469f7838
--- /dev/null
+++ b/chain_capabilities/evm/actions/tx_events.go
@@ -0,0 +1,118 @@
+package actions
+
+import (
+	"context"
+	"encoding/hex"
+	"errors"
+	"time"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	evmtypes "github.com/smartcontractkit/chainlink-common/pkg/types/chains/evm"
+
+	"github.com/smartcontractkit/capabilities/chain_capabilities/evm/txevents"
+)
+
+// txEventsEnabled reports whether emitter actually publishes events.
+func txEventsEnabled(emitter txevents.Emitter) bool {
+	if emitter == nil {
+		return false
+	}
+	_, nop := emitter.(txevents.NopEmitter)
+	return !nop
+}
+
+// txEventReporter emits one ChainTransactionObserved for every transaction a WriteReport request
+// learns was mined, at the moment it learns it: the node's own transaction when the transaction
+// manager returns (it only returns once the transaction is mined), and transactions sent by
+// other nodes when the forwarder's ReportProcessed log reveals their hash. Each hash is emitted
+// once per request; every node emits what it saw and consumers dedup by txevents.Key. Emission
+// never changes the outcome of the write: failures are counted by the emitter and logged here.
+// A nil reporter is a no-op.
+type txEventReporter struct {
+	emitter       txevents.Emitter
+	chainSelector uint64
+	metadata      capabilities.RequestMetadata
+	lggr          logger.SugaredLogger
+	now           func() time.Time
+
+	emitted map[evmtypes.Hash]struct{}
+}
+
+func (e *EVM) newTxEventReporter(lggr logger.SugaredLogger, md capabilities.RequestMetadata) *txEventReporter {
+	if !txEventsEnabled(e.txEvents) {
+		return nil
+	}
+	return &txEventReporter{
+		emitter:       e.txEvents,
+		chainSelector: e.chainSelector,
+		metadata:      md,
+		lggr:          lggr,
+		now:           time.Now,
+		emitted:       map[evmtypes.Hash]struct{}{},
+	}
+}
+
+// sent reports the transaction this node just sent, with the status the transaction manager
+// returned. Fatal results carry no usable hash and are skipped.
+func (r *txEventReporter) sent(ctx context.Context, result *evmtypes.TransactionResult) {
+	if r == nil || result == nil || result.TxStatus == evmtypes.TxFatal {
+		return
+	}
+	r.observed(ctx, result.TxHash, txStatusFromTxmgr(result.TxStatus))
+}
+
+// landed reports a transaction found through the forwarder's ReportProcessed log. Such a log
+// only exists for a transaction that succeeded on chain, so its status is known without the
+// receipt.
+func (r *txEventReporter) landed(ctx context.Context, hash evmtypes.Hash) {
+	r.observed(ctx, hash, txevents.TxStatus_TX_STATUS_SUCCESS)
+}
+
+// observed emits the event for hash unless this request already did.
+func (r *txEventReporter) observed(ctx context.Context, hash evmtypes.Hash, status txevents.TxStatus) {
+	if r == nil || hash == (evmtypes.Hash{}) {
+		return
+	}
+	if _, done := r.emitted[hash]; done {
+		return
+	}
+	r.emitted[hash] = struct{}{}
+
+	ev := newTxEvent(hash, status, r.metadata, r.chainSelector, r.now())
+	// The request context may already be done (the write failed or timed out); the event must
+	// still go out, bounded by the emitter's own timeout.
+	if err := r.emitter.Emit(context.WithoutCancel(ctx), ev); err != nil {
+		if errors.Is(err, txevents.ErrEmitterUnavailable) {
+			return // logged once per process by the emitter
+		}
+		r.lggr.Errorw("Failed to emit ChainTransactionObserved", append(txevents.LogKVs(ev), "error", err.Error())...)
+	}
+}
+
+func txStatusFromTxmgr(s evmtypes.TransactionStatus) txevents.TxStatus {
+	switch s {
+	case evmtypes.TxSuccess:
+		return txevents.TxStatus_TX_STATUS_SUCCESS
+	case evmtypes.TxReverted:
+		return txevents.TxStatus_TX_STATUS_REVERTED
+	default:
+		return txevents.TxStatus_TX_STATUS_UNSPECIFIED
+	}
+}
+
+// formatTxHash renders an EVM hash the way explorers and the Transactions Viewer expect.
+func formatTxHash(h evmtypes.Hash) string {
+	return "0x" + hex.EncodeToString(h[:])
+}
+
+func newTxEvent(hash evmtypes.Hash, status txevents.TxStatus, md capabilities.RequestMetadata, chainSelector uint64, at time.Time) *txevents.ChainTransactionObserved {
+	return &txevents.ChainTransactionObserved{
+		WorkflowExecutionId: md.WorkflowExecutionID,
+		WorkflowId:          md.WorkflowID,
+		ChainSelector:       chainSelector,
+		TxHash:              formatTxHash(hash),
+		TxStatus:            status,
+		ObservedAtMs:        at.UnixMilli(),
+	}
+}
diff --git a/chain_capabilities/evm/actions/tx_events_test.go b/chain_capabilities/evm/actions/tx_events_test.go
new file mode 100644
index 000000000..e9f8ad9f9
--- /dev/null
+++ b/chain_capabilities/evm/actions/tx_events_test.go
@@ -0,0 +1,352 @@
+package actions
+
+import (
+	"context"
+	"errors"
+	"math/big"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+	evmtypes "github.com/smartcontractkit/chainlink-common/pkg/types/chains/evm"
+	commonmocks "github.com/smartcontractkit/chainlink-common/pkg/types/mocks"
+
+	"github.com/smartcontractkit/capabilities/chain_capabilities/common/test"
+	"github.com/smartcontractkit/capabilities/chain_capabilities/evm/internal/contracts"
+	"github.com/smartcontractkit/capabilities/chain_capabilities/evm/txevents"
+)
+
+// recordingTxEventEmitter captures emitted events so tests can assert on them.
+type recordingTxEventEmitter struct {
+	mu     sync.Mutex
+	events []*txevents.ChainTransactionObserved
+	err    error
+}
+
+func (r *recordingTxEventEmitter) Emit(_ context.Context, ev *txevents.ChainTransactionObserved) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.events = append(r.events, ev)
+	return r.err
+}
+
+func (r *recordingTxEventEmitter) recorded() []*txevents.ChainTransactionObserved {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return append([]*txevents.ChainTransactionObserved(nil), r.events...)
+}
+
+func recordingEmitter(t *testing.T, service *EVM) *recordingTxEventEmitter {
+	t.Helper()
+	rec, ok := service.txEvents.(*recordingTxEventEmitter)
+	require.True(t, ok, "service must use the recording emitter")
+	return rec
+}
+
+func recordedTxEvents(t *testing.T, service *EVM) []*txevents.ChainTransactionObserved {
+	t.Helper()
+	return recordingEmitter(t, service).recorded()
+}
+
+type txEventEmitterFunc func(ctx context.Context, ev *txevents.ChainTransactionObserved) error
+
+func (f txEventEmitterFunc) Emit(ctx context.Context, ev *txevents.ChainTransactionObserved) error {
+	return f(ctx, ev)
+}
+
+func testRequestMetadata() capabilities.RequestMetadata {
+	return capabilities.RequestMetadata{
+		WorkflowID:          "wf-id",
+		WorkflowOwner:       "0xowner",
+		WorkflowExecutionID: "exec-1",
+	}
+}
+
+func newTestReporter(t *testing.T, emitter txevents.Emitter) *txEventReporter {
+	t.Helper()
+	svc := &EVM{chainSelector: 1, txEvents: emitter}
+	r := svc.newTxEventReporter(logger.Sugared(logger.Test(t)), testRequestMetadata())
+	require.NotNil(t, r)
+	return r
+}
+
+func TestTxEventReporter(t *testing.T) {
+	t.Parallel()
+	ctx := t.Context()
+	hash := evmtypes.Hash(test.RandomBytes(32))
+	other := evmtypes.Hash(test.RandomBytes(32))
+
+	t.Run("disabled emission yields a nil reporter whose methods are no-ops", func(t *testing.T) {
+		lggr := logger.Sugared(logger.Test(t))
+		require.Nil(t, (&EVM{}).newTxEventReporter(lggr, testRequestMetadata()))
+		r := (&EVM{txEvents: txevents.NopEmitter{}}).newTxEventReporter(lggr, testRequestMetadata())
+		require.Nil(t, r)
+		r.sent(ctx, &evmtypes.TransactionResult{TxHash: hash, TxStatus: evmtypes.TxSuccess})
+		r.landed(ctx, hash)
+		r.observed(ctx, hash, txevents.TxStatus_TX_STATUS_SUCCESS)
+	})
+
+	t.Run("emits one event per transaction, the first time it is observed", func(t *testing.T) {
+		rec := &recordingTxEventEmitter{}
+		r := newTestReporter(t, rec)
+		r.now = func() time.Time { return time.UnixMilli(1_700_000_000_000) }
+
+		r.landed(ctx, hash)
+		r.observed(ctx, hash, txevents.TxStatus_TX_STATUS_REVERTED) // same transaction again: ignored
+		r.sent(ctx, &evmtypes.TransactionResult{TxHash: other, TxStatus: evmtypes.TxReverted})
+		r.landed(ctx, other)
+
+		events := rec.recorded()
+		require.Len(t, events, 2)
+		want := &txevents.ChainTransactionObserved{
+			WorkflowExecutionId: "exec-1",
+			WorkflowId:          "wf-id",
+			ChainSelector:       1,
+			TxHash:              formatTxHash(hash),
+			TxStatus:            txevents.TxStatus_TX_STATUS_SUCCESS,
+			ObservedAtMs:        1_700_000_000_000,
+		}
+		require.True(t, proto.Equal(want, events[0]), "got %v", events[0])
+		require.Equal(t, formatTxHash(other), events[1].TxHash)
+		require.Equal(t, txevents.TxStatus_TX_STATUS_REVERTED, events[1].TxStatus)
+	})
+
+	t.Run("sent ignores nil, fatal and zero-hash results", func(t *testing.T) {
+		rec := &recordingTxEventEmitter{}
+		r := newTestReporter(t, rec)
+		r.sent(ctx, nil)
+		r.sent(ctx, &evmtypes.TransactionResult{TxHash: hash, TxStatus: evmtypes.TxFatal})
+		r.sent(ctx, &evmtypes.TransactionResult{TxStatus: evmtypes.TxSuccess})
+		r.landed(ctx, evmtypes.Hash{})
+		require.Empty(t, rec.recorded())
+	})
+
+	t.Run("survives emitter errors", func(t *testing.T) {
+		rec := &recordingTxEventEmitter{err: errors.New("chip down")}
+		r := newTestReporter(t, rec)
+		r.landed(ctx, hash)
+		require.Len(t, rec.recorded(), 1)
+
+		unavailable := newTestReporter(t, txEventEmitterFunc(func(context.Context, *txevents.ChainTransactionObserved) error {
+			return txevents.ErrEmitterUnavailable
+		}))
+		unavailable.landed(ctx, hash)
+	})
+
+	t.Run("emits even after the request context is done", func(t *testing.T) {
+		var gotCtx context.Context
+		r := newTestReporter(t, txEventEmitterFunc(func(ctx context.Context, _ *txevents.ChainTransactionObserved) error {
+			gotCtx = ctx
+			return nil
+		}))
+		cancelled, cancel := context.WithCancel(t.Context())
+		cancel()
+		r.landed(cancelled, hash)
+		require.NotNil(t, gotCtx)
+		require.NoError(t, gotCtx.Err())
+	})
+}
+
+func TestWriteReport_EmitsChainTransactionObserved(t *testing.T) {
+	t.Parallel()
+
+	expectReceipt := func(t *testing.T, evmServiceMock *commonmocks.EVMService, txHash evmtypes.Hash) {
+		t.Helper()
+		receipt := evmtypes.Receipt{Status: 1, TxHash: txHash, GasUsed: 1000, EffectiveGasPrice: big.NewInt(2), BlockNumber: big.NewInt(1234)}
+		evmServiceMock.EXPECT().
+			GetTransactionReceipt(mock.Anything, evmtypes.GeTransactionReceiptRequest{Hash: txHash, IsExternal: false}).
+			Return(&receipt, nil)
+		evmServiceMock.EXPECT().
+			CalculateTransactionFee(mock.Anything, toReceiptGasInfo(receipt)).
+			Return(&evmtypes.TransactionFee{TransactionFee: big.NewInt(2000)}, nil)
+	}
+
+	assertEvent := func(t *testing.T, ev *txevents.ChainTransactionObserved, fixture writeReportTestFixture, txHash evmtypes.Hash, status txevents.TxStatus) {
+		t.Helper()
+		require.Equal(t, fixture.metadata.WorkflowExecutionID, ev.WorkflowExecutionId)
+		require.Equal(t, fixture.metadata.WorkflowID, ev.WorkflowId)
+		require.Equal(t, uint64(1), ev.ChainSelector)
+		require.Equal(t, formatTxHash(txHash), ev.TxHash)
+		require.Equal(t, status, ev.TxStatus)
+		require.NotZero(t, ev.ObservedAtMs)
+	}
+
+	t.Run("prior successful transmission is reported", func(t *testing.T) {
+		ctx := t.Context()
+		evmServiceMock, mockForwarderClient, service := createMocksAndCapability(t, logger.Test(t))
+		fixture := newWriteReportTestFixture(t)
+		mockForwarderClient.On("GetTransmissionInfo", mock.Anything, mock.Anything).
+			Return(contracts.TransmissionInfo{Success: true, State: contracts.TransmissionStateSucceeded}, nil).Once()
+
+		txHash := evmtypes.Hash(test.RandomBytes(32))
+		expectSuccessfulTransmissionEvent(t, mockForwarderClient, txHash)
+		expectReceipt(t, evmServiceMock, txHash)
+
+		_, err := service.WriteReport(ctx, fixture.metadata, fixture.request)
+		require.NoError(t, err)
+
+		events := recordedTxEvents(t, service)
+		require.Len(t, events, 1)
+		assertEvent(t, events[0], fixture, txHash, txevents.TxStatus_TX_STATUS_SUCCESS)
+	})
+
+	t.Run("prior transmission with an invalid receiver is reported as a mined transaction", func(t *testing.T) {
+		ctx := t.Context()
+		evmServiceMock, mockForwarderClient, service := createMocksAndCapability(t, logger.Test(t))
+		fixture := newWriteReportTestFixture(t)
+		mockForwarderClient.On("GetTransmissionInfo", mock.Anything, mock.Anything).
+			Return(contracts.TransmissionInfo{InvalidReceiver: true, State: contracts.TransmissionStateInvalidReceiver, GasLimit: big.NewInt(EnoughReceiverGas)}, nil).Once()
+
+		txHash := evmtypes.Hash(test.RandomBytes(32))
+		expectFailedTransmissionEvent(t, mockForwarderClient, txHash)
+		expectReceipt(t, evmServiceMock, txHash)
+
+		_, err := service.WriteReport(ctx, fixture.metadata, fixture.request)
+		require.NoError(t, err)
+
+		events := recordedTxEvents(t, service)
+		require.Len(t, events, 1)
+		assertEvent(t, events[0], fixture, txHash, txevents.TxStatus_TX_STATUS_SUCCESS)
+		mockForwarderClient.AssertNotCalled(t, "InvokeOnReport", mock.Anything, mock.Anything, mock.Anything, mock.Anything)
+	})
+
+	t.Run("own transmission is reported as soon as the transaction manager returns, before the receipt", func(t *testing.T) {
+		ctx := t.Context()
+		evmServiceMock, mockForwarderClient, service := createMocksAndCapability(t, logger.Test(t))
+		fixture := newWriteReportTestFixture(t)
+		mockForwarderClient.On("GetTransmissionInfo", mock.Anything, mock.Anything).
+			Return(contracts.TransmissionInfo{State: contracts.TransmissionStateNotAttempted}, nil).Once()
+
+		txHash := evmtypes.Hash(test.RandomBytes(32))
+		emittedBeforeReceipt := -1
+		mockForwarderClient.
+			On("InvokeOnReport", mock.Anything, fixture.receiver, fixture.request.Report, nonNilPositiveGasCfgMatcher()).
+			Return(&evmtypes.TransactionResult{TxHash: txHash, TxStatus: evmtypes.TxSuccess, TxIdempotencyKey: "key"}, nil).Once()
+		mockForwarderClient.On("GetTransmissionInfo", mock.Anything, mock.Anything).
+			Run(func(mock.Arguments) { emittedBeforeReceipt = len(recordedTxEvents(t, service)) }).
+			Return(contracts.TransmissionInfo{Success: true, State: contracts.TransmissionStateSucceeded, GasLimit: big.NewInt(EnoughReceiverGas)}, nil).Once()
+		expectSuccessfulTransmissionEvent(t, mockForwarderClient, txHash)
+		expectReceipt(t, evmServiceMock, txHash)
+
+		_, err := service.WriteReport(ctx, fixture.metadata, fixture.request)
+		require.NoError(t, err)
+		require.Equal(t, 1, emittedBeforeReceipt, "the event goes out when the transaction manager returns the mined transaction")
+
+		events := recordedTxEvents(t, service)
+		require.Len(t, events, 1, "seeing our own hash again in the forwarder log does not emit twice")
+		assertEvent(t, events[0], fixture, txHash, txevents.TxStatus_TX_STATUS_SUCCESS)
+	})
+
+	t.Run("duplicate transmission reports our reverted transaction and the one that landed", func(t *testing.T) {
+		ctx := t.Context()
+		evmServiceMock, mockForwarderClient, service := createMocksAndCapability(t, logger.Test(t))
+		fixture := newWriteReportTestFixture(t)
+		mockForwarderClient.On("GetTransmissionInfo", mock.Anything, mock.Anything).
+			Return(contracts.TransmissionInfo{State: contracts.TransmissionStateNotAttempted}, nil).Once()
+
+		ownHash := evmtypes.Hash(test.RandomBytes(32))
+		landedHash := evmtypes.Hash(test.RandomBytes(32))
+		mockForwarderClient.
+			On("InvokeOnReport", mock.Anything, fixture.receiver, fixture.request.Report, nonNilPositiveGasCfgMatcher()).
+			Return(&evmtypes.TransactionResult{TxHash: ownHash, TxStatus: evmtypes.TxReverted, TxIdempotencyKey: "dup"}, nil).Once()
+		mockForwarderClient.On("GetTransmissionInfo", mock.Anything, mock.Anything).
+			Return(contracts.TransmissionInfo{Success: true, State: contracts.TransmissionStateSucceeded, GasLimit: big.NewInt(EnoughReceiverGas)}, nil).Once()
+		expectSuccessfulTransmissionEvent(t, mockForwarderClient, landedHash)
+		expectReceipt(t, evmServiceMock, landedHash)
+
+		_, err := service.WriteReport(ctx, fixture.metadata, fixture.request)
+		require.NoError(t, err)
+
+		events := recordedTxEvents(t, service)
+		require.Len(t, events, 2)
+		assertEvent(t, events[0], fixture, ownHash, txevents.TxStatus_TX_STATUS_REVERTED)
+		assertEvent(t, events[1], fixture, landedHash, txevents.TxStatus_TX_STATUS_SUCCESS)
+		require.NotEqual(t, txevents.Key(events[0]), txevents.Key(events[1]))
+	})
+
+	t.Run("receipt lookup failure does not lose the event", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(t.Context(), 500*time.Millisecond)
+		defer cancel()
+		evmServiceMock, mockForwarderClient, service := createMocksAndCapability(t, logger.Test(t))
+		fixture := newWriteReportTestFixture(t)
+		mockForwarderClient.On("GetTransmissionInfo", mock.Anything, mock.Anything).
+			Return(contracts.TransmissionInfo{Success: true, State: contracts.TransmissionStateSucceeded}, nil)
+
+		txHash := evmtypes.Hash(test.RandomBytes(32))
+		expectSuccessfulTransmissionEvent(t, mockForwarderClient, txHash)
+		evmServiceMock.EXPECT().
+			GetTransactionReceipt(mock.Anything, evmtypes.GeTransactionReceiptRequest{Hash: txHash, IsExternal: false}).
+			Return(nil, errors.New("rpc down"))
+
+		_, err := service.WriteReport(ctx, fixture.metadata, fixture.request)
+		require.Error(t, err)
+
+		events := recordedTxEvents(t, service)
+		require.Len(t, events, 1)
+		assertEvent(t, events[0], fixture, txHash, txevents.TxStatus_TX_STATUS_SUCCESS)
+	})
+
+	t.Run("request failing after the transaction was sent still reports it", func(t *testing.T) {
+		ctx, cancel := context.WithTimeout(t.Context(), 500*time.Millisecond)
+		defer cancel()
+		_, mockForwarderClient, service := createMocksAndCapability(t, logger.Test(t))
+		fixture := newWriteReportTestFixture(t)
+		mockForwarderClient.On("GetTransmissionInfo", mock.Anything, mock.Anything).
+			Return(contracts.TransmissionInfo{State: contracts.TransmissionStateNotAttempted}, nil).Once()
+
+		txHash := evmtypes.Hash(test.RandomBytes(32))
+		mockForwarderClient.
+			On("InvokeOnReport", mock.Anything, fixture.receiver, fixture.request.Report, nonNilPositiveGasCfgMatcher()).
+			Return(&evmtypes.TransactionResult{TxHash: txHash, TxStatus: evmtypes.TxSuccess}, nil).Once()
+		mockForwarderClient.On("GetTransmissionInfo", mock.Anything, mock.Anything).
+			Return(contracts.TransmissionInfo{}, errors.New("rpc down"))
+
+		_, err := service.WriteReport(ctx, fixture.metadata, fixture.request)
+		require.Error(t, err)
+
+		events := recordedTxEvents(t, service)
+		require.Len(t, events, 1)
+		assertEvent(t, events[0], fixture, txHash, txevents.TxStatus_TX_STATUS_SUCCESS)
+	})
+
+	t.Run("emitter failure does not fail the write", func(t *testing.T) {
+		ctx := t.Context()
+		evmServiceMock, mockForwarderClient, service := createMocksAndCapability(t, logger.Test(t))
+		recordingEmitter(t, service).err = errors.New("chip unreachable")
+		fixture := newWriteReportTestFixture(t)
+		mockForwarderClient.On("GetTransmissionInfo", mock.Anything, mock.Anything).
+			Return(contracts.TransmissionInfo{Success: true, State: contracts.TransmissionStateSucceeded}, nil).Once()
+
+		txHash := evmtypes.Hash(test.RandomBytes(32))
+		expectSuccessfulTransmissionEvent(t, mockForwarderClient, txHash)
+		expectReceipt(t, evmServiceMock, txHash)
+
+		txResult, err := service.WriteReport(ctx, fixture.metadata, fixture.request)
+		require.NoError(t, err)
+		require.Equal(t, txHash[:], txResult.Response.TxHash)
+		require.Len(t, recordedTxEvents(t, service), 1)
+	})
+
+	t.Run("disabled emission leaves the write untouched", func(t *testing.T) {
+		ctx := t.Context()
+		evmServiceMock, mockForwarderClient, service := createMocksAndCapability(t, logger.Test(t))
+		service.txEvents = txevents.NopEmitter{}
+		fixture := newWriteReportTestFixture(t)
+		mockForwarderClient.On("GetTransmissionInfo", mock.Anything, mock.Anything).
+			Return(contracts.TransmissionInfo{Success: true, State: contracts.TransmissionStateSucceeded}, nil).Once()
+
+		txHash := evmtypes.Hash(test.RandomBytes(32))
+		expectSuccessfulTransmissionEvent(t, mockForwarderClient, txHash)
+		expectReceipt(t, evmServiceMock, txHash)
+
+		txResult, err := service.WriteReport(ctx, fixture.metadata, fixture.request)
+		require.NoError(t, err)
+		require.Equal(t, txHash[:], txResult.Response.TxHash)
+	})
+}
diff --git a/chain_capabilities/evm/actions/write_report.go b/chain_capabilities/evm/actions/write_report.go
index 418b4f9ae..5e4e38788 100644
--- a/chain_capabilities/evm/actions/write_report.go
+++ b/chain_capabilities/evm/actions/write_report.go
@@ -55,6 +55,10 @@ type WriteReport struct {
 	executionTimestamp     time.Time
 
 	gasMeter *gasmeter.Meter
+
+	// txReporter emits ChainTransactionObserved the moment a transaction is known to be mined.
+	// Nil when transaction events are disabled.
+	txReporter *txEventReporter
 }
 
 func (e *EVM) WriteReport(ctx context.Context, metadata capabilities.RequestMetadata, input *evm.WriteReportRequest) (*capabilities.ResponseAndMetadata[*evm.WriteReportReply], caperrors.Error) {
@@ -85,13 +89,14 @@ func (e *EVM) WriteReport(ctx context.Context, metadata capabilities.RequestMeta
 }
 
 func (e *EVM) executeWriteReport(ctx context.Context, request *evm.WriteReportRequest, metadata capabilities.RequestMetadata, telemetryContext monitoring.TelemetryContext) (*evm.WriteReportReply, capabilities.ResponseMetadata, error) {
+	lggr := e.messageBuilder.RequestLggr(e.lggr, telemetryContext)
 	wr := &WriteReport{
 		EVMService:         e.EVMService,
 		forwarderClient:    e.forwarderClient,
 		ReceiverGasMinimum: e.ReceiverGasMinimum,
 		chainSelector:      e.chainSelector,
 
-		lggr:              e.messageBuilder.RequestLggr(e.lggr, telemetryContext),
+		lggr:              lggr,
 		beholderProcessor: e.beholderProcessor,
 		messageBuilder:    e.messageBuilder,
 
@@ -101,6 +106,7 @@ func (e *EVM) executeWriteReport(ctx context.Context, request *evm.WriteReportRe
 		transmissionScheduler:  e.transmissionScheduler,
 		executionTimestamp:     metadata.ExecutionTimestamp,
 		gasMeter:               e.gasMeter,
+		txReporter:             e.newTxEventReporter(lggr, metadata),
 	}
 
 	return wr.executeWriteReport(ctx, request, metadata, telemetryContext)
@@ -147,6 +153,7 @@ func (e *WriteReport) executeWriteReport(ctx context.Context, request *evm.Write
 		}
 
 		e.lggr.Infow("Returning without a transmission attempt - prior transmission succeeded", "txHash", common.Bytes2Hex(txHash[:]))
+		e.txReporter.landed(ctx, *txHash)
 		reply, err := e.buildSuccessReply(ctx, *txHash)
 		if err != nil {
 			return nil, capabilities.ResponseMetadata{}, err
@@ -164,6 +171,7 @@ func (e *WriteReport) executeWriteReport(ctx context.Context, request *evm.Write
 		}
 
 		e.lggr.Infow("Returning without a transmission attempt - prior transmission marked receiver invalid", "txHash", common.Bytes2Hex(txHash[:]))
+		e.txReporter.landed(ctx, *txHash)
 		reply, err := e.buildRevertReplyFromTx(ctx, *txHash, transmissionInfo, transmissionID)
 		if err != nil {
 			// Surface the known revert reason so the user learns their receiver is invalid even
@@ -189,6 +197,7 @@ func (e *WriteReport) executeWriteReport(ctx context.Context, request *evm.Write
 				"receiverGasBudget", calculatedReceiverGasBudget,
 				"transmissionReceiverGasBudget", transmissionInfo.GasLimit,
 			)
+			e.txReporter.landed(ctx, *txHash)
 			reply, err := e.buildRevertReplyFromTx(ctx, *txHash, transmissionInfo, transmissionID)
 			if err != nil {
 				// The receiver reverted despite sufficient gas (user contract fault); surface the
@@ -216,6 +225,7 @@ func (e *WriteReport) executeWriteReport(ctx context.Context, request *evm.Write
 		e.lggr.Errorw("Transaction failed", "error", err.Error())
 		return nil, capabilities.ResponseMetadata{}, err
 	}
+	e.txReporter.sent(ctx, transactionResult)
 
 	newTransmissionInfo, err := capcommon.WithPollingRetry(ctx, e.lggr, func(ctx context.Context) (contracts.TransmissionInfo, error) {
 		readTransmissionInfo, readTransmissionErr := e.forwarderClient.GetTransmissionInfo(ctx, transmissionID)
@@ -240,6 +250,7 @@ func (e *WriteReport) executeWriteReport(ctx context.Context, request *evm.Write
 		if err != nil {
 			return nil, capabilities.ResponseMetadata{}, err
 		}
+		e.txReporter.landed(ctx, *txHash)
 		switch transactionResult.TxStatus {
 		case evmtypes.TxReverted:
 			// Report for this transaction has already been submitted and we sent a duplicate tx onchain which is fine, but wastes ethereum gas
@@ -271,6 +282,7 @@ func (e *WriteReport) executeWriteReport(ctx context.Context, request *evm.Write
 			txHash = originalTxHash
 		}
 		e.lggr.Errorw("Made a new transmission attempt - transmission failed", "txHash", common.Bytes2Hex(txHash[:]), "transmissionState", newTransmissionInfo.State.String())
+		e.txReporter.landed(ctx, *txHash)
 
 		reply, err := e.buildRevertReplyFromTx(ctx, *txHash, newTransmissionInfo, transmissionID)
 		if err != nil {
@@ -512,7 +524,6 @@ func (e *WriteReport) replyFromReceipt(ctx context.Context, txHash evmtypes.Hash
 	if txReceipt.Status == 0 {
 		txStatus = evm.TxStatus_TX_STATUS_REVERTED
 	}
-
 	e.lggr.Infow("Successfully fetched tx receipt",
 		"txHash", hex.EncodeToString(txHash[:]),
 		"txStatus", txReceipt.Status,
@@ -684,7 +695,6 @@ func (thr *TxHashRetriever) fetchAndParseLogs(ctx context.Context) (logDetailsLi
 	if err != nil {
 		return nil, fmt.Errorf("malformed log data for transmission %s: %w", thr.transmissionID.String(), err)
 	}
-
 	return details, nil
 }
 
diff --git a/chain_capabilities/evm/actions/write_report_test.go b/chain_capabilities/evm/actions/write_report_test.go
index f4bc2d10b..a6b61b8c0 100644
--- a/chain_capabilities/evm/actions/write_report_test.go
+++ b/chain_capabilities/evm/actions/write_report_test.go
@@ -2249,6 +2249,7 @@ func createMocksAndCapability(t *testing.T, lggr logger.Logger) (*mocks2.EVMServ
 		beholderProcessor:        test.NopBeholderProcessor{},
 		messageBuilder:           monitoring.NewMessageBuilder(types.ChainInfo{}, capabilities.CapabilityInfo{}, ""),
 		transmissionScheduler:    transmissionScheduler,
+		txEvents:                 &recordingTxEventEmitter{},
 	}
 	require.NoError(t, service.initLimiters(limits.Factory{Logger: lggr}))
 	t.Cleanup(func() { assert.NoError(t, service.Close()) })
@@ -2275,6 +2276,7 @@ func createMocksAndCapabilityWithScheduler(
 		beholderProcessor:        test.NopBeholderProcessor{},
 		messageBuilder:           monitoring.NewMessageBuilder(types.ChainInfo{}, capabilities.CapabilityInfo{}, ""),
 		transmissionScheduler:    scheduler,
+		txEvents:                 &recordingTxEventEmitter{},
 	}
 	require.NoError(t, service.initLimiters(limits.Factory{Logger: lggr}))
 	t.Cleanup(func() { assert.NoError(t, service.Close()) })
diff --git a/chain_capabilities/evm/config/config.go b/chain_capabilities/evm/config/config.go
index 8cb56045f..9f6bbbc18 100644
--- a/chain_capabilities/evm/config/config.go
+++ b/chain_capabilities/evm/config/config.go
@@ -21,4 +21,17 @@ type Config struct {
 	// DeltaStage for staggered transmission scheduling
 	DeltaStage time.Duration `json:"deltaStage"`
 	IsLocaL    bool          `json:"isLocal"` // use in integration-test to skip transmission scheduler initialization for log_trigger
+
+	// TxEvents controls emission of ChainTransactionObserved events, which feed the CRE
+	// Transactions Viewer through ChIP and the Blockchain Finality Service (PLEX-3409).
+	TxEvents TxEventsConfig `json:"txEvents"`
+}
+
+// TxEventsConfig configures ChainTransactionObserved emission.
+type TxEventsConfig struct {
+	// Enabled turns emission on. The node must also run with CL_CHIP_INGRESS_DURABLE_EMITTER_ENABLED
+	// for the events to leave the process; otherwise they are counted as dropped.
+	Enabled bool `json:"enabled"`
+	// Domain is the ChIP source domain the events are published to. Defaults to txevents.DefaultDomain.
+	Domain string `json:"domain"`
 }
diff --git a/chain_capabilities/evm/go.mod b/chain_capabilities/evm/go.mod
index b39f7d6ef..53ce837ae 100644
--- a/chain_capabilities/evm/go.mod
+++ b/chain_capabilities/evm/go.mod
@@ -16,6 +16,7 @@ require (
 	github.com/smartcontractkit/libocr v0.0.0-20260810200708-618b5bf7f342
 	github.com/stretchr/testify v1.12.1
 	go.opentelemetry.io/otel v1.46.0
+	go.opentelemetry.io/otel/metric v1.46.0
 	go.opentelemetry.io/otel/sdk/metric v1.46.0
 	go.uber.org/zap v1.28.0
 	google.golang.org/protobuf v1.36.12
@@ -182,7 +183,6 @@ require (
 	go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.46.0 // indirect
 	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.46.0 // indirect
 	go.opentelemetry.io/otel/log v0.22.0 // indirect
-	go.opentelemetry.io/otel/metric v1.46.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.46.0 // indirect
 	go.opentelemetry.io/otel/sdk/log v0.22.0 // indirect
 	go.opentelemetry.io/otel/trace v1.46.0 // indirect
diff --git a/chain_capabilities/evm/main.go b/chain_capabilities/evm/main.go
index 833218732..7550ec179 100644
--- a/chain_capabilities/evm/main.go
+++ b/chain_capabilities/evm/main.go
@@ -29,11 +29,13 @@ import (
 	"github.com/smartcontractkit/capabilities/chain_capabilities/evm/config"
 	"github.com/smartcontractkit/capabilities/chain_capabilities/evm/monitoring"
 	"github.com/smartcontractkit/capabilities/chain_capabilities/evm/trigger"
+	"github.com/smartcontractkit/capabilities/chain_capabilities/evm/txevents"
 	"github.com/smartcontractkit/capabilities/libs/loopserver"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
 	evmcappb "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/chain-capabilities/evm"
 	evmcapserver "github.com/smartcontractkit/chainlink-common/pkg/capabilities/v2/chain-capabilities/evm/server"
+	"github.com/smartcontractkit/chainlink-common/pkg/durableemitter"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/loop"
 	"github.com/smartcontractkit/chainlink-common/pkg/resourcemanager"
@@ -178,7 +180,11 @@ func (c *capabilityGRPCService) Initialise(ctx context.Context, dependencies cor
 		c.lggr.Infow("DeltaStage not configured, transmission scheduling disabled")
 	}
 
-	c.EVM, err = actions.NewEVM(*cfg, evmRelayer, c.lggr, processor, messageBuilder, c.consensusHandler, c.chainSelector, c.limitsFactory, scheduler)
+	txEventsOpt, err := c.txEventsOption(cfg)
+	if err != nil {
+		return err
+	}
+	c.EVM, err = actions.NewEVM(*cfg, evmRelayer, c.lggr, processor, messageBuilder, c.consensusHandler, c.chainSelector, c.limitsFactory, scheduler, txEventsOpt)
 	if err != nil {
 		return fmt.Errorf("failed to init evm relayer for chainID %d from relayer: %w", cfg.ChainID, err)
 	}
@@ -227,6 +233,25 @@ func (c *capabilityGRPCService) Initialise(ctx context.Context, dependencies cor
 	return nil
 }
 
+// txEventsOption wires ChainTransactionObserved emission. Events go to ChIP through the
+// process-global durable emitter (txevents.GlobalSink), the path the platform uses for its own
+// ChIP events; the LOOP server sets it up when the node runs with
+// CL_CHIP_INGRESS_DURABLE_EMITTER_ENABLED. Without it the events are dropped and counted.
+func (c *capabilityGRPCService) txEventsOption(cfg *config.Config) (actions.Option, error) {
+	if !cfg.TxEvents.Enabled {
+		return actions.WithTxEvents(txevents.NopEmitter{}), nil
+	}
+	if durableemitter.GetGlobalEmitter() == nil {
+		c.lggr.Errorw("Transaction events enabled but this LOOP has no durable emitter; events will not be delivered", "domain", cfg.TxEvents.Domain)
+	}
+	emitter, err := txevents.NewDurableEmitter(c.lggr, cfg.TxEvents.Domain, txevents.GlobalSink())
+	if err != nil {
+		return nil, fmt.Errorf("failed to create tx events emitter: %w", err)
+	}
+	c.lggr.Infow("Transaction events enabled", "domain", cfg.TxEvents.Domain)
+	return actions.WithTxEvents(emitter), nil
+}
+
 func (c *capabilityGRPCService) unmarshalConfig(configStr string) (*config.Config, error) {
 	var cfg config.Config
 	if err := json.Unmarshal([]byte(configStr), &cfg); err != nil {
@@ -270,6 +295,10 @@ func (c *capabilityGRPCService) unmarshalConfig(configStr string) (*config.Confi
 		c.lggr.Infof("MaxUnknownRequestsCacheSize is zero, setting to %d.", cfg.MaxUnknownRequestsCacheSize)
 	}
 
+	if cfg.TxEvents.Domain == "" {
+		cfg.TxEvents.Domain = txevents.DefaultDomain
+	}
+
 	// DeltaStage is optional - if not set, transmission scheduling will be disabled
 	return &cfg, nil
 }
diff --git a/chain_capabilities/evm/txevents/chain_transaction_observed.pb.go b/chain_capabilities/evm/txevents/chain_transaction_observed.pb.go
new file mode 100644
index 000000000..3e5cf2537
--- /dev/null
+++ b/chain_capabilities/evm/txevents/chain_transaction_observed.pb.go
@@ -0,0 +1,241 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.36.10
+// 	protoc        v5.29.3
+// source: chain_capabilities/evm/txevents/chain_transaction_observed.proto
+
+// Package txevents defines the transaction event every chain capability emits for the CRE
+// Transactions Viewer. The package name is intentionally free of the chain family so Aptos,
+// Solana and Stellar can emit the same schema; the file lives in the EVM module only until the
+// ChIP domain and schema registration are settled with Data Platform (PLEX-3409).
+
+package txevents
+
+import (
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	reflect "reflect"
+	sync "sync"
+	unsafe "unsafe"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+// TxStatus is the on-chain outcome of the transaction as known at the moment it was observed:
+// from the transaction manager for the node's own transaction, or implied by the forwarder's
+// ReportProcessed log (which only exists for successful transactions) for transactions sent by
+// other nodes. Consumers fetch the receipt for anything beyond that.
+type TxStatus int32
+
+const (
+	TxStatus_TX_STATUS_UNSPECIFIED TxStatus = 0
+	TxStatus_TX_STATUS_SUCCESS     TxStatus = 1
+	TxStatus_TX_STATUS_REVERTED    TxStatus = 2
+)
+
+// Enum value maps for TxStatus.
+var (
+	TxStatus_name = map[int32]string{
+		0: "TX_STATUS_UNSPECIFIED",
+		1: "TX_STATUS_SUCCESS",
+		2: "TX_STATUS_REVERTED",
+	}
+	TxStatus_value = map[string]int32{
+		"TX_STATUS_UNSPECIFIED": 0,
+		"TX_STATUS_SUCCESS":     1,
+		"TX_STATUS_REVERTED":    2,
+	}
+)
+
+func (x TxStatus) Enum() *TxStatus {
+	p := new(TxStatus)
+	*p = x
+	return p
+}
+
+func (x TxStatus) String() string {
+	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
+}
+
+func (TxStatus) Descriptor() protoreflect.EnumDescriptor {
+	return file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_enumTypes[0].Descriptor()
+}
+
+func (TxStatus) Type() protoreflect.EnumType {
+	return &file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_enumTypes[0]
+}
+
+func (x TxStatus) Number() protoreflect.EnumNumber {
+	return protoreflect.EnumNumber(x)
+}
+
+// Deprecated: Use TxStatus.Descriptor instead.
+func (TxStatus) EnumDescriptor() ([]byte, []int) {
+	return file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_rawDescGZIP(), []int{0}
+}
+
+// ChainTransactionObserved is emitted by a chain capability as soon as it learns that a
+// transaction belonging to a workflow execution was mined, including reverted and duplicate
+// ones. It carries only what is needed to attribute the transaction to the execution; the
+// Blockchain Finality Service dedups events by (chain_selector, tx_hash) and resolves the
+// receipt, fee and finality itself.
+type ChainTransactionObserved struct {
+	state               protoimpl.MessageState `protogen:"open.v1"`
+	WorkflowExecutionId string                 `protobuf:"bytes,1,opt,name=workflow_execution_id,json=workflowExecutionId,proto3" json:"workflow_execution_id,omitempty"`
+	WorkflowId          string                 `protobuf:"bytes,2,opt,name=workflow_id,json=workflowId,proto3" json:"workflow_id,omitempty"`
+	ChainSelector       uint64                 `protobuf:"varint,3,opt,name=chain_selector,json=chainSelector,proto3" json:"chain_selector,omitempty"`
+	// Chain-native encoding. EVM: 0x-prefixed lowercase hex.
+	TxHash   string   `protobuf:"bytes,4,opt,name=tx_hash,json=txHash,proto3" json:"tx_hash,omitempty"`
+	TxStatus TxStatus `protobuf:"varint,5,opt,name=tx_status,json=txStatus,proto3,enum=chain_capabilities.txevents.v1.TxStatus" json:"tx_status,omitempty"`
+	// Unix milliseconds at which the node observed the transaction.
+	ObservedAtMs  int64 `protobuf:"varint,6,opt,name=observed_at_ms,json=observedAtMs,proto3" json:"observed_at_ms,omitempty"`
+	unknownFields protoimpl.UnknownFields
+	sizeCache     protoimpl.SizeCache
+}
+
+func (x *ChainTransactionObserved) Reset() {
+	*x = ChainTransactionObserved{}
+	mi := &file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_msgTypes[0]
+	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+	ms.StoreMessageInfo(mi)
+}
+
+func (x *ChainTransactionObserved) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*ChainTransactionObserved) ProtoMessage() {}
+
+func (x *ChainTransactionObserved) ProtoReflect() protoreflect.Message {
+	mi := &file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_msgTypes[0]
+	if x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use ChainTransactionObserved.ProtoReflect.Descriptor instead.
+func (*ChainTransactionObserved) Descriptor() ([]byte, []int) {
+	return file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *ChainTransactionObserved) GetWorkflowExecutionId() string {
+	if x != nil {
+		return x.WorkflowExecutionId
+	}
+	return ""
+}
+
+func (x *ChainTransactionObserved) GetWorkflowId() string {
+	if x != nil {
+		return x.WorkflowId
+	}
+	return ""
+}
+
+func (x *ChainTransactionObserved) GetChainSelector() uint64 {
+	if x != nil {
+		return x.ChainSelector
+	}
+	return 0
+}
+
+func (x *ChainTransactionObserved) GetTxHash() string {
+	if x != nil {
+		return x.TxHash
+	}
+	return ""
+}
+
+func (x *ChainTransactionObserved) GetTxStatus() TxStatus {
+	if x != nil {
+		return x.TxStatus
+	}
+	return TxStatus_TX_STATUS_UNSPECIFIED
+}
+
+func (x *ChainTransactionObserved) GetObservedAtMs() int64 {
+	if x != nil {
+		return x.ObservedAtMs
+	}
+	return 0
+}
+
+var File_chain_capabilities_evm_txevents_chain_transaction_observed_proto protoreflect.FileDescriptor
+
+const file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_rawDesc = "" +
+	"\n" +
+	"@chain_capabilities/evm/txevents/chain_transaction_observed.proto\x12\x1echain_capabilities.txevents.v1\"\x9c\x02\n" +
+	"\x18ChainTransactionObserved\x122\n" +
+	"\x15workflow_execution_id\x18\x01 \x01(\tR\x13workflowExecutionId\x12\x1f\n" +
+	"\vworkflow_id\x18\x02 \x01(\tR\n" +
+	"workflowId\x12%\n" +
+	"\x0echain_selector\x18\x03 \x01(\x04R\rchainSelector\x12\x17\n" +
+	"\atx_hash\x18\x04 \x01(\tR\x06txHash\x12E\n" +
+	"\ttx_status\x18\x05 \x01(\x0e2(.chain_capabilities.txevents.v1.TxStatusR\btxStatus\x12$\n" +
+	"\x0eobserved_at_ms\x18\x06 \x01(\x03R\fobservedAtMs*T\n" +
+	"\bTxStatus\x12\x19\n" +
+	"\x15TX_STATUS_UNSPECIFIED\x10\x00\x12\x15\n" +
+	"\x11TX_STATUS_SUCCESS\x10\x01\x12\x16\n" +
+	"\x12TX_STATUS_REVERTED\x10\x02BSZQgithub.com/smartcontractkit/capabilities/chain_capabilities/evm/txevents;txeventsb\x06proto3"
+
+var (
+	file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_rawDescOnce sync.Once
+	file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_rawDescData []byte
+)
+
+func file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_rawDescGZIP() []byte {
+	file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_rawDescOnce.Do(func() {
+		file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_rawDesc), len(file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_rawDesc)))
+	})
+	return file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_rawDescData
+}
+
+var file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
+var file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
+var file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_goTypes = []any{
+	(TxStatus)(0),                    // 0: chain_capabilities.txevents.v1.TxStatus
+	(*ChainTransactionObserved)(nil), // 1: chain_capabilities.txevents.v1.ChainTransactionObserved
+}
+var file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_depIdxs = []int32{
+	0, // 0: chain_capabilities.txevents.v1.ChainTransactionObserved.tx_status:type_name -> chain_capabilities.txevents.v1.TxStatus
+	1, // [1:1] is the sub-list for method output_type
+	1, // [1:1] is the sub-list for method input_type
+	1, // [1:1] is the sub-list for extension type_name
+	1, // [1:1] is the sub-list for extension extendee
+	0, // [0:1] is the sub-list for field type_name
+}
+
+func init() { file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_init() }
+func file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_init() {
+	if File_chain_capabilities_evm_txevents_chain_transaction_observed_proto != nil {
+		return
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: unsafe.Slice(unsafe.StringData(file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_rawDesc), len(file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_rawDesc)),
+			NumEnums:      1,
+			NumMessages:   1,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_goTypes,
+		DependencyIndexes: file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_depIdxs,
+		EnumInfos:         file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_enumTypes,
+		MessageInfos:      file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_msgTypes,
+	}.Build()
+	File_chain_capabilities_evm_txevents_chain_transaction_observed_proto = out.File
+	file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_goTypes = nil
+	file_chain_capabilities_evm_txevents_chain_transaction_observed_proto_depIdxs = nil
+}
diff --git a/chain_capabilities/evm/txevents/chain_transaction_observed.proto b/chain_capabilities/evm/txevents/chain_transaction_observed.proto
new file mode 100644
index 000000000..b1bc5e7b3
--- /dev/null
+++ b/chain_capabilities/evm/txevents/chain_transaction_observed.proto
@@ -0,0 +1,35 @@
+syntax = "proto3";
+
+// Package txevents defines the transaction event every chain capability emits for the CRE
+// Transactions Viewer. The package name is intentionally free of the chain family so Aptos,
+// Solana and Stellar can emit the same schema; the file lives in the EVM module only until the
+// ChIP domain and schema registration are settled with Data Platform (PLEX-3409).
+package chain_capabilities.txevents.v1;
+
+option go_package = "github.com/smartcontractkit/capabilities/chain_capabilities/evm/txevents;txevents";
+
+// TxStatus is the on-chain outcome of the transaction as known at the moment it was observed:
+// from the transaction manager for the node's own transaction, or implied by the forwarder's
+// ReportProcessed log (which only exists for successful transactions) for transactions sent by
+// other nodes. Consumers fetch the receipt for anything beyond that.
+enum TxStatus {
+  TX_STATUS_UNSPECIFIED = 0;
+  TX_STATUS_SUCCESS = 1;
+  TX_STATUS_REVERTED = 2;
+}
+
+// ChainTransactionObserved is emitted by a chain capability as soon as it learns that a
+// transaction belonging to a workflow execution was mined, including reverted and duplicate
+// ones. It carries only what is needed to attribute the transaction to the execution; the
+// Blockchain Finality Service dedups events by (chain_selector, tx_hash) and resolves the
+// receipt, fee and finality itself.
+message ChainTransactionObserved {
+  string workflow_execution_id = 1;
+  string workflow_id = 2;
+  uint64 chain_selector = 3;
+  // Chain-native encoding. EVM: 0x-prefixed lowercase hex.
+  string tx_hash = 4;
+  TxStatus tx_status = 5;
+  // Unix milliseconds at which the node observed the transaction.
+  int64 observed_at_ms = 6;
+}
diff --git a/chain_capabilities/evm/txevents/emitter.go b/chain_capabilities/evm/txevents/emitter.go
new file mode 100644
index 000000000..78026edf2
--- /dev/null
+++ b/chain_capabilities/evm/txevents/emitter.go
@@ -0,0 +1,209 @@
+package txevents
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"sync/atomic"
+	"time"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
+	"github.com/smartcontractkit/chainlink-common/pkg/durableemitter"
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+)
+
+const (
+	// DefaultDomain is the ChIP source domain (and therefore the Kafka topic) the events are
+	// published to. Placeholder until Data Platform provisions the dedicated domain (PLEX-3409).
+	DefaultDomain = "cre-chain-transactions"
+
+	// Attribute names understood by chainlink-common's durable emitter and chipingress client.
+	attrSource         = "source"
+	attrType           = "type"
+	attrIdempotencyKey = "idempotencykey" // chipingress.IdempotencyKeyAttr
+	attrSubject        = "subject"
+
+	// emitTimeout bounds the synchronous durable-emitter insert so a slow node database cannot
+	// stall the WriteReport that produced the event.
+	emitTimeout = 2 * time.Second
+)
+
+// EntityType returns the CloudEvent type, and the schema full name, of ChainTransactionObserved.
+// It is a function because the generated proto registers itself in an init function that runs
+// after package-level variables are initialised.
+func EntityType() string {
+	return string(proto.MessageName(&ChainTransactionObserved{}))
+}
+
+// ErrEmitterUnavailable is returned when the process has no durable emitter, which is the case
+// when the node runs without CL_CHIP_INGRESS_DURABLE_EMITTER_ENABLED. Events are dropped and
+// counted.
+var ErrEmitterUnavailable = errors.New("durable emitter not available in this process")
+
+// Sink delivers an encoded event with its routing attributes. Production uses GlobalSink; tests
+// substitute a fake.
+type Sink interface {
+	Emit(ctx context.Context, body []byte, attrKVs ...any) error
+}
+
+// SinkFunc adapts a function to Sink.
+type SinkFunc func(ctx context.Context, body []byte, attrKVs ...any) error
+
+// Emit implements Sink.
+func (f SinkFunc) Emit(ctx context.Context, body []byte, attrKVs ...any) error {
+	return f(ctx, body, attrKVs...)
+}
+
+// GlobalSink delivers events through the process-global durable emitter, the same path the
+// platform uses for its own ChIP events (workflow trigger events, relayer config). In a LOOP
+// that emitter is set up by loop.Server when the node runs with
+// CL_CHIP_INGRESS_DURABLE_EMITTER_ENABLED; it persists events in the node database and the host
+// delivers them to ChIP Ingress at least once. The emitter is resolved on every call, so the
+// sink can be built before the LOOP has finished starting. Without a durable emitter it returns
+// ErrEmitterUnavailable.
+func GlobalSink() Sink {
+	return SinkFunc(func(ctx context.Context, body []byte, attrKVs ...any) error {
+		err := durableemitter.GlobalEmit(ctx, body, attrKVs...)
+		if errors.Is(err, durableemitter.ErrNotInitialized) {
+			return ErrEmitterUnavailable
+		}
+		return err
+	})
+}
+
+// Emitter publishes ChainTransactionObserved events.
+type Emitter interface {
+	Emit(ctx context.Context, event *ChainTransactionObserved) error
+}
+
+// NopEmitter discards events. Used when transaction events are disabled.
+type NopEmitter struct{}
+
+// Emit implements Emitter.
+func (NopEmitter) Emit(context.Context, *ChainTransactionObserved) error { return nil }
+
+// Key identifies the transaction independently of the node that observed it. Every node that
+// saw the same transaction produces the same key, so ChIP and the Blockchain Finality Service
+// can dedup on it. It is used both as the CloudEvent subject and as the idempotency key.
+func Key(ev *ChainTransactionObserved) string {
+	return fmt.Sprintf("%d:%s", ev.GetChainSelector(), ev.GetTxHash())
+}
+
+// LogKVs returns the event's identifying fields as logger key/values.
+func LogKVs(ev *ChainTransactionObserved) []any {
+	return []any{
+		"txHash", ev.GetTxHash(),
+		"chainSelector", ev.GetChainSelector(),
+		"txStatus", ev.GetTxStatus().String(),
+		"workflowExecutionID", ev.GetWorkflowExecutionId(),
+		"workflowID", ev.GetWorkflowId(),
+	}
+}
+
+// DurableEmitter encodes events as ChIP CloudEvents and hands them to a Sink, counting the
+// outcome.
+type DurableEmitter struct {
+	lggr    logger.Logger
+	domain  string
+	sink    Sink
+	metrics *emitterMetrics
+
+	unavailableLogged atomic.Bool
+}
+
+// NewDurableEmitter returns an emitter publishing to the given ChIP source domain through sink.
+func NewDurableEmitter(lggr logger.Logger, domain string, sink Sink) (*DurableEmitter, error) {
+	if sink == nil {
+		return nil, errors.New("nil tx events sink")
+	}
+	if domain == "" {
+		domain = DefaultDomain
+	}
+	m, err := newEmitterMetrics()
+	if err != nil {
+		return nil, err
+	}
+	return &DurableEmitter{
+		lggr:    logger.Named(lggr, "TxEvents"),
+		domain:  domain,
+		sink:    sink,
+		metrics: m,
+	}, nil
+}
+
+// Emit implements Emitter. When the sink reports ErrEmitterUnavailable the event is counted as
+// dropped and the condition is logged once per process.
+func (d *DurableEmitter) Emit(ctx context.Context, ev *ChainTransactionObserved) error {
+	if ev == nil {
+		return errors.New("nil ChainTransactionObserved")
+	}
+	body, err := proto.Marshal(ev)
+	if err != nil {
+		d.metrics.recordError(ctx, ev, "marshal")
+		return fmt.Errorf("marshal ChainTransactionObserved: %w", err)
+	}
+
+	emitCtx, cancel := context.WithTimeout(ctx, emitTimeout)
+	defer cancel()
+	key := Key(ev)
+	err = d.sink.Emit(emitCtx, body,
+		attrSource, d.domain,
+		attrType, EntityType(),
+		attrIdempotencyKey, key,
+		attrSubject, key,
+	)
+	switch {
+	case errors.Is(err, ErrEmitterUnavailable):
+		d.metrics.recordError(ctx, ev, "unavailable")
+		if d.unavailableLogged.CompareAndSwap(false, true) {
+			d.lggr.Warnw("Transaction events are enabled but this process has no durable emitter; events are dropped", "domain", d.domain)
+		}
+		return ErrEmitterUnavailable
+	case err != nil:
+		d.metrics.recordError(ctx, ev, "emit")
+		return fmt.Errorf("emit ChainTransactionObserved: %w", err)
+	}
+	d.metrics.recordEmitted(ctx, ev)
+	d.lggr.Debugw("Emitted ChainTransactionObserved", LogKVs(ev)...)
+	return nil
+}
+
+type emitterMetrics struct {
+	emitted metric.Int64Counter
+	errors  metric.Int64Counter
+}
+
+func newEmitterMetrics() (*emitterMetrics, error) {
+	meter := beholder.GetMeter()
+	emitted, err := meter.Int64Counter("chain_capability_tx_events_emitted_count",
+		metric.WithDescription("Count of ChainTransactionObserved events accepted by the durable emitter"))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create tx events emitted counter: %w", err)
+	}
+	errs, err := meter.Int64Counter("chain_capability_tx_events_emit_error_count",
+		metric.WithDescription("Count of ChainTransactionObserved events that could not be emitted, by reason"))
+	if err != nil {
+		return nil, fmt.Errorf("failed to create tx events error counter: %w", err)
+	}
+	return &emitterMetrics{emitted: emitted, errors: errs}, nil
+}
+
+func (m *emitterMetrics) recordEmitted(ctx context.Context, ev *ChainTransactionObserved) {
+	m.emitted.Add(ctx, 1, metric.WithAttributes(eventAttrs(ev)...))
+}
+
+func (m *emitterMetrics) recordError(ctx context.Context, ev *ChainTransactionObserved, reason string) {
+	m.errors.Add(ctx, 1, metric.WithAttributes(append(eventAttrs(ev), attribute.String("reason", reason))...))
+}
+
+func eventAttrs(ev *ChainTransactionObserved) []attribute.KeyValue {
+	return []attribute.KeyValue{
+		attribute.String("chain_selector", strconv.FormatUint(ev.GetChainSelector(), 10)),
+		attribute.String("tx_status", ev.GetTxStatus().String()),
+	}
+}
diff --git a/chain_capabilities/evm/txevents/emitter_test.go b/chain_capabilities/evm/txevents/emitter_test.go
new file mode 100644
index 000000000..d93ab9aa3
--- /dev/null
+++ b/chain_capabilities/evm/txevents/emitter_test.go
@@ -0,0 +1,124 @@
+package txevents
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/durableemitter"
+	"github.com/smartcontractkit/chainlink-common/pkg/logger"
+)
+
+func testEvent() *ChainTransactionObserved {
+	return &ChainTransactionObserved{
+		WorkflowExecutionId: "exec-1",
+		WorkflowId:          "wf-1",
+		ChainSelector:       16015286601757825753,
+		TxHash:              "0xabc",
+		TxStatus:            TxStatus_TX_STATUS_REVERTED,
+		ObservedAtMs:        1,
+	}
+}
+
+func TestEntityType(t *testing.T) {
+	require.Equal(t, "chain_capabilities.txevents.v1.ChainTransactionObserved", EntityType())
+}
+
+func TestKey(t *testing.T) {
+	ev := testEvent()
+	require.Equal(t, "16015286601757825753:0xabc", Key(ev))
+
+	// The key depends only on the transaction, not on who observed it, when, or what it reported.
+	other := proto.Clone(ev).(*ChainTransactionObserved)
+	other.ObservedAtMs = 42
+	other.TxStatus = TxStatus_TX_STATUS_SUCCESS
+	other.WorkflowExecutionId = "exec-2"
+	require.Equal(t, Key(ev), Key(other))
+
+	other.TxHash = "0xdef"
+	require.NotEqual(t, Key(ev), Key(other))
+}
+
+func TestGlobalSink(t *testing.T) {
+	// The test process never sets up a durable emitter, which is exactly the situation of a node
+	// running without CL_CHIP_INGRESS_DURABLE_EMITTER_ENABLED.
+	require.Nil(t, durableemitter.GetGlobalEmitter())
+	err := GlobalSink().Emit(t.Context(), []byte("body"), "source", DefaultDomain)
+	require.ErrorIs(t, err, ErrEmitterUnavailable)
+}
+
+func TestDurableEmitter_Emit(t *testing.T) {
+	newEmitter := func(t *testing.T, emit SinkFunc) *DurableEmitter {
+		t.Helper()
+		d, err := NewDurableEmitter(logger.Test(t), "", emit)
+		require.NoError(t, err)
+		require.Equal(t, DefaultDomain, d.domain)
+		return d
+	}
+
+	t.Run("requires a sink", func(t *testing.T) {
+		_, err := NewDurableEmitter(logger.Test(t), "", nil)
+		require.Error(t, err)
+	})
+
+	t.Run("passes source, type, keys and the marshalled body", func(t *testing.T) {
+		var gotBody []byte
+		var gotAttrs []any
+		d := newEmitter(t, func(_ context.Context, body []byte, attrKVs ...any) error {
+			gotBody, gotAttrs = body, attrKVs
+			return nil
+		})
+		ev := testEvent()
+		require.NoError(t, d.Emit(t.Context(), ev))
+
+		var decoded ChainTransactionObserved
+		require.NoError(t, proto.Unmarshal(gotBody, &decoded))
+		require.True(t, proto.Equal(ev, &decoded))
+		require.Equal(t, []any{
+			"source", DefaultDomain,
+			"type", EntityType(),
+			"idempotencykey", Key(ev),
+			"subject", Key(ev),
+		}, gotAttrs)
+	})
+
+	t.Run("bounds the emit context", func(t *testing.T) {
+		d := newEmitter(t, func(ctx context.Context, _ []byte, _ ...any) error {
+			_, ok := ctx.Deadline()
+			require.True(t, ok, "emit context must carry a deadline")
+			return nil
+		})
+		require.NoError(t, d.Emit(t.Context(), testEvent()))
+	})
+
+	t.Run("reports ErrEmitterUnavailable when the process has no durable emitter", func(t *testing.T) {
+		d, err := NewDurableEmitter(logger.Test(t), "custom-domain", GlobalSink())
+		require.NoError(t, err)
+		require.Equal(t, "custom-domain", d.domain)
+		require.ErrorIs(t, d.Emit(t.Context(), testEvent()), ErrEmitterUnavailable)
+		require.ErrorIs(t, d.Emit(t.Context(), testEvent()), ErrEmitterUnavailable)
+		require.True(t, d.unavailableLogged.Load())
+	})
+
+	t.Run("wraps other emit errors", func(t *testing.T) {
+		cause := errors.New("db down")
+		d := newEmitter(t, func(context.Context, []byte, ...any) error { return cause })
+		err := d.Emit(t.Context(), testEvent())
+		require.ErrorIs(t, err, cause)
+		require.NotErrorIs(t, err, ErrEmitterUnavailable)
+		require.False(t, d.unavailableLogged.Load())
+	})
+
+	t.Run("rejects a nil event", func(t *testing.T) {
+		d := newEmitter(t, func(context.Context, []byte, ...any) error { return nil })
+		require.Error(t, d.Emit(t.Context(), nil))
+	})
+}
+
+func TestNopEmitter(t *testing.T) {
+	require.NoError(t, NopEmitter{}.Emit(t.Context(), testEvent()))
+	require.NoError(t, NopEmitter{}.Emit(t.Context(), nil))
+}
diff --git a/chain_capabilities/evm/txevents/generate.go b/chain_capabilities/evm/txevents/generate.go
new file mode 100644
index 000000000..927c957eb
--- /dev/null
+++ b/chain_capabilities/evm/txevents/generate.go
@@ -0,0 +1,3 @@
+package txevents
+
+//go:generate protoc --proto_path=../../../ --go_out=paths=source_relative:../../../ chain_capabilities/evm/txevents/chain_transaction_observed.proto