feat(xugu): report structured agent errors

agents/drivers/xugu/main.go

@@ -225,11 +225,6 @@ type response struct {
 	Error   *rpcError       `json:"error,omitempty"`
 }
 
-type rpcError struct {
-	Code    int    `json:"code"`
-	Message string `json:"message"`
-}
-
 type connectParams struct {
 	Host             string `json:"host"`
 	Port             int    `json:"port"`
@@ -539,6 +534,7 @@ type server struct {
 	activeRows        map[*sql.Rows]context.CancelFunc
 	activeTimer       *time.Timer
 	activeTimedOut    bool
+	activeCanceled    bool
 	// killSession, if non-nil, is called to force-kill the current
 	// statement on the database server. Tests may replace it with a
 	// stub. The real implementation is set during connectWithControl.
@@ -619,14 +615,14 @@ func newRuntimeServer() *runtimeServer {
 func (r *runtimeServer) handleLine(line string) (response, bool) {
 	var req request
 	if err := json.Unmarshal([]byte(line), &req); err != nil {
-		return errorResponse(nil, err), false
+		return errorResponse(nil, "", "", err), false
 	}
 	if len(req.ID) == 0 {
 		req.ID = json.RawMessage("1")
 	}
 	result, shutdown, err := r.dispatch(req.Method, req.Params)
 	if err != nil {
-		return errorResponse(req.ID, err), false
+		return errorResponse(req.ID, req.Method, stringParam(req.Params, "agentSessionId"), err), false
 	}
 	return response{JSONRPC: "2.0", ID: req.ID, Result: result}, shutdown
 }
@@ -637,7 +633,7 @@ func (r *runtimeServer) dispatch(method string, params map[string]json.RawMessag
 		return map[string]any{
 			"protocolVersion":      multiSessionProtocolVersion,
 			"agentProtocolVersion": multiSessionProtocolVersion,
-			"capabilities":         []string{"connect", "test_connection", "metadata", "query", "ddl", "multi_session"},
+			"capabilities":         []string{"connect", "test_connection", "metadata", "query", "ddl", "structured_error_v1", "multi_session"},
 		}, false, nil
 	case "open_session":
 		agentSessionID := stringParam(params, "agentSessionId")
@@ -711,7 +707,7 @@ func (r *runtimeServer) openSession(agentSessionID string, params connectParams)
 	}
 	if len(r.sessions) >= maxAgentSessions {
 		r.mu.Unlock()
-		return fmt.Errorf("agent session limit reached: %d", maxAgentSessions)
+		return fmt.Errorf("%w: %d", errAgentSessionLimit, maxAgentSessions)
 	}
 	r.mu.Unlock()
 
@@ -777,7 +773,7 @@ func (r *runtimeServer) registerSession(agentSessionID string, server *server, c
 	if len(r.sessions) >= maxAgentSessions {
 		_ = server.disconnect()
 		r.releaseControl(controlKey)
-		return fmt.Errorf("agent session limit reached: %d", maxAgentSessions)
+		return fmt.Errorf("%w: %d", errAgentSessionLimit, maxAgentSessions)
 	}
 	r.sessions[agentSessionID] = session
 	return nil
@@ -811,7 +807,7 @@ func (r *runtimeServer) session(agentSessionID string) (*agentSession, error) {
 	session := r.sessions[agentSessionID]
 	r.mu.RUnlock()
 	if session == nil {
-		return nil, fmt.Errorf("agent session not found: %s", agentSessionID)
+		return nil, fmt.Errorf("%w: %s", errAgentSessionNotFound, agentSessionID)
 	}
 	return session, nil
 }
@@ -897,14 +893,14 @@ func (r *runtimeServer) closeAllSessions() error {
 func (s *server) handleLine(line string) (response, bool) {
 	var req request
 	if err := json.Unmarshal([]byte(line), &req); err != nil {
-		return errorResponse(nil, err), false
+		return errorResponse(nil, "", "", err), false
 	}
 	if len(req.ID) == 0 {
 		req.ID = json.RawMessage("1")
 	}
 	result, shutdown, err := s.dispatch(req.Method, req.Params)
 	if err != nil {
-		return errorResponse(req.ID, err), false
+		return errorResponse(req.ID, req.Method, stringParam(req.Params, "agentSessionId"), err), false
 	}
 	return response{JSONRPC: "2.0", ID: req.ID, Result: result}, shutdown
 }
@@ -3311,7 +3307,7 @@ func (s *server) getExplainInfo(sqlText string) (string, error) {
 	return strings.TrimSpace(builder.String()), rows.Err()
 }
 
-func (s *server) executeTransaction(params map[string]json.RawMessage) (queryResult, error) {
+func (s *server) executeTransaction(params map[string]json.RawMessage) (result queryResult, err error) {
 	var payload struct {
 		Statements []string `json:"statements"`
 		Schema     string   `json:"schema"`
@@ -3324,7 +3320,9 @@ func (s *server) executeTransaction(params map[string]json.RawMessage) (queryRes
 		return queryResult{}, err
 	}
 	ctx, cancel := s.beginActiveOperation()
-	defer s.endActiveOperation(cancel)
+	defer func() {
+		err = s.finishActiveOperation(cancel, 0, err)
+	}()
 	tx, err := db.BeginTx(ctx, nil)
 	if err != nil {
 		return queryResult{}, err
@@ -3342,12 +3340,12 @@ func (s *server) executeTransaction(params map[string]json.RawMessage) (queryRes
 		if statement == "" {
 			continue
 		}
-		result, err := tx.ExecContext(ctx, statement)
+		execResult, err := tx.ExecContext(ctx, statement)
 		if err != nil {
 			tx.Rollback()
 			return queryResult{}, err
 		}
-		count, _ := result.RowsAffected()
+		count, _ := execResult.RowsAffected()
 		affected += count
 	}
 	if err := tx.Commit(); err != nil {
@@ -3520,8 +3518,8 @@ func (s *server) executeQuery(opts queryOptions) (queryResult, error) {
 		return queryResult{}, err
 	}
 	ctx, cancel := s.beginActiveOperationWithTimeout(opts.TimeoutSecs)
-	defer s.endActiveOperation(cancel)
 	execResult, err := db.ExecContext(ctx, sqlText)
+	err = s.finishActiveOperation(cancel, opts.TimeoutSecs, err)
 	if err != nil {
 		return queryResult{}, err
 	}
@@ -3618,18 +3616,20 @@ func (s *server) queryRowsWithTimeoutOnce(sqlText string, args []any, timeoutSec
 		s.activeTimer = nil
 	}
 	timedOut := s.activeTimedOut
-	if queryErr != nil {
-		cancel()
-	} else if timedOut {
-		cancel()
-		if rows != nil {
-			rows.Close()
-		}
-		queryErr = fmt.Errorf("query timed out after %ds", timeoutSecs)
-	} else {
+	canceled := s.activeCanceled
+	s.activeTimedOut = false
+	s.activeCanceled = false
+	if queryErr == nil && !timedOut && !canceled {
 		s.activeRows[rows] = cancel
 	}
 	s.activeCancelMu.Unlock()
+	if queryErr != nil || timedOut || canceled {
+		cancel()
+		if rows != nil {
+			_ = rows.Close()
+		}
+		queryErr = xuguOperationResultError(timedOut, canceled, timeoutSecs, queryErr)
+	}
 	return rows, queryErr
 }
 
@@ -3659,7 +3659,7 @@ func (s *server) execWithReconnect(statement string) error {
 	}
 	ctx, cancel := s.beginActiveOperation()
 	_, execErr := db.ExecContext(ctx, statement)
-	s.endActiveOperation(cancel)
+	execErr = s.finishActiveOperation(cancel, 0, execErr)
 	if execErr == nil || !isXuguConnectionClosedError(execErr) {
 		return execErr
 	}
@@ -3672,7 +3672,7 @@ func (s *server) execWithReconnect(statement string) error {
 	}
 	ctx, cancel = s.beginActiveOperation()
 	_, execErr = db.ExecContext(ctx, statement)
-	s.endActiveOperation(cancel)
+	execErr = s.finishActiveOperation(cancel, 0, execErr)
 	return execErr
 }
 
@@ -3702,25 +3702,60 @@ func (s *server) beginActiveOperationWithTimeout(timeoutSecs int) (context.Conte
 	s.activeCancel = cancel
 	s.activeTimer = timer
 	s.activeTimedOut = false
+	s.activeCanceled = false
 	s.activeCancelMu.Unlock()
 	return ctx, cancel
 }
 
 func (s *server) endActiveOperation(cancel context.CancelFunc) {
-	cancel()
+	_ = s.finishActiveOperation(cancel, 0, nil)
+}
+
+func (s *server) finishActiveOperation(cancel context.CancelFunc, timeoutSecs int, operationErr error) error {
 	s.activeCancelMu.Lock()
 	s.activeCancel = nil
 	if s.activeTimer != nil {
 		s.activeTimer.Stop()
 		s.activeTimer = nil
 	}
+	timedOut := s.activeTimedOut
+	canceled := s.activeCanceled
+	s.activeTimedOut = false
+	s.activeCanceled = false
 	s.activeCancelMu.Unlock()
+	cancel()
+	return xuguOperationResultError(timedOut, canceled, timeoutSecs, operationErr)
+}
+
+func xuguOperationResultError(timedOut, canceled bool, timeoutSecs int, operationErr error) error {
+	if timedOut {
+		if operationErr != nil {
+			return fmt.Errorf("%w after %ds: %v", errXuguOperationTimeout, timeoutSecs, operationErr)
+		}
+		return fmt.Errorf("%w after %ds", errXuguOperationTimeout, timeoutSecs)
+	}
+	if canceled {
+		if operationErr != nil {
+			return fmt.Errorf("%w: %v", errXuguOperationCanceled, operationErr)
+		}
+		return errXuguOperationCanceled
+	}
+	return operationErr
 }
 
 func (s *server) cancelActiveQuery() {
 	s.activeCancelMu.Lock()
 	cancels := make([]context.CancelFunc, 0, len(s.activeRows)+1)
 	if s.activeCancel != nil {
+		if !s.activeTimedOut {
+			s.activeCanceled = true
+			// Explicit cancellation won the race. Disable the watchdog so a
+			// slow driver return cannot relabel this operation as a timeout.
+			if s.activeTimer != nil {
+				s.activeTimer.Stop()
+				s.activeTimer = nil
+			}
+		}
 		cancels = append(cancels, s.activeCancel)
 	}
 	for _, cancel := range s.activeRows {
@@ -5105,8 +5140,8 @@ func stringSliceParam(params map[string]json.RawMessage, key string) []string {
 	return nil
 }
 
-func errorResponse(id json.RawMessage, err error) response {
-	return response{JSONRPC: "2.0", ID: id, Error: &rpcError{Code: -1, Message: err.Error()}}
+func errorResponse(id json.RawMessage, method, agentSessionID string, err error) response {
+	return response{JSONRPC: "2.0", ID: id, Error: classifyRPCError(method, agentSessionID, err)}
 }
 
 func trimStatementSQL(sqlText string) string {

agents/drivers/xugu/main_test.go

@@ -71,7 +71,8 @@ func TestRuntimeHandshakeAdvertisesMultiSessionProtocol(t *testing.T) {
 	if err := json.Unmarshal(data, &result); err != nil {
 		t.Fatal(err)
 	}
-	if result.ProtocolVersion != multiSessionProtocolVersion || !contains(result.Capabilities, "multi_session") {
+	if result.ProtocolVersion != multiSessionProtocolVersion || !contains(result.Capabilities, "multi_session") ||
+		!contains(result.Capabilities, "structured_error_v1") {
 		t.Fatalf("unexpected runtime handshake: %+v", result)
 	}
 }
@@ -3402,13 +3403,176 @@ func TestXuguWatchdogCallsKillOnBlockingQuery(t *testing.T) {
 
 	select {
 	case err := <-errCh:
-		if err == nil {
-			t.Fatal("expected non-nil error after kill")
+		if !errors.Is(err, errXuguOperationTimeout) || !strings.Contains(err.Error(), "killed") {
+			t.Fatalf("expected recorded timeout preserving the driver error, got: %v", err)
 		}
-		if !strings.Contains(err.Error(), "killed") && !strings.Contains(err.Error(), "timed out") {
-			t.Fatalf("expected killed or timeout error, got: %v", err)
+		rpcErr := classifyRPCError("execute_query", "watchdog-query", err)
+		if rpcErr.Data.Category != "timeout" || rpcErr.Data.SessionDisposition != "quarantine" {
+			t.Fatalf("unexpected query timeout contract: %+v", rpcErr.Data)
 		}
 	case <-time.After(3 * time.Second):
 		t.Fatal("query did not return after unblocking driver")
 	}
 }
+
+func TestXuguWatchdogClassifiesBlockingExecAsTimeout(t *testing.T) {
+	resetXuguBlockingDriver()
+	s := newServer()
+	killCh := make(chan struct{})
+	s.killSession = func() { close(killCh) }
+	db, err := sql.Open("xugu-test-blocking", "dsn")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	s.db = db
+
+	errCh := make(chan error, 1)
+	go func() {
+		_, err := s.executeQuery(queryOptions{SQL: "UPDATE DBX_TIMEOUT_TEST SET VALUE = 1", TimeoutSecs: 1})
+		errCh <- err
+	}()
+
+	select {
+	case <-killCh:
+	case <-time.After(3 * time.Second):
+		t.Fatal("killSession was not called for the blocking exec")
+	}
+	close(xuguBlockingUnblock)
+
+	select {
+	case err := <-errCh:
+		if !errors.Is(err, errXuguOperationTimeout) || !strings.Contains(err.Error(), "killed") {
+			t.Fatalf("expected recorded exec timeout preserving the driver error, got: %v", err)
+		}
+		rpcErr := classifyRPCError("execute_query", "watchdog-exec", err)
+		if rpcErr.Data.Category != "timeout" || rpcErr.Data.SessionDisposition != "quarantine" {
+			t.Fatalf("unexpected exec timeout contract: %+v", rpcErr.Data)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("exec did not return after unblocking driver")
+	}
+}
+
+func TestXuguExplicitCancelClassifiesBlockingQueryAndExec(t *testing.T) {
+	for _, test := range []struct {
+		name string
+		run  func(*server) error
+	}{
+		{
+			name: "query",
+			run: func(s *server) error {
+				rows, err := s.queryRowsWithTimeout("SELECT 1", nil, 0)
+				if rows != nil {
+					_ = rows.Close()
+				}
+				return err
+			},
+		},
+		{
+			name: "exec",
+			run: func(s *server) error {
+				_, err := s.executeQuery(queryOptions{SQL: "UPDATE DBX_CANCEL_TEST SET VALUE = 1"})
+				return err
+			},
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			resetXuguBlockingDriver()
+			s := newServer()
+			killCh := make(chan struct{})
+			s.killSession = func() { close(killCh) }
+			db, err := sql.Open("xugu-test-blocking", "dsn")
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer db.Close()
+			s.db = db
+
+			errCh := make(chan error, 1)
+			go func() { errCh <- test.run(s) }()
+			waitForXuguActiveOperation(t, s)
+			s.cancelActiveQuery()
+			select {
+			case <-killCh:
+			case <-time.After(time.Second):
+				t.Fatal("killSession was not called for explicit cancellation")
+			}
+			close(xuguBlockingUnblock)
+
+			select {
+			case err := <-errCh:
+				if !errors.Is(err, errXuguOperationCanceled) || !strings.Contains(err.Error(), "killed") {
+					t.Fatalf("expected recorded cancellation preserving the driver error, got: %v", err)
+				}
+				rpcErr := classifyRPCError("execute_query", "explicit-cancel", err)
+				if rpcErr.Data.Category != "canceled" || rpcErr.Data.SessionDisposition != "quarantine" {
+					t.Fatalf("unexpected cancellation contract: %+v", rpcErr.Data)
+				}
+			case <-time.After(3 * time.Second):
+				t.Fatal("operation did not return after explicit cancellation")
+			}
+		})
+	}
+}
+
+func TestXuguExplicitCancelWinsWatchdogRace(t *testing.T) {
+	resetXuguBlockingDriver()
+	s := newServer()
+	var killMu sync.Mutex
+	killCount := 0
+	s.killSession = func() {
+		killMu.Lock()
+		killCount++
+		killMu.Unlock()
+	}
+	db, err := sql.Open("xugu-test-blocking", "dsn")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	s.db = db
+
+	errCh := make(chan error, 1)
+	go func() {
+		_, err := s.queryRowsWithTimeout("SELECT 1", nil, 1)
+		errCh <- err
+	}()
+	waitForXuguActiveOperation(t, s)
+	s.cancelActiveQuery()
+
+	// Keep the driver blocked beyond the original watchdog deadline. The
+	// explicit cancel happened first, so the timer must not fire or relabel it.
+	time.Sleep(1200 * time.Millisecond)
+	killMu.Lock()
+	gotKillCount := killCount
+	killMu.Unlock()
+	if gotKillCount != 1 {
+		t.Fatalf("expected exactly one kill from explicit cancel, got %d", gotKillCount)
+	}
+	close(xuguBlockingUnblock)
+
+	select {
+	case err := <-errCh:
+		if !errors.Is(err, errXuguOperationCanceled) || errors.Is(err, errXuguOperationTimeout) {
+			t.Fatalf("explicit cancel was relabeled after its watchdog deadline: %v", err)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("query did not return after unblocking driver")
+	}
+}
+
+func waitForXuguActiveOperation(t *testing.T, s *server) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		s.activeCancelMu.Lock()
+		active := s.activeCancel != nil
+		s.activeCancelMu.Unlock()
+		if active {
+			return
+		}
+		time.Sleep(time.Millisecond)
+	}
+	t.Fatal("operation did not become active")
+}

agents/drivers/xugu/protocol_error.go

@@ -0,0 +1,233 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"database/sql/driver"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"regexp"
+	"strconv"
+	"strings"
+)
+
+var (
+	errAgentSessionLimit     = errors.New("agent session limit reached")
+	errAgentSessionNotFound  = errors.New("agent session not found")
+	errXuguOperationTimeout  = errors.New("xugu operation timed out")
+	errXuguOperationCanceled = errors.New("xugu operation canceled")
+
+	// go-xugu-driver exposes server errors as plain strings rather than a typed
+	// error. Match only the stable server error header and keep the complete
+	// original message for diagnostics. A response can contain more than one
+	// Xugu error; vendorCode intentionally records the first/top-level code.
+	xuguServerErrorHeader = regexp.MustCompile(`(?im)^[\t ]*(?:error:[\t ]*)?\[[\t ]*E([0-9]{1,9})(?:[\t ]+L([0-9]+))?(?:[\t ]+C([0-9]+))?[\t ]*\]`)
+	xuguQueryTimeout      = regexp.MustCompile(`(?i)^query timed out after [1-9][0-9]*s$`)
+)
+
+type rpcError struct {
+	Code    int           `json:"code"`
+	Message string        `json:"message"`
+	Data    *rpcErrorData `json:"data,omitempty"`
+}
+
+type rpcErrorData struct {
+	Category           string `json:"category"`
+	Retryable          bool   `json:"retryable"`
+	SessionDisposition string `json:"sessionDisposition"`
+	Stage              string `json:"stage"`
+	ContractVersion    int    `json:"contractVersion"`
+	OperationOutcome   string `json:"operationOutcome"`
+	VendorCode         int32  `json:"vendorCode,omitempty"`
+	ExceptionClass     string `json:"exceptionClass,omitempty"`
+	AgentSessionID     string `json:"agentSessionId,omitempty"`
+}
+
+func classifyRPCError(method, agentSessionID string, err error) *rpcError {
+	stage := rpcErrorStage(method)
+	data := &rpcErrorData{
+		Category:           "protocol",
+		Retryable:          false,
+		SessionDisposition: "keep",
+		Stage:              stage,
+		ContractVersion:    1,
+		OperationOutcome:   rpcOperationOutcome(stage),
+		// The structured-error contract requires an exact echo of the request
+		// session identifier. Validation belongs at the request boundary; an
+		// error response must never normalize the identifier it received.
+		AgentSessionID: agentSessionID,
+	}
+	if err == nil {
+		return &rpcError{Code: -1, Message: "unknown agent error", Data: data}
+	}
+	data.ExceptionClass = safeRPCDiagnostic(fmt.Sprintf("%T", err), 160)
+
+	if vendorCode, ok := xuguVendorCode(err); ok {
+		data.VendorCode = vendorCode
+	}
+
+	switch {
+	case errors.Is(err, errAgentSessionLimit):
+		// Session capacity is checked before a database operation starts, so
+		// keeping the shared runtime is safe and the caller may retry later.
+		if data.OperationOutcome == "not_started" {
+			data.Category = "resource"
+			data.Retryable = true
+		}
+	case errors.Is(err, errXuguOperationCanceled) || errors.Is(err, context.Canceled):
+		data.Category = "canceled"
+		data.SessionDisposition = "quarantine"
+	case errors.Is(err, errXuguOperationTimeout) || errors.Is(err, context.DeadlineExceeded) || isXuguTimeoutError(err):
+		data.Category = "timeout"
+		data.Retryable = stage == "connect" || stage == "validate"
+		data.SessionDisposition = "quarantine"
+	case errors.Is(err, errAgentSessionNotFound):
+		data.Category = "protocol"
+		data.SessionDisposition = "quarantine"
+	case isXuguTypedConnectionError(err):
+		data.Category = "connection"
+		data.Retryable = stage == "connect" || stage == "validate"
+		if stage != "connect" {
+			data.SessionDisposition = "quarantine"
+		}
+	case data.VendorCode != 0:
+		if stage == "connect" || stage == "validate" {
+			// The strict Agent contract does not allow category=sql during
+			// connect/validate. Preserve the Xugu code while reporting the
+			// failed connection stage accurately.
+			data.Category = "connection"
+			data.Retryable = true
+			if stage == "validate" {
+				data.SessionDisposition = "quarantine"
+			}
+		} else if stage == "request" {
+			data.Category = "protocol"
+		} else {
+			data.Category = "sql"
+		}
+	case isXuguWireProtocolError(err):
+		// go-xugu-driver v1.0.12 discards the error from its first socket
+		// read. A zero-byte EOF consequently surfaces as this parser text,
+		// and the read buffer is no longer trustworthy. Treat it as a broken
+		// connection rather than asking the caller to reuse the session.
+		data.Category = "connection"
+		data.Retryable = stage == "connect" || stage == "validate"
+		if stage != "connect" {
+			data.SessionDisposition = "quarantine"
+		}
+	case isXuguConnectionError(err):
+		data.Category = "connection"
+		data.Retryable = stage == "connect" || stage == "validate"
+		if stage != "connect" {
+			data.SessionDisposition = "quarantine"
+		}
+	}
+
+	return &rpcError{Code: -1, Message: err.Error(), Data: data}
+}
+
+func rpcErrorStage(method string) string {
+	switch method {
+	case "connect", "open_session", "test_connection":
+		return "connect"
+	case "validate_connection", "validate_session":
+		return "validate"
+	case "cancel_session":
+		return "cancel"
+	case "close_session", "disconnect", "close_query_session", "close_table_read_session", "shutdown":
+		return "close"
+	case "fetch_query_page", "fetch_table_read_page":
+		return "fetch"
+	case "handshake", "":
+		return "request"
+	default:
+		return "execute"
+	}
+}
+
+func rpcOperationOutcome(stage string) string {
+	switch stage {
+	case "request", "connect", "validate":
+		return "not_started"
+	default:
+		return "unknown"
+	}
+}
+
+func xuguVendorCode(err error) (int32, bool) {
+	if err == nil {
+		return 0, false
+	}
+	message := strings.TrimRight(err.Error(), "\x00")
+	match := xuguServerErrorHeader.FindStringSubmatch(message)
+	if len(match) < 2 {
+		return 0, false
+	}
+	value, parseErr := strconv.ParseInt(match[1], 10, 32)
+	if parseErr != nil || value <= 0 {
+		return 0, false
+	}
+	return int32(value), true
+}
+
+func isXuguTimeoutError(err error) bool {
+	var timeout interface{ Timeout() bool }
+	if errors.As(err, &timeout) && timeout.Timeout() {
+		return true
+	}
+	return xuguQueryTimeout.MatchString(strings.TrimSpace(err.Error()))
+}
+
+func isXuguConnectionError(err error) bool {
+	if isXuguTypedConnectionError(err) {
+		return true
+	}
+	lower := strings.ToLower(strings.TrimSpace(err.Error()))
+	for _, marker := range []string{
+		"connection refused",
+		"connection reset",
+		"broken pipe",
+		"connection closed",
+		"connection lost",
+		"driver: bad connection",
+		"unexpected eof",
+		"no route to host",
+		"agent is not connected",
+		"向数据库发起连接失败",
+		"接收数据库连接失败",
+		"数据库连接失败",
+	} {
+		if strings.Contains(lower, marker) {
+			return true
+		}
+	}
+	return lower == "not connected"
+}
+
+func isXuguTypedConnectionError(err error) bool {
+	if errors.Is(err, driver.ErrBadConn) || errors.Is(err, sql.ErrConnDone) || errors.Is(err, io.EOF) ||
+		errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, net.ErrClosed) {
+		return true
+	}
+	var networkError *net.OpError
+	return errors.As(err, &networkError)
+}
+
+func isXuguWireProtocolError(err error) bool {
+	return strings.Contains(strings.ToLower(err.Error()), "parsemsg: unknown message type")
+}
+
+func safeRPCDiagnostic(value string, maxLength int) string {
+	var result strings.Builder
+	for _, char := range value {
+		if result.Len() >= maxLength {
+			break
+		}
+		if char >= 0x21 && char <= 0x7e {
+			result.WriteRune(char)
+		}
+	}
+	return result.String()
+}

agents/drivers/xugu/protocol_error_live_test.go

@@ -0,0 +1,89 @@
+package main
+
+import (
+	"encoding/json"
+	"os"
+	"testing"
+)
+
+// TestLiveXuguStructuredErrorRoundTrip verifies the v2 runtime contract with a
+// real Xugu server. It is opt-in because public CI does not provide XuguDB.
+// The test is read-only: it triggers an object-not-found error, confirms the
+// structured response, and then proves that the same logical session remains
+// usable for a successful query.
+func TestLiveXuguStructuredErrorRoundTrip(t *testing.T) {
+	if os.Getenv("XUGU_LIVE_TEST") != "1" {
+		t.Skip("set XUGU_LIVE_TEST=1 with XUGU_LIVE_* connection settings to run the XuguDB integration test")
+	}
+	params := connectParams{
+		Host:     os.Getenv("XUGU_LIVE_HOST"),
+		Port:     parsePort(os.Getenv("XUGU_LIVE_PORT")),
+		Database: os.Getenv("XUGU_LIVE_DATABASE"),
+		Username: os.Getenv("XUGU_LIVE_USERNAME"),
+		Password: os.Getenv("XUGU_LIVE_PASSWORD"),
+	}
+	if params.Host == "" || params.Database == "" || params.Username == "" || params.Password == "" {
+		t.Fatal("XUGU_LIVE_HOST, XUGU_LIVE_DATABASE, XUGU_LIVE_USERNAME, and XUGU_LIVE_PASSWORD are required")
+	}
+
+	runtime := newRuntimeServer()
+	const agentSessionID = "xugu-structured-error-live"
+	openParams := map[string]any{
+		"agentSessionId": agentSessionID,
+		"host":           params.Host,
+		"port":           params.Port,
+		"database":       params.Database,
+		"username":       params.Username,
+		"password":       params.Password,
+	}
+	open := liveXuguRPCRequest(t, runtime, 1, "open_session", openParams)
+	if open.Error != nil {
+		t.Fatalf("open live Xugu session: %s", open.Error.Message)
+	}
+	defer runtime.closeSession(agentSessionID)
+
+	invalid := liveXuguRPCRequest(t, runtime, 2, "execute_query", map[string]any{
+		"agentSessionId": agentSessionID,
+		"database":       params.Database,
+		"schema":         params.Username,
+		"sql":            `SELECT * FROM "DBX_JAR12_OBJECT_THAT_MUST_NOT_EXIST"`,
+		"maxRows":        1,
+	})
+	if invalid.Error == nil {
+		t.Fatal("expected the missing object query to fail")
+	}
+	assertXuguStructuredErrorContract(t, invalid.Error.Data)
+	if invalid.Error.Data.Category != "sql" || invalid.Error.Data.SessionDisposition != "keep" ||
+		invalid.Error.Data.AgentSessionID != agentSessionID || invalid.Error.Data.VendorCode == 0 {
+		t.Fatalf("unexpected live Xugu error contract: %+v", invalid.Error.Data)
+	}
+
+	valid := liveXuguRPCRequest(t, runtime, 3, "execute_query", map[string]any{
+		"agentSessionId": agentSessionID,
+		"database":       params.Database,
+		"schema":         params.Username,
+		"sql":            "SELECT 1",
+		"maxRows":        1,
+	})
+	if valid.Error != nil {
+		t.Fatalf("live Xugu session was not reusable after SQL error: %s", valid.Error.Message)
+	}
+}
+
+func liveXuguRPCRequest(t *testing.T, runtime *runtimeServer, id int, method string, params map[string]any) response {
+	t.Helper()
+	payload, err := json.Marshal(map[string]any{
+		"jsonrpc": "2.0",
+		"id":      id,
+		"method":  method,
+		"params":  params,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	result, shutdown := runtime.handleLine(string(payload))
+	if shutdown {
+		t.Fatalf("%s unexpectedly shut down the Xugu runtime", method)
+	}
+	return result
+}

agents/drivers/xugu/protocol_error_test.go

@@ -0,0 +1,386 @@
+package main
+
+import (
+	"context"
+	"database/sql/driver"
+	"encoding/json"
+	"errors"
+	"io"
+	"strings"
+	"testing"
+)
+
+func TestXuguStructuredRPCErrorClassification(t *testing.T) {
+	tests := []struct {
+		name        string
+		method      string
+		err         error
+		category    string
+		retryable   bool
+		disposition string
+		stage       string
+		outcome     string
+		vendorCode  int32
+	}{
+		{
+			name:        "permission remains sql",
+			method:      "get_object_source",
+			err:         errors.New("[E18012] 权限不够"),
+			category:    "sql",
+			disposition: "keep",
+			stage:       "execute",
+			outcome:     "unknown",
+			vendorCode:  18012,
+		},
+		{
+			name:        "first syntax code and location text are preserved",
+			method:      "execute_query",
+			err:         errors.New("[E19132] 语法错误 [E19260 L6 C29] unexpected end of file"),
+			category:    "sql",
+			disposition: "keep",
+			stage:       "execute",
+			outcome:     "unknown",
+			vendorCode:  19132,
+		},
+		{
+			name:        "server code with trailing nul",
+			method:      "list_objects",
+			err:         errors.New("[E10049 L2 C57] 字段 ON_NULL 不存在\x00"),
+			category:    "sql",
+			disposition: "keep",
+			stage:       "execute",
+			outcome:     "unknown",
+			vendorCode:  10049,
+		},
+		{
+			name:        "server error during connect",
+			method:      "open_session",
+			err:         errors.New("[E18012] login rejected"),
+			category:    "connection",
+			retryable:   true,
+			disposition: "keep",
+			stage:       "connect",
+			outcome:     "not_started",
+			vendorCode:  18012,
+		},
+		{
+			name:        "server error during validation",
+			method:      "validate_session",
+			err:         errors.New("[E18012] validation failed"),
+			category:    "connection",
+			retryable:   true,
+			disposition: "quarantine",
+			stage:       "validate",
+			outcome:     "not_started",
+			vendorCode:  18012,
+		},
+		{
+			name:        "context canceled",
+			method:      "execute_query",
+			err:         context.Canceled,
+			category:    "canceled",
+			disposition: "quarantine",
+			stage:       "execute",
+			outcome:     "unknown",
+		},
+		{
+			name:        "deadline exceeded",
+			method:      "fetch_query_page",
+			err:         context.DeadlineExceeded,
+			category:    "timeout",
+			disposition: "quarantine",
+			stage:       "fetch",
+			outcome:     "unknown",
+		},
+		{
+			name:        "agent watchdog timeout",
+			method:      "execute_query",
+			err:         errors.New("query timed out after 2s"),
+			category:    "timeout",
+			disposition: "quarantine",
+			stage:       "execute",
+			outcome:     "unknown",
+		},
+		{
+			name:        "eof while executing",
+			method:      "execute_query",
+			err:         io.EOF,
+			category:    "connection",
+			disposition: "quarantine",
+			stage:       "execute",
+			outcome:     "unknown",
+		},
+		{
+			name:        "bad connection while connecting",
+			method:      "test_connection",
+			err:         driver.ErrBadConn,
+			category:    "connection",
+			retryable:   true,
+			disposition: "keep",
+			stage:       "connect",
+			outcome:     "not_started",
+		},
+		{
+			name:        "driver connection wrapper",
+			method:      "list_tables",
+			err:         errors.New("接收数据库连接失败: EOF"),
+			category:    "connection",
+			disposition: "quarantine",
+			stage:       "execute",
+			outcome:     "unknown",
+		},
+		{
+			name:        "wire parser failure quarantines live session",
+			method:      "list_tables",
+			err:         errors.New("parseMsg: unknown message type"),
+			category:    "connection",
+			disposition: "quarantine",
+			stage:       "execute",
+			outcome:     "unknown",
+		},
+		{
+			name:        "server error text wins over eof words",
+			method:      "execute_query",
+			err:         errors.New("[E19260 L6 C29] syntax error: unexpected EOF"),
+			category:    "sql",
+			disposition: "keep",
+			stage:       "execute",
+			outcome:     "unknown",
+			vendorCode:  19260,
+		},
+		{
+			name:        "killed is not guessed as cancellation",
+			method:      "execute_query",
+			err:         errors.New("killed"),
+			category:    "protocol",
+			disposition: "keep",
+			stage:       "execute",
+			outcome:     "unknown",
+		},
+		{
+			name:        "session capacity",
+			method:      "open_session",
+			err:         errAgentSessionLimit,
+			category:    "resource",
+			retryable:   true,
+			disposition: "keep",
+			stage:       "connect",
+			outcome:     "not_started",
+		},
+		{
+			name:        "missing runtime session",
+			method:      "list_tables",
+			err:         errAgentSessionNotFound,
+			category:    "protocol",
+			disposition: "quarantine",
+			stage:       "execute",
+			outcome:     "unknown",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			rpcErr := classifyRPCError(test.method, "session-1", test.err)
+			data := rpcErr.Data
+			assertXuguStructuredErrorContract(t, data)
+			if data.Category != test.category || data.Retryable != test.retryable ||
+				data.SessionDisposition != test.disposition || data.Stage != test.stage ||
+				data.OperationOutcome != test.outcome || data.VendorCode != test.vendorCode {
+				t.Fatalf("unexpected classification: %+v", data)
+			}
+			if data.ContractVersion != 1 || data.AgentSessionID != "session-1" {
+				t.Fatalf("missing structured error identity: %+v", data)
+			}
+			if rpcErr.Message != test.err.Error() {
+				t.Fatalf("error message changed: got %q want %q", rpcErr.Message, test.err.Error())
+			}
+		})
+	}
+}
+
+func TestXuguStructuredRPCErrorContainsRequiredContractFields(t *testing.T) {
+	rpcErr := classifyRPCError("fetch_query_page", "session-2", context.DeadlineExceeded)
+	assertXuguStructuredErrorContract(t, rpcErr.Data)
+	payload, err := json.Marshal(rpcErr)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var decoded struct {
+		Data map[string]any `json:"data"`
+	}
+	if err := json.Unmarshal(payload, &decoded); err != nil {
+		t.Fatal(err)
+	}
+	for _, key := range []string{
+		"category", "retryable", "sessionDisposition", "stage", "contractVersion", "operationOutcome", "agentSessionId",
+	} {
+		if _, ok := decoded.Data[key]; !ok {
+			t.Fatalf("structured error missing %s: %s", key, payload)
+		}
+	}
+	if decoded.Data["stage"] != "fetch" || decoded.Data["operationOutcome"] != "unknown" {
+		t.Fatalf("unexpected fetch error contract: %s", payload)
+	}
+}
+
+func TestXuguStructuredRPCErrorEchoesSessionIDExactly(t *testing.T) {
+	const sessionID = " session-with-boundary-whitespace "
+	rpcErr := classifyRPCError("list_tables", sessionID, errAgentSessionNotFound)
+	if rpcErr.Data == nil || rpcErr.Data.AgentSessionID != sessionID {
+		t.Fatalf("agentSessionId was normalized: got %q want %q", rpcErr.Data.AgentSessionID, sessionID)
+	}
+}
+
+func TestXuguVendorCodeUsesStableHeaderOnly(t *testing.T) {
+	tests := []struct {
+		name    string
+		message string
+		code    int32
+		ok      bool
+	}{
+		{name: "simple", message: "[E18012] 权限不够", code: 18012, ok: true},
+		{name: "leading whitespace", message: "  [E19260 L6 C29] syntax error", code: 19260, ok: true},
+		{name: "second line", message: "wrapper\n[E10049 L2 C57] missing", code: 10049, ok: true},
+		{name: "first of multiple", message: "[E19132] syntax [E19260 L6 C29] detail", code: 19132, ok: true},
+		{name: "short code", message: "[E1 L1 C8] syntax error", code: 1, ok: true},
+		{name: "unstructured mention", message: "metadata request failed with E18012", ok: false},
+		{name: "bracket cannot span lines", message: "[E18012\n] denied", ok: false},
+		{name: "too many digits", message: "[E9999999999] invalid", ok: false},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			code, ok := xuguVendorCode(errors.New(test.message))
+			if code != test.code || ok != test.ok {
+				t.Fatalf("xuguVendorCode(%q) = (%d, %t), want (%d, %t)", test.message, code, ok, test.code, test.ok)
+			}
+		})
+	}
+}
+
+func TestXuguRPCErrorStageMapping(t *testing.T) {
+	tests := map[string]string{
+		"":                         "request",
+		"handshake":                "request",
+		"open_session":             "connect",
+		"test_connection":          "connect",
+		"validate_session":         "validate",
+		"cancel_session":           "cancel",
+		"close_session":            "close",
+		"disconnect":               "close",
+		"close_query_session":      "close",
+		"close_table_read_session": "close",
+		"shutdown":                 "close",
+		"fetch_query_page":         "fetch",
+		"fetch_table_read_page":    "fetch",
+		"list_tables":              "execute",
+	}
+	for method, want := range tests {
+		if got := rpcErrorStage(method); got != want {
+			t.Fatalf("rpcErrorStage(%q) = %q, want %q", method, got, want)
+		}
+	}
+}
+
+func TestXuguSafeRPCDiagnosticIsBoundedASCII(t *testing.T) {
+	got := safeRPCDiagnostic("*errors.errorString\n权限", 12)
+	if got != "*errors.erro" {
+		t.Fatalf("unexpected diagnostic sanitization: %q", got)
+	}
+}
+
+func TestXuguRuntimeErrorsCarryRequestStageAndSession(t *testing.T) {
+	runtime := newRuntimeServer()
+
+	malformed, shutdown := runtime.handleLine(`{"jsonrpc":`)
+	if shutdown || malformed.Error == nil {
+		t.Fatalf("unexpected malformed request response: shutdown=%v response=%+v", shutdown, malformed)
+	}
+	if malformed.Error.Data.Stage != "request" || malformed.Error.Data.OperationOutcome != "not_started" ||
+		malformed.Error.Data.AgentSessionID != "" {
+		t.Fatalf("unexpected malformed request contract: %+v", malformed.Error.Data)
+	}
+
+	missing, shutdown := runtime.handleLine(`{"jsonrpc":"2.0","id":9,"method":"list_tables","params":{"agentSessionId":"missing-session"}}`)
+	if shutdown || missing.Error == nil {
+		t.Fatalf("unexpected missing session response: shutdown=%v response=%+v", shutdown, missing)
+	}
+	if missing.Error.Data.AgentSessionID != "missing-session" || missing.Error.Data.Category != "protocol" ||
+		missing.Error.Data.SessionDisposition != "quarantine" {
+		t.Fatalf("unexpected missing session contract: %+v", missing.Error.Data)
+	}
+}
+
+func TestXuguLegacyHandshakeDoesNotAdvertiseStructuredErrors(t *testing.T) {
+	s := newServer()
+	resp, _ := s.handleLine(`{"jsonrpc":"2.0","id":1,"method":"handshake","params":{}}`)
+	encoded, err := json.Marshal(resp.Result)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if strings.Contains(string(encoded), "structured_error_v1") {
+		t.Fatalf("protocol v1 handshake must not advertise v2 structured errors: %s", encoded)
+	}
+}
+
+func TestXuguLegacyErrorPathsRemainContractCompatible(t *testing.T) {
+	s := newServer()
+
+	malformed, shutdown := s.handleLine(`{"jsonrpc":`)
+	if shutdown || malformed.Error == nil {
+		t.Fatalf("unexpected malformed legacy response: shutdown=%v response=%+v", shutdown, malformed)
+	}
+	assertXuguStructuredErrorContract(t, malformed.Error.Data)
+	if malformed.Error.Data.Stage != "request" || malformed.Error.Data.AgentSessionID != "" {
+		t.Fatalf("unexpected malformed legacy contract: %+v", malformed.Error.Data)
+	}
+
+	dispatch, shutdown := s.handleLine(`{"jsonrpc":"2.0","id":2,"method":"list_tables","params":{}}`)
+	if shutdown || dispatch.Error == nil {
+		t.Fatalf("unexpected legacy dispatch response: shutdown=%v response=%+v", shutdown, dispatch)
+	}
+	assertXuguStructuredErrorContract(t, dispatch.Error.Data)
+	if dispatch.Error.Data.Stage != "execute" || dispatch.Error.Data.Category != "connection" {
+		t.Fatalf("unexpected legacy dispatch contract: %+v", dispatch.Error.Data)
+	}
+}
+
+func assertXuguStructuredErrorContract(t *testing.T, data *rpcErrorData) {
+	t.Helper()
+	if data == nil || data.ContractVersion != 1 {
+		t.Fatalf("missing structured error contract: %+v", data)
+	}
+	expectedOutcome := "unknown"
+	if data.Stage == "request" || data.Stage == "connect" || data.Stage == "validate" {
+		expectedOutcome = "not_started"
+	}
+	if data.OperationOutcome != expectedOutcome {
+		t.Fatalf("stage/outcome violates the Agent contract: %+v", data)
+	}
+	switch data.Category {
+	case "connection":
+		if data.SessionDisposition == "replace_runtime" {
+			t.Fatalf("connection error cannot replace the runtime: %+v", data)
+		}
+	case "sql":
+		if data.Stage != "execute" && data.Stage != "fetch" && data.Stage != "cancel" && data.Stage != "close" {
+			t.Fatalf("sql error has invalid stage: %+v", data)
+		}
+		if data.SessionDisposition == "replace_runtime" {
+			t.Fatalf("sql error cannot replace the runtime: %+v", data)
+		}
+	case "resource":
+		if data.SessionDisposition != "replace_runtime" && data.OperationOutcome != "not_started" {
+			t.Fatalf("resource error has invalid disposition/outcome: %+v", data)
+		}
+	case "protocol":
+		if data.SessionDisposition == "replace_runtime" && data.OperationOutcome == "not_started" {
+			t.Fatalf("protocol error has unsafe disposition/outcome: %+v", data)
+		}
+	case "timeout", "canceled":
+		if data.SessionDisposition != "quarantine" {
+			t.Fatalf("%s error must quarantine its session: %+v", data.Category, data)
+		}
+	default:
+		t.Fatalf("unknown structured error category: %+v", data)
+	}
+}