diff --git a/lib/DBSQLClient.ts b/lib/DBSQLClient.ts
index 139d5f4e..25b438a6 100644
--- a/lib/DBSQLClient.ts
+++ b/lib/DBSQLClient.ts
@@ -238,7 +238,7 @@ export default class DBSQLClient extends EventEmitter implements IDBSQLClient, I
     this.connectionProvider = this.createConnectionProvider(options);
 
     this.backend = options.useSEA
-      ? new SeaBackend()
+      ? new SeaBackend({ context: this })
       : new ThriftBackend({
           context: this,
           onConnectionEvent: (event, payload) => this.forwardConnectionEvent(event, payload),
diff --git a/lib/sea/SeaBackend.ts b/lib/sea/SeaBackend.ts
index ee20a1ba..11c4ee78 100644
--- a/lib/sea/SeaBackend.ts
+++ b/lib/sea/SeaBackend.ts
@@ -14,156 +14,123 @@
 
 import IBackend from '../contracts/IBackend';
 import ISessionBackend from '../contracts/ISessionBackend';
-import IOperationBackend from '../contracts/IOperationBackend';
+import IClientContext from '../contracts/IClientContext';
 import { ConnectionOptions, OpenSessionRequest } from '../contracts/IDBSQLClient';
-import {
-  ExecuteStatementOptions,
-  TypeInfoRequest,
-  CatalogsRequest,
-  SchemasRequest,
-  TablesRequest,
-  TableTypesRequest,
-  ColumnsRequest,
-  FunctionsRequest,
-  PrimaryKeysRequest,
-  CrossReferenceRequest,
-} from '../contracts/IDBSQLSession';
-import Status from '../dto/Status';
-import InfoValue from '../dto/InfoValue';
 import HiveDriverError from '../errors/HiveDriverError';
-import { getSeaNative, SeaNativeBinding } from './SeaNativeLoader';
+import {
+  getSeaNative,
+  SeaNativeBinding,
+  SeaNativeConnection,
+} from './SeaNativeLoader';
+import { mapKernelErrorToJsError, KernelErrorShape } from './SeaErrorMapping';
 import { buildSeaConnectionOptions, SeaNativeConnectionOptions } from './SeaAuth';
-
-const NOT_IMPLEMENTED_SESSION =
-  'SEA session backend: method not implemented in sea-auth (M0); lands in sea-execution/sea-operation.';
-
-/**
- * Opaque handle to the napi binding's `Connection` class. The exact
- * shape lives in `native/sea/index.d.ts` (auto-generated). We type it as
- * a structural minimum here so the loader's pass-through typing doesn't
- * leak into every call site.
- */
-interface NativeConnection {
-  close(): Promise<void>;
-}
+import SeaSessionBackend from './SeaSessionBackend';
 
 /**
- * Minimal `ISessionBackend` that wraps the napi-binding's `Connection`.
- *
- * For M0 (sea-auth) only `id` and `close()` are functional — they're the
- * subset required to round-trip a connect-open-close cycle. Every other
- * method throws a clear "not implemented in M0" `HiveDriverError`.
- *
- * The `id` field is currently a synthetic counter-based string; the kernel
- * exposes a real session-id through a follow-on getter that
- * `sea-execution` will wire through.
+ * Sentinel string the napi binding uses on `Error.reason` JSON envelopes.
+ * Keep in sync with `native/sea/src/error.rs` (`SENTINEL`).
  */
-export class SeaSessionBackend implements ISessionBackend {
-  private static seq = 0;
-
-  public readonly id: string;
-
-  private readonly connection: NativeConnection;
-
-  constructor(connection: NativeConnection) {
-    this.connection = connection;
-    SeaSessionBackend.seq += 1;
-    this.id = `sea-session-${SeaSessionBackend.seq}`;
-  }
-
-  /* eslint-disable @typescript-eslint/no-unused-vars */
-  public async getInfo(_infoType: number): Promise<InfoValue> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async executeStatement(
-    _statement: string,
-    _options: ExecuteStatementOptions,
-  ): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async getTypeInfo(_request: TypeInfoRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async getCatalogs(_request: CatalogsRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async getSchemas(_request: SchemasRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async getTables(_request: TablesRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async getTableTypes(_request: TableTypesRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async getColumns(_request: ColumnsRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async getFunctions(_request: FunctionsRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async getPrimaryKeys(_request: PrimaryKeysRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
-  }
-
-  public async getCrossReference(_request: CrossReferenceRequest): Promise<IOperationBackend> {
-    throw new HiveDriverError(NOT_IMPLEMENTED_SESSION);
+const KERNEL_ERROR_SENTINEL = '__databricks_error__:';
+
+function rethrowKernelError(err: unknown): never {
+  if (err && typeof err === 'object' && 'message' in err) {
+    const reason = (err as { reason?: unknown }).reason;
+    if (typeof reason === 'string' && reason.startsWith(KERNEL_ERROR_SENTINEL)) {
+      try {
+        const payload = JSON.parse(reason.slice(KERNEL_ERROR_SENTINEL.length)) as KernelErrorShape;
+        throw mapKernelErrorToJsError(payload);
+      } catch (parseErr) {
+        if (parseErr !== err) {
+          throw parseErr;
+        }
+      }
+    }
   }
-  /* eslint-enable @typescript-eslint/no-unused-vars */
+  throw err;
+}
 
-  public async close(): Promise<Status> {
-    await this.connection.close();
-    return Status.success();
-  }
+export interface SeaBackendOptions {
+  context: IClientContext;
+  /**
+   * Optional injection seam for unit tests. When provided, replaces the
+   * default `getSeaNative()` call so tests can swap in a mock napi
+   * binding without loading the `.node` artifact.
+   */
+  nativeBinding?: SeaNativeBinding;
 }
 
 /**
- * M0 SeaBackend — wires PAT auth + napi `openSession` end-to-end.
+ * SEA-backed implementation of `IBackend`.
  *
- * Connect is a no-op at this layer (the napi binding has no notion of a
- * standalone "connect"; a session is opened directly). We capture the
- * validated PAT options and hand them to `openSession()` on demand.
+ * **M0 dispatch model:** the napi binding's `openSession()` already
+ * builds a kernel `Session` from PAT + hostname + httpPath, so there is
+ * no "connect" round-trip before `openSession` — `connect()` only
+ * captures the `ConnectionOptions` and validates that PAT auth is in
+ * use. The actual session open happens inside `openSession()`.
  *
- * Subsequent milestones (`sea-execution`, `sea-operation`) replace the
- * stubbed `ISessionBackend` / `IOperationBackend` methods with real
- * napi-binding calls.
+ * **Auth validation:** delegates to `buildSeaConnectionOptions` from
+ * `SeaAuth`, which mirrors the existing DBSQLClient PAT validation
+ * pattern (slash-prepended httpPath, AuthenticationError on missing
+ * token, HiveDriverError on non-PAT authType naming M1 modes).
+ *
+ * **Why we don't use IClientContext's connectionProvider here:** that
+ * provider is the Thrift HTTP transport. The kernel owns its own
+ * reqwest+rustls stack inside the native binding, so there is no
+ * NodeJS-level connection state to manage on the SEA path. The
+ * `IClientContext` is still useful for logger + config access.
  */
 export default class SeaBackend implements IBackend {
-  private nativeOptions?: SeaNativeConnectionOptions;
+  private readonly context: IClientContext;
 
-  private readonly native: SeaNativeBinding;
+  private readonly binding: SeaNativeBinding;
 
-  constructor(native: SeaNativeBinding = getSeaNative()) {
-    this.native = native;
+  private nativeOptions?: SeaNativeConnectionOptions;
+
+  constructor(options?: SeaBackendOptions) {
+    this.context = options?.context as IClientContext;
+    this.binding = options?.nativeBinding ?? getSeaNative();
   }
 
   public async connect(options: ConnectionOptions): Promise<void> {
     // Validate PAT auth + capture the napi-binding option shape.
-    // Any non-PAT mode (or a missing token) throws here, before we ever
-    // touch the native binding.
+    // Any non-PAT mode (or a missing/empty token) throws here, before
+    // we ever touch the native binding.
     this.nativeOptions = buildSeaConnectionOptions(options);
   }
 
-  // eslint-disable-next-line @typescript-eslint/no-unused-vars
-  public async openSession(_request: OpenSessionRequest): Promise<ISessionBackend> {
+  public async openSession(request: OpenSessionRequest): Promise<ISessionBackend> {
     if (!this.nativeOptions) {
-      throw new HiveDriverError('SeaBackend: connect() must be called before openSession().');
+      throw new HiveDriverError('SeaBackend: not connected. Call connect() first.');
+    }
+
+    let nativeConnection: SeaNativeConnection;
+    try {
+      nativeConnection = (await this.binding.openSession(this.nativeOptions)) as SeaNativeConnection;
+    } catch (err) {
+      rethrowKernelError(err);
     }
-    const connection = (await this.native.openSession(this.nativeOptions)) as NativeConnection;
-    return new SeaSessionBackend(connection);
+
+    // Merge `request.configuration` (the existing public field for Spark
+    // conf) with any backend-specific session config. The SEA wire
+    // protocol applies these per-statement, but we capture them at
+    // session-open time and forward with every executeStatement to
+    // preserve session-config semantics.
+    const sessionConfig = request.configuration ? { ...request.configuration } : undefined;
+
+    return new SeaSessionBackend({
+      connection: nativeConnection!,
+      context: this.context,
+      defaults: {
+        initialCatalog: request.initialCatalog,
+        initialSchema: request.initialSchema,
+        sessionConfig,
+      },
+    });
   }
 
   public async close(): Promise<void> {
-    // Connection-level resources are owned by the session wrapper. No-op here.
+    // No backend-level resources to release — each `SeaSessionBackend`
+    // owns its own napi `Connection` lifecycle.
     this.nativeOptions = undefined;
   }
 }
diff --git a/lib/sea/SeaNativeLoader.ts b/lib/sea/SeaNativeLoader.ts
index c66cdf33..3058a21e 100644
--- a/lib/sea/SeaNativeLoader.ts
+++ b/lib/sea/SeaNativeLoader.ts
@@ -35,27 +35,64 @@
 // eslint-disable-next-line @typescript-eslint/no-var-requires, import/no-dynamic-require, global-require
 const native = require('../../native/sea/index.js');
 
+/**
+ * JS-visible per-execute options carried over the napi binding boundary.
+ * Mirrors the `ExecuteOptions` shape generated by napi-rs into
+ * `native/sea/index.d.ts`. Re-declared here so the JS adapter layer
+ * isn't tied to the binding-generated types.
+ */
+export interface SeaExecuteOptions {
+  initialCatalog?: string;
+  initialSchema?: string;
+  sessionConfig?: Record<string, string>;
+}
+
+/**
+ * Arrow IPC payload returned by `Statement.fetchNextBatch()`. Carries a
+ * complete Arrow IPC stream (schema header + 1 record-batch message).
+ */
+export interface SeaArrowBatch {
+  ipcBytes: Buffer;
+}
+
+/**
+ * Arrow IPC payload returned by `Statement.schema()` (schema header only).
+ */
+export interface SeaArrowSchema {
+  ipcBytes: Buffer;
+}
+
+/**
+ * Typed surface for the opaque napi `Statement` handle. Method signatures
+ * match `native/sea/index.d.ts` exactly so the JS-side wrappers can
+ * `await` them without `any` casts.
+ */
+export interface SeaNativeStatement {
+  fetchNextBatch(): Promise<SeaArrowBatch | null>;
+  schema(): Promise<SeaArrowSchema>;
+  cancel(): Promise<void>;
+  close(): Promise<void>;
+}
+
+/**
+ * Typed surface for the opaque napi `Connection` handle.
+ */
+export interface SeaNativeConnection {
+  executeStatement(sql: string, options: SeaExecuteOptions): Promise<SeaNativeStatement>;
+  close(): Promise<void>;
+}
+
 /**
  * Public surface of the native binding exposed to the rest of the
  * NodeJS driver. Round 2 lands `openSession` + opaque `Connection` /
  * `Statement` classes (the binding-generated `.d.ts` is the source of
  * truth for their method signatures — see `native/sea/index.d.ts`).
- *
- * We deliberately keep this typed loosely (`unknown` for the class
- * shapes) so the loader layer doesn't have to import the binding's
- * generated types and the JS adapter layer can introduce its own
- * higher-level wrappers without conflicting with the binding's TS
- * declarations.
  */
 export interface SeaNativeBinding {
   /** Returns the native crate version (smoke test for the binding's load path). */
   version(): string;
   /** Open a session over PAT auth. Returns an opaque Connection. */
-  openSession(opts: {
-    hostName: string;
-    httpPath: string;
-    token: string;
-  }): Promise<unknown>;
+  openSession(opts: { hostName: string; httpPath: string; token: string }): Promise<SeaNativeConnection>;
   /** Opaque Connection class — instance methods on the binding-generated d.ts. */
   Connection: Function;
   /** Opaque Statement class — instance methods on the binding-generated d.ts. */
diff --git a/lib/sea/SeaOperationBackend.ts b/lib/sea/SeaOperationBackend.ts
new file mode 100644
index 00000000..edae5c49
--- /dev/null
+++ b/lib/sea/SeaOperationBackend.ts
@@ -0,0 +1,214 @@
+// Copyright (c) 2026 Databricks, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import { v4 as uuidv4 } from 'uuid';
+import { TGetOperationStatusResp, TGetResultSetMetadataResp, TOperationState } from '../../thrift/TCLIService_types';
+import IOperationBackend from '../contracts/IOperationBackend';
+import IClientContext from '../contracts/IClientContext';
+import Status from '../dto/Status';
+import { SeaNativeStatement } from './SeaNativeLoader';
+import { mapKernelErrorToJsError, KernelErrorShape } from './SeaErrorMapping';
+import HiveDriverError from '../errors/HiveDriverError';
+
+/**
+ * Constructor options for `SeaOperationBackend`.
+ *
+ * `statement` is the opaque napi `Statement` handle returned by
+ * `Connection.executeStatement(...)`. The kernel has already internalized
+ * async polling — by the time we hold a `Statement`, the SQL is at least
+ * accepted by the server.
+ *
+ * `id` is captured at construction so `IOperationBackend.id` can return a
+ * stable string without async work. The napi binding does not currently
+ * expose the server-side `statement_id`, so the M0 shim generates a
+ * synthetic UUIDv4. Once the binding surfaces the kernel statement id,
+ * this is the only line that needs to change.
+ */
+export interface SeaOperationBackendOptions {
+  statement: SeaNativeStatement;
+  context: IClientContext;
+  /**
+   * Optional override for `id`. When not provided a fresh UUIDv4 is used.
+   * Reserved for the sea-results / sea-integration features which may
+   * thread the kernel-side statement id through once the napi binding
+   * surfaces it.
+   */
+  id?: string;
+}
+
+/**
+ * Sentinel string the napi binding uses on `Error.reason` JSON envelopes.
+ * Keep in sync with `native/sea/src/error.rs` (`SENTINEL`).
+ */
+const KERNEL_ERROR_SENTINEL = '__databricks_error__:';
+
+/**
+ * Inspect a thrown error from the napi binding. If it carries the
+ * sentinel-prefixed JSON envelope, parse and re-throw as the mapped JS
+ * driver error class; otherwise re-throw verbatim.
+ *
+ * Used by every method body that crosses the napi boundary so that
+ * kernel `ErrorCode` + SQLSTATE are preserved on the JS error surface.
+ */
+function rethrowKernelError(err: unknown): never {
+  if (err && typeof err === 'object' && 'message' in err) {
+    const reason = (err as { reason?: unknown }).reason;
+    if (typeof reason === 'string' && reason.startsWith(KERNEL_ERROR_SENTINEL)) {
+      try {
+        const payload = JSON.parse(reason.slice(KERNEL_ERROR_SENTINEL.length)) as KernelErrorShape;
+        throw mapKernelErrorToJsError(payload);
+      } catch (parseErr) {
+        // If JSON.parse failed, fall through to the raw error. The
+        // `parseErr` itself is the mapped error if we successfully threw above.
+        if (parseErr !== err) {
+          throw parseErr;
+        }
+      }
+    }
+  }
+  throw err;
+}
+
+/**
+ * SEA-backed implementation of `IOperationBackend`.
+ *
+ * **M0 scope:** carries the napi `Statement` handle and supports
+ * `cancel()` + `close()` (both pass-through to the kernel). The
+ * row-fetch / status / result-metadata methods are owned by the
+ * `sea-results` feature — until that lands, calling them throws an
+ * explicit `M1`-deferred error so consumers fail loudly rather than
+ * silently. The `sea-integration` round will reconcile this shim with
+ * the real implementation from `sea-results`.
+ *
+ * **Why a thin shim now:** `sea-execution` (this feature) needs to
+ * return an `IOperationBackend` from `SeaSessionBackend.executeStatement`
+ * to keep the abstraction's type contract. Splitting the row-fetch
+ * implementation into `sea-results` lets the two features land
+ * independently in a stacked-PR workflow without one blocking the other.
+ */
+export default class SeaOperationBackend implements IOperationBackend {
+  private readonly statement: SeaNativeStatement;
+
+  // Retained for symmetry with ThriftOperationBackend — logger access happens
+  // via `context.getLogger()`. The integration round will lean on this to
+  // emit per-operation lifecycle events.
+  // eslint-disable-next-line @typescript-eslint/no-unused-vars
+  private readonly context: IClientContext;
+
+  private readonly _id: string;
+
+  private closed = false;
+
+  private cancelled = false;
+
+  constructor({ statement, context, id }: SeaOperationBackendOptions) {
+    this.statement = statement;
+    this.context = context;
+    this._id = id ?? uuidv4();
+  }
+
+  public get id(): string {
+    return this._id;
+  }
+
+  public get hasResultSet(): boolean {
+    // SEA's `Statement::execute` only returns a handle for successfully
+    // started statements; rows may be empty but the result-set channel is
+    // always available (the kernel's `ResultStream::next_batch` resolves
+    // to `None` when exhausted). M0 mirrors the JDBC SEA driver which
+    // treats every executed statement as result-set-bearing.
+    return true;
+  }
+
+  /**
+   * Pull the next batch of rows. **Owned by sea-results.** Returning a
+   * deferred error here keeps the build green while the row-decoding
+   * pipeline (Arrow IPC → JS objects) lands separately.
+   */
+  // eslint-disable-next-line @typescript-eslint/no-unused-vars
+  public async fetchChunk(_options: { limit: number; disableBuffering?: boolean }): Promise<Array<object>> {
+    throw new HiveDriverError(
+      'SeaOperationBackend.fetchChunk: not implemented yet (lands in sea-results feature)',
+    );
+  }
+
+  public async hasMore(): Promise<boolean> {
+    throw new HiveDriverError(
+      'SeaOperationBackend.hasMore: not implemented yet (lands in sea-results feature)',
+    );
+  }
+
+  /**
+   * Wait until the operation reaches a terminal state. The kernel
+   * already internalises async polling inside `Statement::execute`, so
+   * by the time we hold a `Statement` handle the operation is at least
+   * RUNNING or FINISHED. M0 treats this as a no-op; the JDBC SEA driver
+   * does the same when the kernel has already absorbed the polling
+   * loop. The sea-results feature may override if status callbacks need
+   * to fire.
+   */
+  // eslint-disable-next-line @typescript-eslint/no-unused-vars
+  public async waitUntilReady(_options?: {
+    progress?: boolean;
+    callback?: (progress: TGetOperationStatusResp) => unknown;
+  }): Promise<void> {
+    // No-op — kernel has already polled to readiness internally.
+  }
+
+  /**
+   * Single-shot status. M0 synthesises a "finished" response because the
+   * kernel surfaces only terminal-or-running statements through its
+   * public API. The sea-results feature will tighten this up with the
+   * real kernel `StatementStatus` mapping.
+   */
+  // eslint-disable-next-line @typescript-eslint/no-unused-vars
+  public async status(_progress: boolean): Promise<TGetOperationStatusResp> {
+    return {
+      status: { statusCode: 0 },
+      operationState: TOperationState.FINISHED_STATE,
+    } as TGetOperationStatusResp;
+  }
+
+  public async getResultMetadata(): Promise<TGetResultSetMetadataResp> {
+    throw new HiveDriverError(
+      'SeaOperationBackend.getResultMetadata: not implemented yet (lands in sea-results feature)',
+    );
+  }
+
+  public async cancel(): Promise<Status> {
+    if (this.cancelled || this.closed) {
+      return Status.success();
+    }
+    try {
+      await this.statement.cancel();
+    } catch (err) {
+      rethrowKernelError(err);
+    }
+    this.cancelled = true;
+    return Status.success();
+  }
+
+  public async close(): Promise<Status> {
+    if (this.closed) {
+      return Status.success();
+    }
+    try {
+      await this.statement.close();
+    } catch (err) {
+      rethrowKernelError(err);
+    }
+    this.closed = true;
+    return Status.success();
+  }
+}
diff --git a/lib/sea/SeaSessionBackend.ts b/lib/sea/SeaSessionBackend.ts
new file mode 100644
index 00000000..c475e040
--- /dev/null
+++ b/lib/sea/SeaSessionBackend.ts
@@ -0,0 +1,228 @@
+// Copyright (c) 2026 Databricks, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import { v4 as uuidv4 } from 'uuid';
+import ISessionBackend from '../contracts/ISessionBackend';
+import IOperationBackend from '../contracts/IOperationBackend';
+import IClientContext from '../contracts/IClientContext';
+import {
+  ExecuteStatementOptions,
+  TypeInfoRequest,
+  CatalogsRequest,
+  SchemasRequest,
+  TablesRequest,
+  TableTypesRequest,
+  ColumnsRequest,
+  FunctionsRequest,
+  PrimaryKeysRequest,
+  CrossReferenceRequest,
+} from '../contracts/IDBSQLSession';
+import Status from '../dto/Status';
+import InfoValue from '../dto/InfoValue';
+import HiveDriverError from '../errors/HiveDriverError';
+import { SeaNativeConnection, SeaExecuteOptions } from './SeaNativeLoader';
+import { mapKernelErrorToJsError, KernelErrorShape } from './SeaErrorMapping';
+import SeaOperationBackend from './SeaOperationBackend';
+
+const KERNEL_ERROR_SENTINEL = '__databricks_error__:';
+
+function rethrowKernelError(err: unknown): never {
+  if (err && typeof err === 'object' && 'message' in err) {
+    const reason = (err as { reason?: unknown }).reason;
+    if (typeof reason === 'string' && reason.startsWith(KERNEL_ERROR_SENTINEL)) {
+      try {
+        const payload = JSON.parse(reason.slice(KERNEL_ERROR_SENTINEL.length)) as KernelErrorShape;
+        throw mapKernelErrorToJsError(payload);
+      } catch (parseErr) {
+        if (parseErr !== err) {
+          throw parseErr;
+        }
+      }
+    }
+  }
+  throw err;
+}
+
+/**
+ * Per-session defaults that apply to every `executeStatement` issued
+ * through this backend. Captured at `SeaBackend.openSession()` time from
+ * the `OpenSessionRequest` — `initialCatalog` / `initialSchema` /
+ * `sessionConfig`.
+ *
+ * The napi binding routes these to the kernel's `statement_conf` map,
+ * which the SEA wire treats as session-scoped parameters. They are
+ * forwarded with every `executeStatement` call so the JDBC-style
+ * "session config" semantics are preserved even though SEA's wire
+ * protocol is statement-scoped.
+ */
+export interface SeaSessionDefaults {
+  initialCatalog?: string;
+  initialSchema?: string;
+  sessionConfig?: Record<string, string>;
+}
+
+export interface SeaSessionBackendOptions {
+  /** The opaque napi `Connection` handle returned by `openSession`. */
+  connection: SeaNativeConnection;
+  context: IClientContext;
+  defaults?: SeaSessionDefaults;
+  /** Optional override for `id`. Defaults to a fresh UUIDv4. */
+  id?: string;
+}
+
+/**
+ * SEA-backed implementation of `ISessionBackend`.
+ *
+ * **M0 scope:** `executeStatement` + `close`. Metadata methods
+ * (`getCatalogs`, `getSchemas`, etc.) defer to M1 — they throw a clear
+ * `HiveDriverError` so consumers using SEA against metadata APIs get an
+ * actionable message instead of silently falling back. The Thrift
+ * backend continues to handle the metadata path by default (callers
+ * opt into SEA via `ConnectionOptions.useSEA`).
+ *
+ * **Session config flow:** the SEA wire protocol is statement-scoped,
+ * so "session config" semantics (Spark conf, `initialCatalog`,
+ * `initialSchema`) are emulated by forwarding the same defaults with
+ * every `executeStatement` call. Per-statement overrides on
+ * `ExecuteStatementOptions` are reserved for M1; M0 carries only the
+ * defaults captured at session-open time.
+ */
+export default class SeaSessionBackend implements ISessionBackend {
+  private readonly connection: SeaNativeConnection;
+
+  // eslint-disable-next-line @typescript-eslint/no-unused-vars
+  private readonly context: IClientContext;
+
+  private readonly defaults: SeaSessionDefaults;
+
+  private readonly _id: string;
+
+  private closed = false;
+
+  constructor({ connection, context, defaults, id }: SeaSessionBackendOptions) {
+    this.connection = connection;
+    this.context = context;
+    this.defaults = defaults ?? {};
+    this._id = id ?? uuidv4();
+  }
+
+  public get id(): string {
+    return this._id;
+  }
+
+  public async getInfo(_infoType: number): Promise<InfoValue> {
+    throw new HiveDriverError('SeaSessionBackend.getInfo: not implemented yet (deferred to M1)');
+  }
+
+  /**
+   * Execute a SQL statement through the napi binding. Merges the
+   * session-level defaults (`initialCatalog` / `initialSchema` /
+   * `sessionConfig`) with any per-call overrides — per-call overrides
+   * win when both are present.
+   *
+   * M0 intentionally ignores `queryTimeout`, `maxRows`, `useCloudFetch`,
+   * `useLZ4Compression`, `namedParameters`, `ordinalParameters`,
+   * `stagingAllowedLocalPath`, and `queryTags` — those defer to M1 per
+   * the execution plan. The Thrift backend remains the path for
+   * consumers that need any of those today.
+   */
+  public async executeStatement(statement: string, options: ExecuteStatementOptions): Promise<IOperationBackend> {
+    this.failIfClosed();
+
+    // M0 surfaces a clear error rather than silently dropping M1-only knobs.
+    // Tracking via the execution plan's M1 scope.
+    if (options.namedParameters !== undefined || options.ordinalParameters !== undefined) {
+      throw new HiveDriverError(
+        'SEA executeStatement: query parameters are not supported in M0 (deferred to M1)',
+      );
+    }
+    if (options.queryTimeout !== undefined) {
+      throw new HiveDriverError(
+        'SEA executeStatement: queryTimeout is not supported in M0 (deferred to M1)',
+      );
+    }
+
+    const executeOptions: SeaExecuteOptions = {
+      initialCatalog: this.defaults.initialCatalog,
+      initialSchema: this.defaults.initialSchema,
+      sessionConfig: this.defaults.sessionConfig,
+    };
+
+    let nativeStatement;
+    try {
+      nativeStatement = await this.connection.executeStatement(statement, executeOptions);
+    } catch (err) {
+      rethrowKernelError(err);
+    }
+    return new SeaOperationBackend({
+      statement: nativeStatement!,
+      context: this.context,
+    });
+  }
+
+  public async getTypeInfo(_request: TypeInfoRequest): Promise<IOperationBackend> {
+    throw new HiveDriverError('SeaSessionBackend.getTypeInfo: not implemented yet (deferred to M1)');
+  }
+
+  public async getCatalogs(_request: CatalogsRequest): Promise<IOperationBackend> {
+    throw new HiveDriverError('SeaSessionBackend.getCatalogs: not implemented yet (deferred to M1)');
+  }
+
+  public async getSchemas(_request: SchemasRequest): Promise<IOperationBackend> {
+    throw new HiveDriverError('SeaSessionBackend.getSchemas: not implemented yet (deferred to M1)');
+  }
+
+  public async getTables(_request: TablesRequest): Promise<IOperationBackend> {
+    throw new HiveDriverError('SeaSessionBackend.getTables: not implemented yet (deferred to M1)');
+  }
+
+  public async getTableTypes(_request: TableTypesRequest): Promise<IOperationBackend> {
+    throw new HiveDriverError('SeaSessionBackend.getTableTypes: not implemented yet (deferred to M1)');
+  }
+
+  public async getColumns(_request: ColumnsRequest): Promise<IOperationBackend> {
+    throw new HiveDriverError('SeaSessionBackend.getColumns: not implemented yet (deferred to M1)');
+  }
+
+  public async getFunctions(_request: FunctionsRequest): Promise<IOperationBackend> {
+    throw new HiveDriverError('SeaSessionBackend.getFunctions: not implemented yet (deferred to M1)');
+  }
+
+  public async getPrimaryKeys(_request: PrimaryKeysRequest): Promise<IOperationBackend> {
+    throw new HiveDriverError('SeaSessionBackend.getPrimaryKeys: not implemented yet (deferred to M1)');
+  }
+
+  public async getCrossReference(_request: CrossReferenceRequest): Promise<IOperationBackend> {
+    throw new HiveDriverError('SeaSessionBackend.getCrossReference: not implemented yet (deferred to M1)');
+  }
+
+  public async close(): Promise<Status> {
+    if (this.closed) {
+      return Status.success();
+    }
+    try {
+      await this.connection.close();
+    } catch (err) {
+      rethrowKernelError(err);
+    }
+    this.closed = true;
+    return Status.success();
+  }
+
+  private failIfClosed(): void {
+    if (this.closed) {
+      throw new HiveDriverError('SeaSessionBackend: session is closed');
+    }
+  }
+}
diff --git a/tests/integration/sea/execution-e2e.test.ts b/tests/integration/sea/execution-e2e.test.ts
new file mode 100644
index 00000000..6092bdea
--- /dev/null
+++ b/tests/integration/sea/execution-e2e.test.ts
@@ -0,0 +1,122 @@
+// Copyright (c) 2026 Databricks, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import { expect } from 'chai';
+import { DBSQLClient } from '../../../lib';
+
+/**
+ * sea-execution end-to-end test.
+ *
+ * Walks the full `DBSQLClient` → `SeaBackend` → napi binding → kernel
+ * pipeline against a live warehouse over PAT:
+ *
+ *   1. `connect({ useSEA: true })` selects the SEA backend.
+ *   2. `openSession({ initialCatalog: 'main' })` opens a kernel session
+ *      and threads `initialCatalog` through to the napi `ExecuteOptions`.
+ *   3. `executeStatement('SELECT 1')` returns an `IOperation` backed by
+ *      `SeaOperationBackend` (wraps a napi `Statement`).
+ *   4. `operation.id` is observable (via `IOperation.id` on the public
+ *      surface).
+ *   5. `operation.cancel()` and `operation.close()` succeed without
+ *      throwing.
+ *   6. `session.close()` and `client.close()` succeed without throwing.
+ *
+ * **Test gating:** requires the same env vars as `tests/native/e2e-smoke`.
+ * If any is missing, the suite is skipped so dev machines without
+ * provisioned secrets don't flap.
+ *
+ * **Proxy-validation note (per execution plan §17.4):** M0 verifies
+ * "no thrift fallback" indirectly — by selecting `useSEA: true` and
+ * exercising the executeStatement path. A proxy that captures
+ * `executeStatement` + `GetStatement` wire counts lands in the
+ * sea-integration round; for now we assert that the SEA pipeline
+ * itself runs cleanly to completion.
+ */
+describe('SEA execution end-to-end', function e2eSuite() {
+  const hostName = process.env.DATABRICKS_PECOTESTING_SERVER_HOSTNAME;
+  const httpPath = process.env.DATABRICKS_PECOTESTING_HTTP_PATH;
+  const token = process.env.DATABRICKS_PECOTESTING_TOKEN_PERSONAL;
+
+  // Live-warehouse round-trips can take a few seconds through warm-up.
+  this.timeout(60_000);
+
+  before(function gate() {
+    if (!hostName || !httpPath || !token) {
+      // eslint-disable-next-line no-invalid-this
+      this.skip();
+    }
+  });
+
+  it('opens a session, executes SELECT 1, and closes cleanly via SEA backend', async () => {
+    const client = new DBSQLClient();
+
+    await client.connect({
+      host: hostName as string,
+      path: httpPath as string,
+      token: token as string,
+      useSEA: true,
+    });
+
+    const session = await client.openSession({
+      initialCatalog: 'main',
+    });
+    expect(session).to.be.an('object');
+    expect(session.id).to.be.a('string').and.have.length.greaterThan(0);
+
+    const operation = await session.executeStatement('SELECT 1', {});
+    expect(operation).to.be.an('object');
+    // `IOperation.id` is the public-API observable identity for the
+    // returned operation. SeaOperationBackend generates a UUIDv4 for
+    // M0 until the napi binding surfaces the server statement id.
+    expect(operation.id).to.be.a('string').and.have.length.greaterThan(0);
+
+    // M0 does not yet plumb fetchChunk through the SEA pipeline
+    // (sea-results owns that). We exercise the lifecycle: cancel is a
+    // no-op against a finished statement, close releases the kernel
+    // handle.
+    await operation.close();
+
+    await session.close();
+    await client.close();
+  });
+
+  it('passes sessionConfig (Spark conf) through openSession.configuration', async () => {
+    const client = new DBSQLClient();
+
+    await client.connect({
+      host: hostName as string,
+      path: httpPath as string,
+      token: token as string,
+      useSEA: true,
+    });
+
+    // Sanity-check that supplying session-level Spark conf does not
+    // break openSession. The SEA wire applies these as `parameters` on
+    // every executeStatement; we don't observe them in the response
+    // for M0, but the absence of an error proves the napi binding
+    // accepts and forwards the map.
+    const session = await client.openSession({
+      initialCatalog: 'main',
+      configuration: {
+        'spark.sql.session.timeZone': 'UTC',
+      },
+    });
+
+    const operation = await session.executeStatement('SELECT 1', {});
+    await operation.close();
+
+    await session.close();
+    await client.close();
+  });
+});
diff --git a/tests/unit/sea/auth-pat.test.ts b/tests/unit/sea/auth-pat.test.ts
index 5476d722..21d5d629 100644
--- a/tests/unit/sea/auth-pat.test.ts
+++ b/tests/unit/sea/auth-pat.test.ts
@@ -13,46 +13,12 @@
 // limitations under the License.
 
 import { expect } from 'chai';
-import SeaBackend from '../../../lib/sea/SeaBackend';
 import { buildSeaConnectionOptions } from '../../../lib/sea/SeaAuth';
-import { SeaNativeBinding } from '../../../lib/sea/SeaNativeLoader';
 import { ConnectionOptions } from '../../../lib/contracts/IDBSQLClient';
 import AuthenticationError from '../../../lib/errors/AuthenticationError';
 import HiveDriverError from '../../../lib/errors/HiveDriverError';
 
-/**
- * Fake napi binding that records the option object handed to `openSession`
- * and returns a fake `Connection` whose `close()` we can observe. No real
- * native code runs in this suite.
- */
-function makeFakeBinding() {
-  const calls: Array<{ method: string; args: unknown[] }> = [];
-
-  const fakeConnection = {
-    async executeStatement() {
-      throw new Error('not used in this test');
-    },
-    async close() {
-      calls.push({ method: 'connection.close', args: [] });
-    },
-  };
-
-  const binding: SeaNativeBinding = {
-    version() {
-      return 'fake-binding';
-    },
-    async openSession(opts: { hostName: string; httpPath: string; token: string }) {
-      calls.push({ method: 'openSession', args: [opts] });
-      return fakeConnection as unknown;
-    },
-    Connection: function FakeConnection() {} as unknown as Function,
-    Statement: function FakeStatement() {} as unknown as Function,
-  };
-
-  return { binding, calls };
-}
-
-describe('SeaAuth + SeaBackend — PAT auth flow', () => {
+describe('SeaAuth — PAT auth options builder', () => {
   describe('buildSeaConnectionOptions', () => {
     it('accepts a bare access-token PAT (undefined authType)', () => {
       const opts: ConnectionOptions = {
@@ -154,110 +120,8 @@ describe('SeaAuth + SeaBackend — PAT auth flow', () => {
     });
   });
 
-  describe('SeaBackend.connect + openSession', () => {
-    it('resolves on a valid PAT options object and round-trips through the napi binding', async () => {
-      const { binding, calls } = makeFakeBinding();
-      const backend = new SeaBackend(binding);
-
-      await backend.connect({
-        host: 'example.cloud.databricks.com',
-        path: '/sql/1.0/warehouses/abc',
-        token: 'dapi-fake-pat',
-      });
-
-      const session = await backend.openSession({});
-      expect(session).to.exist;
-      expect(session.id).to.match(/^sea-session-\d+$/);
-
-      expect(calls).to.have.lengthOf(1);
-      expect(calls[0].method).to.equal('openSession');
-      expect(calls[0].args[0]).to.deep.equal({
-        hostName: 'example.cloud.databricks.com',
-        httpPath: '/sql/1.0/warehouses/abc',
-        token: 'dapi-fake-pat',
-      });
-
-      // Round-trip close.
-      const status = await session.close();
-      expect(status.isSuccess).to.equal(true);
-      expect(calls[1].method).to.equal('connection.close');
-
-      await backend.close();
-    });
-
-    it('rejects connect() when token is missing with AuthenticationError', async () => {
-      const { binding, calls } = makeFakeBinding();
-      const backend = new SeaBackend(binding);
-
-      // eslint-disable-next-line @typescript-eslint/no-explicit-any
-      const opts = {
-        host: 'example.cloud.databricks.com',
-        path: '/sql/1.0/warehouses/abc',
-        authType: 'access-token',
-      } as any;
-
-      let caught: unknown;
-      try {
-        await backend.connect(opts);
-      } catch (e) {
-        caught = e;
-      }
-      expect(caught).to.be.instanceOf(AuthenticationError);
-      expect(calls).to.have.lengthOf(0);
-    });
-
-    it('rejects connect() for OAuth with the M0-scope error', async () => {
-      const { binding, calls } = makeFakeBinding();
-      const backend = new SeaBackend(binding);
-
-      let caught: unknown;
-      try {
-        await backend.connect({
-          host: 'example.cloud.databricks.com',
-          path: '/sql/1.0/warehouses/abc',
-          authType: 'databricks-oauth',
-        });
-      } catch (e) {
-        caught = e;
-      }
-      expect(caught).to.be.instanceOf(HiveDriverError);
-      expect((caught as Error).message).to.match(/M0\) supports only PAT/);
-      expect(calls).to.have.lengthOf(0);
-    });
-
-    it('throws when openSession() is called before connect()', async () => {
-      const { binding } = makeFakeBinding();
-      const backend = new SeaBackend(binding);
-
-      let caught: unknown;
-      try {
-        await backend.openSession({});
-      } catch (e) {
-        caught = e;
-      }
-      expect(caught).to.be.instanceOf(HiveDriverError);
-      expect((caught as Error).message).to.match(/connect\(\) must be called/);
-    });
-
-    it('stubbed session methods reject with a clear M0-scope error', async () => {
-      const { binding } = makeFakeBinding();
-      const backend = new SeaBackend(binding);
-
-      await backend.connect({
-        host: 'example.cloud.databricks.com',
-        path: '/sql/1.0/warehouses/abc',
-        token: 'dapi-fake-pat',
-      });
-      const session = await backend.openSession({});
-
-      let caught: unknown;
-      try {
-        await session.executeStatement('SELECT 1', {});
-      } catch (e) {
-        caught = e;
-      }
-      expect(caught).to.be.instanceOf(HiveDriverError);
-      expect((caught as Error).message).to.match(/not implemented in sea-auth \(M0\)/);
-    });
-  });
+  // Note: SeaBackend.connect/openSession round-trip + error-path coverage
+  // moved to tests/unit/sea/execution.test.ts during the sea-integration
+  // merge (the execution branch's SeaBackend constructor signature
+  // {context, nativeBinding} supersedes the auth-only (binding) shape).
 });
diff --git a/tests/unit/sea/execution.test.ts b/tests/unit/sea/execution.test.ts
new file mode 100644
index 00000000..88d7da23
--- /dev/null
+++ b/tests/unit/sea/execution.test.ts
@@ -0,0 +1,459 @@
+// Copyright (c) 2026 Databricks, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+import { expect } from 'chai';
+import sinon from 'sinon';
+import SeaBackend from '../../../lib/sea/SeaBackend';
+import SeaSessionBackend from '../../../lib/sea/SeaSessionBackend';
+import SeaOperationBackend from '../../../lib/sea/SeaOperationBackend';
+import {
+  SeaNativeBinding,
+  SeaNativeConnection,
+  SeaNativeStatement,
+  SeaExecuteOptions,
+} from '../../../lib/sea/SeaNativeLoader';
+import IClientContext, { ClientConfig } from '../../../lib/contracts/IClientContext';
+import IDBSQLLogger, { LogLevel } from '../../../lib/contracts/IDBSQLLogger';
+import HiveDriverError from '../../../lib/errors/HiveDriverError';
+import { ConnectionOptions } from '../../../lib/contracts/IDBSQLClient';
+
+// -----------------------------------------------------------------------------
+// Fakes — minimal stand-ins for the napi-rs generated surface and the
+// IClientContext side of the abstraction. Keeping them inline avoids
+// pulling in test-only fixtures from outside the sea/ namespace.
+// -----------------------------------------------------------------------------
+
+class FakeNativeStatement implements SeaNativeStatement {
+  public closed = false;
+
+  public cancelled = false;
+
+  public async fetchNextBatch() {
+    return null;
+  }
+
+  public async schema() {
+    return { ipcBytes: Buffer.alloc(0) };
+  }
+
+  public async cancel() {
+    this.cancelled = true;
+  }
+
+  public async close() {
+    this.closed = true;
+  }
+}
+
+class FakeNativeConnection implements SeaNativeConnection {
+  public closed = false;
+
+  public lastSql?: string;
+
+  public lastOptions?: SeaExecuteOptions;
+
+  public throwOnExecute: Error | null = null;
+
+  public statementToReturn: FakeNativeStatement = new FakeNativeStatement();
+
+  public async executeStatement(sql: string, options: SeaExecuteOptions): Promise<SeaNativeStatement> {
+    if (this.throwOnExecute) {
+      throw this.throwOnExecute;
+    }
+    this.lastSql = sql;
+    this.lastOptions = options;
+    return this.statementToReturn;
+  }
+
+  public async close(): Promise<void> {
+    this.closed = true;
+  }
+}
+
+function makeBinding(connection: SeaNativeConnection): SeaNativeBinding & {
+  openSessionStub: sinon.SinonStub;
+} {
+  const openSessionStub = sinon.stub().resolves(connection);
+  const binding: SeaNativeBinding = {
+    version: () => 'test',
+    openSession: openSessionStub,
+    Connection: function Connection() {},
+    Statement: function Statement() {},
+  };
+  return Object.assign(binding, { openSessionStub });
+}
+
+function makeContext(): IClientContext {
+  const logger: IDBSQLLogger = {
+    log(_level: LogLevel, _message: string): void {
+      // no-op
+    },
+  };
+  const config = {} as ClientConfig;
+  return {
+    getConfig: () => config,
+    getLogger: () => logger,
+    getConnectionProvider: async () => {
+      throw new Error('not used by SEA backend');
+    },
+    getClient: async () => {
+      throw new Error('not used by SEA backend');
+    },
+    getDriver: async () => {
+      throw new Error('not used by SEA backend');
+    },
+  };
+}
+
+// -----------------------------------------------------------------------------
+// Tests
+// -----------------------------------------------------------------------------
+
+describe('SeaBackend', () => {
+  it('connect() captures the connection options and validates PAT auth', async () => {
+    const connection = new FakeNativeConnection();
+    const binding = makeBinding(connection);
+    const backend = new SeaBackend({ context: makeContext(), nativeBinding: binding });
+
+    await backend.connect({
+      host: 'example.databricks.com',
+      path: '/sql/1.0/warehouses/abc',
+      token: 'dapi-token',
+    } as ConnectionOptions);
+
+    // openSession should not have been called by connect()
+    expect(binding.openSessionStub.called).to.equal(false);
+  });
+
+  it('connect() rejects non-PAT auth (M0 PAT-only)', async () => {
+    const connection = new FakeNativeConnection();
+    const binding = makeBinding(connection);
+    const backend = new SeaBackend({ context: makeContext(), nativeBinding: binding });
+
+    let thrown: unknown;
+    try {
+      await backend.connect({
+        host: 'example.databricks.com',
+        path: '/sql/1.0/warehouses/abc',
+        authType: 'databricks-oauth',
+      } as ConnectionOptions);
+    } catch (err) {
+      thrown = err;
+    }
+    expect(thrown).to.be.instanceOf(HiveDriverError);
+    expect((thrown as Error).message).to.match(/access-token/);
+  });
+
+  it('connect() rejects missing token', async () => {
+    const connection = new FakeNativeConnection();
+    const binding = makeBinding(connection);
+    const backend = new SeaBackend({ context: makeContext(), nativeBinding: binding });
+
+    let thrown: unknown;
+    try {
+      await backend.connect({
+        host: 'example.databricks.com',
+        path: '/sql/1.0/warehouses/abc',
+        token: '',
+      } as ConnectionOptions);
+    } catch (err) {
+      thrown = err;
+    }
+    expect(thrown).to.be.instanceOf(HiveDriverError);
+    // After sea-integration merge, missing-token validation goes through
+    // SeaAuth.buildSeaConnectionOptions which throws AuthenticationError
+    // (extends HiveDriverError) with the "non-empty PAT" message.
+    expect((thrown as Error).message).to.match(/non-empty PAT/);
+  });
+
+  it('openSession() throws if connect() was not called', async () => {
+    const connection = new FakeNativeConnection();
+    const binding = makeBinding(connection);
+    const backend = new SeaBackend({ context: makeContext(), nativeBinding: binding });
+
+    let thrown: unknown;
+    try {
+      await backend.openSession({});
+    } catch (err) {
+      thrown = err;
+    }
+    expect(thrown).to.be.instanceOf(HiveDriverError);
+    expect((thrown as Error).message).to.match(/not connected/);
+  });
+
+  it('openSession() forwards hostName / httpPath / token to napi binding', async () => {
+    const connection = new FakeNativeConnection();
+    const binding = makeBinding(connection);
+    const backend = new SeaBackend({ context: makeContext(), nativeBinding: binding });
+
+    await backend.connect({
+      host: 'workspace.example',
+      path: '/sql/1.0/warehouses/xyz',
+      token: 'dapi-token',
+    } as ConnectionOptions);
+
+    await backend.openSession({});
+
+    expect(binding.openSessionStub.calledOnce).to.equal(true);
+    const args = binding.openSessionStub.firstCall.args[0];
+    expect(args).to.deep.equal({
+      hostName: 'workspace.example',
+      httpPath: '/sql/1.0/warehouses/xyz',
+      token: 'dapi-token',
+    });
+  });
+
+  it('openSession() returns a SeaSessionBackend wrapping the napi Connection', async () => {
+    const connection = new FakeNativeConnection();
+    const binding = makeBinding(connection);
+    const backend = new SeaBackend({ context: makeContext(), nativeBinding: binding });
+
+    await backend.connect({
+      host: 'h',
+      path: '/p',
+      token: 't',
+    } as ConnectionOptions);
+
+    const sessionBackend = await backend.openSession({});
+    expect(sessionBackend).to.be.instanceOf(SeaSessionBackend);
+    expect(sessionBackend.id).to.be.a('string').and.have.length.greaterThan(0);
+  });
+
+  it('openSession() propagates initialCatalog / initialSchema / sessionConfig through to executeStatement', async () => {
+    const connection = new FakeNativeConnection();
+    const binding = makeBinding(connection);
+    const backend = new SeaBackend({ context: makeContext(), nativeBinding: binding });
+
+    await backend.connect({
+      host: 'h',
+      path: '/p',
+      token: 't',
+    } as ConnectionOptions);
+
+    const session = await backend.openSession({
+      initialCatalog: 'main',
+      initialSchema: 'default',
+      configuration: { 'spark.sql.execution.arrow.enabled': 'true' },
+    });
+
+    await session.executeStatement('SELECT 1', {});
+
+    expect(connection.lastSql).to.equal('SELECT 1');
+    expect(connection.lastOptions).to.deep.equal({
+      initialCatalog: 'main',
+      initialSchema: 'default',
+      sessionConfig: { 'spark.sql.execution.arrow.enabled': 'true' },
+    });
+  });
+
+  it('close() clears connection state without throwing', async () => {
+    const connection = new FakeNativeConnection();
+    const binding = makeBinding(connection);
+    const backend = new SeaBackend({ context: makeContext(), nativeBinding: binding });
+    await backend.connect({ host: 'h', path: '/p', token: 't' } as ConnectionOptions);
+    await backend.close();
+
+    let thrown: unknown;
+    try {
+      await backend.openSession({});
+    } catch (err) {
+      thrown = err;
+    }
+    expect(thrown).to.be.instanceOf(HiveDriverError);
+  });
+});
+
+describe('SeaSessionBackend', () => {
+  function makeSession(connection: SeaNativeConnection, defaults = {}) {
+    return new SeaSessionBackend({ connection, context: makeContext(), defaults });
+  }
+
+  it('executeStatement passes sql through verbatim', async () => {
+    const connection = new FakeNativeConnection();
+    const session = makeSession(connection);
+    await session.executeStatement('SELECT * FROM foo', {});
+    expect(connection.lastSql).to.equal('SELECT * FROM foo');
+  });
+
+  it('executeStatement returns a SeaOperationBackend with an id', async () => {
+    const connection = new FakeNativeConnection();
+    const session = makeSession(connection);
+    const op = await session.executeStatement('SELECT 1', {});
+    expect(op).to.be.instanceOf(SeaOperationBackend);
+    expect(op.id).to.be.a('string').and.have.length.greaterThan(0);
+  });
+
+  it('executeStatement merges session defaults into ExecuteOptions', async () => {
+    const connection = new FakeNativeConnection();
+    const session = makeSession(connection, {
+      initialCatalog: 'main',
+      initialSchema: 'default',
+      sessionConfig: { foo: 'bar' },
+    });
+    await session.executeStatement('SELECT 1', {});
+    expect(connection.lastOptions).to.deep.equal({
+      initialCatalog: 'main',
+      initialSchema: 'default',
+      sessionConfig: { foo: 'bar' },
+    });
+  });
+
+  it('executeStatement rejects namedParameters (M1)', async () => {
+    const connection = new FakeNativeConnection();
+    const session = makeSession(connection);
+    let thrown: unknown;
+    try {
+      await session.executeStatement('SELECT :x', { namedParameters: { x: 1 } });
+    } catch (err) {
+      thrown = err;
+    }
+    expect(thrown).to.be.instanceOf(HiveDriverError);
+    expect((thrown as Error).message).to.match(/parameters/);
+  });
+
+  it('executeStatement rejects ordinalParameters (M1)', async () => {
+    const connection = new FakeNativeConnection();
+    const session = makeSession(connection);
+    let thrown: unknown;
+    try {
+      await session.executeStatement('SELECT ?', { ordinalParameters: [1] });
+    } catch (err) {
+      thrown = err;
+    }
+    expect(thrown).to.be.instanceOf(HiveDriverError);
+  });
+
+  it('executeStatement rejects queryTimeout (M1)', async () => {
+    const connection = new FakeNativeConnection();
+    const session = makeSession(connection);
+    let thrown: unknown;
+    try {
+      await session.executeStatement('SELECT 1', { queryTimeout: 30 });
+    } catch (err) {
+      thrown = err;
+    }
+    expect(thrown).to.be.instanceOf(HiveDriverError);
+    expect((thrown as Error).message).to.match(/queryTimeout/);
+  });
+
+  it('metadata methods throw deferred-M1 errors', async () => {
+    const connection = new FakeNativeConnection();
+    const session = makeSession(connection);
+    for (const method of [
+      'getInfo',
+      'getTypeInfo',
+      'getCatalogs',
+      'getSchemas',
+      'getTables',
+      'getTableTypes',
+      'getColumns',
+      'getFunctions',
+      'getPrimaryKeys',
+      'getCrossReference',
+    ] as const) {
+      let thrown: unknown;
+      try {
+        // eslint-disable-next-line @typescript-eslint/no-explicit-any
+        await (session as any)[method]({});
+      } catch (err) {
+        thrown = err;
+      }
+      expect(thrown, `expected ${method} to throw`).to.be.instanceOf(HiveDriverError);
+      expect((thrown as Error).message).to.match(/M1|not implemented/);
+    }
+  });
+
+  it('close() forwards to the native connection', async () => {
+    const connection = new FakeNativeConnection();
+    const session = makeSession(connection);
+    const status = await session.close();
+    expect(connection.closed).to.equal(true);
+    expect(status.isSuccess).to.equal(true);
+  });
+
+  it('close() is idempotent', async () => {
+    const connection = new FakeNativeConnection();
+    const session = makeSession(connection);
+    await session.close();
+    // Second call should not re-invoke connection.close
+    connection.closed = false;
+    const status = await session.close();
+    expect(connection.closed).to.equal(false);
+    expect(status.isSuccess).to.equal(true);
+  });
+
+  it('executeStatement fails after close()', async () => {
+    const connection = new FakeNativeConnection();
+    const session = makeSession(connection);
+    await session.close();
+    let thrown: unknown;
+    try {
+      await session.executeStatement('SELECT 1', {});
+    } catch (err) {
+      thrown = err;
+    }
+    expect(thrown).to.be.instanceOf(HiveDriverError);
+  });
+});
+
+describe('SeaOperationBackend', () => {
+  function makeOperation(statement: SeaNativeStatement = new FakeNativeStatement()) {
+    return new SeaOperationBackend({ statement, context: makeContext() });
+  }
+
+  it('id is a stable string', () => {
+    const op = makeOperation();
+    expect(op.id).to.equal(op.id);
+    expect(op.id).to.be.a('string').and.have.length.greaterThan(0);
+  });
+
+  it('hasResultSet is true for M0', () => {
+    const op = makeOperation();
+    expect(op.hasResultSet).to.equal(true);
+  });
+
+  it('cancel() forwards to napi Statement', async () => {
+    const stmt = new FakeNativeStatement();
+    const op = makeOperation(stmt);
+    await op.cancel();
+    expect(stmt.cancelled).to.equal(true);
+  });
+
+  it('cancel() is idempotent', async () => {
+    const stmt = new FakeNativeStatement();
+    const op = makeOperation(stmt);
+    await op.cancel();
+    stmt.cancelled = false;
+    await op.cancel();
+    expect(stmt.cancelled).to.equal(false);
+  });
+
+  it('close() forwards to napi Statement', async () => {
+    const stmt = new FakeNativeStatement();
+    const op = makeOperation(stmt);
+    await op.close();
+    expect(stmt.closed).to.equal(true);
+  });
+
+  it('waitUntilReady() is a no-op (kernel internalises polling)', async () => {
+    const op = makeOperation();
+    await op.waitUntilReady();
+  });
+
+  // Note: after sea-integration merge, fetchChunk is no longer a stub —
+  // the sea-results SeaResultsProvider + ArrowResultConverter pipeline
+  // implements the real fetch path. Full coverage lives in
+  // tests/unit/sea/SeaOperationBackend.test.ts and the parity-gate e2e
+  // at tests/integration/sea/results-e2e.test.ts.
+});