diff --git a/cmd/proxy/server.go b/cmd/proxy/server.go
index a8a1320..fb8d384 100644
--- a/cmd/proxy/server.go
+++ b/cmd/proxy/server.go
@@ -128,12 +128,16 @@ func (s *server) addr() string {
 // relays — Accept blocks on a semaphore so the kernel queue absorbs the
 // backpressure instead of the process opening unbounded sessions.
 func (s *server) serve() error {
+	s.mu.Lock()
+	ln := s.listener
+	s.mu.Unlock()
+
 	var sem chan struct{}
 	if s.cfg.MaxConnections > 0 {
 		sem = make(chan struct{}, s.cfg.MaxConnections)
 	}
 	for {
-		conn, err := s.listener.Accept()
+		conn, err := ln.Accept()
 		if err != nil {
 			if errors.Is(err, net.ErrClosed) {
 				return nil
@@ -147,7 +151,22 @@ func (s *server) serve() error {
 		if s.cfg.HandshakeTimeout > 0 {
 			_ = conn.SetDeadline(time.Now().Add(s.cfg.HandshakeTimeout))
 		}
-		s.wg.Add(1)
+		// Register the connection under the lock close() takes before
+		// waiting, so a conn accepted concurrently with close either
+		// increments the WaitGroup before close waits or is dropped.
+		s.mu.Lock()
+		closed := s.listener == nil
+		if !closed {
+			s.wg.Add(1)
+		}
+		s.mu.Unlock()
+		if closed {
+			_ = conn.Close()
+			if sem != nil {
+				<-sem
+			}
+			return nil
+		}
 		go func() {
 			defer s.wg.Done()
 			defer func() {
@@ -164,6 +183,7 @@ func (s *server) serve() error {
 func (s *server) close() error {
 	s.mu.Lock()
 	ln := s.listener
+	s.listener = nil
 	s.mu.Unlock()
 	if ln == nil {
 		return nil
diff --git a/go.mod b/go.mod
index 8c35b72..189eef9 100644
--- a/go.mod
+++ b/go.mod
@@ -19,6 +19,7 @@ require (
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
 	github.com/kr/text v0.2.0 // indirect
+	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.66.1 // indirect
diff --git a/internal/observe/metrics.go b/internal/observe/metrics.go
index 5f7fae7..d603ff6 100644
--- a/internal/observe/metrics.go
+++ b/internal/observe/metrics.go
@@ -45,11 +45,30 @@ var (
 		Name: "kkp_kms_calls_total",
 		Help: "Calls to the KMS (Encrypt/Decrypt of DEKs), by op and outcome.",
 	}, []string{"op", "outcome"})
+
+	// CiphertextPassthrough counts DataRows that left the proxy still
+	// carrying a ciphertext envelope because no decrypt plan matched, by
+	// reason. Any non-zero value is an incident: the client saw raw $KKP$
+	// blobs (the verify-email failure class).
+	CiphertextPassthrough = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: "kkp_ciphertext_passthrough_total",
+		Help: "DataRows passed through undecrypted despite carrying a ciphertext envelope, by reason.",
+	}, []string{"reason"})
+
+	// DoubleEncrypted counts decrypted values that still carry a ciphertext
+	// envelope: rows whose stored value IS a leaked ciphertext (written back
+	// by the client during a passthrough window and encrypted again). Used
+	// to inventory corrupted rows for cleanup.
+	DoubleEncrypted = prometheus.NewCounterVec(prometheus.CounterOpts{
+		Name: "kkp_double_encrypted_total",
+		Help: "Decrypted values still carrying a ciphertext envelope (double-encrypted rows), by table/column.",
+	}, []string{"table", "column"})
 )
 
 func init() {
 	prometheus.MustRegister(
 		DecryptTotal, DecryptFailures, EncryptTotal,
-		UnrecognizedPIISQL, KMSCallTotal,
+		UnrecognizedPIISQL, KMSCallTotal, CiphertextPassthrough,
+		DoubleEncrypted,
 	)
 }
diff --git a/internal/rewrite/analyze.go b/internal/rewrite/analyze.go
index 3cf4582..21e2be9 100644
--- a/internal/rewrite/analyze.go
+++ b/internal/rewrite/analyze.go
@@ -49,6 +49,10 @@ type ColumnParam struct {
 // to the filtered columns on the read/search path.
 type Analysis struct {
 	Kind StmtKind
+	// SQL is this statement's own text within a multi-statement
+	// simple-protocol query string. Set by AnalyzeAll only; empty for
+	// Analyze, whose callers already hold the (single) statement text.
+	SQL string
 	// Table is the target table (INSERT/UPDATE/DELETE) or the first FROM table
 	// (SELECT). Empty if no single table applies.
 	Table string
@@ -76,19 +80,56 @@ func Analyze(sql string) (*Analysis, error) {
 	if len(stmts) != 1 {
 		return nil, fmt.Errorf("rewrite: expected exactly one statement, got %d", len(stmts))
 	}
+	return analyzeNode(stmts[0].GetStmt()), nil
+}
+
+// AnalyzeAll parses a simple-protocol query string, which may carry several
+// semicolon-separated statements, and returns one Analysis per statement in
+// order, each carrying its own statement text. The backend answers each
+// statement with its own result cycle, so the wire layer needs one
+// result-queue entry per statement to stay in sync.
+func AnalyzeAll(sql string) ([]*Analysis, error) {
+	result, err := pg.Parse(sql)
+	if err != nil {
+		return nil, fmt.Errorf("rewrite: parse: %w", err)
+	}
+	stmts := result.GetStmts()
+	analyses := make([]*Analysis, 0, len(stmts))
+	for _, st := range stmts {
+		a := analyzeNode(st.GetStmt())
+		a.SQL = statementText(sql, st)
+		analyses = append(analyses, a)
+	}
+	return analyses, nil
+}
+
+// statementText slices one statement's own text out of a multi-statement
+// query string using the parser's location/length, so per-statement logging
+// and PII heuristics do not see the neighbouring statements.
+func statementText(sql string, st *pg.RawStmt) string {
+	start := int(st.GetStmtLocation())
+	if start < 0 || start > len(sql) {
+		return sql
+	}
+	end := len(sql)
+	if l := int(st.GetStmtLen()); l > 0 && start+l <= len(sql) {
+		end = start + l
+	}
+	return strings.TrimLeft(sql[start:end], " \t\r\n;")
+}
 
-	node := stmts[0].GetStmt()
+func analyzeNode(node *pg.Node) *Analysis {
 	switch {
 	case node.GetInsertStmt() != nil:
-		return analyzeInsert(node.GetInsertStmt()), nil
+		return analyzeInsert(node.GetInsertStmt())
 	case node.GetUpdateStmt() != nil:
-		return analyzeUpdate(node.GetUpdateStmt()), nil
+		return analyzeUpdate(node.GetUpdateStmt())
 	case node.GetSelectStmt() != nil:
-		return analyzeSelect(node.GetSelectStmt()), nil
+		return analyzeSelect(node.GetSelectStmt())
 	case node.GetDeleteStmt() != nil:
-		return analyzeDelete(node.GetDeleteStmt()), nil
+		return analyzeDelete(node.GetDeleteStmt())
 	default:
-		return &Analysis{Kind: KindOther}, nil
+		return &Analysis{Kind: KindOther}
 	}
 }
 
diff --git a/internal/rewrite/analyze_test.go b/internal/rewrite/analyze_test.go
index 3ee383c..e79f4a7 100644
--- a/internal/rewrite/analyze_test.go
+++ b/internal/rewrite/analyze_test.go
@@ -163,3 +163,24 @@ func TestAnalyzeErrors(t *testing.T) {
 		t.Error("Analyze accepted multiple statements")
 	}
 }
+
+func TestAnalyzeAllPerStatementSQL(t *testing.T) {
+	t.Parallel()
+
+	analyses, err := AnalyzeAll("select id from realm; select email from user_entity where id = 'u1'")
+	if err != nil {
+		t.Fatalf("AnalyzeAll: %v", err)
+	}
+	if len(analyses) != 2 {
+		t.Fatalf("statements = %d, want 2", len(analyses))
+	}
+	if analyses[0].SQL != "select id from realm" {
+		t.Errorf("first statement SQL = %q", analyses[0].SQL)
+	}
+	if analyses[1].SQL != "select email from user_entity where id = 'u1'" {
+		t.Errorf("second statement SQL = %q", analyses[1].SQL)
+	}
+	if analyses[0].Table != "realm" || analyses[1].Table != "user_entity" {
+		t.Errorf("tables = %q, %q", analyses[0].Table, analyses[1].Table)
+	}
+}
diff --git a/internal/wire/pgjdbc_flow_test.go b/internal/wire/pgjdbc_flow_test.go
new file mode 100644
index 0000000..c9c21d3
--- /dev/null
+++ b/internal/wire/pgjdbc_flow_test.go
@@ -0,0 +1,435 @@
+package wire
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/jackc/pgproto3/v2"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+
+	"github.com/cozystack/keycloak-kms-proxy/internal/observe"
+	"github.com/cozystack/keycloak-kms-proxy/internal/rewrite"
+)
+
+// Regression tests for the read-path flows behind the stage verify-email
+// incident: pgjdbc describes a server-prepared statement once (Describe 'S')
+// and afterwards executes it with bare Bind/Execute cycles that carry no
+// Describe at all, so no RowDescription flows during those Executes; and
+// simple-protocol queries share the connection with extended traffic, where
+// an untracked CommandComplete desyncs the result queue. Both paths used to
+// pass raw $KKP$ ciphertext through to Keycloak.
+
+// keycloakUserSelect mirrors the Hibernate SQL captured live on the stage
+// incident (statement S_22).
+const keycloakUserSelect = "select ue1_0.ID,ue1_0.EMAIL,ue1_0.USERNAME from USER_ENTITY ue1_0 where ue1_0.USERNAME=$1 and ue1_0.REALM_ID=$2"
+
+func userSelectRowDescription() *pgproto3.RowDescription {
+	return &pgproto3.RowDescription{Fields: []pgproto3.FieldDescription{
+		{Name: []byte("id")}, {Name: []byte("email")}, {Name: []byte("username")},
+	}}
+}
+
+// storedEmail produces the backend-stored ciphertext for an email value.
+func storedEmail(t *testing.T, s *Session, plaintext string) []byte {
+	t.Helper()
+	stored, err := s.cipher.Encrypt(0 /*deterministic*/, []byte(plaintext), rewrite.AAD("USER_ENTITY", "EMAIL"))
+	if err != nil {
+		t.Fatalf("Encrypt: %v", err)
+	}
+	return []byte(stored)
+}
+
+// TestDescribeStatementFlowDecrypts covers pgjdbc's first execution of a
+// server-prepared statement: Parse + Describe('S') + Bind + Execute in one
+// batch. The RowDescription answers the statement Describe — not a portal
+// Describe — and the DataRow must still decrypt.
+func TestDescribeStatementFlowDecrypts(t *testing.T) {
+	t.Parallel()
+
+	s := newEncryptingSession(t)
+	if err := s.OnParse(&pgproto3.Parse{Name: "S_1", Query: keycloakUserSelect}); err != nil {
+		t.Fatalf("OnParse: %v", err)
+	}
+	s.OnDescribe(&pgproto3.Describe{ObjectType: 'S', Name: "S_1"})
+	s.OnBind(&pgproto3.Bind{DestinationPortal: "C_1", PreparedStatement: "S_1"})
+	s.OnExecute(&pgproto3.Execute{Portal: "C_1"})
+	s.OnSync()
+
+	// Backend answers: ParseComplete, ParameterDescription, RowDescription
+	// (for the Describe), then the Execute's rows.
+	s.OnRowDescription(userSelectRowDescription())
+	dr := &pgproto3.DataRow{Values: [][]byte{[]byte("u1"), storedEmail(t, s, "dora@example.com"), []byte("dora")}}
+	if err := s.DecryptDataRow(dr); err != nil {
+		t.Fatalf("DecryptDataRow: %v", err)
+	}
+	if string(dr.Values[1]) != "dora@example.com" {
+		t.Fatalf("email not decrypted on describe-statement flow: %q", dr.Values[1])
+	}
+	s.OnCommandComplete()
+	s.OnReadyForQuery()
+}
+
+// TestServerPreparedReuseDecrypts is the verify-email regression: once pgjdbc
+// has the statement's metadata it re-executes with Bind + Execute only — no
+// Describe, no RowDescription. The decrypt plan must come from the columns
+// cached at the initial Describe.
+func TestServerPreparedReuseDecrypts(t *testing.T) {
+	t.Parallel()
+
+	s := newEncryptingSession(t)
+	if err := s.OnParse(&pgproto3.Parse{Name: "S_1", Query: keycloakUserSelect}); err != nil {
+		t.Fatalf("OnParse: %v", err)
+	}
+	// First batch: describe-statement flow, complete cycle.
+	s.OnDescribe(&pgproto3.Describe{ObjectType: 'S', Name: "S_1"})
+	s.OnBind(&pgproto3.Bind{DestinationPortal: "C_1", PreparedStatement: "S_1"})
+	s.OnExecute(&pgproto3.Execute{Portal: "C_1"})
+	s.OnSync()
+	s.OnRowDescription(userSelectRowDescription())
+	s.OnCommandComplete()
+	s.OnReadyForQuery()
+
+	// Warm reuse: bare Bind + Execute. The backend sends BindComplete,
+	// DataRow, CommandComplete — no RowDescription at all.
+	s.OnBind(&pgproto3.Bind{DestinationPortal: "C_2", PreparedStatement: "S_1"})
+	s.OnExecute(&pgproto3.Execute{Portal: "C_2"})
+	s.OnSync()
+	dr := &pgproto3.DataRow{Values: [][]byte{[]byte("u1"), storedEmail(t, s, "erin@example.com"), []byte("erin")}}
+	if err := s.DecryptDataRow(dr); err != nil {
+		t.Fatalf("DecryptDataRow: %v", err)
+	}
+	if string(dr.Values[1]) != "erin@example.com" {
+		t.Fatalf("email not decrypted on server-prepared reuse: %q", dr.Values[1])
+	}
+}
+
+// TestSimpleQuerySelectDecrypts covers the simple protocol ('Q'): its
+// RowDescription arrives with no Describe outstanding and its rows must
+// decrypt through the same read-plan path.
+func TestSimpleQuerySelectDecrypts(t *testing.T) {
+	t.Parallel()
+
+	s := newEncryptingSession(t)
+	if err := s.OnQuery(&pgproto3.Query{String: "select id, email from user_entity where id = 'u1'"}); err != nil {
+		t.Fatalf("OnQuery: %v", err)
+	}
+	s.OnRowDescription(&pgproto3.RowDescription{Fields: []pgproto3.FieldDescription{
+		{Name: []byte("id")}, {Name: []byte("email")},
+	}})
+	dr := &pgproto3.DataRow{Values: [][]byte{[]byte("u1"), storedEmail(t, s, "faye@example.com")}}
+	if err := s.DecryptDataRow(dr); err != nil {
+		t.Fatalf("DecryptDataRow: %v", err)
+	}
+	if string(dr.Values[1]) != "faye@example.com" {
+		t.Fatalf("email not decrypted on simple-query flow: %q", dr.Values[1])
+	}
+	s.OnCommandComplete()
+	s.OnReadyForQuery()
+}
+
+// TestSimpleQueryKeepsResultQueueInSync: an untracked simple query's
+// CommandComplete used to pop a pending extended-protocol portal, shifting
+// every later result onto the wrong portal (the "NO executing portal"
+// desyncs observed on stage).
+func TestSimpleQueryKeepsResultQueueInSync(t *testing.T) {
+	t.Parallel()
+
+	s := newEncryptingSession(t)
+	if err := s.OnQuery(&pgproto3.Query{String: "COMMIT"}); err != nil {
+		t.Fatalf("OnQuery: %v", err)
+	}
+	if err := s.OnParse(&pgproto3.Parse{Name: "S_1", Query: keycloakUserSelect}); err != nil {
+		t.Fatalf("OnParse: %v", err)
+	}
+	s.OnDescribe(&pgproto3.Describe{ObjectType: 'S', Name: "S_1"})
+	s.OnBind(&pgproto3.Bind{DestinationPortal: "C_1", PreparedStatement: "S_1"})
+	s.OnExecute(&pgproto3.Execute{Portal: "C_1"})
+	s.OnSync()
+
+	// Backend: COMMIT's CommandComplete must pop the synthetic simple-query
+	// entry (and its ReadyForQuery the implicit boundary), leaving the select
+	// portal to receive its own results.
+	s.OnCommandComplete()
+	s.OnReadyForQuery()
+	s.OnRowDescription(userSelectRowDescription())
+	dr := &pgproto3.DataRow{Values: [][]byte{[]byte("u1"), storedEmail(t, s, "gino@example.com"), []byte("gino")}}
+	if err := s.DecryptDataRow(dr); err != nil {
+		t.Fatalf("DecryptDataRow: %v", err)
+	}
+	if string(dr.Values[1]) != "gino@example.com" {
+		t.Fatalf("email not decrypted after interleaved simple query: %q", dr.Values[1])
+	}
+}
+
+// TestSimpleQueryMultiStatement: each statement in a multi-statement simple
+// query gets its own result cycle, so each needs its own queue entry.
+func TestSimpleQueryMultiStatement(t *testing.T) {
+	t.Parallel()
+
+	s := newEncryptingSession(t)
+	err := s.OnQuery(&pgproto3.Query{String: "select id from realm; select id, email from user_entity where id = 'u1'"})
+	if err != nil {
+		t.Fatalf("OnQuery: %v", err)
+	}
+	// First statement's cycle: realm rows pass through.
+	s.OnRowDescription(&pgproto3.RowDescription{Fields: []pgproto3.FieldDescription{{Name: []byte("id")}}})
+	if err := s.DecryptDataRow(&pgproto3.DataRow{Values: [][]byte{[]byte("r1")}}); err != nil {
+		t.Fatalf("DecryptDataRow(realm): %v", err)
+	}
+	s.OnCommandComplete()
+	// Second statement's cycle: user rows decrypt.
+	s.OnRowDescription(&pgproto3.RowDescription{Fields: []pgproto3.FieldDescription{
+		{Name: []byte("id")}, {Name: []byte("email")},
+	}})
+	dr := &pgproto3.DataRow{Values: [][]byte{[]byte("u1"), storedEmail(t, s, "hana@example.com")}}
+	if err := s.DecryptDataRow(dr); err != nil {
+		t.Fatalf("DecryptDataRow(user): %v", err)
+	}
+	if string(dr.Values[1]) != "hana@example.com" {
+		t.Fatalf("email not decrypted on second simple statement: %q", dr.Values[1])
+	}
+}
+
+// TestSimpleQueryLiteralPIIWriteFailsLoud: a simple-protocol write of a PII
+// column carries the value as a literal the proxy cannot encrypt; it must be
+// refused, mirroring OnParse.
+func TestSimpleQueryLiteralPIIWriteFailsLoud(t *testing.T) {
+	t.Parallel()
+
+	s := newEncryptingSession(t)
+	err := s.OnQuery(&pgproto3.Query{String: "insert into user_entity (id, email) values ('u1', 'raw@example.com')"})
+	if err == nil || !strings.Contains(err.Error(), "not a bound parameter") {
+		t.Fatalf("OnQuery err=%v, want unencryptable-PII failure", err)
+	}
+}
+
+// TestNoDataConsumesPendingDescribe: a Describe of a row-less statement is
+// answered with NoData; it must consume its queue slot so the next
+// RowDescription lands on the right statement.
+func TestNoDataConsumesPendingDescribe(t *testing.T) {
+	t.Parallel()
+
+	s := newEncryptingSession(t)
+	if err := s.OnParse(&pgproto3.Parse{Name: "S_up", Query: "update user_entity set email = $1 where id = $2"}); err != nil {
+		t.Fatalf("OnParse(update): %v", err)
+	}
+	if err := s.OnParse(&pgproto3.Parse{Name: "S_sel", Query: keycloakUserSelect}); err != nil {
+		t.Fatalf("OnParse(select): %v", err)
+	}
+	s.OnDescribe(&pgproto3.Describe{ObjectType: 'S', Name: "S_up"})
+	s.OnDescribe(&pgproto3.Describe{ObjectType: 'S', Name: "S_sel"})
+
+	s.OnNoData()                                   // answers S_up
+	s.OnRowDescription(userSelectRowDescription()) // answers S_sel
+
+	up, _ := s.Statement("S_up")
+	sel, _ := s.Statement("S_sel")
+	if len(up.Columns) != 0 {
+		t.Fatalf("update statement got columns: %v", up.Columns)
+	}
+	if len(sel.Columns) != 3 {
+		t.Fatalf("select statement columns = %v, want 3", sel.Columns)
+	}
+}
+
+// TestPortalSuspendedAdvancesQueue: a row-limited Execute ends in
+// PortalSuspended; the resumed Execute re-enqueues the portal and its rows
+// keep decrypting.
+func TestPortalSuspendedAdvancesQueue(t *testing.T) {
+	t.Parallel()
+
+	s := newEncryptingSession(t)
+	if err := s.OnParse(&pgproto3.Parse{Name: "S_1", Query: keycloakUserSelect}); err != nil {
+		t.Fatalf("OnParse: %v", err)
+	}
+	s.OnDescribe(&pgproto3.Describe{ObjectType: 'S', Name: "S_1"})
+	s.OnBind(&pgproto3.Bind{DestinationPortal: "C_1", PreparedStatement: "S_1"})
+	s.OnExecute(&pgproto3.Execute{Portal: "C_1", MaxRows: 1})
+	s.OnRowDescription(userSelectRowDescription())
+
+	dr := &pgproto3.DataRow{Values: [][]byte{[]byte("u1"), storedEmail(t, s, "ivy@example.com"), []byte("ivy")}}
+	if err := s.DecryptDataRow(dr); err != nil {
+		t.Fatalf("DecryptDataRow(first): %v", err)
+	}
+	s.OnPortalSuspended()
+
+	s.OnExecute(&pgproto3.Execute{Portal: "C_1", MaxRows: 1})
+	dr2 := &pgproto3.DataRow{Values: [][]byte{[]byte("u2"), storedEmail(t, s, "june@example.com"), []byte("june")}}
+	if err := s.DecryptDataRow(dr2); err != nil {
+		t.Fatalf("DecryptDataRow(resumed): %v", err)
+	}
+	if string(dr2.Values[1]) != "june@example.com" {
+		t.Fatalf("email not decrypted after portal resume: %q", dr2.Values[1])
+	}
+}
+
+// TestErrorResponseClearsPipeline: after an error the backend skips to Sync,
+// so nothing still queued will produce results; keeping entries would shift
+// the next batch's rows onto the wrong portals.
+func TestErrorResponseClearsPipeline(t *testing.T) {
+	t.Parallel()
+
+	s := newEncryptingSession(t)
+	if err := s.OnParse(&pgproto3.Parse{Name: "S_1", Query: keycloakUserSelect}); err != nil {
+		t.Fatalf("OnParse: %v", err)
+	}
+	s.OnDescribe(&pgproto3.Describe{ObjectType: 'S', Name: "S_1"})
+	s.OnBind(&pgproto3.Bind{DestinationPortal: "C_1", PreparedStatement: "S_1"})
+	s.OnExecute(&pgproto3.Execute{Portal: "C_1"})
+	s.OnBind(&pgproto3.Bind{DestinationPortal: "C_2", PreparedStatement: "S_1"})
+	s.OnExecute(&pgproto3.Execute{Portal: "C_2"})
+
+	s.OnErrorResponse()
+	if s.CurrentExecuting() != nil {
+		t.Fatal("exec queue not cleared by ErrorResponse")
+	}
+	s.OnReadyForQuery()
+
+	// A fresh cycle decrypts normally.
+	s.OnBind(&pgproto3.Bind{DestinationPortal: "C_3", PreparedStatement: "S_1"})
+	s.OnExecute(&pgproto3.Execute{Portal: "C_3"})
+	s.OnRowDescription(userSelectRowDescription())
+	dr := &pgproto3.DataRow{Values: [][]byte{[]byte("u1"), storedEmail(t, s, "kate@example.com"), []byte("kate")}}
+	if err := s.DecryptDataRow(dr); err != nil {
+		t.Fatalf("DecryptDataRow: %v", err)
+	}
+	if string(dr.Values[1]) != "kate@example.com" {
+		t.Fatalf("email not decrypted after pipeline reset: %q", dr.Values[1])
+	}
+}
+
+// TestPIIWarnIdentifierBoundary: RESET_CREDENTIALS_FLOW must not count as a
+// mention of the CREDENTIAL table (the stage false-positive WARN), while a
+// real USER_ENTITY select with a PII column must.
+func TestPIIWarnIdentifierBoundary(t *testing.T) {
+	t.Parallel()
+
+	realmSQL := "select re1_0.ID, re1_0.RESET_CREDENTIALS_FLOW, a1_0.VALUE from REALM re1_0 left join REALM_ATTRIBUTE a1_0 on re1_0.ID=a1_0.REALM_ID"
+	if piiTouchedButNotPlanned(realmSQL, []string{"id", "reset_credentials_flow", "value"}) {
+		t.Fatal("REALM select flagged as PII-touching via RESET_CREDENTIALS_FLOW substring")
+	}
+	if !piiTouchedButNotPlanned(keycloakUserSelect, []string{"id", "email", "username"}) {
+		t.Fatal("USER_ENTITY select not flagged as PII-touching")
+	}
+}
+
+// TestPipelinedBatchAcrossReadyForQuery: clients pipeline the next batch
+// before consuming the previous batch's ReadyForQuery. The boundary handling
+// must only close the finished batch — clearing the queues wholesale here
+// dropped the live entries and leaked ciphertext (observed on stage as
+// "ready-for-query with entries still pending").
+func TestPipelinedBatchAcrossReadyForQuery(t *testing.T) {
+	t.Parallel()
+
+	s := newEncryptingSession(t)
+	if err := s.OnParse(&pgproto3.Parse{Name: "S_1", Query: keycloakUserSelect}); err != nil {
+		t.Fatalf("OnParse: %v", err)
+	}
+	// Batch 1 and batch 2 both sent before any backend answer arrives.
+	s.OnDescribe(&pgproto3.Describe{ObjectType: 'S', Name: "S_1"})
+	s.OnBind(&pgproto3.Bind{DestinationPortal: "C_1", PreparedStatement: "S_1"})
+	s.OnExecute(&pgproto3.Execute{Portal: "C_1"})
+	s.OnSync()
+	s.OnBind(&pgproto3.Bind{DestinationPortal: "C_2", PreparedStatement: "S_1"})
+	s.OnExecute(&pgproto3.Execute{Portal: "C_2"})
+	s.OnSync()
+
+	// Backend answers batch 1.
+	s.OnRowDescription(userSelectRowDescription())
+	dr1 := &pgproto3.DataRow{Values: [][]byte{[]byte("u1"), storedEmail(t, s, "lena@example.com"), []byte("lena")}}
+	if err := s.DecryptDataRow(dr1); err != nil {
+		t.Fatalf("DecryptDataRow(batch1): %v", err)
+	}
+	s.OnCommandComplete()
+	s.OnReadyForQuery()
+
+	// Backend answers batch 2 — bare rows, no RowDescription.
+	dr2 := &pgproto3.DataRow{Values: [][]byte{[]byte("u2"), storedEmail(t, s, "mira@example.com"), []byte("mira")}}
+	if err := s.DecryptDataRow(dr2); err != nil {
+		t.Fatalf("DecryptDataRow(batch2): %v", err)
+	}
+	if string(dr2.Values[1]) != "mira@example.com" {
+		t.Fatalf("email not decrypted in pipelined batch: %q", dr2.Values[1])
+	}
+	s.OnCommandComplete()
+	s.OnReadyForQuery()
+}
+
+// TestErrorInBatchKeepsPipelinedBatch: an error aborts only the failing
+// batch (the backend skips to Sync); a batch pipelined behind it must keep
+// its entries and still decrypt.
+func TestErrorInBatchKeepsPipelinedBatch(t *testing.T) {
+	t.Parallel()
+
+	s := newEncryptingSession(t)
+	// Batch 1: a passthrough statement that will fail on the backend.
+	if err := s.OnParse(&pgproto3.Parse{Name: "S_bad", Query: "SELECT broken FROM nowhere"}); err != nil {
+		t.Fatalf("OnParse(bad): %v", err)
+	}
+	s.OnDescribe(&pgproto3.Describe{ObjectType: 'S', Name: "S_bad"})
+	s.OnBind(&pgproto3.Bind{DestinationPortal: "C_1", PreparedStatement: "S_bad"})
+	s.OnExecute(&pgproto3.Execute{Portal: "C_1"})
+	s.OnSync()
+	// Batch 2 pipelined behind it: the user select.
+	if err := s.OnParse(&pgproto3.Parse{Name: "S_1", Query: keycloakUserSelect}); err != nil {
+		t.Fatalf("OnParse: %v", err)
+	}
+	s.OnDescribe(&pgproto3.Describe{ObjectType: 'S', Name: "S_1"})
+	s.OnBind(&pgproto3.Bind{DestinationPortal: "C_2", PreparedStatement: "S_1"})
+	s.OnExecute(&pgproto3.Execute{Portal: "C_2"})
+	s.OnSync()
+
+	// Backend: batch 1 errors, its ReadyForQuery closes the batch.
+	s.OnErrorResponse()
+	s.OnReadyForQuery()
+
+	// Batch 2 proceeds normally and must decrypt.
+	s.OnRowDescription(userSelectRowDescription())
+	dr := &pgproto3.DataRow{Values: [][]byte{[]byte("u1"), storedEmail(t, s, "nora@example.com"), []byte("nora")}}
+	if err := s.DecryptDataRow(dr); err != nil {
+		t.Fatalf("DecryptDataRow: %v", err)
+	}
+	if string(dr.Values[1]) != "nora@example.com" {
+		t.Fatalf("email not decrypted in batch after error: %q", dr.Values[1])
+	}
+	s.OnCommandComplete()
+	s.OnReadyForQuery()
+}
+
+// TestDecryptDataRowFlagsDoubleEncrypted: a stored value that decrypts into
+// another envelope is a corrupted row — ciphertext leaked during a
+// passthrough window and written back as the value. The proxy decrypts one
+// layer, returns the inner envelope, and inventories the row via the
+// kkp_double_encrypted_total counter.
+func TestDecryptDataRowFlagsDoubleEncrypted(t *testing.T) {
+	t.Parallel()
+
+	s := newEncryptingSession(t)
+	if err := s.OnParse(&pgproto3.Parse{Name: "S_1", Query: keycloakUserSelect}); err != nil {
+		t.Fatalf("OnParse: %v", err)
+	}
+	s.OnDescribe(&pgproto3.Describe{ObjectType: 'S', Name: "S_1"})
+	s.OnBind(&pgproto3.Bind{DestinationPortal: "C_1", PreparedStatement: "S_1"})
+	s.OnExecute(&pgproto3.Execute{Portal: "C_1"})
+	s.OnSync()
+	s.OnRowDescription(userSelectRowDescription())
+
+	inner := string(storedEmail(t, s, "olga@example.com"))
+	outer, err := s.cipher.Encrypt(0 /*deterministic*/, []byte(inner), rewrite.AAD("USER_ENTITY", "EMAIL"))
+	if err != nil {
+		t.Fatalf("Encrypt(outer): %v", err)
+	}
+
+	before := testutil.ToFloat64(observe.DoubleEncrypted.WithLabelValues("USER_ENTITY", "EMAIL"))
+	dr := &pgproto3.DataRow{Values: [][]byte{[]byte("u1"), []byte(outer), []byte("olga")}}
+	if err := s.DecryptDataRow(dr); err != nil {
+		t.Fatalf("DecryptDataRow: %v", err)
+	}
+	if string(dr.Values[1]) != inner {
+		t.Fatalf("double-encrypted value = %q, want one decryption layer removed (%q)", dr.Values[1], inner)
+	}
+	after := testutil.ToFloat64(observe.DoubleEncrypted.WithLabelValues("USER_ENTITY", "EMAIL"))
+	if after != before+1 {
+		t.Fatalf("kkp_double_encrypted_total = %v, want %v", after, before+1)
+	}
+}
diff --git a/internal/wire/relay.go b/internal/wire/relay.go
index d9e749f..1544b0e 100644
--- a/internal/wire/relay.go
+++ b/internal/wire/relay.go
@@ -65,8 +65,14 @@ func (s *Session) observeFrontend(msg pgproto3.FrontendMessage) error {
 	case *pgproto3.Bind:
 		s.OnBind(m)
 		return s.EncryptBind(m)
+	case *pgproto3.Describe:
+		s.OnDescribe(m)
 	case *pgproto3.Execute:
 		s.OnExecute(m)
+	case *pgproto3.Sync:
+		s.OnSync()
+	case *pgproto3.Query:
+		return s.OnQuery(m)
 	case *pgproto3.Close:
 		s.OnClose(m)
 	}
@@ -83,14 +89,20 @@ func (s *Session) observeBackend(msg pgproto3.BackendMessage) error {
 	switch m := msg.(type) {
 	case *pgproto3.RowDescription:
 		s.OnRowDescription(m)
+	case *pgproto3.NoData:
+		s.OnNoData()
 	case *pgproto3.DataRow:
 		return s.DecryptDataRow(m)
 	case *pgproto3.CommandComplete:
 		s.OnCommandComplete()
+	case *pgproto3.PortalSuspended:
+		s.OnPortalSuspended()
 	case *pgproto3.EmptyQueryResponse:
 		s.OnEmptyQueryResponse()
 	case *pgproto3.ErrorResponse:
 		s.OnErrorResponse()
+	case *pgproto3.ReadyForQuery:
+		s.OnReadyForQuery()
 	}
 	return nil
 }
diff --git a/internal/wire/session.go b/internal/wire/session.go
index d6bd741..95df02f 100644
--- a/internal/wire/session.go
+++ b/internal/wire/session.go
@@ -25,10 +25,12 @@ func debugf(format string, args ...any) {
 	}
 }
 
-func truncate(s string, n int) string {
+const truncateLimit = 4000
+
+func truncate(s string) string {
 	s = strings.Join(strings.Fields(s), " ")
-	if len(s) > n {
-		return s[:n] + "..."
+	if len(s) > truncateLimit {
+		return s[:truncateLimit] + "..."
 	}
 	return s
 }
@@ -41,6 +43,12 @@ type PreparedStatement struct {
 	SQL       string
 	Analysis  *rewrite.Analysis
 	WritePlan *rewrite.WritePlan
+	// Columns caches the result-set column names learned from the backend's
+	// RowDescription (statement or portal Describe). pgjdbc describes a
+	// statement once and afterwards binds/executes it with no further
+	// Describe, so no RowDescription flows on those executions; the read plan
+	// must be derivable from this cache or ciphertext leaks to the client.
+	Columns []string
 }
 
 // Portal is a bound instance of a prepared statement (a Bind destination). It
@@ -65,6 +73,7 @@ type Session struct {
 	statements map[string]*PreparedStatement
 	portals    map[string]*Portal
 	execQueue  []*Portal
+	describes  []pendingDescribe
 
 	// mu serializes state access between the two relay pump goroutines (the
 	// frontend and backend directions). The pump takes it via observeFrontend /
@@ -73,6 +82,22 @@ type Session struct {
 	mu sync.Mutex
 }
 
+// pendingDescribe is a frontend Describe awaiting its backend answer
+// (RowDescription or NoData). The backend answers Describes strictly in
+// request order, so a FIFO attributes each answer to the described object.
+// A boundary entry marks a Sync boundary instead of a Describe.
+type pendingDescribe struct {
+	stmt     *PreparedStatement
+	portal   *Portal
+	boundary bool
+}
+
+// syncBoundary marks a Sync boundary in the execute queue. Clients may
+// pipeline the next batch before consuming the previous batch's
+// ReadyForQuery, so error/RFQ handling must only affect entries of the
+// batch the backend is actually answering — everything up to the boundary.
+var syncBoundary = &Portal{Name: "(sync)"}
+
 // NewSession creates a Session that plans against the given planner and applies
 // the cipher on Bind/DataRow. The cipher may be nil for state-only use (no
 // transformation); transforming a non-passthrough statement then errors.
@@ -97,10 +122,10 @@ func (s *Session) OnParse(p *pgproto3.Parse) error {
 	if a, err := rewrite.Analyze(p.Query); err == nil {
 		plan, perr := s.planner.PlanWrite(a)
 		if perr != nil {
-			debugf("kkp: parse %q FAIL plan: kind=%v table=%q err=%v sql=%q", p.Name, a.Kind, a.Table, perr, truncate(p.Query, 4000))
+			debugf("kkp: parse %q FAIL plan: kind=%v table=%q err=%v sql=%q", p.Name, a.Kind, a.Table, perr, truncate(p.Query))
 			return fmt.Errorf("wire: prepare %q: %w", p.Name, perr)
 		}
-		debugf("kkp: parse %q kind=%v table=%q writeParams=%d sql=%q", p.Name, a.Kind, a.Table, len(plan.Params), truncate(p.Query, 4000))
+		debugf("kkp: parse %q kind=%v table=%q writeParams=%d sql=%q", p.Name, a.Kind, a.Table, len(plan.Params), truncate(p.Query))
 		ps.Analysis = a
 		ps.WritePlan = plan
 
@@ -152,6 +177,74 @@ func (s *Session) OnExecute(e *pgproto3.Execute) {
 	}
 }
 
+// OnDescribe records a frontend Describe so its backend answer (RowDescription
+// or NoData) can be attributed to the described statement or portal. An
+// unknown name still occupies a queue slot to keep the FIFO aligned with the
+// backend's answers.
+func (s *Session) OnDescribe(d *pgproto3.Describe) {
+	pd := pendingDescribe{}
+	switch d.ObjectType {
+	case 'S':
+		pd.stmt = s.statements[d.Name]
+	case 'P':
+		if p, ok := s.portals[d.Name]; ok {
+			pd.portal = p
+			pd.stmt = p.Stmt
+		}
+	}
+	s.describes = append(s.describes, pd)
+}
+
+// simplePortalName labels synthetic result-queue entries for simple-protocol
+// queries in debug logs.
+const simplePortalName = "(simple)"
+
+// OnQuery tracks a simple-protocol query ('Q'). The backend answers each
+// statement in the string with its own result cycle (RowDescription, DataRows,
+// CommandComplete), so each statement enqueues one synthetic executing entry —
+// without them a simple query's CommandComplete pops a pending
+// extended-protocol portal and desyncs every later result on the connection.
+// Statements are planned like OnParse: a literal PII write is refused
+// (fail-loud) instead of stored as plaintext, and SELECT results decrypt
+// through the same read-plan path once the RowDescription arrives. A simple
+// query implicitly syncs (its response ends with ReadyForQuery), so it also
+// pushes the Sync boundaries.
+func (s *Session) OnQuery(q *pgproto3.Query) error {
+	analyses, err := rewrite.AnalyzeAll(q.String)
+	switch {
+	case err != nil || len(analyses) == 0:
+		// Unparsable (dialect edge) or an empty query string: the backend
+		// answers with a single result cycle (or EmptyQueryResponse).
+		debugf("kkp: query passthrough (stmts=%d, err=%v) sql=%q", len(analyses), err, truncate(q.String))
+		s.execQueue = append(s.execQueue, &Portal{Name: simplePortalName, Stmt: &PreparedStatement{SQL: q.String}})
+	default:
+		for _, a := range analyses {
+			plan, perr := s.planner.PlanWrite(a)
+			if perr != nil {
+				debugf("kkp: query FAIL plan: kind=%v table=%q err=%v sql=%q", a.Kind, a.Table, perr, truncate(a.SQL))
+				return fmt.Errorf("wire: simple query: %w", perr)
+			}
+			debugf("kkp: query kind=%v table=%q sql=%q", a.Kind, a.Table, truncate(a.SQL))
+			// Each entry carries its own statement text so per-statement
+			// logging and the PII heuristics do not see its neighbours.
+			s.execQueue = append(s.execQueue, &Portal{
+				Name: simplePortalName,
+				Stmt: &PreparedStatement{SQL: a.SQL, Analysis: a, WritePlan: plan},
+			})
+		}
+	}
+	s.OnSync()
+	return nil
+}
+
+// OnSync pushes a Sync boundary onto both result queues. The backend's
+// matching ReadyForQuery consumes it, so batches pipelined behind an
+// unconsumed ReadyForQuery keep their entries.
+func (s *Session) OnSync() {
+	s.execQueue = append(s.execQueue, syncBoundary)
+	s.describes = append(s.describes, pendingDescribe{boundary: true})
+}
+
 // OnClose drops a closed prepared statement or portal.
 func (s *Session) OnClose(c *pgproto3.Close) {
 	switch c.ObjectType {
@@ -162,34 +255,114 @@ func (s *Session) OnClose(c *pgproto3.Close) {
 	}
 }
 
-// OnRowDescription learns the result columns for the executing portal and
-// builds its decrypt-on-DataRow plan.
+// OnRowDescription learns the result columns of the statement being described
+// or executed. A RowDescription answers the oldest outstanding Describe when
+// one is pending (extended protocol); only the simple protocol emits a
+// RowDescription without a Describe, directly before that statement's
+// DataRows. Either way the columns are cached on the statement so later
+// Bind/Execute cycles that carry no Describe at all (pgjdbc's server-prepared
+// reuse) can still build their decrypt plan.
 func (s *Session) OnRowDescription(rd *pgproto3.RowDescription) {
+	columns := make([]string, len(rd.Fields))
+	for i, f := range rd.Fields {
+		columns[i] = string(f.Name)
+	}
+
+	if len(s.describes) > 0 && !s.describes[0].boundary {
+		pd := s.popDescribe()
+		if pd.stmt == nil {
+			debugf("kkp: rowdesc — describe of unknown statement/portal, skipping decrypt-plan build")
+			return
+		}
+		s.learnColumns(pd.stmt, columns)
+		if pd.portal != nil {
+			pd.portal.ReadPlan = s.planRead(pd.stmt, columns)
+		}
+		return
+	}
+
+	// No Describe outstanding: simple-protocol result shape for the current
+	// executing entry.
 	p := s.CurrentExecuting()
-	if p == nil || p.Stmt == nil || p.Stmt.Analysis == nil {
+	if p == nil || p.Stmt == nil {
 		debugf("kkp: rowdesc — no executing portal/analysis, skipping decrypt-plan build")
 		return
 	}
-	columns := make([]string, len(rd.Fields))
-	for i, f := range rd.Fields {
-		columns[i] = string(f.Name)
+	s.learnColumns(p.Stmt, columns)
+	p.ReadPlan = s.planRead(p.Stmt, columns)
+}
+
+// learnColumns caches a statement's result columns and fail-louds when a
+// PII-touching SELECT yields no read plan — the silent-ciphertext failure
+// mode the runtime-SQL conformance contract is meant to catch. The check is a
+// hint, not an error — production alerts on the metric.
+func (s *Session) learnColumns(ps *PreparedStatement, columns []string) {
+	ps.Columns = columns
+	plan := s.planRead(ps, columns)
+	table := ""
+	if ps.Analysis != nil {
+		table = ps.Analysis.Table
+	}
+	fields := 0
+	if plan != nil {
+		fields = len(plan.Fields)
 	}
-	p.ReadPlan = s.planner.PlanRead(p.Stmt.Analysis.Table, columns)
-	debugf("kkp: rowdesc portal=%q stmt=%q table=%q cols=%v readPlanFields=%d",
-		p.Name, p.Stmt.Name, p.Stmt.Analysis.Table, columns, len(p.ReadPlan.Fields))
+	debugf("kkp: rowdesc stmt=%q table=%q cols=%v readPlanFields=%d", ps.Name, table, columns, fields)
 
-	// Observability fail-loud: log when a SELECT mentions a PII
-	// table and returns a PII column, but the analyser produced no read
-	// plan. This is exactly the silent-ciphertext failure mode the runtime-
-	// SQL conformance contract is meant to catch. The check is a
-	// hint, not an error — production turns this into an alerted metric.
-	if p.ReadPlan.IsEmpty() && piiTouchedButNotPlanned(p.Stmt.SQL, columns) {
+	if (plan == nil || plan.IsEmpty()) && piiTouchedButNotPlanned(ps.SQL, columns) {
 		observe.UnrecognizedPIISQL.Inc()
 		log.Printf("kkp: WARN unrecognised PII-touching SELECT — passthrough — stmt=%q cols=%v sql=%q",
-			p.Stmt.Name, columns, truncate(p.Stmt.SQL, 4000))
+			ps.Name, columns, truncate(ps.SQL))
 	}
 }
 
+// planRead builds the decrypt plan for a statement's result columns; nil for
+// a statement without analysis (unparsable SQL → passthrough).
+func (s *Session) planRead(ps *PreparedStatement, columns []string) *rewrite.ReadPlan {
+	if ps.Analysis == nil {
+		return nil
+	}
+	return s.planner.PlanRead(ps.Analysis.Table, columns)
+}
+
+// ensureReadPlan resolves a portal's decrypt plan, deriving it from the
+// statement's cached columns when no RowDescription flowed for this portal:
+// pgjdbc describes a server-prepared statement once and then executes it many
+// times with a bare Bind/Execute, so the portal itself is never described
+// (the verify-email incident: those rows left the proxy undecrypted).
+func (s *Session) ensureReadPlan(p *Portal) *rewrite.ReadPlan {
+	if p.ReadPlan == nil && p.Stmt != nil && len(p.Stmt.Columns) > 0 {
+		p.ReadPlan = s.planRead(p.Stmt, p.Stmt.Columns)
+	}
+	return p.ReadPlan
+}
+
+// OnNoData consumes the pending Describe of a statement or portal that
+// returns no rows, keeping later RowDescriptions attributed correctly.
+func (s *Session) OnNoData() {
+	if len(s.describes) > 0 && !s.describes[0].boundary {
+		s.popDescribe()
+	}
+}
+
+// popDescribe removes and returns the head describe entry, zeroing the slot
+// so the backing array does not pin the referenced statement and portal.
+func (s *Session) popDescribe() pendingDescribe {
+	pd := s.describes[0]
+	s.describes[0] = pendingDescribe{}
+	s.describes = s.describes[1:]
+	return pd
+}
+
+// popExec removes and returns the head execute entry, zeroing the slot so
+// the backing array does not pin the popped portal.
+func (s *Session) popExec() *Portal {
+	p := s.execQueue[0]
+	s.execQueue[0] = nil
+	s.execQueue = s.execQueue[1:]
+	return p
+}
+
 // piiTouchedButNotPlanned reports whether the SQL textually mentions one of
 // the PII tables and the result set contains at least one PII column name.
 // Used purely to log a warning when a read sneaks past the analyser; it does
@@ -198,7 +371,7 @@ func piiTouchedButNotPlanned(sql string, cols []string) bool {
 	su := strings.ToUpper(sql)
 	mentioned := false
 	for _, t := range []string{"USER_ENTITY", "USER_ATTRIBUTE", "CREDENTIAL", "FED_USER_", "FEDERATED_IDENTITY"} {
-		if strings.Contains(su, t) {
+		if containsIdent(su, t) {
 			mentioned = true
 			break
 		}
@@ -217,24 +390,86 @@ func piiTouchedButNotPlanned(sql string, cols []string) bool {
 	return false
 }
 
+// containsIdent reports whether ident occurs in the upper-cased SQL text as a
+// standalone identifier (or identifier prefix when ident ends in '_'), so
+// that e.g. RESET_CREDENTIALS_FLOW does not count as a mention of CREDENTIAL.
+func containsIdent(su, ident string) bool {
+	for from := 0; ; {
+		i := strings.Index(su[from:], ident)
+		if i < 0 {
+			return false
+		}
+		i += from
+		before := i == 0 || !isIdentChar(su[i-1])
+		end := i + len(ident)
+		after := strings.HasSuffix(ident, "_") || end == len(su) || !isIdentChar(su[end])
+		if before && after {
+			return true
+		}
+		from = i + 1
+	}
+}
+
+func isIdentChar(c byte) bool {
+	return c == '_' || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9')
+}
+
 // OnCommandComplete advances past the executing portal once its result ends.
 func (s *Session) OnCommandComplete() { s.popExecuting() }
 
 // OnEmptyQueryResponse advances past an empty-query result.
 func (s *Session) OnEmptyQueryResponse() { s.popExecuting() }
 
-// OnErrorResponse advances past a failed command's result.
-func (s *Session) OnErrorResponse() { s.popExecuting() }
+// OnPortalSuspended advances past a row-limited Execute; the client resumes
+// with another Execute for the same portal, which re-enqueues it (the portal
+// keeps its read plan).
+func (s *Session) OnPortalSuspended() { s.popExecuting() }
+
+// OnErrorResponse drops the rest of the failing batch: after an error the
+// backend skips all subsequent messages until Sync, so no Execute or Describe
+// queued before the boundary will ever produce its results. Entries behind
+// the boundary belong to pipelined later batches and stay.
+func (s *Session) OnErrorResponse() {
+	for len(s.execQueue) > 0 && s.execQueue[0] != syncBoundary {
+		s.popExec()
+	}
+	for len(s.describes) > 0 && !s.describes[0].boundary {
+		s.popDescribe()
+	}
+}
+
+// OnReadyForQuery consumes one Sync boundary from both queues, together with
+// any dead entries of the batch it closes (a desync leftover: an entry that
+// never saw its CommandComplete). Entries of pipelined later batches stay.
+func (s *Session) OnReadyForQuery() {
+	dead := 0
+	for len(s.execQueue) > 0 {
+		if s.popExec() == syncBoundary {
+			break
+		}
+		dead++
+	}
+	for len(s.describes) > 0 {
+		if s.popDescribe().boundary {
+			break
+		}
+		dead++
+	}
+	if dead > 0 {
+		debugf("kkp: ready-for-query dropped %d dead pipeline entries — resync", dead)
+	}
+}
 
 func (s *Session) popExecuting() {
-	if len(s.execQueue) > 0 {
-		s.execQueue = s.execQueue[1:]
+	if len(s.execQueue) > 0 && s.execQueue[0] != syncBoundary {
+		s.popExec()
 	}
 }
 
-// CurrentExecuting returns the portal at the head of the execute queue, or nil.
+// CurrentExecuting returns the portal at the head of the execute queue, or
+// nil when the queue is empty or paused at a Sync boundary.
 func (s *Session) CurrentExecuting() *Portal {
-	if len(s.execQueue) == 0 {
+	if len(s.execQueue) == 0 || s.execQueue[0] == syncBoundary {
 		return nil
 	}
 	return s.execQueue[0]
diff --git a/internal/wire/transform.go b/internal/wire/transform.go
index c3ab095..eee6e74 100644
--- a/internal/wire/transform.go
+++ b/internal/wire/transform.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"errors"
 	"fmt"
+	"log"
 	"strings"
 
 	"github.com/jackc/pgproto3/v2"
@@ -250,26 +251,30 @@ func (s *Session) DecryptDataRow(dr *pgproto3.DataRow) error {
 	portal := s.CurrentExecuting()
 	if portal == nil {
 		debugf("kkp: datarow — NO executing portal (race? popped early?) — passthrough %d values", len(dr.Values))
+		flagCiphertextLeak(dr, "no-portal")
 		return nil
 	}
-	if portal.ReadPlan == nil {
+	plan := s.ensureReadPlan(portal)
+	if plan == nil {
 		debugf("kkp: datarow portal=%q stmt=%q — NO read plan built — passthrough", portal.Name, func() string {
 			if portal.Stmt != nil {
 				return portal.Stmt.Name
 			}
 			return ""
 		}())
+		flagCiphertextLeak(dr, "no-plan")
 		return nil
 	}
-	if portal.ReadPlan.IsEmpty() {
+	if plan.IsEmpty() {
+		flagCiphertextLeak(dr, "empty-plan")
 		return nil
 	}
 	if s.cipher == nil {
 		return errNoCipher
 	}
 
-	debugf("kkp: datarow portal=%q decrypting %d fields", portal.Name, len(portal.ReadPlan.Fields))
-	for _, f := range portal.ReadPlan.Fields {
+	debugf("kkp: datarow portal=%q decrypting %d fields", portal.Name, len(plan.Fields))
+	for _, f := range plan.Fields {
 		if f.Index < 0 || f.Index >= len(dr.Values) {
 			continue
 		}
@@ -283,11 +288,36 @@ func (s *Session) DecryptDataRow(dr *pgproto3.DataRow) error {
 			return fmt.Errorf("wire: decrypt field %d (%s.%s): %w", f.Index, f.Table, f.Column, err)
 		}
 		observe.DecryptTotal.WithLabelValues(f.Table, f.Column).Inc()
+		// A decrypted value that still parses as an envelope is a
+		// double-encrypted row: ciphertext leaked to the client during a
+		// passthrough window and was written back as the value. Surface it
+		// for cleanup — the proxy intentionally decrypts only one layer.
+		if _, ok, _ := crypto.Parse(string(plaintext)); ok {
+			observe.DoubleEncrypted.WithLabelValues(f.Table, f.Column).Inc()
+			log.Printf("kkp: WARN decrypted %s.%s still carries a ciphertext envelope — double-encrypted row (written back during a passthrough window)", f.Table, f.Column)
+		}
 		dr.Values[f.Index] = plaintext
 	}
 	return nil
 }
 
+// flagCiphertextLeak fail-louds (metric + WARN) when a DataRow that could not
+// be matched to a decrypt plan carries what looks like one of our ciphertext
+// envelopes — the silent-passthrough failure mode behind the verify-email
+// incident (raw $KKP$ blobs reaching Keycloak).
+func flagCiphertextLeak(dr *pgproto3.DataRow, reason string) {
+	for _, v := range dr.Values {
+		if v == nil {
+			continue
+		}
+		if _, ok, err := crypto.Parse(string(v)); ok || err != nil {
+			observe.CiphertextPassthrough.WithLabelValues(reason).Inc()
+			log.Printf("kkp: WARN ciphertext passed through undecrypted (%s) — read-path gap", reason)
+			return
+		}
+	}
+}
+
 // classifyErr buckets decrypt errors into low-cardinality reasons for the
 // kkp_decrypt_failures_total metric. Refine as new failure modes appear.
 func classifyErr(err error) string {