feat(sdk): let a v2 session carry its own environment (#2268)

* feat(sdk): let a v2 session carry its own environment

`QueryOptions` accepts `env`; `SDKSessionOptions` did not, so a host that
keeps several sessions alive in one process had to share one environment
between them.

That matters when the variable selects the provider. `ANTHROPIC_BASE_URL`,
`OPENAI_BASE_URL` and the matching keys are read per request, so two sessions
aimed at different endpoints silently used whichever had been set last — and
the symptom is not an error but an answer from the wrong place, or billing
against the wrong account.

Applied around each turn and restored afterwards, reusing the process-wide env
mutex `query()` already takes (SEC-1), so one session's variables cannot leak
into another session's turn. The wrapping sits around the whole iteration
rather than inside the generator, so a consumer that breaks out of the loop
still restores: the generator's return runs the finally with it. A session
that sets no `env` takes no mutex and costs nothing.

The mutex means turns belonging to sessions with different environments
serialise rather than overlap. With one process environment there is no honest
alternative, and serialising is the behaviour that is correct rather than
merely concurrent; the type documents it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(sdk): apply a session's env after init, and never hold the mutex on failure

Review follow-up on the placement and on two ways the lock could be kept.

APPLIED AFTER init(), INSIDE THE GENERATOR. `query()` takes the mutex and
applies its overrides after `init()` for a reason: `init` is memoized, and it
configures the global agents, the proxies and mTLS once for the process. Doing
it before meant whichever session happened to run first configured all of that
out of its own variables. The apply now sits where query's does — right after
init, with the restore in a `finally` around the rest of the turn.

An early exit still restores. That was the stated reason for wrapping `yield*
inner` instead, and it was not a reason: a consumer that breaks out of the loop
calls the generator's `return`, which runs the generator's own `finally` — the
way the same pattern already works in `query()`.

THE MUTEX IS RELEASED IF APPLYING THROWS. `Object.entries(overrides)` invokes
getters, so an `env` whose properties are accessors could throw after the lock
was taken and before `sendMessage` entered its `try`. The lock stayed held and
every later env-setting turn in the process waited on it forever. Application
is now wrapped: what was already applied goes back and the lock is released
before the error leaves.

THE SNAPSHOT HAS A NULL PROTOTYPE. `__proto__` is a variable name like any
other, and on a plain object `snapshot['__proto__'] = value` is swallowed by
the prototype setter instead of being remembered — so the one key that must be
restored would be the one key missing from `Object.keys(snapshot)`.

The tests now drive `sendMessage()` itself, not only the private apply and
restore: the engine is a stand-in that reports what it sees in `process.env`,
and `init` is stubbed because the real one reaches the network. They prove the
variables are in place when the engine asks, that an abandoned iterator puts
them back and frees the lock, and that two turns with different environments
never interleave — three tests, all five of the earlier ones kept, red against
main (0 of 8).

Documented in the type what this does NOT do, since it would otherwise be
found out instead: a turn of a session without `env` takes no mutex, exactly as
in `query()`, so it can run while another session's variables are in place.
Isolating readers too would mean serialising every turn in the process.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* docs(sdk): say how a session's env can deadlock, since the lock is held all turn

Review follow-up. Holding the environment mutex for the whole turn is what
makes the isolation real, and it has two consequences a host author would
otherwise meet as a hang:

A PAUSED ITERATOR STILL HOLDS IT. Stopping half way through `for await (… of
session.sendMessage(…))` to await another env-carrying turn waits forever: the
first turn cannot finish while the second waits for the lock it holds.
Finishing the turn, or leaving the loop with `break`/`return` — which calls the
generator's `return` and frees the lock — is the way out.

A `canUseTool` CALLBACK IS INSIDE THE TURN. Awaiting an env-carrying `query()`
from it is the same deadlock one level down.

Both are documented on the option rather than discovered. Neither applies to a
session without `env`, which takes no lock at all.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(sdk): enumerate a session's env once, and say what an env-less turn sees

Three findings from review.

Object.keys fed the snapshot loop and Object.entries the apply loop. A plain
host object gives both the same keys; a Proxy need not. A key seen only by the
apply loop would be set and never restored. Enumerated once now, and before
the mutex is taken, so a throwing getter cannot leave it held either.

mock.module registrations are process-global and mock.restore does not remove
them, so the stubbed init() went on standing in for the real one in every test
file that ran afterwards in the same bun process. The real module is captured
first and re-registered in afterAll.

The review also asked for a concurrent env-less turn, expecting isolation.
There is none, and the test now says so: a turn that sets no environment takes
no mutex, so it runs alongside one that does and reads the same process.env
while the variables are in place. Serialising every turn instead would stall
an embedder's whole process the moment one session used env. What does hold —
that nothing survives the turn — is asserted, and the limit is documented on
SDKSessionOptions.env.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix(sdk): restore session environment across provider and close paths

* test(sdk): run Node environment regression through package harness

* test(sdk): rebuild current sources before runtime assertions

* fix(sdk): release closed session references immediately

* fix(sdk): restore authentication state between session turns

* fix(sdk): invalidate helper credentials at session env boundaries

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
Co-authored-by: jatmn <the@jat.mn>

src/entrypoints/sdk.d.ts

@@ -407,6 +407,28 @@ export type SDKSessionOptions = {
   permissionMode?: QueryPermissionMode
   allowDangerouslySkipPermissions?: boolean
   abortController?: AbortController
+  /**
+   * Environment variables to apply while this session's turns run.
+   *
+   * Applied around each turn and restored afterwards, under the same
+   * process-wide env mutex `query()` uses. Turns belonging to sessions with
+   * different environments therefore serialise rather than overlap.
+   *
+   * A turn that sets no environment takes no mutex, so it may run alongside
+   * one that does — and, since `process.env` is one object per process, it
+   * can read that turn's variables while they are in place. Nothing survives
+   * the turn either way. Give a session its own `env` if its turns must not
+   * see another's.
+   *
+   * Provider-derived routing/authentication variables are restored too.
+   * A value of `undefined` removes the variable for the turn.
+   *
+   * A paused iterator holds the mutex: finish the turn or leave the loop
+   * with `break`/`return` before awaiting another env-carrying turn.
+   * A `canUseTool` callback runs inside the turn and must not await an
+   * env-carrying `query()` or session turn, which would wait on its mutex.
+   */
+  env?: Record<string, string | undefined>
   /**
    * Callback invoked before each tool use. Return `{ behavior: 'allow' }` to
    * permit the call or `{ behavior: 'deny', message?: string }` to reject it.

src/entrypoints/sdk/v2.sessionEnv.test.ts

@@ -0,0 +1,130 @@
+/**
+ * A session's own environment.
+ *
+ * `QueryOptions` has accepted `env` for a while; `SDKSessionOptions` could
+ * not, so a host keeping several sessions in one process had to share one
+ * environment between them. That matters when the variable selects the
+ * provider — `ANTHROPIC_BASE_URL`, `OPENAI_BASE_URL` and their keys are read
+ * per request, so two sessions aimed at different endpoints silently used
+ * whichever had been set last.
+ *
+ * These tests drive the apply/restore directly rather than running a turn:
+ * what needs proving is that a session's variables are in place for its own
+ * turn and gone afterwards, and that no turn can see another session's.
+ */
+
+import { describe, expect, test, afterEach } from 'bun:test'
+
+import { unstable_v2_createSession } from './v2'
+import { resetEnvMutexForTesting } from './shared'
+
+const KEY = 'OPENCLAUDE_TEST_SESSION_ENV'
+
+afterEach(() => {
+  delete process.env[KEY]
+  resetEnvMutexForTesting()
+})
+
+/** Reach the private apply/restore the turn uses, without running a turn. */
+function envHooks(session: unknown): {
+  apply: () => Promise<Record<string, string | undefined> | null>
+  restore: (snapshot: Record<string, string | undefined> | null) => void
+} {
+  const impl = session as {
+    applySessionEnv: () => Promise<Record<string, string | undefined> | null>
+    restoreSessionEnv: (snapshot: Record<string, string | undefined> | null) => void
+  }
+  return {
+    apply: () => impl.applySessionEnv(),
+    restore: snapshot => impl.restoreSessionEnv(snapshot),
+  }
+}
+
+describe('SDKSessionOptions.env', () => {
+  test('applies for the turn and restores afterwards', async () => {
+    process.env[KEY] = 'before'
+    const session = unstable_v2_createSession({ cwd: process.cwd(), env: { [KEY]: 'during' } })
+    const hooks = envHooks(session)
+
+    const snapshot = await hooks.apply()
+    expect(process.env[KEY]).toBe('during')
+    hooks.restore(snapshot)
+    expect(process.env[KEY]).toBe('before')
+    session.close()
+  })
+
+  test('a variable the environment did not have is removed again, not emptied', async () => {
+    expect(process.env[KEY]).toBeUndefined()
+    const session = unstable_v2_createSession({ cwd: process.cwd(), env: { [KEY]: 'during' } })
+    const hooks = envHooks(session)
+
+    const snapshot = await hooks.apply()
+    expect(process.env[KEY]).toBe('during')
+    hooks.restore(snapshot)
+    expect(KEY in process.env).toBe(false)
+    session.close()
+  })
+
+  test('undefined removes a variable for the turn', async () => {
+    process.env[KEY] = 'before'
+    const session = unstable_v2_createSession({ cwd: process.cwd(), env: { [KEY]: undefined } })
+    const hooks = envHooks(session)
+
+    const snapshot = await hooks.apply()
+    expect(KEY in process.env).toBe(false)
+    hooks.restore(snapshot)
+    expect(process.env[KEY]).toBe('before')
+    session.close()
+  })
+
+  test('a session without env takes no mutex and changes nothing', async () => {
+    process.env[KEY] = 'before'
+    const session = unstable_v2_createSession({ cwd: process.cwd() })
+    const hooks = envHooks(session)
+
+    const snapshot = await hooks.apply()
+    expect(snapshot).toBeNull()
+    expect(process.env[KEY]).toBe('before')
+    // Restoring a null snapshot must not release a mutex nobody took: a
+    // stray release would let two env-setting turns overlap.
+    hooks.restore(snapshot)
+    expect(process.env[KEY]).toBe('before')
+    session.close()
+  })
+
+  /**
+   * The point of the mutex. One process has one environment, so two sessions
+   * wanting different values cannot both have them at once — the honest
+   * behaviour is to serialise, not to interleave and let the last writer win.
+   */
+  test('TWO SESSIONS WITH DIFFERENT ENVIRONMENTS SERIALISE', async () => {
+    const first = unstable_v2_createSession({ cwd: process.cwd(), env: { [KEY]: 'first' } })
+    const second = unstable_v2_createSession({ cwd: process.cwd(), env: { [KEY]: 'second' } })
+
+    const firstHooks = envHooks(first)
+    const secondHooks = envHooks(second)
+
+    const firstSnapshot = await firstHooks.apply()
+    expect(process.env[KEY]).toBe('first')
+
+    let secondEntered = false
+    const secondTurn = secondHooks.apply().then(snapshot => {
+      secondEntered = true
+      return snapshot
+    })
+
+    await new Promise(resolve => setTimeout(resolve, 20))
+    expect(secondEntered).toBe(false)
+    expect(process.env[KEY]).toBe('first')
+
+    firstHooks.restore(firstSnapshot)
+    const secondSnapshot = await secondTurn
+    expect(secondEntered).toBe(true)
+    expect(process.env[KEY]).toBe('second')
+
+    secondHooks.restore(secondSnapshot)
+    expect(KEY in process.env).toBe(false)
+    first.close()
+    second.close()
+  })
+})

src/entrypoints/sdk/v2.sessionEnvTurn.test.ts

@@ -0,0 +1,634 @@
+/**
+ * A session's own environment, through a real turn.
+ *
+ * `v2.sessionEnv.test.ts` drives the apply/restore directly; these tests
+ * iterate `sendMessage()` itself, because that is where the behaviour has to
+ * hold: the variables must be in place by the time the engine asks the
+ * provider anything, and gone once the turn is over — including when the
+ * consumer walks away from the iterator half way through.
+ *
+ * The engine is a stand-in that reports what it sees in `process.env` instead
+ * of calling a provider, and `init()` is stubbed because the real one
+ * configures proxies, mTLS and remote settings process-wide.
+ */
+
+import { afterAll, afterEach, beforeAll, describe, expect, mock, spyOn, test } from 'bun:test'
+import { mkdtempSync, rmSync, writeFileSync } from 'fs'
+import { tmpdir } from 'os'
+import { join } from 'path'
+
+import { resetEnvMutexForTesting } from './shared'
+import type { SDKMessage } from './coreTypes'
+import { getAnthropicClient } from '../../services/api/client'
+import { snapshotEnv } from '../../test/envSnapshot'
+import { clearApiKeyHelperCache, clearAwsCredentialsCache, clearOAuthTokenCache, refreshAndGetAwsCredentials } from '../../utils/auth'
+import * as settings from '../../utils/settings/settings'
+import * as aws from '../../utils/aws'
+import * as secureStorage from '../../utils/secureStorage/index'
+
+const KEY = 'OPENCLAUDE_TEST_SESSION_ENV_TURN'
+
+const authEnvKeys = [
+  'OPENCLAUDE_CONFIG_DIR', 'CLAUDE_CODE_SIMPLE', 'CLAUDE_CODE_OAUTH_TOKEN',
+  'ANTHROPIC_API_KEY', 'ANTHROPIC_AUTH_TOKEN', 'ANTHROPIC_BASE_URL',
+  'CLAUDE_CODE_USE_OPENAI', 'CLAUDE_CODE_USE_BEDROCK', 'CLAUDE_CODE_USE_VERTEX',
+  'CLAUDE_CODE_USE_FOUNDRY', 'CLAUDE_CODE_USE_GITHUB', 'CLAUDE_CODE_USE_GEMINI',
+  'CLAUDE_CODE_USE_GEMINI_VERTEX', 'CLAUDE_CODE_USE_MISTRAL',
+  'MINIMAX_API_KEY', 'MIMO_API_KEY', 'XAI_API_KEY', 'NEARAI_API_KEY',
+  'FIREWORKS_API_KEY', 'LONGCAT_API_KEY', 'AIMLAPI_API_KEY', 'APISMART_API_KEY',
+  'CONCENTRATE_API_KEY', 'API_ROUTE_API_KEY', 'GEMINI_AUTH_MODE',
+  'GEMINI_ACCESS_TOKEN', 'GITHUB_TOKEN', 'GH_TOKEN', 'GITHUB_COPILOT_KEY',
+  'CLAUDE_CODE_GITHUB_TOKEN_HYDRATED', 'OPENAI_API_KEY', 'OPENAI_API_KEYS',
+]
+
+/** What the engine saw in the environment, one entry per yielded message. */
+let seen: (string | undefined)[] = []
+
+/** An engine that answers out of the environment and never leaves the process. */
+function engineSeeingEnv(messages = 2) {
+  return {
+    // Everything the turn and `close()` touch on an engine, doing nothing.
+    injectAgents: () => {},
+    setMcpClients: () => {},
+    updateTools: () => {},
+    getMcpClients: () => [],
+    getMessages: () => [],
+    interrupt: () => {},
+    submitMessage: async function* (): AsyncGenerator<SDKMessage> {
+      for (let i = 0; i < messages; i++) {
+        seen.push(process.env[KEY])
+        yield { type: 'assistant', message: { content: [] }, session_id: 's' } as unknown as SDKMessage
+        // Hand control back, so a second turn could interleave if it were let in.
+        await new Promise(resolve => setTimeout(resolve, 0))
+      }
+    },
+  }
+}
+
+let createSession: typeof import('./v2').unstable_v2_createSession
+let realInit: typeof import('../init.js')
+
+beforeAll(async () => {
+  // Held so the registration can be put back. Raised in review: a
+  // `mock.module` registration is process-global and `mock.restore()` does
+  // not remove it, so a stubbed `init` would go on standing in for the real
+  // one in every test file that runs after this one in the same bun process.
+  realInit = { ...(await import('../init.js')) }
+  mock.module('../init.js', () => ({ init: async () => {} }))
+  createSession = (await import('./v2')).unstable_v2_createSession
+})
+
+afterAll(() => {
+  mock.module('../init.js', () => realInit)
+})
+
+afterEach(() => {
+  delete process.env[KEY]
+  seen = []
+  resetEnvMutexForTesting()
+})
+
+/** A session whose turn is answered by `engineSeeingEnv`. */
+function session(env?: Record<string, string | undefined>, messages = 2) {
+  const made = createSession({ cwd: process.cwd(), ...(env ? { env } : {}) })
+  ;(made as unknown as { setEngine: (engine: unknown) => void }).setEngine(engineSeeingEnv(messages))
+  return made
+}
+
+describe('SDKSessionOptions.env through a turn', () => {
+  test('THE TURN SEES ITS OWN VARIABLES, AND AFTERWARDS NOBODY DOES', async () => {
+    process.env[KEY] = 'before'
+    const one = session({ [KEY]: 'during' })
+
+    for await (const _ of one.sendMessage('hi')) {
+      // The point is what the engine saw, not what came back.
+    }
+
+    expect(seen).toEqual(['during', 'during'])
+    expect(process.env[KEY]).toBe('before')
+    one.close()
+  })
+
+  /**
+   * The case a direct apply/restore test cannot reach: a consumer that stops
+   * iterating. Breaking out of the loop calls the generator's `return`, which
+   * must run the restore — otherwise the variables, and the mutex, outlive the
+   * turn and every later env-setting turn waits forever.
+   */
+  test('AN ABANDONED TURN STILL PUTS THE ENVIRONMENT BACK', async () => {
+    process.env[KEY] = 'before'
+    const one = session({ [KEY]: 'during' }, 5)
+
+    for await (const _ of one.sendMessage('hi')) {
+      break
+    }
+
+    expect(seen).toEqual(['during'])
+    expect(process.env[KEY]).toBe('before')
+
+    // And the mutex came back with it: a second turn gets in without waiting.
+    const two = session({ [KEY]: 'second' })
+    for await (const _ of two.sendMessage('hi')) {
+      break
+    }
+    expect(seen).toEqual(['during', 'second'])
+    expect(process.env[KEY]).toBe('before')
+
+    one.close()
+    two.close()
+  })
+
+  test('close returns a paused turn and admits the next scoped turn', async () => {
+    process.env[KEY] = 'before'
+    const one = session({ [KEY]: 'during' })
+    const iterator = one.sendMessage('hi')[Symbol.asyncIterator]()
+    try {
+      await iterator.next()
+      one.close()
+      await new Promise(resolve => setTimeout(resolve, 20))
+      expect(process.env[KEY]).toBe('before')
+      const two = session({ [KEY]: 'next' })
+      try {
+        for await (const _ of two.sendMessage('hi')) break
+        expect(seen).toEqual(['during', 'next'])
+      } finally {
+        two.close()
+      }
+    } finally {
+      await iterator.return?.()
+      one.close()
+    }
+  })
+
+  test('close waits for active engine cleanup before restoring the environment', async () => {
+    let finish!: () => void
+    const cleanup = new Promise<void>(resolve => { finish = resolve })
+    const one = session({ [KEY]: 'during' })
+    const engine = engineSeeingEnv(1)
+    engine.submitMessage = async function* () {
+      try {
+        yield* engineSeeingEnv(1).submitMessage()
+      } finally {
+        await cleanup
+        seen.push(process.env[KEY])
+      }
+    }
+    ;(one as unknown as { setEngine: (engine: unknown) => void }).setEngine(engine)
+    const iterator = one.sendMessage('hi')[Symbol.asyncIterator]()
+    try {
+      await iterator.next()
+      one.close()
+      await new Promise(resolve => setTimeout(resolve, 10))
+      expect(process.env[KEY]).toBe('during')
+      finish()
+      await new Promise(resolve => setTimeout(resolve, 20))
+      expect(seen).toEqual(['during', 'during'])
+      expect(Object.hasOwn(process.env, KEY)).toBe(false)
+    } finally {
+      finish()
+      await iterator.return?.()
+      one.close()
+    }
+  })
+
+  test('close during a pending next waits for work without yielding another message', async () => {
+    let finish!: () => void
+    let entered!: () => void
+    const started = new Promise<void>(resolve => { entered = resolve })
+    const work = new Promise<void>(resolve => { finish = resolve })
+    const one = session({ [KEY]: 'during' })
+    const engine = engineSeeingEnv(1)
+    engine.submitMessage = async function* () {
+      entered()
+      await work
+      yield* engineSeeingEnv(1).submitMessage()
+    }
+    ;(one as unknown as { setEngine: (engine: unknown) => void }).setEngine(engine)
+    const iterator = one.sendMessage('hi')[Symbol.asyncIterator]()
+    const next = iterator.next()
+    try {
+      await started
+      one.close()
+      one.close()
+      expect(process.env[KEY]).toBe('during')
+      finish()
+      expect((await next).done).toBe(true)
+      expect(seen).toEqual(['during'])
+      expect(Object.hasOwn(process.env, KEY)).toBe(false)
+    } finally {
+      finish()
+      await iterator.return?.()
+      one.close()
+    }
+  })
+
+  test('closing a turn queued for the mutex never submits its engine', async () => {
+    const first = session({ [KEY]: 'first' })
+    const second = session({ [KEY]: 'second' })
+    const firstIterator = first.sendMessage('hi')[Symbol.asyncIterator]()
+    const secondIterator = second.sendMessage('hi')[Symbol.asyncIterator]()
+    try {
+      await firstIterator.next()
+      const pending = secondIterator.next()
+      await new Promise(resolve => setTimeout(resolve, 10))
+      second.close()
+      expect(() => second.getMessages()).toThrow('engine not initialized')
+      first.close()
+      expect((await pending).done).toBe(true)
+      expect(seen).toEqual(['first'])
+      expect(Object.hasOwn(process.env, KEY)).toBe(false)
+      const third = session({ [KEY]: 'third' })
+      try {
+        for await (const _ of third.sendMessage('hi')) break
+        expect(seen).toEqual(['first', 'third'])
+      } finally {
+        third.close()
+      }
+    } finally {
+      await firstIterator.return?.()
+      await secondIterator.return?.()
+      first.close()
+      second.close()
+    }
+  })
+
+  for (const route of [
+    { key: 'XAI_API_KEY', model: 'grok-4', host: 'api.x.ai', native: false },
+    { key: 'MINIMAX_API_KEY', model: 'MiniMax-M2.5', host: 'api.minimax.io', native: true },
+    { key: 'API_ROUTE_API_KEY', model: 'claude-sonnet-4-6', host: 'global.api-route.com', native: false },
+  ]) {
+    test(`restores provider-derived environment after a ${route.key} request`, async () => {
+      const keys = [
+        'ANTHROPIC_API_KEY', 'ANTHROPIC_BASE_URL', 'ANTHROPIC_MODEL',
+        'ANTHROPIC_CUSTOM_HEADERS', 'OPENAI_API_KEY', 'OPENAI_API_KEYS',
+        'OPENAI_BASE_URL', 'OPENAI_MODEL', 'OPENAI_API_FORMAT',
+        'OPENAI_AZURE_STYLE', 'OPENAI_AUTH_HEADER', 'OPENAI_AUTH_SCHEME',
+        'OPENAI_AUTH_HEADER_VALUE', 'CLAUDE_CODE_USE_OPENAI', route.key,
+      ]
+      const restore = snapshotEnv(keys, { clear: true })
+      const originalFetch = globalThis.fetch
+      const globals = globalThis as Record<string, unknown>
+      const originalMacro = globals.MACRO
+      globals.MACRO = { VERSION: 'test-version' }
+      const requests: { url: string; headers: Headers }[] = []
+      const one = session({ [route.key]: 'session-test-secret' })
+      try {
+        // These ambient values are deleted or overwritten by normalization.
+        process.env.OPENAI_AUTH_HEADER = 'X-Ambient-Key'
+        process.env.OPENAI_AUTH_SCHEME = 'Bearer'
+        process.env.OPENAI_AUTH_HEADER_VALUE = 'ambient-auth'
+        process.env.OPENAI_AZURE_STYLE = '1'
+        process.env.OPENAI_API_FORMAT = 'responses'
+        if (route.native) {
+          process.env.ANTHROPIC_API_KEY = 'ambient-native-key'
+          process.env.ANTHROPIC_MODEL = 'ambient-model'
+          process.env.CLAUDE_CODE_USE_OPENAI = '1'
+        }
+        if (route.key === 'API_ROUTE_API_KEY') {
+          process.env.OPENAI_API_KEYS = 'ambient-pool-key'
+          process.env.ANTHROPIC_CUSTOM_HEADERS = 'X-Ambient: before'
+        }
+        const before = Object.fromEntries(keys.map(key => [key, process.env[key]]))
+        globalThis.fetch = (async (input, init) => {
+          requests.push({ url: String(input), headers: new Headers(init?.headers) })
+          return Response.json(route.native
+            ? { id: 'msg-test', type: 'message', role: 'assistant', model: route.model,
+                content: [{ type: 'text', text: 'ok' }], stop_reason: 'end_turn',
+                usage: { input_tokens: 1, output_tokens: 1 } }
+            : { id: 'chat-test', object: 'chat.completion', model: route.model,
+                choices: [{ index: 0, message: { role: 'assistant', content: 'ok' }, finish_reason: 'stop' }],
+                usage: { prompt_tokens: 1, completion_tokens: 1 } })
+        }) as typeof fetch
+        const engine = engineSeeingEnv(1)
+        engine.submitMessage = async function* () {
+          const client = await getAnthropicClient({ maxRetries: 0, model: route.model })
+          await client.beta.messages.create({
+            model: route.model, max_tokens: 16,
+            messages: [{ role: 'user', content: 'hello' }],
+          })
+          yield* engineSeeingEnv(1).submitMessage()
+        }
+        ;(one as unknown as { setEngine: (engine: unknown) => void }).setEngine(engine)
+        for await (const _ of one.sendMessage('hi')) { /* drain */ }
+        expect(requests).toHaveLength(1)
+        expect(new URL(requests[0]!.url).hostname).toBe(route.host)
+        expect(requests[0]!.headers.get(route.native ? 'x-api-key' : 'authorization'))
+          .toBe(route.native ? 'session-test-secret' : 'Bearer session-test-secret')
+        expect(Object.fromEntries(keys.map(key => [key, process.env[key]]))).toEqual(before)
+      } finally {
+        one.close()
+        globalThis.fetch = originalFetch
+        if (originalMacro === undefined) delete globals.MACRO
+        else globals.MACRO = originalMacro
+        restore()
+      }
+    })
+  }
+
+
+  for (const exit of ['complete', 'break', 'throw'] as const) {
+    test(`OAuth identity follows scoped and ambient environments after ${exit}`, async () => {
+      const restore = snapshotEnv(authEnvKeys, { clear: true })
+      const configDir = mkdtempSync(join(tmpdir(), 'sdk-env-oauth-'))
+      const globals = globalThis as Record<string, unknown>
+      const originalMacro = globals.MACRO
+      globals.MACRO = { VERSION: 'test-version' }
+      const authorization: (string | null)[] = []
+      try {
+        process.env.OPENCLAUDE_CONFIG_DIR = configDir
+        // A stable credentials file prevents ENOENT/mtime invalidation from
+        // hiding a token cached under another environment.
+        writeFileSync(join(configDir, '.credentials.json'), '{}')
+        process.env.CLAUDE_CODE_OAUTH_TOKEN = 'ambient-token'
+        clearOAuthTokenCache()
+        const request = async () => {
+          const client = await getAnthropicClient({
+            maxRetries: 0,
+            fetchOverride: async (_input, init) => {
+              authorization.push(new Headers(init?.headers).get('authorization'))
+              return Response.json({
+                id: 'msg-test', type: 'message', role: 'assistant',
+                model: 'claude-sonnet-4-6', content: [], stop_reason: 'end_turn',
+                usage: { input_tokens: 1, output_tokens: 0 },
+              })
+            },
+          })
+          await client.beta.messages.create({
+            model: 'claude-sonnet-4-6', max_tokens: 16,
+            messages: [{ role: 'user', content: 'hello' }],
+          })
+        }
+        // Prewarm ambient auth, then check entry AND exit without manually
+        // clearing caches between any of the observed requests.
+        await request()
+        for (const token of ['first-token', 'second-token']) {
+          const one = session({ CLAUDE_CODE_OAUTH_TOKEN: token })
+          const engine = engineSeeingEnv(1)
+          engine.submitMessage = async function* () {
+            await request()
+            if (exit === 'throw') throw new Error('engine-turn-failure')
+            yield* engineSeeingEnv(1).submitMessage()
+          }
+          ;(one as unknown as { setEngine: (engine: unknown) => void }).setEngine(engine)
+          try {
+            const drain = async () => {
+              for await (const _ of one.sendMessage('hi')) {
+                if (exit === 'break') break
+              }
+            }
+            if (exit === 'throw') await expect(drain()).rejects.toThrow('engine-turn-failure')
+            else await drain()
+          } finally {
+            one.close()
+          }
+          expect(process.env.CLAUDE_CODE_OAUTH_TOKEN).toBe('ambient-token')
+          await request()
+        }
+        expect(authorization).toEqual([
+          'Bearer ambient-token', 'Bearer first-token', 'Bearer ambient-token',
+          'Bearer second-token', 'Bearer ambient-token',
+        ])
+      } finally {
+        restore()
+        clearOAuthTokenCache()
+        if (originalMacro === undefined) delete globals.MACRO
+        else globals.MACRO = originalMacro
+        rmSync(configDir, { recursive: true, force: true })
+      }
+    })
+  }
+
+  for (const credentialSource of ['apiKeyHelper', 'awsCredentialExport'] as const) {
+    for (const exit of ['complete', 'break', 'throw'] as const) {
+      test(`${credentialSource} identity follows scoped and ambient environments after ${exit}`, async () => {
+        const restore = snapshotEnv([...authEnvKeys, KEY, 'NODE_ENV', 'CI'], { clear: true })
+        const configDir = mkdtempSync(join(tmpdir(), 'sdk-env-helper-'))
+        const globals = globalThis as Record<string, unknown>
+        const originalMacro = globals.MACRO
+        globals.MACRO = { VERSION: 'test-version' }
+        process.env.OPENCLAUDE_CONFIG_DIR = configDir
+        process.env.NODE_ENV = 'production'
+        process.env[KEY] = 'ambient'
+        const helperPath = join(configDir, 'credential-helper.cjs')
+        // Real, cross-platform helper execution: the command inherits the turn
+        // environment; only network/storage dependencies are substituted.
+        writeFileSync(helperPath, credentialSource === 'apiKeyHelper'
+          ? `process.stdout.write(process.env.${KEY})`
+          : `process.stdout.write(JSON.stringify({ Credentials: {
+              AccessKeyId: process.env.${KEY}, SecretAccessKey: 'test-secret', SessionToken: 'test-token',
+            } }))`)
+        const helperSettings = { [credentialSource]: `"${process.execPath}" "${helperPath}"` }
+        const mergedSettings = spyOn(settings, 'getSettings_DEPRECATED').mockReturnValue(helperSettings)
+        const sourceSettings = spyOn(settings, 'getSettingsForSource')
+          .mockImplementation(source => source === 'userSettings' ? helperSettings : null)
+        const storage = spyOn(secureStorage, 'getSecureStorage').mockReturnValue({
+          name: 'session-env-helper-fixture', read: () => null, readAsync: async () => null,
+          update: () => ({ success: true }), delete: () => true,
+        })
+        const sts = credentialSource === 'awsCredentialExport'
+          ? spyOn(aws, 'checkStsCallerIdentity').mockRejectedValue(new Error('export required'))
+          : undefined
+        const ini = credentialSource === 'awsCredentialExport'
+          ? spyOn(aws, 'clearAwsIniCache').mockResolvedValue(undefined)
+          : undefined
+        clearApiKeyHelperCache()
+        clearAwsCredentialsCache()
+        clearOAuthTokenCache()
+        const identities: (string | null | undefined)[] = []
+        const request = async () => {
+          if (credentialSource === 'awsCredentialExport') {
+            // This is the credential object getAnthropicClient passes to
+            // Bedrock; STS and INI operations above avoid live AWS access.
+            identities.push((await refreshAndGetAwsCredentials())?.accessKeyId)
+            return
+          }
+          const client = await getAnthropicClient({
+            maxRetries: 0,
+            fetchOverride: async (_input, init) => {
+              identities.push(new Headers(init?.headers).get('authorization'))
+              return Response.json({
+                id: 'msg-helper', type: 'message', role: 'assistant',
+                model: 'claude-sonnet-4-6', content: [], stop_reason: 'end_turn',
+                usage: { input_tokens: 1, output_tokens: 0 },
+              })
+            },
+          })
+          await client.beta.messages.create({
+            model: 'claude-sonnet-4-6', max_tokens: 16,
+            messages: [{ role: 'user', content: 'hello' }],
+          })
+        }
+        try {
+          // A warm ambient cache must not supply the scoped request. Conversely,
+          // each scope's cached result must disappear before the ambient read.
+          await request()
+          for (const identity of ['first', 'second']) {
+            const one = session({ [KEY]: identity })
+            const engine = engineSeeingEnv(1)
+            engine.submitMessage = async function* () {
+              await request()
+              if (exit === 'throw') throw new Error('helper-turn-failure')
+              yield* engineSeeingEnv(1).submitMessage()
+            }
+            ;(one as unknown as { setEngine: (engine: unknown) => void }).setEngine(engine)
+            try {
+              const drain = async () => {
+                for await (const _ of one.sendMessage('hi')) {
+                  if (exit === 'break') break
+                }
+              }
+              if (exit === 'throw') await expect(drain()).rejects.toThrow('helper-turn-failure')
+              else await drain()
+            } finally {
+              one.close()
+            }
+            expect(process.env[KEY]).toBe('ambient')
+            await request()
+          }
+          expect(identities).toEqual(credentialSource === 'apiKeyHelper'
+            ? ['Bearer ambient', 'Bearer first', 'Bearer ambient', 'Bearer second', 'Bearer ambient']
+            : ['ambient', 'first', 'ambient', 'second', 'ambient'])
+        } finally {
+          ini?.mockRestore()
+          sts?.mockRestore()
+          storage.mockRestore()
+          sourceSettings.mockRestore()
+          mergedSettings.mockRestore()
+          restore()
+          clearApiKeyHelperCache()
+          clearAwsCredentialsCache()
+          clearOAuthTokenCache()
+          if (originalMacro === undefined) delete globals.MACRO
+          else globals.MACRO = originalMacro
+          rmSync(configDir, { recursive: true, force: true })
+        }
+      })
+    }
+  }
+
+  for (const route of ['gemini', 'github-token', 'github-key'] as const) {
+    for (const original of [undefined, ' '] as const) {
+      test(`restores ${route} hydration outputs originally ${original === undefined ? 'absent' : 'present'}`, async () => {
+        const restore = snapshotEnv(authEnvKeys, { clear: true })
+        const globals = globalThis as Record<string, unknown>
+        const originalMacro = globals.MACRO
+        globals.MACRO = { VERSION: 'test-version' }
+        const data = {
+          claudeAiOauth: undefined,
+          gemini: { accessToken: 'stored-gemini-secret' },
+          githubModels: {
+            accessToken: 'stored-github-secret',
+            credentialType: route === 'github-key' ? 'copilot_key' : 'copilot_token',
+          },
+        }
+        const storage = spyOn(secureStorage, 'getSecureStorage').mockReturnValue({
+          name: 'session-env-fixture', read: () => data, readAsync: async () => data,
+          update: () => ({ success: true }), delete: () => true,
+        })
+        const outputs = [
+          'GEMINI_ACCESS_TOKEN', 'GITHUB_TOKEN', 'GITHUB_COPILOT_KEY',
+          'CLAUDE_CODE_GITHUB_TOKEN_HYDRATED',
+        ]
+        for (const key of outputs) {
+          if (original !== undefined) process.env[key] = original
+        }
+        const before = outputs.map(key => [Object.hasOwn(process.env, key), process.env[key]])
+        const one = session(route === 'gemini'
+          ? { CLAUDE_CODE_USE_GEMINI: '1', GEMINI_AUTH_MODE: 'access-token' }
+          : { CLAUDE_CODE_USE_GITHUB: '1' })
+        try {
+          const engine = engineSeeingEnv(1)
+          engine.submitMessage = async function* () {
+            await getAnthropicClient({
+              maxRetries: 0, model: route === 'gemini' ? 'gemini-2.5-flash' : 'gpt-4.1',
+            })
+            // Prove real hydration ran; the storage double only supplies data.
+            const key = route === 'gemini' ? 'GEMINI_ACCESS_TOKEN'
+              : route === 'github-key' ? 'GITHUB_COPILOT_KEY' : 'GITHUB_TOKEN'
+            expect(process.env[key]).toBe(route === 'gemini'
+              ? 'stored-gemini-secret' : 'stored-github-secret')
+            yield* engineSeeingEnv(1).submitMessage()
+          }
+          ;(one as unknown as { setEngine: (engine: unknown) => void }).setEngine(engine)
+          for await (const _ of one.sendMessage('hi')) { /* drain */ }
+          expect(outputs.map(key => [Object.hasOwn(process.env, key), process.env[key]])).toEqual(before)
+        } finally {
+          one.close()
+          storage.mockRestore()
+          restore()
+          clearOAuthTokenCache()
+          if (originalMacro === undefined) delete globals.MACRO
+          else globals.MACRO = originalMacro
+        }
+      })
+    }
+  }
+
+  /**
+   * Two turns, two environments, one process. They may not interleave, and the
+   * proof is in what each engine saw: never the other session's value.
+   */
+  test('TWO TURNS WITH DIFFERENT ENVIRONMENTS DO NOT INTERLEAVE', async () => {
+    const first = session({ [KEY]: 'first' }, 3)
+    const second = session({ [KEY]: 'second' }, 3)
+
+    const drain = async (one: ReturnType<typeof session>) => {
+      for await (const _ of one.sendMessage('hi')) {
+        // Drain, slowly enough that an unguarded second turn would slip in.
+        await new Promise(resolve => setTimeout(resolve, 5))
+      }
+    }
+
+    await Promise.all([drain(first), drain(second)])
+
+    expect(seen.length).toBe(6)
+    const halves = [seen.slice(0, 3), seen.slice(3)].map(half => new Set(half))
+    expect(halves[0]!.size).toBe(1)
+    expect(halves[1]!.size).toBe(1)
+    expect([...halves[0]!][0]).not.toBe([...halves[1]!][0])
+    expect(KEY in process.env).toBe(false)
+
+    first.close()
+    second.close()
+  })
+
+  /**
+   * THE LIMIT, WRITTEN DOWN AS A TEST.
+   *
+   * A turn that asked for no environment takes no mutex — nothing to take and
+   * nothing to put back — so it is free to run beside an env-scoped turn. And
+   * `process.env` is one object for the whole process: while the scoped turn
+   * holds its variables, the other turn is looking at the same object and
+   * sees them.
+   *
+   * This is the price of the design and not a defect in it. The alternative —
+   * making every turn take the mutex — would serialise an embedder's whole
+   * process the moment one session used `env`, which is a far worse trade for
+   * the host this exists for. What IS guaranteed, and asserted below, is that
+   * nothing survives the turn.
+   *
+   * The suggestion to cover a concurrent env-less turn came from review; it
+   * was written expecting isolation, and this is what it actually found.
+   */
+  test('AN ENV-LESS TURN RUNNING ALONGSIDE ONE CAN SEE ITS VARIABLES, BUT NOT OUTLIVE THEM', async () => {
+    const scoped = session({ [KEY]: 'scoped' }, 3)
+    const plain = session(undefined, 3)
+
+    const drain = async (one: ReturnType<typeof session>) => {
+      for await (const _ of one.sendMessage('hi')) {
+        await new Promise(resolve => setTimeout(resolve, 5))
+      }
+    }
+
+    await Promise.all([drain(scoped), drain(plain)])
+
+    expect(seen.length).toBe(6)
+    // Both turns were looking at the same `process.env` while it was set.
+    expect(seen.every(one => one === 'scoped')).toBe(true)
+    // And the guarantee that does hold: the turn took its variables with it.
+    expect(KEY in process.env).toBe(false)
+
+    scoped.close()
+    plain.close()
+  })
+})

src/entrypoints/sdk/v2.ts

@@ -19,6 +19,11 @@ import {
 } from '../../Tool.js'
 import { assembleToolPool, getTools } from '../../tools.js'
 import { createFileStateCacheWithSizeLimit } from '../../utils/fileStateCache.js'
+import {
+  clearApiKeyHelperCache,
+  clearAwsCredentialsCache,
+  clearOAuthTokenCache,
+} from '../../utils/auth.js'
 import { requestSdkRootAbort } from './interruption.js'
 import { init } from '../init.js'
 import {
@@ -53,8 +58,10 @@ import type {
   SDKPermissionRequestMessage,
 } from './shared.js'
 import {
+  acquireEnvMutex,
   assertValidSessionId,
   mapMessageToSDK,
+  releaseEnvMutex,
 } from './shared.js'
 import {
   buildPermissionContext,
@@ -97,6 +104,50 @@ export type SDKSessionOptions = {
   allowDangerouslySkipPermissions?: boolean
   /** AbortController to cancel the session. */
   abortController?: AbortController
+  /**
+   * Environment variables to apply while this session's turns run.
+   *
+   * `QueryOptions` already accepts this; a session could not, so a host that
+   * keeps several sessions in one process had to share one environment
+   * between them. That is a problem when the variable in question selects the
+   * provider: `ANTHROPIC_BASE_URL`, `OPENAI_BASE_URL` and the matching keys
+   * are read per request, so two sessions pointed at different endpoints
+   * silently used whichever was set last.
+   *
+   * Applied around each turn and restored afterwards, under the same
+   * process-wide env mutex `query()` uses, so a session's variables cannot
+   * leak into another session's turn. That mutex also means turns with
+   * different environments SERIALISE rather than overlap: with one process
+   * environment there is no honest alternative, and serialising is the
+   * behaviour that is correct rather than merely concurrent.
+   *
+   * Provider-derived routing/authentication variables are restored too.
+   * A value of `undefined` removes the variable for the turn.
+   *
+   * What it does NOT do, exactly as in `query()`: a turn belonging to a
+   * session without `env` takes no mutex, so it can run while another
+   * session's variables are in place. In one process there is one
+   * environment; isolating readers as well as writers would mean serialising
+   * every turn in the process. So set `env` on every session that cares which
+   * provider it reaches, not only on the odd one out.
+   *
+   * TWO WAYS TO DEADLOCK YOURSELF, because the lock is held for the whole
+   * turn and not just while the variables are written:
+   *
+   * 1. A PAUSED ITERATOR STILL HOLDS IT. Stopping half way through
+   *    `for await (… of session.sendMessage(…))` to await another
+   *    env-carrying turn waits forever: the first turn cannot finish while
+   *    the second waits for the lock it holds. Finish the turn, or leave the
+   *    loop with `break`/`return` — which calls the generator's `return` and
+   *    frees the lock — before awaiting another one.
+   * 2. A `canUseTool` CALLBACK IS INSIDE THE TURN. Awaiting an env-carrying
+   *    `query()` or session turn from it is the same deadlock, one level
+   *    down. Answer the permission from what the callback already has.
+   *
+   * Neither applies to sessions without `env`: they take no lock.
+   */
+  env?: Record<string, string | undefined>
+
   /**
    * Callback invoked before each tool use. Return `{ behavior: 'allow' }` to
    * permit the call or `{ behavior: 'deny', message?: string }` to reject it.
@@ -152,7 +203,7 @@ export interface SDKSession {
   getMessages(): SDKMessage[]
   /** Abort the current in-flight query. */
   interrupt(): void
-  /** Close the session and release resources. */
+  /** Close the session; active turns unwind before their environment is restored. */
   close(): void
   /**
    * Respond to a pending permission prompt asynchronously.
@@ -210,6 +261,8 @@ class SDKSessionImpl implements SDKSession {
     return this._appStateStore
   }
   private _abortController: AbortController | null = null
+  private closed = false
+  private activeEnvTurns = new Set<AsyncGenerator<SDKMessage>>()
   private agentsLoaded = false
   private mcpServers?: Record<string, unknown>
   private mcpConnected = false
@@ -260,6 +313,87 @@ class SDKSessionImpl implements SDKSession {
     return this._sessionId
   }
 
+  /**
+   * Apply this session's `env`, remembering what was there.
+   *
+   * Returns the snapshot to restore, or null when the session set no `env` —
+   * in which case the mutex is never taken and a session without `env` costs
+   * nothing.
+   *
+   * If reading or writing a variable throws half way through — an `env` whose
+   * properties are getters, say — what was already applied is put back and the
+   * mutex is released before the error leaves. Otherwise the lock would be
+   * held by a turn that never started, and every later env-setting turn in the
+   * process would wait on it forever.
+   */
+  private async applySessionEnv(): Promise<Record<string, string | undefined> | null> {
+    const overrides = this.options.env
+    if (!overrides) return null
+    // ENUMERATED ONCE, AND BEFORE THE MUTEX. Raised in review: the snapshot
+    // loop read `Object.keys` and the apply loop `Object.entries`, which a
+    // Proxy may answer differently — a key seen only by the apply loop would
+    // be set and never restored. Reading the host's object before the lock
+    // also means a throwing getter cannot leave the mutex held: nothing has
+    // been taken yet.
+    const entries = Object.entries(overrides)
+    if (entries.length === 0) return null
+    await acquireEnvMutex()
+    // Null-prototype: `__proto__` is a variable name like any other here, and
+    // on a plain object it would be swallowed by the prototype setter instead
+    // of being remembered — leaving the one key we must restore unrestorable.
+    const snapshot: Record<string, string | undefined> = Object.create(null)
+    try {
+      // Provider construction normalizes routing and hydrates saved credentials.
+      // These outputs belong to the turn even when not explicit overrides.
+      const keys = new Set([
+        ...entries.map(([key]) => key),
+        'ANTHROPIC_API_KEY', 'ANTHROPIC_BASE_URL', 'ANTHROPIC_MODEL',
+        'ANTHROPIC_CUSTOM_HEADERS', 'CLAUDE_CODE_USE_OPENAI',
+        'OPENAI_API_KEY', 'OPENAI_API_KEYS', 'OPENAI_BASE_URL', 'OPENAI_MODEL',
+        'OPENAI_API_FORMAT', 'OPENAI_AZURE_STYLE', 'OPENAI_AUTH_HEADER',
+        'OPENAI_AUTH_SCHEME', 'OPENAI_AUTH_HEADER_VALUE',
+        'GEMINI_ACCESS_TOKEN', 'GITHUB_TOKEN', 'GITHUB_COPILOT_KEY',
+        'CLAUDE_CODE_GITHUB_TOKEN_HYDRATED',
+      ])
+      for (const key of keys) {
+        snapshot[key] = Object.hasOwn(process.env, key) ? process.env[key] : undefined
+      }
+      for (const [key, value] of entries) {
+        if (value === undefined) {
+          delete process.env[key]
+        } else {
+          process.env[key] = value
+        }
+      }
+      // Authentication readers cache results independently of process.env.
+      // A scoped turn must not reuse credentials from the previous environment.
+      clearOAuthTokenCache()
+      clearApiKeyHelperCache()
+      clearAwsCredentialsCache()
+    } catch (error) {
+      this.restoreSessionEnv(snapshot)
+      throw error
+    }
+    return snapshot
+  }
+
+  /** Put back what `applySessionEnv` found, and let the next turn in. */
+  private restoreSessionEnv(snapshot: Record<string, string | undefined> | null): void {
+    if (!snapshot) return
+    for (const key of Object.keys(snapshot)) {
+      const original = snapshot[key]
+      if (original === undefined) {
+        delete process.env[key]
+      } else {
+        process.env[key] = original
+      }
+    }
+    clearOAuthTokenCache()
+    clearApiKeyHelperCache()
+    clearAwsCredentialsCache()
+    releaseEnvMutex()
+  }
+
   async *sendMessage(content: string): AsyncIterable<SDKMessage> {
     const sdkContext = {
       sessionId: this._sessionId as SessionId,
@@ -274,116 +408,140 @@ class SDKSessionImpl implements SDKSession {
       (async function* (): AsyncGenerator<SDKMessage> {
         // Fast exit: if the caller's AbortController was already aborted
         // before iteration starts, do not initialize or submit a turn.
-        if (self._abortController?.signal.aborted) return
+        if (self.closed || self._abortController?.signal.aborted) return
 
         await runOutsideSdkContext(init)
+        if (self.closed) return
 
-        // Load agent definitions once (not on every sendMessage call)
-        if (!self.agentsLoaded) {
-          let agentDefs: AgentDefinitionsResult = { activeAgents: [], allAgents: [] }
-          const reportDefinitionFailure = (errorMessage: string) => {
-            self.pushAgentFailure({
-              type: 'agent_load_failure',
-              stage: 'definitions',
-              error_message: errorMessage,
-            })
-          }
-          try {
-            agentDefs = await getAgentDefinitionsWithOverrides(self.options.cwd)
-            for (const failedFile of agentDefs.failedFiles ?? []) {
-              reportDefinitionFailure(
-                `Failed to load agent definition '${failedFile.path}': ${failedFile.error}`,
-              )
-            }
-          } catch (err) {
-            // Agent loading failed — continue without filesystem agents but emit failure event
-            const errorMessage = err instanceof Error ? err.message : String(err)
-            console.warn('SDK: agent loading failed:', errorMessage)
-            reportDefinitionFailure(errorMessage)
-          }
-
-          const userAgents = buildSdkUserAgents(
-            self.options.agents,
-            (name, errorMessage) => {
+        // The session's own environment, for the length of the turn and no
+        // longer. Applied AFTER init() with a full-duration mutex, the way
+        // query() does it (SEC-1): init is memoized and configures proxies,
+        // mTLS and the global agents process-wide, so it must not be
+        // configured out of one session's variables.
+        //
+        // Inside the generator rather than around `yield* inner`, so an
+        // abandoned iterator still restores: breaking out of the loop calls
+        // the generator's return, which runs this finally with it.
+        const envSnapshot = await self.applySessionEnv()
+        try {
+          if (self.closed || self._abortController?.signal.aborted) return
+          // Load agent definitions once (not on every sendMessage call)
+          if (!self.agentsLoaded) {
+            let agentDefs: AgentDefinitionsResult = { activeAgents: [], allAgents: [] }
+            const reportDefinitionFailure = (errorMessage: string) => {
               self.pushAgentFailure({
                 type: 'agent_load_failure',
-                stage: 'injection',
-                error_message: `Invalid SDK agent '${name}': ${errorMessage}`,
-              })
-            },
-          )
-          const mergedAgentDefs = mergeSdkUserAgents(agentDefs, userAgents)
-          if (mergedAgentDefs.activeAgents.length > 0) {
-            self.appStateStore.setState(prev => ({
-              ...prev,
-              agentDefinitions: mergedAgentDefs,
-            }))
-            try {
-              self.engine.injectAgents(mergedAgentDefs.activeAgents)
-            } catch (err) {
-              // Agent injection failed — continue without agents but emit failure event
-              const errorMessage = err instanceof Error ? err.message : String(err)
-              console.warn('SDK: agent injection failed:', errorMessage)
-              self.appStateStore.setState(prev => ({
-                ...prev,
-                agentDefinitions: { activeAgents: [], allAgents: [] },
-              }))
-              self.pushAgentFailure({
-                type: 'agent_load_failure',
-                stage: 'injection',
+                stage: 'definitions',
                 error_message: errorMessage,
               })
             }
-          } else {
-            self.appStateStore.setState(prev => ({
-              ...prev,
-              agentDefinitions: mergedAgentDefs,
-            }))
+            try {
+              agentDefs = await getAgentDefinitionsWithOverrides(self.options.cwd)
+              if (self.closed) return
+              for (const failedFile of agentDefs.failedFiles ?? []) {
+                reportDefinitionFailure(
+                  `Failed to load agent definition '${failedFile.path}': ${failedFile.error}`,
+                )
+              }
+            } catch (err) {
+              // Agent loading failed — continue without filesystem agents but emit failure event
+              const errorMessage = err instanceof Error ? err.message : String(err)
+              console.warn('SDK: agent loading failed:', errorMessage)
+              reportDefinitionFailure(errorMessage)
+            }
+
+            if (self.closed) return
+            const userAgents = buildSdkUserAgents(
+              self.options.agents,
+              (name, errorMessage) => {
+                self.pushAgentFailure({
+                  type: 'agent_load_failure',
+                  stage: 'injection',
+                  error_message: `Invalid SDK agent '${name}': ${errorMessage}`,
+                })
+              },
+            )
+            const mergedAgentDefs = mergeSdkUserAgents(agentDefs, userAgents)
+            if (mergedAgentDefs.activeAgents.length > 0) {
+              self.appStateStore.setState(prev => ({
+                ...prev,
+                agentDefinitions: mergedAgentDefs,
+              }))
+              try {
+                self.engine.injectAgents(mergedAgentDefs.activeAgents)
+              } catch (err) {
+                // Agent injection failed — continue without agents but emit failure event
+                const errorMessage = err instanceof Error ? err.message : String(err)
+                console.warn('SDK: agent injection failed:', errorMessage)
+                self.appStateStore.setState(prev => ({
+                  ...prev,
+                  agentDefinitions: { activeAgents: [], allAgents: [] },
+                }))
+                self.pushAgentFailure({
+                  type: 'agent_load_failure',
+                  stage: 'injection',
+                  error_message: errorMessage,
+                })
+              }
+            } else {
+              self.appStateStore.setState(prev => ({
+                ...prev,
+                agentDefinitions: mergedAgentDefs,
+              }))
+            }
+            self.agentsLoaded = true
           }
-          self.agentsLoaded = true
-        }
+
+          // Connect MCP servers once (lazy, on first message)
+          if (!self.mcpConnected && self.mcpServers && Object.keys(self.mcpServers).length > 0) {
+            try {
+              const { clients: mcpClients, tools: mcpTools } = await connectSdkMcpServers(self.mcpServers)
+              if (self.closed) return
+              if (mcpClients.length > 0) {
+                self.engine.setMcpClients(mcpClients)
+              }
+              if (mcpTools.length > 0) {
+                const permissionContext = self.appStateStore.getState().toolPermissionContext
+                self.engine.updateTools(assembleToolPool(permissionContext, mcpTools))
+              }
+            } catch (err) {
+              // MCP connection failed — continue without MCP tools
+              console.warn('SDK: MCP server connection failed:', err instanceof Error ? err.message : String(err))
+            }
+            self.mcpConnected = true
+          }
+
+          // Switch session for transcript writes using session's own resolved dir
+          switchSession(self._sessionId as SessionId, self._sessionProjectDir)
 
-        // Connect MCP servers once (lazy, on first message)
-        if (!self.mcpConnected && self.mcpServers && Object.keys(self.mcpServers).length > 0) {
           try {
-            const { clients: mcpClients, tools: mcpTools } = await connectSdkMcpServers(self.mcpServers)
-            if (mcpClients.length > 0) {
-              self.engine.setMcpClients(mcpClients)
+            if (self.closed || self._abortController?.signal.aborted) return
+            yield* self.drainAgentFailureQueue()
+            for await (const engineMsg of self.engine.submitMessage(content)) {
+              if (self.closed || self._abortController?.signal.aborted) break
+              yield engineMsg
+              yield* self.drainTimeoutQueue()
+              yield* self.drainAgentFailureQueue()
             }
-            if (mcpTools.length > 0) {
-              const permissionContext = self.appStateStore.getState().toolPermissionContext
-              self.engine.updateTools(assembleToolPool(permissionContext, mcpTools))
-            }
-          } catch (err) {
-            // MCP connection failed — continue without MCP tools
-            console.warn('SDK: MCP server connection failed:', err instanceof Error ? err.message : String(err))
-          }
-          self.mcpConnected = true
-        }
-
-        // Switch session for transcript writes using session's own resolved dir
-        switchSession(self._sessionId as SessionId, self._sessionProjectDir)
-
-        try {
-          if (self._abortController?.signal.aborted) return
-          yield* self.drainAgentFailureQueue()
-          for await (const engineMsg of self.engine.submitMessage(content)) {
-            if (self._abortController?.signal.aborted) break
-            yield engineMsg
+            // Final drain for timeout/failure messages that fired on the last engine yield
             yield* self.drainTimeoutQueue()
             yield* self.drainAgentFailureQueue()
+          } finally {
+            self.timeoutQueue.length = 0
+            self.agentFailureQueue.length = 0
           }
-          // Final drain for timeout/failure messages that fired on the last engine yield
-          yield* self.drainTimeoutQueue()
-          yield* self.drainAgentFailureQueue()
         } finally {
-          self.timeoutQueue.length = 0
-          self.agentFailureQueue.length = 0
+          self.restoreSessionEnv(envSnapshot)
         }
       })(),
     )
 
-    yield* inner
+    if (this.options.env) this.activeEnvTurns.add(inner)
+    try {
+      yield* inner
+    } finally {
+      this.activeEnvTurns.delete(inner)
+    }
   }
 
   getMessages(): SDKMessage[] {
@@ -413,6 +571,8 @@ class SDKSessionImpl implements SDKSession {
   }
 
   close(): void {
+    if (this.closed) return
+    this.closed = true
     const wrapperController = this._abortController
     this.interruptWithSource('sdk_close')
     if (wrapperController) {
@@ -429,7 +589,15 @@ class SDKSessionImpl implements SDKSession {
         })
       }
     }
-    // Clear engine and store references to prevent memory leaks
+    // Return paused turns and let pending engine work unwind before their
+    // finally restores the environment. A queued turn has not acquired an
+    // environment and must not keep the closed session's engine/store alive.
+    for (const turn of this.activeEnvTurns) {
+      void turn.return(undefined).catch(error => {
+        console.warn('SDK: turn cleanup error:', error)
+      })
+    }
+    this.activeEnvTurns.clear()
     this._engine = null
     this._appStateStore = null
   }

tests/sdk/package-consumer-types.test.ts

@@ -10,7 +10,7 @@
  * - Self-referential type wrappers
  */
 import { afterAll, describe, expect, test } from 'bun:test'
-import { execFile, execSync } from 'child_process'
+import { execFile, execFileSync, execSync } from 'child_process'
 import { promisify } from 'node:util'
 import { existsSync, mkdirSync, rmSync, writeFileSync, cpSync, readFileSync } from 'fs'
 import { join } from 'path'
@@ -112,17 +112,18 @@ async function tsc(tmpDir: string, compiler = TSC_BIN, timeout = 20_000): Promis
   }
 }
 
-/** Build the CLI and SDK when package-export validation needs missing distribution artifacts. */
+let artifactsBuilt = false
+
+/** Build current sources once before validating package exports and runtime behavior. */
 function ensureBuildArtifacts(): void {
-  const distCli = join(ROOT, 'dist', 'cli.mjs')
-  const distSdk = join(ROOT, 'dist', 'sdk.mjs')
-  if (existsSync(distCli) && existsSync(distSdk)) return
+  if (artifactsBuilt) return
 
   execSync('bun run build', {
     cwd: ROOT,
     stdio: ['ignore', 'pipe', 'pipe'],
     timeout: 180000,
   })
+  artifactsBuilt = true
 }
 
 afterAll(() => {
@@ -361,5 +362,26 @@ describe('package exports resolution', () => {
     expect(existsSync(join(ROOT, 'dist', 'cli.mjs'))).toBe(true)
     expect(existsSync(join(ROOT, 'dist', 'sdk.mjs'))).toBe(true)
     expect(existsSync(join(ROOT, 'src', 'entrypoints', 'sdk.d.ts'))).toBe(true)
-  })
+  }, 180_000)
 })
+
+test('built Node SDK preserves absence of inherited environment names', () => {
+  ensureBuildArtifacts()
+  const sdkUrl = new URL('../../dist/sdk.mjs', import.meta.url).href
+  const result = execFileSync('node', ['--input-type=module', '-e', `
+    import { unstable_v2_createSession } from ${JSON.stringify(sdkUrl)};
+    const keys = ['__proto__', 'constructor', 'toString'];
+    for (const key of keys) delete process.env[key];
+    const session = unstable_v2_createSession({
+      cwd: process.cwd(), env: Object.fromEntries(keys.map(key => [key, 'during'])),
+    });
+    const snapshot = await session.applySessionEnv();
+    session.restoreSessionEnv(snapshot);
+    session.close();
+    if (keys.some(key => Object.hasOwn(process.env, key))) {
+      throw new Error('restoration created an originally absent environment entry');
+    }
+    console.log('restored absence');
+  `], { cwd: ROOT, encoding: 'utf8', timeout: 20_000 })
+  expect(result.trim()).toBe('restored absence')
+}, 180_000)