[SEA-NodeJS] Address review: test the AsyncStatement arm; type-guard; null + stale comments

- F1: add coverage for the directResults Running (AsyncStatement) arm — the
  branch the PR exists to add. Three tests via the fake's `directReturnsRunning`:
  (a) a still-running query routes through the AsyncStatement arm and is driven
  via `status()`/`waitUntilReady()`; (b) `op.cancel()` reaches the running
  statement's `cancel()`; (c) contrast — a fast query routes through the terminal
  `Statement` arm and cancel reaches it there. Previously `directReturnsRunning`
  was never set, so the async arm had zero coverage.
- F4: replace the `as unknown as` double-casts with a cast-free user-defined type
  guard `isSeaAsyncStatement` (the napi `Statement`/`AsyncStatement` are the exact
  alias types, so no laundering is needed). Idiomatic, narrows both arms.
- F5: reject a null/undefined `direct` up front via `logAndMapError`
  (HiveDriverError) instead of the inconsistent `direct !== null` guard that let a
  null fall through to the `{statement}` arm and defer an opaque TypeError.
- F3: update the stale `runAsync` selector comment (still described
  `executeStatementCancellable` + blocking `result()`) to directResults; fix the
  stale `executeStatementCancellable` comment in the test.
- Document the `queryTimeout`→`wait_timeout=Ns`+CANCEL interaction (a timeout
  shorter than the query cancels it rather than returning the Running handle).

SEA unit suite 263 passing (3 new); eslint clean.

Co-authored-by: Isaac
Signed-off-by: Madhavendra Rathore <madhavendra.rathore@databricks.com>

Author: msrathore-db

lib/sea/SeaSessionBackend.ts

@@ -67,6 +67,19 @@ export interface SeaSessionBackendOptions {
  * needed — that pattern was removed when the napi binding moved these
  * onto `openSession` to match pyo3.
  */
+
+/**
+ * Narrow the directResults union (`executeStatementDirect`'s
+ * `Statement | AsyncStatement`) to the Running `AsyncStatement` arm. Only that
+ * arm exposes `awaitResult`; the terminal `Statement` (Completed arm) does not.
+ * Mirrors the kernel `DirectStatement::{Completed, Running}` discriminant, which
+ * the opaque napi classes can't carry on the wire — the `awaitResult` probe is
+ * the load-bearing feature-detect (see databricks-sql-kernel#140).
+ */
+function isSeaAsyncStatement(x: SeaStatement | SeaNativeAsyncStatement): x is SeaNativeAsyncStatement {
+  return typeof (x as SeaNativeAsyncStatement).awaitResult === 'function';
+}
+
 export default class SeaSessionBackend implements ISessionBackend {
   private readonly connection: SeaConnection;
 
@@ -164,13 +177,16 @@ export default class SeaSessionBackend implements ISessionBackend {
     // JSDoc in `IDBSQLSession` for the cross-backend contract.
     //
     //   - DEFAULT (`runAsync` false/undefined) — SYNC. Route through
-    //     `executeStatementCancellable`: the kernel blocks on `execute()`
-    //     (server-side direct-results / poll-to-terminal), which is faster and,
-    //     with the napi sync canceller, fully cancellable mid-COMPUTE. The
-    //     blocking drive runs in the operation backend's `result()` (inside
-    //     `waitUntilReady`, which the facade invokes lazily at first fetch).
-    //     `queryTimeoutSecs` IS honoured on this path (forwarded to the napi
-    //     options below) since the kernel `execute()` consults it.
+    //     `executeStatementDirect` (the directResults model): the kernel sends
+    //     ExecuteStatement with the server's inline wait and returns WITHOUT
+    //     polling past it — a terminal `Statement` (result inline) for a fast
+    //     query, or a still-running `AsyncStatement` (poll/cancel handle) for a
+    //     slow one. The handle is server-owned, so a long query stays cancellable
+    //     via `op.cancel()` and `close()` is a clean release (no client-side
+    //     drive-to-terminal). `queryTimeoutSecs` IS honoured here (forwarded to
+    //     the napi options below) — note the kernel sends it as `wait_timeout=Ns`
+    //     + CANCEL, so a timeout shorter than the query cancels it (eager error)
+    //     rather than returning the `Running` handle.
     //
     //   - `runAsync: true` — ASYNC. Submit (`wait_timeout=0s`): the server
     //     returns a pending `AsyncStatement` immediately while the query runs;
@@ -193,7 +209,7 @@ export default class SeaSessionBackend implements ISessionBackend {
       // release (no client-side drive-to-terminal). Fire-and-forget DDL/DML
       // commits because the server runs it inline during the POST.
       const execOptions = this.buildExecuteOptions(options, queryTimeoutSecs);
-      let direct;
+      let direct: SeaStatement | SeaNativeAsyncStatement;
       try {
         direct =
           execOptions === undefined
@@ -202,17 +218,22 @@ export default class SeaSessionBackend implements ISessionBackend {
       } catch (err) {
         throw this.logAndMapError('executeStatement', err);
       }
-      // Feature-detect the arm: only `AsyncStatement` (the Running arm) exposes
-      // `awaitResult`; the terminal `Statement` (Completed arm) does not.
-      const asAsync = direct as unknown as SeaNativeAsyncStatement;
-      if (direct !== null && typeof asAsync.awaitResult === 'function') {
-        return new SeaOperationBackend({ asyncStatement: asAsync, context: this.context, queryTimeoutSecs });
+      // The kernel contract (`Promise<Statement | AsyncStatement>`) never yields
+      // null/undefined; reject a non-handle up front so a contract violation
+      // surfaces as a mapped driver error here rather than an opaque `TypeError`
+      // deferred to first fetch/close.
+      if (direct === null || direct === undefined) {
+        throw this.logAndMapError(
+          'executeStatement',
+          new HiveDriverError('SEA executeStatementDirect returned no statement handle'),
+        );
+      }
+      // Feature-detect the arm via a type guard: only the Running `AsyncStatement`
+      // exposes `awaitResult`; the terminal `Statement` (Completed arm) does not.
+      if (isSeaAsyncStatement(direct)) {
+        return new SeaOperationBackend({ asyncStatement: direct, context: this.context, queryTimeoutSecs });
       }
-      return new SeaOperationBackend({
-        statement: direct as unknown as SeaStatement,
-        context: this.context,
-        queryTimeoutSecs,
-      });
+      return new SeaOperationBackend({ statement: direct, context: this.context, queryTimeoutSecs });
     }
 
     // Async path: do NOT forward `queryTimeoutSecs` (the kernel ignores it on

tests/unit/sea/execution.test.ts

@@ -567,6 +567,45 @@ describe('SeaSessionBackend', () => {
     expect(op.id).to.be.a('string').and.have.length.greaterThan(0);
   });
 
+  it('executeStatement (sync default) routes a still-running query through the AsyncStatement arm', async () => {
+    // The directResults Running arm — a query that did NOT finish within the
+    // server inline wait comes back as an AsyncStatement (poll/cancel handle).
+    // This is the branch the whole PR exists to add (Node mid-run cancel).
+    const connection = new FakeNativeConnection();
+    connection.directReturnsRunning = true;
+    const session = makeSession(connection);
+    const op = await session.executeStatement('SELECT slow', {});
+    expect(op).to.be.instanceOf(SeaOperationBackend);
+    // The Running arm was taken: an AsyncStatement was constructed + wired
+    // (not the terminal `statement` arm).
+    expect(connection.lastAsyncStatement, 'AsyncStatement (Running) arm should be taken').to.not.equal(undefined);
+    // Driving the op polls the async handle's status() — the polling arm.
+    await op.waitUntilReady();
+    expect(connection.lastAsyncStatement!.statusCalls, 'async handle polled via status()').to.be.greaterThan(0);
+  });
+
+  it('executeStatement (sync default) AsyncStatement arm: op.cancel() reaches the running statement', async () => {
+    // The point of directResults on single-threaded Node: the returned op holds
+    // a handle to the still-running statement, so op.cancel() can abort it.
+    const connection = new FakeNativeConnection();
+    connection.directReturnsRunning = true;
+    const session = makeSession(connection);
+    const op = await session.executeStatement('SELECT slow', {});
+    await op.cancel();
+    expect(connection.lastAsyncStatement!.cancelled, 'cancel reaches the running statement').to.equal(true);
+  });
+
+  it('executeStatement (sync default) routes a fast query through the terminal Statement arm', async () => {
+    // Contrast: a query that finished within the inline wait comes back as a
+    // terminal Statement (result inline) — no AsyncStatement is created.
+    const connection = new FakeNativeConnection(); // directReturnsRunning = false (default)
+    const session = makeSession(connection);
+    const op = await session.executeStatement('SELECT 1', {});
+    expect(connection.lastAsyncStatement, 'no AsyncStatement arm for a terminal query').to.equal(undefined);
+    await op.cancel();
+    expect(connection.statementToReturn.cancelled, 'cancel reaches the terminal statement').to.equal(true);
+  });
+
   it('executeStatement forwards ordinalParameters as napi positionalParams', async () => {
     const connection = new FakeNativeConnection();
     const session = makeSession(connection);
@@ -696,7 +735,7 @@ describe('SeaSessionBackend', () => {
     const envelope = `__databricks_error__:${JSON.stringify({ code: 'SqlError', message: 'SUBMIT_BOOM' })}`;
     for (const opts of [{}, { runAsync: true }]) {
       const connection = new FakeNativeConnection();
-      connection.throwOnExecute = new Error(envelope); // fails executeStatementCancellable / submitStatement
+      connection.throwOnExecute = new Error(envelope); // fails executeStatementDirect / submitStatement
       const session = makeSession(connection);
       let thrown: unknown;
       try {