diff --git a/.env.example b/.env.example
index 3eff20b..f7aca61 100644
--- a/.env.example
+++ b/.env.example
@@ -3,14 +3,19 @@
 
 # ---------------------------------------------------------------- team card --
 
-# Service-account API key (API Key v2). One key is used everywhere: the Orca
-# Agent Engine API, Kafka, Schema Registry, and the StreamNative MCP server.
+# Service-account API key (API Key v2) for the hosted Agent Engine API,
+# Kafka and Schema Registry. OAuth MCP servers use a separate browser login.
 SN_API_KEY=
 
 # Service-account principal, used as the Kafka SASL username.
 # Looks like: <service-account>@<org>.auth.streamnative.cloud
 SN_SERVICE_ACCOUNT=
 
+# Optional separate Registry workspace key for ork local (sent as x-api-key).
+# Leave empty for a hosted team card; SN_API_KEY then authenticates Agent Engine.
+# This key does not authenticate Kafka, Schema Registry, or StreamNative MCP.
+ORCA_API_KEY=
+
 # Agent Engine registry endpoint (the External one). Host root only, no /v1.
 # Looks like: https://<workspace-host>
 ORCA_BASE_URL=
@@ -22,10 +27,24 @@ SCHEMA_REGISTRY_URL=
 # StreamNative MCP server for your SQL Workspace (the agent's data tools).
 SN_MCP_URL=
 
+# MCP authentication: oauth (default) or static_bearer for API-key MCP servers.
+# All three tutorial paths call ork for the first OAuth login, then reuse the
+# credential stored in the vault. Install ork with the OAuth discovery support
+# from orca-cli PR #8 (or current main).
+SN_MCP_AUTH=oauth
+
+# Optional authorization server selection. Leave empty for automatic discovery.
+# If multiple servers are advertised, copy one exact authorization_servers value
+# from the MCP protected-resource metadata; do not substitute the metadata issuer.
+SN_MCP_OAUTH_ISSUER=
+SN_MCP_OAUTH_SCOPE="openid profile email offline_access"
+
 # ------------------------------------------------------------- your choices --
 
-# The preloaded login topic.
-LOGIN_TOPIC=avro.security.login_events
+# The Kafka topic name. Injectors and doctor read this value.
+# L2 SQL files use the default below: edit their quoted "avro.<LOGIN_TOPIC>"
+# source name to match this value before running them in SQL Workspace.
+LOGIN_TOPIC=security.login_events
 
 # The model your agent runs on (served by the event's AI gateway).
 ORCA_MODEL=claude-sonnet-4-6
diff --git a/README.md b/README.md
index d4dcb32..f573602 100644
--- a/README.md
+++ b/README.md
@@ -18,7 +18,7 @@ them from live data, and flag the account once you say so.
 
 ```mermaid
 flowchart LR
-    K["Kafka topic<br/>avro.security.login_events"] --> S["SQL Workspace<br/>materialized view<br/>login_failures"]
+    K["Kafka topic<br/>security.login_events"] --> S["SQL Workspace<br/>materialized view<br/>login_failures"]
     J["inject<br/>(you, in L3)"] -- "new login burst" --> K
     S -- "StreamNative MCP<br/>sql_workspace_query" --> A["Orca agent<br/>hello-agent-&lt;you&gt;"]
     A -- "sql_workspace_insert_rows<br/>(only if you approve)" --> F["SQL table<br/>flagged_accounts"]
@@ -26,7 +26,7 @@ flowchart LR
 
 | Step | Time | Where | You | The idea |
 |---|---|---|---|---|
-| [0. Connect](#step-0-connect-3-min) | 3 min | terminal | Fill in `.env`, run the doctor | One key, checked end to end |
+| [0. Connect](#step-0-connect-3-min) | 3 min | terminal | Fill in `.env`, run the doctor | Check service access; authorize MCP with OAuth |
 | [L1. Hello, agent](#l1-hello-agent-5-min) | 5 min | CLI / Python / TS | Create an agent and chat | Agent, environment, session, events |
 | [L2. Hello, streaming SQL](#l2-hello-streaming-sql-8-min) | 8 min | SQL Workspace | Build a materialized view over the topic | Context that keeps itself fresh |
 | [L3. Agent + live context](#l3-agent--live-context-9-min) | 9 min | CLI / Python / TS | Give the agent SQL tools, inject new data | The answer changes with the data |
@@ -54,8 +54,52 @@ Go to your path's folder and run the doctor:
 | TypeScript | `cd typescript && npm run doctor` |
 | CLI | `cd cli`, and run the doctor from your helper language: `(cd ../python && .venv/bin/python doctor.py)` or `(cd ../typescript && npm run doctor)` |
 
-Every line should say `PASS`. A failed check prints its fix. Still stuck after
-two tries? Raise your hand.
+The service checks should say `PASS`. Before the first OAuth login, the MCP
+check asks you to run L3; that script opens your browser and stores the credential
+in a vault. After completing L2 and running L3, rerun the doctor to validate the
+stored OAuth credential. This check verifies MCP initialization; L3/L4 exercise
+the actual SQL tools. A failed check prints its fix. Still stuck after two tries?
+Raise your hand.
+
+For StreamNative SQL Workspace MCP, keep `SN_MCP_AUTH=oauth`, leave
+`SN_MCP_OAUTH_ISSUER` empty for automatic discovery, and use the scope from
+`.env.example`. Use an `ork` build containing [PR #8](https://github.com/orca-ae/orca-cli/pull/8)
+or current main. Its discovery accepts HTTPS issuer aliases within the same
+registrable domain and port. Only set `SN_MCP_OAUTH_ISSUER` when selecting one
+of multiple advertised `authorization_servers`; copy that advertised value
+exactly rather than the final issuer in authorization-server metadata.
+`SN_API_KEY` authenticates the hosted Agent Engine,
+Kafka and Schema Registry; it is not the OAuth MCP access token. All three paths
+use `ork` for the first MCP login, then reuse the live credential for the same URL
+and auth type from `.orca-state/<participant>.json`. Tokens stay in the server-side
+vault, where they can be refreshed; they are never written to `.env` or local state.
+Set `SN_MCP_AUTH=static_bearer` only when your MCP server accepts `SN_API_KEY`.
+Changing the auth mode archives the previous live credential for that same URL
+before creating its replacement (the Registry permits one active credential per
+URL in a vault). If authorization fails, rerun L3/L4 to finish setup; other URLs'
+credentials are preserved. A local Agent Engine with OAuth MCP needs only
+`ORCA_API_KEY` for Registry authentication; `SN_API_KEY` is still needed for Kafka
+and Schema Registry.
+
+### Use a local Agent Engine
+
+Start the CLI's stack with a provider key in your shell:
+
+```bash
+export ANTHROPIC_API_KEY='<your-provider-key>'
+ork local start --with-gateway
+```
+
+Set `ORCA_BASE_URL=http://127.0.0.1:8080` in the tutorial's `.env`, and copy the
+workspace key from the file printed by `ork local start` into `ORCA_API_KEY`.
+The tutorial sends this key as `x-api-key`. A hosted team card continues to use
+`SN_API_KEY` as a Bearer token when `ORCA_API_KEY` is empty.
+
+For L1, run `python doctor.py --agent-only` or `npm run doctor -- --agent-only`.
+This checks the Agent Engine without requiring Kafka, Schema Registry, or MCP.
+The local stack provides the Agent Engine and AI Gateway; L2–L4 still need the
+streaming data services from your team card. For L3/L4, keep `SN_API_KEY` set to
+the MCP service key, separately from the local Registry's `ORCA_API_KEY`.
 
 ## L1: Hello, agent (5 min)
 
@@ -150,6 +194,19 @@ create a new version when its definition changes.
 
 In the StreamNative Cloud console, open **SQL Workspace**, select the hackathon
 workspace, and pick your team's database. Use a new query tab for each step.
+The default Kafka topic is `security.login_events`; SQL Workspace exposes its
+Avro source as `"avro.security.login_events"`.
+
+**Align the SQL with your `.env` before running it.** The injectors and doctor use
+`LOGIN_TOPIC`, but the SQL files and examples below contain a fixed source name:
+SQL Workspace does not read your local `.env`. Check `LOGIN_TOPIC`, then replace
+`"avro.security.login_events"` with `"avro.<your LOGIN_TOPIC>"` in both
+[`sql/01_explore.sql`](sql/01_explore.sql) and
+[`sql/02_login_failures.sql`](sql/02_login_failures.sql), and in any query copied
+from this page. For example, `LOGIN_TOPIC=security.team07_logins` requires
+`FROM "avro.security.team07_logins"`. Keep the double quotes around the entire
+source name and confirm that SQL Workspace imported that topic as an Avro source.
+Keep the `login_failures` view name: L3/L4 query that view.
 
 **1. Peek at the stream** ([`sql/01_explore.sql`](sql/01_explore.sql)). Each row
 is one login attempt. The topic name contains dots, so it's double-quoted.
@@ -228,8 +285,10 @@ event landed in Kafka, the view updated itself, and the agent read the view.
 - `mcp_servers`: the StreamNative MCP server for your SQL Workspace.
 - `tools`: an allow-list. Two read-only tools run without asking
   (`always_allow`); every other tool on that server is disabled.
-- A **vault**: the MCP server's credential (your team key) is stored server-side.
-  The session references the vault by id, so the key never enters the prompt.
+- A **vault**: the MCP server's OAuth credential is created through `ork` and
+  stored server-side. Approve the browser login on the first run. The session
+  references the vault by id, so tokens never enter the prompt. Later runs reuse
+  the credential without another browser login.
 
 <details>
 <summary>The code (Python)</summary>
@@ -238,7 +297,7 @@ event landed in Kafka, the view updated itself, and the agent read the view.
 layer = load_layer("l3-live-context")
 agent = ensure_agent(client, state, agent_params(layer, config))
 
-vault_id = ensure_vault(client, state, f"hello-vault-{config.participant}", config["SN_MCP_URL"], config["SN_API_KEY"])
+vault_id = ensure_vault(client, state, f"hello-vault-{config.participant}", config)
 session = client.sessions.create(
     environment_id=environment_id,
     agent={"type": "agent", "id": agent.id, "version": agent.version},
@@ -256,7 +315,7 @@ chat(client, session.id, QUESTION)
 const layer = loadLayer('l3-live-context');
 const agent = await ensureAgent(client, state, agentParams(layer, config));
 
-const vaultId = await ensureVault(client, state, `hello-vault-${config.participant}`, config.get('SN_MCP_URL'), config.get('SN_API_KEY'));
+const vaultId = await ensureVault(client, state, `hello-vault-${config.participant}`, config);
 const session = await client.sessions.create({
   environment_id: environmentId,
   agent: { type: 'agent', id: agent.id, version: agent.version },
@@ -278,7 +337,8 @@ ork agent update "$AGENT_ID" --version 1 --model "$ORCA_MODEL" \
 
 ork agent vaults create --display-name hello-vault-ana -o json
 ork agent vaults credentials create --vault "$VAULT_ID" --display-name streamnative-mcp \
-  --auth-json '{"type":"static_bearer","mcp_server_url":"<SN_MCP_URL>","token":"<SN_API_KEY>"}'
+  --mcp-server-url "$SN_MCP_URL" \
+  --oauth-scope "$SN_MCP_OAUTH_SCOPE" -o json
 
 ork agent sessions create --agent "$AGENT_ID" --agent-version 2 \
   --environment-id "$ENVIRONMENT_ID" --vault-id "$VAULT_ID" --title "L3: live context" -o json
@@ -292,13 +352,22 @@ ork agent sessions create --agent "$AGENT_ID" --agent-version 2 \
 | `./l4_act.sh` | `python l4_act.py` | `npm run l4` |
 
 The agent (version 3) gets one write tool, and it can only use it with your
-approval. It queries the view, then proposes an insert, and the session pauses:
+approval. It queries the view, describes the flag table, and reads the database
+time before proposing an insert. The MCP insert tool requires every writable
+column, including nullable columns; it does not apply table defaults. The
+session pauses before the proposed row is written:
 
 ```
 [approve?] The agent wants to run sql_workspace_insert_rows with:
 {
+  "database": "<your database>",
+  "schema": "public",
   "table": "flagged_accounts",
-  "rows": [{"account_id": "acct_9…", "reason": "6 failed logins then a success from one new IP"}]
+  "rows": [{
+    "account_id": "acct_9…",
+    "reason": "6 failed logins then a success from one new IP",
+    "flagged_at": "2026-09-30T12:00:00Z"
+  }]
 }
 Allow it? [y/N]
 ```
@@ -309,11 +378,12 @@ Type `y`, then check in SQL Workspace:
 SELECT * FROM flagged_accounts;
 ```
 
-Ask again, and answer `n` this time. The agent is told a human denied the insert,
+Ask the agent to flag a different account, and answer `n` this time. The agent is told a human denied the insert,
 and it does not retry.
 
-**What changed** ([`agent/l4-act.json`](agent/l4-act.json)): one more tool,
-`sql_workspace_insert_rows`, with `permission_policy: always_ask`. When the agent
+**What changed** ([`agent/l4-act.json`](agent/l4-act.json)): the read-only
+`sql_workspace_describe_table` checks the required columns, and
+`sql_workspace_insert_rows` uses `permission_policy: always_ask`. When the agent
 calls it, the session emits `agent.mcp_tool_use` and goes idle with
 `stop_reason: requires_action`. Your script answers with a
 `user.tool_confirmation`: `allow`, or `deny` with a reason. On the CLI that is:
@@ -341,9 +411,10 @@ action, and you have your hackathon project. Ideas and next steps:
 | Doctor: `Agent Engine HTTP 401/403` | The key was rejected. A key created before its permissions must be re-created: ask a facilitator. |
 | Doctor: `Kafka ... authentication` | `SN_SERVICE_ACCOUNT` must be the full principal, `<name>@<org>.auth.streamnative.cloud`; `SN_API_KEY` is the raw key. |
 | The login topic isn't listed in SQL Workspace | Only topics with a registered Avro schema appear. Ask a facilitator. |
-| `relation "avro.security.login_events" does not exist` | Select your team's database, and keep the double quotes around the name. |
+| `relation "avro.security.login_events" does not exist` | Select your team's database and update the quoted Avro source in both L2 SQL files to match `LOGIN_TOPIC` in `.env`. |
 | The agent can't find `login_failures` | Create the view in your team's database (L2, step 2); the agent looks it up there. |
-| `[error]` lines from MCP tools in L3 | Check `SN_MCP_URL` against your team card, then rerun the doctor. |
+| `[error]` lines from MCP tools in L3 | Check `SN_MCP_URL` and `SN_MCP_AUTH`, finish the OAuth login, then rerun the doctor. |
+| OAuth issuer mismatch / unsupported client authentication | Use current `ork` main or PR #8 and leave `SN_MCP_OAUTH_ISSUER` empty for StreamNative discovery. An explicit issuer must match an advertised authorization server. `--oauth-allow-issuer-mismatch` is only for trusted servers whose metadata issuer crosses registrable domains; StreamNative does not need it. |
 | `Cannot reach the Agent Engine` | `ORCA_BASE_URL` must be the host root from your card, with no `/v1`. |
 | The agent answers from memory instead of querying | Ask again, "check the view first". The system prompt tells it to always query. |
 
@@ -353,8 +424,9 @@ action, and you have your hackathon project. Ideas and next steps:
 |---|---|---|
 | `./cleanup.sh` | `python cleanup.py` | `npm run cleanup` |
 
-This archives your agent and deletes your vault and environment. To start L2
-over, run [`sql/99_reset.sql`](sql/99_reset.sql).
+This archives your agent and environment, and deletes your vault. An environment
+with session history cannot be deleted; archiving keeps that history available.
+To start L2 over, run [`sql/99_reset.sql`](sql/99_reset.sql).
 
 ## What's in this repository
 
diff --git a/agent/l4-act.json b/agent/l4-act.json
index 081afdd..addf6dc 100644
--- a/agent/l4-act.json
+++ b/agent/l4-act.json
@@ -1,19 +1,20 @@
 {
   "layer": "L4",
   "summary": "+ insert into flagged_accounts, only with human approval",
-  "system": "You are a security analyst for Aegis Financial, a fictional bank. Your context is live: logins stream into Kafka, and a streaming SQL materialized view keeps a running summary in StreamNative SQL Workspace, which you query with the streamnative tools.\n\nRules:\n1. Always query before you answer. Never guess or reuse numbers from earlier answers: the data changes while you talk.\n2. First call sql_workspace_list_databases, then use the database that contains login_failures.\n3. login_failures has one row per account: account_id, failed_logins, successful_logins, distinct_ips, last_seen.\n4. Several failed logins (5 or more) plus at least one success is a likely account takeover.\n5. Cite the numbers you used. Keep answers under 120 words.\n\nActing:\n6. When asked to flag an account, insert exactly one row into the table flagged_accounts with sql_workspace_insert_rows: account_id, and reason (one sentence that cites the numbers).\n7. A human approves every insert. If an insert is denied, say so, and do not retry it.",
+  "system": "You are a security analyst for Aegis Financial, a fictional bank. Your context is live: logins stream into Kafka, and a streaming SQL materialized view keeps a running summary in StreamNative SQL Workspace, which you query with the streamnative tools.\n\nRules:\n1. Always query before you answer. Never guess or reuse numbers from earlier answers: the data changes while you talk.\n2. First call sql_workspace_list_databases, then use the database that contains login_failures.\n3. login_failures has one row per account: account_id, failed_logins, successful_logins, distinct_ips, last_seen.\n4. Several failed logins (5 or more) plus at least one success is a likely account takeover.\n5. Cite the numbers you used. Keep answers under 120 words.\n\nActing:\n6. When asked to flag an account, first call sql_workspace_describe_table for public.flagged_accounts. Every inserted row must include all writable columns, including nullable columns: this MCP tool does not apply table defaults. In this tutorial, supply account_id, reason (one sentence citing fresh login counts), and flagged_at. Read CURRENT_TIMESTAMP AS flagged_at with sql_workspace_query and use the returned RFC3339 timestamp as a literal; never invent a time or pass a SQL expression.\n7. Submit exactly one sql_workspace_insert_rows tool call with database, schema=public, table=flagged_accounts, and one complete row. Calling the tool proposes the action: Orca pauses it for human approval. Do not ask for chat approval before submitting the tool call.\n8. If approval is denied, say so and do not retry. A tool rejection is different from denied human approval. Check the tool outcome and visibility, and verify flagged_accounts with a read-only query after an accepted insert. Never automatically resubmit an insert after an error, warning, or unknown outcome.",
   "mcp_servers": [
-    { "name": "streamnative", "type": "url", "url": "${SN_MCP_URL}" }
+    {"name": "streamnative", "type": "url", "url": "${SN_MCP_URL}"}
   ],
   "tools": [
     {
       "type": "mcp_toolset",
       "mcp_server_name": "streamnative",
-      "default_config": { "enabled": false },
+      "default_config": {"enabled": false},
       "configs": [
-        { "name": "sql_workspace_list_databases", "enabled": true, "permission_policy": { "type": "always_allow" } },
-        { "name": "sql_workspace_query", "enabled": true, "permission_policy": { "type": "always_allow" } },
-        { "name": "sql_workspace_insert_rows", "enabled": true, "permission_policy": { "type": "always_ask" } }
+        {"name": "sql_workspace_list_databases", "enabled": true, "permission_policy": {"type": "always_allow"}},
+        {"name": "sql_workspace_query", "enabled": true, "permission_policy": {"type": "always_allow"}},
+        {"name": "sql_workspace_describe_table", "enabled": true, "permission_policy": {"type": "always_allow"}},
+        {"name": "sql_workspace_insert_rows", "enabled": true, "permission_policy": {"type": "always_ask"}}
       ]
     }
   ]
diff --git a/cli/cleanup.sh b/cli/cleanup.sh
index 5240f3d..064d443 100755
--- a/cli/cleanup.sh
+++ b/cli/cleanup.sh
@@ -8,7 +8,7 @@
 set -euo pipefail
 # shellcheck source=lib.sh
 . "$(dirname "$0")/lib.sh"
-hello_setup ORCA_BASE_URL SN_API_KEY
+hello_setup ORCA_BASE_URL
 
 remove() {  # remove <label> <state key> <ork command...>
   local id
@@ -26,5 +26,5 @@ remove() {  # remove <label> <state key> <ork command...>
 # Agents cannot be deleted, only archived.
 remove agent agent_id agent archive
 remove vault vault_id agent vaults delete
-remove environment environment_id agent environments delete
+remove environment environment_id agent environments archive
 rm -f "$HELLO_STATE_FILE"
diff --git a/cli/env.sh b/cli/env.sh
index 811fdaa..b8ae592 100644
--- a/cli/env.sh
+++ b/cli/env.sh
@@ -75,13 +75,17 @@ hello_setup() {
     hello_die "jq is not installed. Install it (brew install jq, apt install jq, or winget install jqlang.jq) and try again."
 
   hello_load_dotenv
-  hello_require "$@"
-
-  # ork reads ORCA_REGISTRY_URL and ORCA_ACCESS_TOKEN (sent as a Bearer token),
-  # and refuses to start if ORCA_API_KEY is also set.
+  # Registry workspace keys use x-api-key; team-card keys use Bearer.
+  # Keep the two CLI credentials mutually exclusive.
+  if [ -n "$(hello_trim "${ORCA_API_KEY:-}")" ]; then
+    hello_require "$@" ORCA_API_KEY
+    unset ORCA_ACCESS_TOKEN
+  else
+    hello_require "$@" SN_API_KEY
+    export ORCA_ACCESS_TOKEN="$SN_API_KEY"
+    unset ORCA_API_KEY
+  fi
   export ORCA_REGISTRY_URL="$ORCA_BASE_URL"
-  export ORCA_ACCESS_TOKEN="$SN_API_KEY"
-  unset ORCA_API_KEY
 
   local who
   who=$(printenv PARTICIPANT || true)
diff --git a/cli/l1_hello.sh b/cli/l1_hello.sh
index 9f2237d..d823b66 100755
--- a/cli/l1_hello.sh
+++ b/cli/l1_hello.sh
@@ -11,7 +11,7 @@
 set -euo pipefail
 # shellcheck source=lib.sh
 . "$(dirname "$0")/lib.sh"
-hello_setup ORCA_BASE_URL SN_API_KEY ORCA_MODEL
+hello_setup ORCA_BASE_URL ORCA_MODEL
 
 QUESTION="Hi! What is the Data + Agent Hackathon, and what can you see right now?"
 [ $# -eq 0 ] || QUESTION="$*"
diff --git a/cli/l3_live_context.sh b/cli/l3_live_context.sh
index 6680c45..2ab3c77 100755
--- a/cli/l3_live_context.sh
+++ b/cli/l3_live_context.sh
@@ -12,7 +12,7 @@
 set -euo pipefail
 # shellcheck source=lib.sh
 . "$(dirname "$0")/lib.sh"
-hello_setup ORCA_BASE_URL SN_API_KEY ORCA_MODEL SN_MCP_URL
+hello_setup ORCA_BASE_URL ORCA_MODEL SN_MCP_URL
 
 QUESTION="Which accounts look like an account takeover right now?"
 
@@ -25,8 +25,8 @@ echo "$AGENT_NAME v$AGENT_VERSION: $(jq -r .summary "$(layer_file l3-live-contex
 
 # The MCP server needs a credential. It goes in a vault, never in the prompt.
 #      ork agent vaults create --display-name hello-vault-<you>
-#      ork agent vaults credentials create --vault <id> --auth-json '{"type":"static_bearer",...}'
-ensure_vault "hello-vault-$HELLO_PARTICIPANT" "$SN_MCP_URL" "$SN_API_KEY"
+#      ork agent vaults credentials create --vault <id> --mcp-server-url <SN_MCP_URL>
+ensure_vault "hello-vault-$HELLO_PARTICIPANT"
 
 #      ork agent sessions create ... --vault-id <id>
 create_session "L3: live context" "$VAULT_ID"
diff --git a/cli/l4_act.sh b/cli/l4_act.sh
index c7f3597..5811b87 100755
--- a/cli/l4_act.sh
+++ b/cli/l4_act.sh
@@ -11,7 +11,7 @@
 set -euo pipefail
 # shellcheck source=lib.sh
 . "$(dirname "$0")/lib.sh"
-hello_setup ORCA_BASE_URL SN_API_KEY ORCA_MODEL SN_MCP_URL
+hello_setup ORCA_BASE_URL ORCA_MODEL SN_MCP_URL
 
 REQUEST="Flag the account most likely to be under attack right now."
 
@@ -21,7 +21,7 @@ ensure_environment "hello-env-$HELLO_PARTICIPANT"
 ensure_agent l4-act
 echo "$AGENT_NAME v$AGENT_VERSION: $(jq -r .summary "$(layer_file l4-act)")"
 
-ensure_vault "hello-vault-$HELLO_PARTICIPANT" "$SN_MCP_URL" "$SN_API_KEY"
+ensure_vault "hello-vault-$HELLO_PARTICIPANT"
 create_session "L4: act with approval" "$VAULT_ID"
 
 # When the session pauses for approval, you decide, and the script answers with
diff --git a/cli/lib.sh b/cli/lib.sh
index 243ff28..c9e893e 100644
--- a/cli/lib.sh
+++ b/cli/lib.sh
@@ -188,8 +188,14 @@ ensure_environment() {  # ensure_environment <name>
 
 # Sets VAULT_ID: a vault holding the credential the agent uses to call the MCP
 # server. The token stays in the vault; it never enters a prompt (or your terminal).
-ensure_vault() {  # ensure_vault <name> <mcp url> <token>
-  local json credentials auth
+ensure_vault() {  # ensure_vault <name>
+  local json credentials auth auth_type credential_id
+  local -a oauth_args
+  case "${SN_MCP_AUTH:-oauth}" in
+    oauth) auth_type=mcp_oauth ;;
+    static_bearer) auth_type=static_bearer; hello_require SN_API_KEY ;;
+    *) hello_die "SN_MCP_AUTH must be oauth or static_bearer." ;;
+  esac
   VAULT_ID=$(state_get vault_id)
   if [ -z "$VAULT_ID" ] || ! ork_get_live agent vaults get "$VAULT_ID"; then
     json=$(ork_json agent vaults create --display-name "$1") || ork_fail
@@ -197,11 +203,25 @@ ensure_vault() {  # ensure_vault <name> <mcp url> <token>
     state_set vault_id "$VAULT_ID"
   fi
   credentials=$(ork_json agent vaults credentials list --vault "$VAULT_ID") || ork_fail
-  if ! jq -e --arg url "$2" '[(.data? // .)[]? | select(.auth.mcp_server_url == $url and .archived_at == null)] | length > 0' \
+  if ! jq -e --arg url "$SN_MCP_URL" --arg type "$auth_type" '[(.data? // .)[]? | select(.auth.mcp_server_url == $url and .auth.type == $type and .archived_at == null)] | length > 0' \
     <<<"$credentials" >/dev/null; then
-    auth=$(jq -cn --arg url "$2" --arg token "$3" '{type: "static_bearer", mcp_server_url: $url, token: $token}')
-    ork_json agent vaults credentials create --vault "$VAULT_ID" --display-name streamnative-mcp --auth-json "$auth" \
-      >/dev/null || ork_fail
+    # Registry permits one active credential per MCP URL: retire the old auth mode.
+    while IFS= read -r credential_id; do
+      [ -z "$credential_id" ] || ork agent vaults credentials archive "$credential_id" --vault "$VAULT_ID" >/dev/null ||
+        hello_die "Cannot archive the previous MCP credential. Check the ork error above."
+    done < <(jq -r --arg url "$SN_MCP_URL" '(.data? // .)[]? | select(.auth.mcp_server_url == $url and .archived_at == null) | .id' <<<"$credentials")
+    if [ "$auth_type" = mcp_oauth ]; then
+      oauth_args=(--vault "$VAULT_ID" --display-name streamnative-mcp --mcp-server-url "$SN_MCP_URL")
+      [ -z "${SN_MCP_OAUTH_ISSUER:-}" ] || oauth_args+=(--oauth-issuer "$SN_MCP_OAUTH_ISSUER")
+      [ -z "${SN_MCP_OAUTH_SCOPE:-}" ] || oauth_args+=(--oauth-scope "$SN_MCP_OAUTH_SCOPE")
+      # Keep the browser URL and callback progress visible; tokens go directly to the vault.
+      ork agent vaults credentials create "${oauth_args[@]}" -o json ||
+        hello_die "MCP OAuth authorization failed. Check the ork error above; normally leave SN_MCP_OAUTH_ISSUER empty for discovery, then rerun L3/L4."
+    else
+      auth=$(jq -cn --arg url "$SN_MCP_URL" --arg token "$SN_API_KEY" '{type: "static_bearer", mcp_server_url: $url, token: $token}')
+      ork_json agent vaults credentials create --vault "$VAULT_ID" --display-name streamnative-mcp --auth-json "$auth" \
+        >/dev/null || ork_fail
+    fi
   fi
 }
 
diff --git a/cli/tests/fake-ork b/cli/tests/fake-ork
index 8498a60..ffd4258 100755
--- a/cli/tests/fake-ork
+++ b/cli/tests/fake-ork
@@ -19,6 +19,12 @@ set -euo pipefail
 dir=${FAKE_ORK_DIR:?FAKE_ORK_DIR is not set}
 mkdir -p "$dir/agents" "$dir/envs" "$dir/vaults" "$dir/creds" "$dir/sessions" "$dir/reactions"
 jq -cn '$ARGS.positional' --args -- "$@" >>"$dir/calls.log"
+api_key_set=false
+access_token_set=false
+[ -z "${ORCA_API_KEY:-}" ] || api_key_set=true
+[ -z "${ORCA_ACCESS_TOKEN:-}" ] || access_token_set=true
+jq -cn --argjson api "$api_key_set" --argjson bearer "$access_token_set" \
+  '{api_key_set: $api, access_token_set: $bearer}' >"$dir/auth.json"
 
 fail() {  # fail <status> <method> <path>
   printf 'Error: failed: %s http://fake.invalid%s returned status %s: {"type":"error","error":{"type":"error","message":"fake %s"}}\n' \
@@ -42,6 +48,7 @@ now() {
 
 words=()
 name="" model="" system="" version="" display="" auth="" vault="" agent="" agent_version="" env="" title=""
+mcp_url=""
 session="" text="" tool_use="" decision="" deny="" cursor="" timeout=""
 mcp=() tools=() metadata=() vault_ids=()
 while [ $# -gt 0 ]; do
@@ -55,6 +62,9 @@ while [ $# -gt 0 ]; do
     --metadata) metadata+=("$2") ;;
     --display-name) display=$2 ;;
     --auth-json) auth=$2 ;;
+    --mcp-server-url) mcp_url=$2 ;;
+    --oauth-issuer) : ;;
+    --oauth-scope) : ;;
     --vault) vault=$2 ;;
     --agent) agent=$2 ;;
     --agent-version) agent_version=$2 ;;
@@ -164,6 +174,12 @@ case "$command" in
     [ -f "$dir/envs/${words[3]}.json" ] || fail 404 GET "/v1/environments/${words[3]}"
     cat "$dir/envs/${words[3]}.json"
     ;;
+  "agent environments archive "*)
+    [ -f "$dir/envs/${words[3]}.json" ] || fail 404 POST "/v1/environments/${words[3]}/archive"
+    jq --arg now "$(now)" '.archived_at = $now' "$dir/envs/${words[3]}.json" >"$dir/env.tmp"
+    mv "$dir/env.tmp" "$dir/envs/${words[3]}.json"
+    echo "archived ${words[3]}"
+    ;;
   "agent environments delete "*)
     [ -f "$dir/envs/${words[3]}.json" ] || fail 404 DELETE "/v1/environments/${words[3]}"
     rm "$dir/envs/${words[3]}.json"
@@ -187,8 +203,23 @@ case "$command" in
     [ -f "$dir/creds/$vault.json" ] || echo '[]' >"$dir/creds/$vault.json"
     jq '{data: .}' "$dir/creds/$vault.json"
     ;;
+  "agent vaults credentials archive "*)
+    file="$dir/creds/$vault.json"
+    [ -f "$file" ] || fail 404 POST "/v1/vaults/$vault/credentials/${words[4]}/archive"
+    jq --arg id "${words[4]}" 'map(if .id == $id then .archived_at = "old" else . end)' "$file" >"$file.tmp"
+    mv "$file.tmp" "$file"
+    echo "archived ${words[4]}"
+    ;;
   "agent vaults credentials create")
     [ -f "$dir/creds/$vault.json" ] || echo '[]' >"$dir/creds/$vault.json"
+    if [ -n "$mcp_url" ]; then
+      [ -z "$auth" ] || { echo "fake-ork: auth modes are mutually exclusive" >&2; exit 2; }
+      auth=$(jq -cn --arg url "$mcp_url" '{type: "mcp_oauth", mcp_server_url: $url}')
+    fi
+    # Enforce the real Registry's one-active-credential-per-URL constraint.
+    if jq -e --arg url "$(jq -r .mcp_server_url <<<"$auth")" 'any(.archived_at == null and .auth.mcp_server_url == $url)' "$dir/creds/$vault.json" >/dev/null; then
+      fail 409 POST "/v1/vaults/$vault/credentials"
+    fi
     n=$(counter cred)
     credential=$(jq -c --arg id "cred_$n" --arg vault "$vault" --arg name "$display" \
       '{id: $id, type: "vault_credential", vault_id: $vault, display_name: $name, auth: del(.token), archived_at: null}' <<<"$auth")
diff --git a/cli/tests/run.sh b/cli/tests/run.sh
index 8d21837..935ada0 100755
--- a/cli/tests/run.sh
+++ b/cli/tests/run.sh
@@ -19,13 +19,14 @@ ln -s "$TESTS/fake-ork" "$WORK/bin/ork"
 export PATH="$WORK/bin:$PATH"
 # Nothing from the developer's own shell may leak into the tests.
 unset SN_API_KEY SN_SERVICE_ACCOUNT ORCA_BASE_URL KAFKA_BOOTSTRAP_SERVERS SCHEMA_REGISTRY_URL SN_MCP_URL \
-  LOGIN_TOPIC ORCA_MODEL PARTICIPANT ORCA_API_KEY ORCA_ACCESS_TOKEN ORCA_REGISTRY_URL
+  LOGIN_TOPIC ORCA_MODEL PARTICIPANT ORCA_API_KEY ORCA_ACCESS_TOKEN ORCA_REGISTRY_URL \
++  SN_MCP_AUTH SN_MCP_OAUTH_ISSUER SN_MCP_OAUTH_SCOPE
 export HELLO_ROUND_SECONDS=1 HELLO_TURN_TIMEOUT=5
 
 MCP_URL=https://mcp.example.com/mcp/x/o-test/sqlworkspace/ws-1
 SHA_L1=457f86a77738415f
 SHA_L3=ac4d0b08aca3f336
-SHA_L4=452e3876c295ffc8
+SHA_L4=8d424c704af22671
 
 PASSED=0
 FAILED=0
@@ -51,6 +52,7 @@ SN_API_KEY=test-key
 SN_SERVICE_ACCOUNT=team-07@o-test.auth.streamnative.cloud
 ORCA_BASE_URL=https://ws.example.com
 SN_MCP_URL=$MCP_URL
+SN_MCP_AUTH=static_bearer
 ORCA_MODEL=claude-sonnet-4-6
 PARTICIPANT=jane
 EOF
@@ -106,7 +108,7 @@ test_missing_team_card() {
   run "" l1_hello.sh
   check "exits 1 without a team card" [ "$STATUS" -eq 1 ]
   check "names every missing variable" \
-    err_has "Missing ORCA_BASE_URL, SN_API_KEY, ORCA_MODEL. Copy .env.example to .env in the repo root and fill it in from your team card."
+    err_has "Missing ORCA_BASE_URL, ORCA_MODEL, SN_API_KEY. Copy .env.example to .env in the repo root and fill it in from your team card."
   check "runs no ork command" [ ! -s "$FAKE_ORK_DIR/calls.log" ]
 }
 
@@ -116,6 +118,7 @@ test_l1_creates_everything() {
   reaction 1 "$(message evt_a 'Hello!')" "$(end_turn 1)"
   run "" l1_hello.sh
   check "L1 succeeds" [ "$STATUS" -eq 0 ]
+  check "hosted team cards use only Bearer" jq -e '.access_token_set and (.api_key_set | not)' "$FAKE_ORK_DIR/auth.json"
   check "creates the environment" called '["agent","environments","create","--name","hello-env-jane","-o","json"]'
   check "creates the agent with the L1 fingerprint" called_with "agent create" "definition_sha=$SHA_L1"
   check "names the agent after the participant" called_with "agent create" "hello-agent-jane"
@@ -176,7 +179,7 @@ test_l1_creates_everything() {
   check "cleanup succeeds" [ "$STATUS" -eq 0 ]
   check "archives the agent" out_has "removed agent agent_2"
   check "deletes the vault" out_has "removed vault vlt_1"
-  check "deletes the environment" out_has "removed environment env_1"
+  check "archives the environment" out_has "removed environment env_1"
   check "forgets the ids" [ ! -f "$R/.orca-state/jane.json" ]
 }
 
@@ -323,6 +326,68 @@ test_team_card_parsing() {
   check "state goes to the participant's file" [ -f "$R/.orca-state/jane-doe-42.json" ]
 }
 
+test_local_registry_key() {
+  fresh_repo local-key
+  cat >"$R/.env" <<EOF
+ORCA_BASE_URL=http://127.0.0.1:8080
+ORCA_API_KEY=local-test-key
+ORCA_MODEL=claude-sonnet-4-6
+PARTICIPANT=jane
+EOF
+  export ORCA_ACCESS_TOKEN=stale-bearer
+  reaction 1 "$(message evt_a 'Hello locally!')" "$(end_turn local)"
+  run "" l1_hello.sh
+  unset ORCA_ACCESS_TOKEN
+  check "local L1 needs no team-card key" [ "$STATUS" -eq 0 ]
+  check "local key uses only x-api-key" jq -e '.api_key_set and (.access_token_set | not)' "$FAKE_ORK_DIR/auth.json"
+}
+
+test_mcp_oauth() {
+  fresh_repo oauth
+  card
+  # Omitted SN_MCP_AUTH defaults to OAuth.
+  sed -i.bak '/SN_MCP_AUTH=/d' "$R/.env"
+  cat >>"$R/.env" <<EOF
+SN_MCP_OAUTH_ISSUER=https://auth.example.com/
+SN_MCP_OAUTH_SCOPE="openid profile email offline_access"
+EOF
+  export ORCA_API_KEY=stale-local ORCA_ACCESS_TOKEN=stale-bearer
+  # The configured local key has priority. Unset it to exercise hosted Bearer.
+  unset ORCA_API_KEY
+  reaction 1 "$(message evt_a 'Live data!')" "$(end_turn oauth)"
+  run "" l3_live_context.sh
+  unset ORCA_ACCESS_TOKEN
+  check "OAuth L3 succeeds" [ "$STATUS" -eq 0 ]
+  check "OAuth invokes native discovery and browser flow" called "$(jq -cn --arg url "$MCP_URL" '["agent","vaults","credentials","create","--vault","vlt_1","--display-name","streamnative-mcp","--mcp-server-url",$url,"--oauth-issuer","https://auth.example.com/","--oauth-scope","openid profile email offline_access","-o","json"]')"
+  check "OAuth stores no static bearer credential" jq -e '.[0].auth.type == "mcp_oauth"' "$FAKE_ORK_DIR/creds/vlt_1.json"
+  check "OAuth child uses only Registry Bearer" jq -e '.access_token_set and (.api_key_set | not)' "$FAKE_ORK_DIR/auth.json"
+  # shellcheck disable=SC2016  # the child shell expands its own positional argument
+  check "OAuth state contains no tokens" bash -c '! grep -q "test-key" "$1"' _ "$R/.orca-state/jane.json"
+
+  reaction 2 "$(message evt_b 'Still connected!')" "$(end_turn reuse)"
+  run "" l4_act.sh
+  check "L4 reuses L3 OAuth credential" [ "$(calls_of "agent vaults credentials create")" -eq 1 ]
+  check "OAuth L4 succeeds" [ "$STATUS" -eq 0 ]
+
+  jq '.[0].auth.type = "static_bearer"' "$FAKE_ORK_DIR/creds/vlt_1.json" >"$R/creds.tmp"
+  mv "$R/creds.tmp" "$FAKE_ORK_DIR/creds/vlt_1.json"
+  reaction 3 "$(message evt_c 'New OAuth credential!')" "$(end_turn replace)"
+  run "" l3_live_context.sh
+  check "static credentials do not satisfy OAuth mode" [ "$(calls_of "agent vaults credentials create")" -eq 2 ]
+
+  fresh_repo oauth-fail
+  card
+  export SN_MCP_AUTH=oauth
+  export FAKE_ORK_FAIL="agent vaults credentials create:401"
+  run "" l3_live_context.sh
+  unset SN_MCP_AUTH
+  check "OAuth failure stops L3" [ "$STATUS" -ne 0 ]
+  check "OAuth failure has setup guidance" err_has "MCP OAuth authorization failed"
+  check "no session after OAuth failure" [ "$(calls_of "agent sessions create")" -eq 0 ]
+}
+
+test_mcp_oauth
+test_local_registry_key
 test_missing_team_card
 test_l1_creates_everything
 test_turns_ignore_history_and_duplicates
diff --git a/docs/before-you-arrive.md b/docs/before-you-arrive.md
index 930a3c9..d08a0ff 100644
--- a/docs/before-you-arrive.md
+++ b/docs/before-you-arrive.md
@@ -9,8 +9,16 @@ can pick a different one.
 | **TypeScript** | Node.js 20 or newer |
 | **CLI** | [`ork`](https://github.com/orca-ae/orca-cli) (the Orca CLI) and [`jq`](https://jqlang.org/download/), plus Python 3.11+ *or* Node.js 20+ for two helper scripts |
 
-Everyone also needs `git` and a terminal. On Windows, use WSL or Git Bash for
-the CLI path.
+Everyone also needs `git`, a terminal, and [`ork`](https://github.com/orca-ae/orca-cli)
+for the first OAuth MCP login in L3/L4, including the Python and TypeScript paths.
+Use current `orca-cli` main or a build containing [PR #8](https://github.com/orca-ae/orca-cli/pull/8).
+Check `ork agent vaults credentials create --help` for `--oauth-issuer` and
+`--oauth-allow-issuer-mismatch`. Automatic discovery handles StreamNative's
+same-domain proxy issuer aliases and dynamic client authentication. Leave
+`SN_MCP_OAUTH_ISSUER` empty for this flow; `--oauth-issuer` only selects an
+authorization server advertised by the MCP server when a choice is needed.
+The browser flow stores tokens directly in the vault.
+On Windows, use WSL or Git Bash for the CLI path.
 
 ## 1. Get the code
 
diff --git a/python/cleanup.py b/python/cleanup.py
index 9d2f147..3172cd2 100644
--- a/python/cleanup.py
+++ b/python/cleanup.py
@@ -9,7 +9,7 @@
 
 
 def main() -> None:
-    config = load_config(["ORCA_BASE_URL", "SN_API_KEY"])
+    config = load_config(["ORCA_BASE_URL"])
     cleanup(orca_client(config), state_for(config))
 
 
diff --git a/python/common.py b/python/common.py
index 4979491..5ab860f 100644
--- a/python/common.py
+++ b/python/common.py
@@ -12,6 +12,7 @@
 import os
 import re
 import secrets
+import subprocess
 import sys
 from dataclasses import dataclass, field
 from datetime import datetime
@@ -130,8 +131,12 @@ def _lookup(name: str, config: Config) -> str:
 
 
 def orca_client(config: Config) -> Orca:
-    """One service-account API key, sent as a Bearer token, authenticates everything."""
-    return Orca(base_url=config["ORCA_BASE_URL"], api_key=config["SN_API_KEY"], timeout=600)
+    """Use a Registry workspace key locally, or the team card's hosted Bearer key."""
+    if key := config.values.get("ORCA_API_KEY"):
+        return Orca(base_url=config["ORCA_BASE_URL"], api_key=None, default_headers={"x-api-key": key}, timeout=600)
+    if key := config.values.get("SN_API_KEY"):
+        return Orca(base_url=config["ORCA_BASE_URL"], api_key=key, timeout=600)
+    raise ConfigError("Set ORCA_API_KEY for ork local, or SN_API_KEY from your team card.")
 
 
 def ensure_environment(client: Any, state: State, name: str) -> str:
@@ -159,22 +164,66 @@ def ensure_agent(client: Any, state: State, params: dict[str, Any]) -> Agent:
     return agent
 
 
-def ensure_vault(client: Any, state: State, name: str, mcp_url: str, token: str) -> str:
+def mcp_auth_type(config: Config) -> str:
+    mode = config.values.get("SN_MCP_AUTH", "oauth")
+    if mode not in ("oauth", "static_bearer"):
+        raise ConfigError("SN_MCP_AUTH must be oauth or static_bearer.")
+    return "mcp_oauth" if mode == "oauth" else "static_bearer"
+
+
+def authorize_mcp(vault_id: str, config: Config) -> None:
+    """Let ork handle discovery, browser login, PKCE and server-side token storage."""
+    args = ["ork", "agent", "vaults", "credentials", "create", "--vault", vault_id,
+            "--display-name", "streamnative-mcp", "--mcp-server-url", config["SN_MCP_URL"], "-o", "json"]
+    for name, flag in (("SN_MCP_OAUTH_ISSUER", "--oauth-issuer"), ("SN_MCP_OAUTH_SCOPE", "--oauth-scope")):
+        if value := config.values.get(name):
+            args += [flag, value]
+    env = dict(os.environ)
+    env["ORCA_REGISTRY_URL"] = config["ORCA_BASE_URL"]
+    # Export only the chosen Registry credential, never pass it in argv.
+    env.pop("ORCA_API_KEY", None)
+    env.pop("ORCA_ACCESS_TOKEN", None)
+    if config.values.get("ORCA_API_KEY"):
+        env["ORCA_API_KEY"] = config["ORCA_API_KEY"]
+    else:
+        env["ORCA_ACCESS_TOKEN"] = config["SN_API_KEY"]
+    try:
+        result = subprocess.run(args, env=env, check=False)
+    except OSError:
+        raise ConfigError("Install ork with MCP OAuth proxy support and put it on PATH (see docs/before-you-arrive.md).") from None
+    if result.returncode:
+        raise ConfigError("MCP OAuth authorization failed. Check the ork error above; normally leave SN_MCP_OAUTH_ISSUER empty for discovery, then rerun L3/L4.")
+
+
+def ensure_vault(client: Any, state: State, name: str, config: Config) -> str:
     """A vault holding the credential the agent uses to call the MCP server.
 
     The token lives in the vault on the server side; it never enters a prompt.
     """
+    auth_type = mcp_auth_type(config)
+    mcp_url = config["SN_MCP_URL"]
+    if auth_type == "static_bearer" and not config.values.get("SN_API_KEY"):
+        raise ConfigError("SN_MCP_AUTH=static_bearer requires SN_API_KEY.")
     vault = _live(client.vaults.retrieve, state.get("vault_id"))
     if vault is None:
         vault = client.vaults.create(display_name=name)
         state.set("vault_id", vault.id)
     credentials = client.vaults.credentials.list(vault.id).data
-    if not any(getattr(c.auth, "mcp_server_url", None) == mcp_url and not c.archived_at for c in credentials):
-        client.vaults.credentials.create(
-            vault.id,
-            display_name="streamnative-mcp",
-            auth={"type": "static_bearer", "mcp_server_url": mcp_url, "token": token},
-        )
+    if not any(getattr(c.auth, "mcp_server_url", None) == mcp_url
+               and c.auth.type == auth_type and not c.archived_at for c in credentials):
+        # Registry permits only one active credential per MCP URL in a vault.
+        # Retire the old auth mode before creating its replacement.
+        for credential in credentials:
+            if getattr(credential.auth, "mcp_server_url", None) == mcp_url and not credential.archived_at:
+                client.vaults.credentials.archive(vault.id, credential.id)
+        if auth_type == "mcp_oauth":
+            authorize_mcp(vault.id, config)
+        else:
+            client.vaults.credentials.create(
+                vault.id,
+                display_name="streamnative-mcp",
+                auth={"type": "static_bearer", "mcp_server_url": mcp_url, "token": config["SN_API_KEY"]},
+            )
     return vault.id
 
 
@@ -216,11 +265,11 @@ def ask_human(tool_use: dict[str, Any], *, ask: Callable[[str], str] = input, ou
 
 
 def cleanup(client: Any, state: State, *, out: Callable[[str], None] = print) -> None:
-    """Remove what the scripts created. Agents cannot be deleted, only archived."""
+    """Archive the agent and environment, and delete the vault. Sessions reserve the environment."""
     steps = [
         ("agent", "agent_id", client.agents.archive),
         ("vault", "vault_id", client.vaults.delete),
-        ("environment", "environment_id", client.environments.delete),
+        ("environment", "environment_id", client.environments.archive),
     ]
     for label, key, remove in steps:
         resource_id = state.get(key)
diff --git a/python/doctor.py b/python/doctor.py
index d7a59e5..da1f1a0 100644
--- a/python/doctor.py
+++ b/python/doctor.py
@@ -2,6 +2,7 @@
 
     python doctor.py             # laptop + .env + Agent Engine + Kafka + Schema Registry + MCP
     python doctor.py --offline   # laptop only (run this before the event)
+    python doctor.py --agent-only  # laptop + Agent Engine (for ork local / L1)
 
 Every failed check prints the fix.
 """
@@ -18,7 +19,7 @@
 
 CARD = ("SN_API_KEY", "SN_SERVICE_ACCOUNT", "ORCA_BASE_URL", "KAFKA_BOOTSTRAP_SERVERS", "SCHEMA_REGISTRY_URL", "SN_MCP_URL", "LOGIN_TOPIC", "ORCA_MODEL")
 SQL_FIELDS = ("account_id", "event_time", "ip_address", "result", "failure_reason")
-MCP_TOOLS = ("sql_workspace_list_databases", "sql_workspace_query", "sql_workspace_insert_rows")
+MCP_TOOLS = ("sql_workspace_list_databases", "sql_workspace_query", "sql_workspace_describe_table", "sql_workspace_insert_rows")
 PACKAGES = {"orca": "runorca", "confluent_kafka": "confluent-kafka[avro]", "dotenv": "python-dotenv"}
 
 
@@ -40,9 +41,10 @@ def check_python(version: tuple) -> Check:
 
 def check_orca_base_url(url: str) -> Check:
     parts = urlsplit(url.strip())
-    if parts.scheme != "https" or not parts.netloc:
-        return Check("ORCA_BASE_URL", False, url, "Use the full https:// registry endpoint from your team card.")
-    root = f"https://{parts.netloc}"
+    local_http = parts.scheme == "http" and parts.hostname in ("localhost", "127.0.0.1", "::1")
+    if not parts.netloc or (parts.scheme != "https" and not local_http):
+        return Check("ORCA_BASE_URL", False, url, "Use the https:// registry endpoint from your team card, or http://127.0.0.1:8080 for ork local.")
+    root = f"{parts.scheme}://{parts.netloc}"
     if parts.path.rstrip("/"):
         return Check("ORCA_BASE_URL", False, url, f"Use the host root only: ORCA_BASE_URL={root}")
     return Check("ORCA_BASE_URL", True, root)
@@ -98,18 +100,20 @@ def kafka_hint(error: str) -> str:
 # ---------------------------------------------------------------- probes --
 
 
-def check_packages() -> list[Check]:
+def check_packages(agent_only: bool = False) -> list[Check]:
     checks = []
     for module, package in PACKAGES.items():
+        if agent_only and module == "confluent_kafka":
+            continue
         found = importlib.util.find_spec(module) is not None
         checks.append(Check(f"package {package}", found, "", "" if found else "Run: pip install -r requirements.txt"))
     return checks
 
 
 def check_cli_tools() -> list[Check]:
-    """Only the CLI path needs these, so a missing tool is reported but never fails."""
+    """ork is needed for first OAuth login; jq is only needed by the CLI path."""
     return [
-        Check(f"{tool} (CLI path only)", True, "found" if shutil.which(tool) else "not found: fine unless you take the CLI path")
+        Check(tool, True, "found" if shutil.which(tool) else ("not found: install before first OAuth MCP login" if tool == "ork" else "not found: CLI path only"))
         for tool in ("ork", "jq")
     ]
 
@@ -123,7 +127,7 @@ def probe_orca(config: Any) -> Check:
         orca_client(config).agents.list(limit=1)
     except APIStatusError as err:
         if err.status_code in (401, 403):
-            fix = "The Agent Engine rejected SN_API_KEY. A key created before its rolebinding must be re-created: ask a facilitator."
+            fix = "The Agent Engine rejected the key. For ork local use its generated workspace key as ORCA_API_KEY; for a team card check SN_API_KEY and its rolebinding."
         elif err.status_code == 404:
             fix = "ORCA_BASE_URL is not an Agent Engine registry: copy the registry endpoint from your team card."
         else:
@@ -151,7 +155,8 @@ def probe_kafka(config: Any) -> Check:
     )
     topic_name = config["LOGIN_TOPIC"]
     try:
-        topic = admin.list_topics(topic=topic_name, timeout=15).topics.get(topic_name)
+        # Listing all existing topics avoids metadata requests that can auto-create a missing topic.
+        topic = admin.list_topics(timeout=15).topics.get(topic_name)
     except KafkaException as err:
         reason = errors[0] if errors else str(err)
         return Check("Kafka", False, reason, kafka_hint(reason))
@@ -208,26 +213,50 @@ def post(payload: dict[str, Any]) -> dict[str, Any] | None:
     return names
 
 
-def probe_mcp(config: Any) -> Check:
+def probe_mcp(config: Any, *, client: Any = None, state: Any = None) -> Check:
+    from common import mcp_auth_type, orca_client, state_for
+
     try:
+        if mcp_auth_type(config) == "mcp_oauth":
+            client = client or orca_client(config)
+            state = state or state_for(config)
+            vault_id = state.get("vault_id")
+            fix = "Run L3 to create the MCP OAuth credential with ork, then rerun doctor."
+            if not vault_id:
+                return Check("MCP OAuth", False, "no tutorial vault yet", fix)
+            credentials = client.vaults.credentials.list(vault_id).data
+            credential = next((c for c in credentials if c.auth.type == "mcp_oauth"
+                               and c.auth.mcp_server_url == config["SN_MCP_URL"] and not c.archived_at), None)
+            if credential is None:
+                return Check("MCP OAuth", False, "no matching OAuth credential", fix)
+            result = client.vaults.credentials.validate(vault_id, credential.id)
+            ok = result.status == "valid"
+            if result.status == "unknown":
+                fix = "The MCP probe was inconclusive. Check Registry/MCP connectivity and rerun doctor; keep the existing credential."
+            elif result.status == "invalid":
+                fix = f"Reauthorize: ork agent vaults credentials archive {credential.id} --vault {vault_id}, then rerun L3/L4."
+            return Check("MCP OAuth", ok, f"{result.status}: MCP initialization; SQL tools checked in L3/L4", "" if ok else fix)
         names = mcp_tool_names(config["SN_MCP_URL"], config["SN_API_KEY"])
     except Exception as err:  # HTTP errors, JSON-RPC errors, timeouts
-        return Check("MCP server", False, str(err), "Check SN_MCP_URL, and that MCP is enabled for your key: ask a facilitator.")
+        return Check("MCP server", False, str(err), "Check SN_MCP_URL and SN_MCP_AUTH; for OAuth, finish the L3 browser login and verify the vault credential.")
     return check_mcp_tools(names)
 
 
 # ------------------------------------------------------------------ main --
 
 
-def run_checks(offline: bool) -> list[Check]:
-    checks = [check_python(sys.version_info), *check_packages(), *check_cli_tools()]
+def run_checks(offline: bool, agent_only: bool = False) -> list[Check]:
+    checks = [check_python(sys.version_info), *check_packages(agent_only), *check_cli_tools()]
     if offline or not all(check.ok for check in checks):
         return checks
 
     from common import load_config
 
     config = load_config([])
-    missing = [name for name in CARD if name not in config.values]
+    required = ("ORCA_BASE_URL", "ORCA_MODEL") if agent_only else CARD
+    missing = [name for name in required if name not in config.values]
+    if agent_only and not any(name in config.values for name in ("ORCA_API_KEY", "SN_API_KEY")):
+        missing.append("ORCA_API_KEY or SN_API_KEY")
     if missing:
         fix = "Copy .env.example to .env in the repo root and paste your team card."
         return [*checks, Check("team card (.env)", False, "missing " + ", ".join(missing), fix)]
@@ -237,6 +266,8 @@ def run_checks(offline: bool) -> list[Check]:
     checks.append(base_url)
     if base_url.ok:
         checks.append(probe_orca(config))
+    if agent_only:
+        return checks
     checks.append(probe_kafka(config))
     checks += probe_schema_registry(config)
     checks.append(probe_mcp(config))
@@ -244,7 +275,7 @@ def run_checks(offline: bool) -> list[Check]:
 
 
 def main() -> int:
-    checks = run_checks(offline="--offline" in sys.argv[1:])
+    checks = run_checks(offline="--offline" in sys.argv[1:], agent_only="--agent-only" in sys.argv[1:])
     for check in checks:
         print(f"{'PASS' if check.ok else 'FAIL'}  {check.name:<32} {check.detail}")
         if not check.ok and check.fix:
diff --git a/python/l1_hello.py b/python/l1_hello.py
index b32202e..a156578 100644
--- a/python/l1_hello.py
+++ b/python/l1_hello.py
@@ -14,7 +14,7 @@
 
 
 def main() -> None:
-    config = load_config(["ORCA_BASE_URL", "SN_API_KEY", "ORCA_MODEL"])
+    config = load_config(["ORCA_BASE_URL", "ORCA_MODEL"])
     client = orca_client(config)
     state = state_for(config)
 
diff --git a/python/l3_live_context.py b/python/l3_live_context.py
index e8e04a0..33c6deb 100644
--- a/python/l3_live_context.py
+++ b/python/l3_live_context.py
@@ -24,7 +24,7 @@
 
 
 def main() -> None:
-    config = load_config(["ORCA_BASE_URL", "SN_API_KEY", "ORCA_MODEL", "SN_MCP_URL"])
+    config = load_config(["ORCA_BASE_URL", "ORCA_MODEL", "SN_MCP_URL"])
     client = orca_client(config)
     state = state_for(config)
     environment_id = ensure_environment(client, state, f"hello-env-{config.participant}")
@@ -35,7 +35,7 @@ def main() -> None:
     print(f"{agent.name} v{agent.version}: {layer['summary']}")
 
     # The MCP server needs a credential. It goes in a vault, never in the prompt.
-    vault_id = ensure_vault(client, state, f"hello-vault-{config.participant}", config["SN_MCP_URL"], config["SN_API_KEY"])
+    vault_id = ensure_vault(client, state, f"hello-vault-{config.participant}", config)
 
     session = client.sessions.create(
         environment_id=environment_id,
diff --git a/python/l4_act.py b/python/l4_act.py
index 2fe8f2a..792d1b4 100644
--- a/python/l4_act.py
+++ b/python/l4_act.py
@@ -24,7 +24,7 @@
 
 
 def main() -> None:
-    config = load_config(["ORCA_BASE_URL", "SN_API_KEY", "ORCA_MODEL", "SN_MCP_URL"])
+    config = load_config(["ORCA_BASE_URL", "ORCA_MODEL", "SN_MCP_URL"])
     client = orca_client(config)
     state = state_for(config)
     environment_id = ensure_environment(client, state, f"hello-env-{config.participant}")
@@ -34,7 +34,7 @@ def main() -> None:
     agent = ensure_agent(client, state, agent_params(layer, config))
     print(f"{agent.name} v{agent.version}: {layer['summary']}")
 
-    vault_id = ensure_vault(client, state, f"hello-vault-{config.participant}", config["SN_MCP_URL"], config["SN_API_KEY"])
+    vault_id = ensure_vault(client, state, f"hello-vault-{config.participant}", config)
     session = client.sessions.create(
         environment_id=environment_id,
         agent={"type": "agent", "id": agent.id, "version": agent.version},
diff --git a/python/tests/fakes.py b/python/tests/fakes.py
index 56966df..07a2a1f 100644
--- a/python/tests/fakes.py
+++ b/python/tests/fakes.py
@@ -198,6 +198,13 @@ def retrieve(self, environment_id: str):
             raise _not_found()
         return self.store[environment_id]
 
+    def archive(self, environment_id: str):
+        self.calls.append(("archive", environment_id))
+        if environment_id not in self.store:
+            raise _not_found()
+        self.store[environment_id] = _archived(self.store[environment_id])
+        return self.store[environment_id]
+
     def delete(self, environment_id: str):
         self.calls.append(("delete", environment_id))
         if self.store.pop(environment_id, None) is None:
@@ -224,6 +231,14 @@ def create(self, vault_id: str, *, auth: dict[str, Any], display_name: str | Non
         self.store.setdefault(vault_id, []).append(cred)
         return cred
 
+    def archive(self, vault_id: str, credential_id: str):
+        self.calls.append(("archive", vault_id, credential_id))
+        for credential in self.store.get(vault_id, []):
+            if credential.id == credential_id:
+                credential.archived_at = NOW
+                return credential
+        raise _not_found()
+
     def list(self, vault_id: str, **_: Any):
         self.calls.append(("list", vault_id))
         return SimpleNamespace(data=list(self.store.get(vault_id, [])))
diff --git a/python/tests/test_agent_defs.py b/python/tests/test_agent_defs.py
index 039daae..0187020 100644
--- a/python/tests/test_agent_defs.py
+++ b/python/tests/test_agent_defs.py
@@ -66,10 +66,12 @@ def test_different_layers_carry_different_definition_fingerprints():
     [
         ("l3-live-context", "sql_workspace_list_databases", "always_allow"),
         ("l3-live-context", "sql_workspace_query", "always_allow"),
+        ("l3-live-context", "sql_workspace_describe_table", "disabled"),
         ("l3-live-context", "sql_workspace_insert_rows", "disabled"),
         ("l3-live-context", "sql_workspace_delete_rows", "disabled"),
         ("l3-live-context", "kafka_client_produce", "disabled"),
         ("l4-act", "sql_workspace_query", "always_allow"),
+        ("l4-act", "sql_workspace_describe_table", "always_allow"),
         ("l4-act", "sql_workspace_insert_rows", "always_ask"),
         ("l4-act", "sql_workspace_delete_rows", "disabled"),
         ("l4-act", "sql_workspace_create_materialized_view", "disabled"),
diff --git a/python/tests/test_config.py b/python/tests/test_config.py
index 283b420..ae6a0a4 100644
--- a/python/tests/test_config.py
+++ b/python/tests/test_config.py
@@ -2,7 +2,7 @@
 
 import pytest
 
-from common import ConfigError, load_config
+from common import ConfigError, load_config, orca_client
 
 CARD = {
     "ORCA_BASE_URL": "https://ws.example.com",
@@ -51,3 +51,38 @@ def test_participant_defaults_to_the_os_user(monkeypatch):
     monkeypatch.setattr("getpass.getuser", lambda: "Sam.Lee")
 
     assert load_config([], env={}).participant == "sam-lee"
+
+
+@pytest.mark.parametrize(
+    "credentials, bearer, workspace_key",
+    [
+        ({"SN_API_KEY": "team-key"}, "Bearer team-key", None),
+        ({"ORCA_API_KEY": "local-key"}, None, "local-key"),
+        ({"ORCA_API_KEY": "local-key", "SN_API_KEY": "mcp-key"}, None, "local-key"),
+    ],
+)
+def test_client_sends_exactly_one_registry_credential(monkeypatch, credentials, bearer, workspace_key):
+    import httpx2
+    from orca import Orca
+
+    seen = []
+    def respond(request):
+        seen.append(request)
+        return httpx2.Response(200, json={"data": [], "next_page": None})
+
+    monkeypatch.setenv("ORCA_API_KEY", "ambient-key-must-not-be-used-as-bearer")
+    monkeypatch.setattr("common.Orca", lambda **kwargs: Orca(
+        **kwargs, http_client=httpx2.Client(transport=httpx2.MockTransport(respond))
+    ))
+    config = load_config([], env={"ORCA_BASE_URL": "http://127.0.0.1:8080", **credentials})
+    with orca_client(config) as client:
+        client.agents.list(limit=1)
+
+    assert len(seen) == 1
+    assert seen[0].headers.get("Authorization") == bearer
+    assert seen[0].headers.get("x-api-key") == workspace_key
+
+
+def test_client_explains_how_to_supply_a_missing_registry_credential():
+    with pytest.raises(ConfigError, match="ORCA_API_KEY.*SN_API_KEY"):
+        orca_client(load_config([], env={"ORCA_BASE_URL": "http://127.0.0.1:8080"}))
diff --git a/python/tests/test_doctor.py b/python/tests/test_doctor.py
index efb0bbb..5a1d608 100644
--- a/python/tests/test_doctor.py
+++ b/python/tests/test_doctor.py
@@ -52,7 +52,7 @@ def test_an_outcome_field_instead_of_result_gets_a_specific_fix():
 
 
 def test_the_mcp_server_must_offer_the_three_tools_the_agent_uses():
-    tools = ["sql_workspace_list_databases", "sql_workspace_query", "sql_workspace_insert_rows", "sncloud_context_whoami"]
+    tools = ["sql_workspace_list_databases", "sql_workspace_query", "sql_workspace_describe_table", "sql_workspace_insert_rows", "sncloud_context_whoami"]
 
     assert check_mcp_tools(tools).ok
 
@@ -99,3 +99,18 @@ def test_an_mcp_error_response_raises_with_its_message():
 )
 def test_kafka_errors_map_to_a_concrete_fix(error, advice):
     assert advice in kafka_hint(error)
+
+
+@pytest.mark.parametrize("host", ["127.0.0.1", "localhost", "[::1]"])
+def test_local_http_agent_engine_is_accepted(host):
+    assert check_orca_base_url(f"http://{host}:8080").ok
+
+
+def test_non_local_http_still_requires_https():
+    assert not check_orca_base_url("http://ws.example.com").ok
+
+
+def test_local_api_path_suggests_the_local_host_root():
+    check = check_orca_base_url("http://127.0.0.1:8080/v1")
+    assert not check.ok
+    assert "http://127.0.0.1:8080" in check.fix
diff --git a/python/tests/test_ensure.py b/python/tests/test_ensure.py
index cfa4045..9e04b5e 100644
--- a/python/tests/test_ensure.py
+++ b/python/tests/test_ensure.py
@@ -92,7 +92,7 @@ def test_a_reserved_environment_name_gets_a_fresh_suffix(state):
 def test_the_vault_gets_one_bearer_credential_for_the_mcp_server(state):
     client = fake_client()
 
-    vault_id = ensure_vault(client, state, "hello-vault-jane", MCP_URL, "the-api-key")
+    vault_id = ensure_vault(client, state, "hello-vault-jane", Config(values={"SN_MCP_AUTH": "static_bearer", "SN_MCP_URL": MCP_URL, "SN_API_KEY": "the-api-key"}, participant="jane"))
 
     creates = client.vaults.credentials.calls
     assert [c for c in creates if c[0] == "create"] == [
@@ -102,9 +102,9 @@ def test_the_vault_gets_one_bearer_credential_for_the_mcp_server(state):
 
 def test_an_existing_credential_for_the_same_server_is_reused(state):
     client = fake_client()
-    first = ensure_vault(client, state, "hello-vault-jane", MCP_URL, "the-api-key")
+    first = ensure_vault(client, state, "hello-vault-jane", Config(values={"SN_MCP_AUTH": "static_bearer", "SN_MCP_URL": MCP_URL, "SN_API_KEY": "the-api-key"}, participant="jane"))
 
-    second = ensure_vault(client, state, "hello-vault-jane", MCP_URL, "the-api-key")
+    second = ensure_vault(client, state, "hello-vault-jane", Config(values={"SN_MCP_AUTH": "static_bearer", "SN_MCP_URL": MCP_URL, "SN_API_KEY": "the-api-key"}, participant="jane"))
 
     assert first == second
     assert len([c for c in client.vaults.credentials.calls if c[0] == "create"]) == 1
@@ -112,9 +112,9 @@ def test_an_existing_credential_for_the_same_server_is_reused(state):
 
 def test_a_new_mcp_url_gets_its_own_credential(state):
     client = fake_client()
-    ensure_vault(client, state, "hello-vault-jane", MCP_URL, "the-api-key")
+    ensure_vault(client, state, "hello-vault-jane", Config(values={"SN_MCP_AUTH": "static_bearer", "SN_MCP_URL": MCP_URL, "SN_API_KEY": "the-api-key"}, participant="jane"))
 
-    ensure_vault(client, state, "hello-vault-jane", MCP_URL + "-v2", "the-api-key")
+    ensure_vault(client, state, "hello-vault-jane", Config(values={"SN_MCP_AUTH": "static_bearer", "SN_MCP_URL": MCP_URL + "-v2", "SN_API_KEY": "the-api-key"}, participant="jane"))
 
     urls = [c[2]["mcp_server_url"] for c in client.vaults.credentials.calls if c[0] == "create"]
     assert urls == [MCP_URL, MCP_URL + "-v2"]
diff --git a/python/tests/test_inject.py b/python/tests/test_inject.py
index fd3bbe7..9af4eb2 100644
--- a/python/tests/test_inject.py
+++ b/python/tests/test_inject.py
@@ -83,19 +83,19 @@ def flush(self, timeout=None):
 def test_publish_writes_every_record_keyed_by_account():
     producer = FakeProducer()
 
-    errors = publish(producer, "avro.security.login_events", "acct_9123", burst())
+    errors = publish(producer, "security.login_events", "acct_9123", burst())
 
     assert errors == []
     assert len(producer.produced) == 7
-    assert {(topic, key) for topic, key, _ in producer.produced} == {("avro.security.login_events", "acct_9123")}
+    assert {(topic, key) for topic, key, _ in producer.produced} == {("security.login_events", "acct_9123")}
 
 
 def test_a_schema_registry_failure_is_reported_instead_of_raised():
     from confluent_kafka.error import ValueSerializationError
 
-    producer = FakeProducer(raise_on_produce=ValueSerializationError(Exception("Subject 'avro.security.login_events-value' not found")))
+    producer = FakeProducer(raise_on_produce=ValueSerializationError(Exception("Subject 'security.login_events-value' not found")))
 
-    errors = publish(producer, "avro.security.login_events", "acct_9123", burst())
+    errors = publish(producer, "security.login_events", "acct_9123", burst())
 
     assert any("not found" in e for e in errors)
 
diff --git a/python/tests/test_mcp_oauth.py b/python/tests/test_mcp_oauth.py
new file mode 100644
index 0000000..388f0dc
--- /dev/null
+++ b/python/tests/test_mcp_oauth.py
@@ -0,0 +1,122 @@
+"""OAuth delegation, reuse and doctor: no live browser or credentials required."""
+
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+
+from common import Config, ConfigError, State, authorize_mcp, ensure_vault
+from doctor import probe_mcp
+from fakes import fake_client
+
+URL = "https://mcp.example.com/mcp"
+
+
+def config(**overrides):
+    return Config(values={"ORCA_BASE_URL": "https://registry.example.com", "SN_API_KEY": "registry-secret",
+                          "SN_MCP_URL": URL, **overrides}, participant="jane")
+
+
+@pytest.fixture
+def state(tmp_path):
+    return State(tmp_path / "jane.json")
+
+
+def seed(client, state, auth_type="mcp_oauth", url=URL, archived=False):
+    vault = client.vaults.create(display_name="test")
+    state.set("vault_id", vault.id)
+    credential = SimpleNamespace(id="cred_oauth", auth=SimpleNamespace(type=auth_type, mcp_server_url=url), archived_at="old" if archived else None)
+    client.vaults.credentials.store[vault.id] = [credential]
+    return vault.id
+
+
+@pytest.mark.parametrize("local", [False, True])
+def test_oauth_uses_ork_without_secret_arguments(monkeypatch, local):
+    runner = Mock(return_value=SimpleNamespace(returncode=0))
+    monkeypatch.setattr("common.subprocess.run", runner)
+    monkeypatch.setenv("ORCA_API_KEY", "stale-local")
+    monkeypatch.setenv("ORCA_ACCESS_TOKEN", "stale-bearer")
+    cfg = config(SN_MCP_OAUTH_ISSUER="https://auth.example.com/", SN_MCP_OAUTH_SCOPE="openid offline_access",
+                 **({"ORCA_API_KEY": "local-secret"} if local else {}))
+    authorize_mcp("vlt_1", cfg)
+    args = runner.call_args.args[0]
+    env = runner.call_args.kwargs["env"]
+    assert args == ["ork", "agent", "vaults", "credentials", "create", "--vault", "vlt_1", "--display-name", "streamnative-mcp",
+                    "--mcp-server-url", URL, "-o", "json", "--oauth-issuer", "https://auth.example.com/", "--oauth-scope", "openid offline_access"]
+    assert "secret" not in " ".join(args)
+    assert env["ORCA_REGISTRY_URL"] == "https://registry.example.com"
+    assert env.get("ORCA_API_KEY") == ("local-secret" if local else None)
+    assert env.get("ORCA_ACCESS_TOKEN") == (None if local else "registry-secret")
+
+
+def test_default_oauth_creates_vault_then_delegates(monkeypatch, state):
+    client = fake_client()
+    authorize = Mock()
+    monkeypatch.setattr("common.authorize_mcp", authorize)
+    vault_id = ensure_vault(client, state, "hello-vault-jane", config())
+    authorize.assert_called_once_with(vault_id, config())
+    assert not any(c[0] == "create" for c in client.vaults.credentials.calls)
+    assert state.get("vault_id") == vault_id
+    assert "registry-secret" not in state.path.read_text()
+
+
+def test_existing_oauth_needs_no_ork(monkeypatch, state):
+    client = fake_client()
+    vault_id = seed(client, state)
+    authorize = Mock()
+    monkeypatch.setattr("common.authorize_mcp", authorize)
+    assert ensure_vault(client, state, "hello-vault-jane", config()) == vault_id
+    authorize.assert_not_called()
+
+
+@pytest.mark.parametrize("auth_type,url,archived", [("static_bearer", URL, False), ("mcp_oauth", URL + "-other", False), ("mcp_oauth", URL, True)])
+def test_wrong_type_url_or_archived_credential_is_not_reused(monkeypatch, state, auth_type, url, archived):
+    client = fake_client()
+    vault_id = seed(client, state, auth_type, url, archived)
+    authorize = Mock()
+    monkeypatch.setattr("common.authorize_mcp", authorize)
+    ensure_vault(client, state, "hello-vault-jane", config())
+    authorize.assert_called_once_with(vault_id, config())
+    archives = [c for c in client.vaults.credentials.calls if c[0] == "archive"]
+    assert len(archives) == (1 if auth_type == "static_bearer" and url == URL and not archived else 0)
+
+
+@pytest.mark.parametrize("failure", [FileNotFoundError(), SimpleNamespace(returncode=1)])
+def test_oauth_failures_stop_without_static_fallback(monkeypatch, state, failure):
+    runner = Mock(side_effect=failure) if isinstance(failure, Exception) else Mock(return_value=failure)
+    monkeypatch.setattr("common.subprocess.run", runner)
+    client = fake_client()
+    with pytest.raises(ConfigError):
+        ensure_vault(client, state, "hello-vault-jane", config())
+    assert not any(c[0] == "create" for c in client.vaults.credentials.calls)
+
+
+def test_invalid_auth_mode_fails_before_resources(state):
+    client = fake_client()
+    with pytest.raises(ConfigError, match="SN_MCP_AUTH"):
+        ensure_vault(client, state, "test", config(SN_MCP_AUTH="typo"))
+    assert client.vaults.calls == []
+
+
+@pytest.mark.parametrize("status,ok", [("valid", True), ("invalid", False), ("unknown", False)])
+def test_doctor_validates_vault_oauth_not_service_account_key(monkeypatch, state, status, ok):
+    client = fake_client()
+    vault_id = seed(client, state)
+    client.vaults.credentials.validate = Mock(return_value=SimpleNamespace(status=status))
+    direct = Mock(side_effect=AssertionError("must not send SN_API_KEY to OAuth MCP"))
+    monkeypatch.setattr("doctor.mcp_tool_names", direct)
+    check = probe_mcp(config(), client=client, state=state)
+    assert check.ok is ok
+    assert "initialization" in check.detail
+    if status == "unknown":
+        assert "keep the existing credential" in check.fix
+    if status == "invalid":
+        assert "credentials archive cred_oauth" in check.fix
+    client.vaults.credentials.validate.assert_called_once_with(vault_id, "cred_oauth")
+    direct.assert_not_called()
+
+
+def test_doctor_before_first_oauth_login_gives_setup_hint(state):
+    check = probe_mcp(config(), client=fake_client(), state=state)
+    assert not check.ok
+    assert "Run L3" in check.fix
diff --git a/python/tests/test_scripts_support.py b/python/tests/test_scripts_support.py
index 68c9122..19317ac 100644
--- a/python/tests/test_scripts_support.py
+++ b/python/tests/test_scripts_support.py
@@ -4,7 +4,7 @@
 import pytest
 from orca import AuthenticationError
 
-from common import ConfigError, State, TurnError, ask_human, chat, cleanup, ensure_agent, ensure_environment, ensure_vault, run_main
+from common import Config, ConfigError, State, TurnError, ask_human, chat, cleanup, ensure_agent, ensure_environment, ensure_vault, run_main
 from fakes import FakeSessionEvents, client_with_events, fake_client
 from test_ensure import MCP_URL, params
 
@@ -34,17 +34,18 @@ def test_the_human_must_type_yes_to_approve(typed, allowed):
     assert any("acct_9123" in line for line in shown)
 
 
-def test_cleanup_archives_the_agent_deletes_the_rest_and_forgets_the_ids(tmp_path):
+def test_cleanup_archives_the_agent_and_environment_deletes_the_vault_and_forgets_the_ids(tmp_path):
     client = fake_client()
     state = State(tmp_path / "jane.json")
     agent = ensure_agent(client, state, params("l1-hello"))
     env_id = ensure_environment(client, state, "hello-env-jane")
-    vault_id = ensure_vault(client, state, "hello-vault-jane", MCP_URL, "key")
+    vault_id = ensure_vault(client, state, "hello-vault-jane", Config(values={"SN_MCP_AUTH": "static_bearer", "SN_MCP_URL": MCP_URL, "SN_API_KEY": "key"}, participant="jane"))
 
     cleanup(client, state, out=lambda _l: None)
 
     assert client.agents.store[agent.id].archived_at is not None
-    assert env_id not in client.environments.store
+    assert client.environments.store[env_id].archived_at is not None
+    assert ("delete", env_id) not in client.environments.calls
     assert vault_id not in client.vaults.store
     assert not (tmp_path / "jane.json").exists()
 
diff --git a/sql/01_explore.sql b/sql/01_explore.sql
index 36d3db8..0b8d1a2 100644
--- a/sql/01_explore.sql
+++ b/sql/01_explore.sql
@@ -1,4 +1,6 @@
 -- L2 step 1: peek at the live login stream in your team's Kafka cluster.
+-- Before running: match the quoted "avro.security.login_events" source below
+-- to LOGIN_TOPIC in .env ("avro.<LOGIN_TOPIC>"). SQL Workspace does not load .env.
 --
 -- SQL Workspace imported the topic as a source table. Its name contains dots,
 -- so always wrap it in double quotes.
diff --git a/sql/02_login_failures.sql b/sql/02_login_failures.sql
index b1d6cde..1106ed5 100644
--- a/sql/02_login_failures.sql
+++ b/sql/02_login_failures.sql
@@ -1,4 +1,6 @@
 -- L2 step 2: turn the stream into always-fresh context for the agent.
+-- Before running: match the quoted "avro.security.login_events" source below
+-- to LOGIN_TOPIC in .env ("avro.<LOGIN_TOPIC>"). SQL Workspace does not load .env.
 --
 -- A materialized view is maintained incrementally: every new login event
 -- updates the counts within seconds. No batch job, no refresh.
diff --git a/typescript/src/cleanup.ts b/typescript/src/cleanup.ts
index 998a2f9..91980d6 100644
--- a/typescript/src/cleanup.ts
+++ b/typescript/src/cleanup.ts
@@ -9,7 +9,7 @@
 import { cleanup, loadConfig, orcaClient, runMain, stateFor } from './common.js';
 
 async function main(): Promise<void> {
-  const config = loadConfig(['ORCA_BASE_URL', 'SN_API_KEY']);
+  const config = loadConfig(['ORCA_BASE_URL']);
   await cleanup(orcaClient(config), stateFor(config));
 }
 
diff --git a/typescript/src/common.ts b/typescript/src/common.ts
index 54dfd3e..44ff049 100644
--- a/typescript/src/common.ts
+++ b/typescript/src/common.ts
@@ -5,6 +5,7 @@
  * leaning on these helpers. Read those first; come here when you want the details.
  */
 
+import { spawnSync } from 'node:child_process';
 import { createHash, randomBytes } from 'node:crypto';
 import { existsSync, mkdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
 import os from 'node:os';
@@ -211,7 +212,7 @@ export interface Client {
   environments: {
     create(params: EnvironmentCreateParams): Promise<Environment>;
     retrieve(environmentId: string): Promise<Environment>;
-    delete(environmentId: string): Promise<unknown>;
+    archive(environmentId: string): Promise<Environment>;
   };
   vaults: {
     create(params: VaultCreateParams): Promise<Vault>;
@@ -220,6 +221,7 @@ export interface Client {
     credentials: {
       list(vaultId: string): Promise<{ data: VaultCredential[] }>;
       create(vaultId: string, params: CredentialCreateParams): Promise<VaultCredential>;
+      archive(vaultId: string, credentialId: string): Promise<VaultCredential>;
     };
   };
   sessions: {
@@ -228,9 +230,14 @@ export interface Client {
   };
 }
 
-/** One service-account API key, sent as a Bearer token, authenticates everything. */
-export function orcaClient(config: Config): Client {
-  return new Orca({ baseURL: config.get('ORCA_BASE_URL'), apiKey: config.get('SN_API_KEY'), timeout: 600_000 });
+/** Use a Registry workspace key locally, or the team card's hosted Bearer key. */
+export function orcaClient(config: Config): Orca {
+  const baseURL = config.get('ORCA_BASE_URL');
+  if (config.has('ORCA_API_KEY')) {
+    return new Orca({ baseURL, apiKey: null, defaultHeaders: { 'x-api-key': config.get('ORCA_API_KEY') }, timeout: 600_000 });
+  }
+  if (config.has('SN_API_KEY')) return new Orca({ baseURL, apiKey: config.get('SN_API_KEY'), timeout: 600_000 });
+  throw new ConfigError('Set ORCA_API_KEY for ork local, or SN_API_KEY from your team card.');
 }
 
 /** The sandbox your sessions run in. Created once, then reused. */
@@ -271,25 +278,61 @@ export async function ensureVault(
   client: Pick<Client, 'vaults'>,
   state: State,
   name: string,
-  mcpUrl: string,
-  token: string,
+  config: Config,
 ): Promise<string> {
+  const authType = mcpAuthType(config);
+  const mcpUrl = config.get('SN_MCP_URL');
+  if (authType === 'static_bearer' && !config.has('SN_API_KEY')) throw new ConfigError('SN_MCP_AUTH=static_bearer requires SN_API_KEY.');
   let vault = await live((id) => client.vaults.retrieve(id), state.get('vault_id'));
   if (!vault) {
     vault = await client.vaults.create({ display_name: name });
     state.set('vault_id', vault.id);
   }
   const { data: credentials } = await client.vaults.credentials.list(vault.id);
-  const hasCredential = credentials.some((c) => 'mcp_server_url' in c.auth && c.auth.mcp_server_url === mcpUrl && !c.archived_at);
+  const hasCredential = credentials.some((c) => 'mcp_server_url' in c.auth && c.auth.mcp_server_url === mcpUrl && c.auth.type === authType && !c.archived_at);
   if (!hasCredential) {
-    await client.vaults.credentials.create(vault.id, {
-      display_name: 'streamnative-mcp',
-      auth: { type: 'static_bearer', mcp_server_url: mcpUrl, token },
-    });
+    // Registry permits only one active credential per MCP URL in a vault.
+    for (const credential of credentials) {
+      if ('mcp_server_url' in credential.auth && credential.auth.mcp_server_url === mcpUrl && !credential.archived_at) {
+        await client.vaults.credentials.archive(vault.id, credential.id);
+      }
+    }
+    if (authType === 'mcp_oauth') {
+      authorizeMcp(vault.id, config);
+    } else {
+      await client.vaults.credentials.create(vault.id, {
+        display_name: 'streamnative-mcp',
+        auth: { type: 'static_bearer', mcp_server_url: mcpUrl, token: config.get('SN_API_KEY') },
+      });
+    }
   }
   return vault.id;
 }
 
+export function mcpAuthType(config: Config): 'mcp_oauth' | 'static_bearer' {
+  const mode = config.has('SN_MCP_AUTH') ? config.get('SN_MCP_AUTH') : 'oauth';
+  if (mode !== 'oauth' && mode !== 'static_bearer') throw new ConfigError('SN_MCP_AUTH must be oauth or static_bearer.');
+  return mode === 'oauth' ? 'mcp_oauth' : 'static_bearer';
+}
+
+/** ork owns discovery, browser login, PKCE and server-side token storage. */
+export function authorizeMcp(vaultId: string, config: Config): void {
+  const args = ['agent', 'vaults', 'credentials', 'create', '--vault', vaultId,
+    '--display-name', 'streamnative-mcp', '--mcp-server-url', config.get('SN_MCP_URL'), '-o', 'json'];
+  for (const [name, flag] of [['SN_MCP_OAUTH_ISSUER', '--oauth-issuer'], ['SN_MCP_OAUTH_SCOPE', '--oauth-scope']]) {
+    if (config.has(name)) args.push(flag, config.get(name));
+  }
+  const env: NodeJS.ProcessEnv = { ...process.env, ORCA_REGISTRY_URL: config.get('ORCA_BASE_URL') };
+  delete env.ORCA_API_KEY;
+  delete env.ORCA_ACCESS_TOKEN;
+  if (config.has('ORCA_API_KEY')) env.ORCA_API_KEY = config.get('ORCA_API_KEY');
+  else env.ORCA_ACCESS_TOKEN = config.get('SN_API_KEY');
+  // No shell and no credentials in argv; OAuth tokens are never read by this script.
+  const result = spawnSync('ork', args, { env, stdio: 'inherit' });
+  if (result.error) throw new ConfigError('Install ork with MCP OAuth proxy support and put it on PATH (see docs/before-you-arrive.md).');
+  if (result.status !== 0) throw new ConfigError('MCP OAuth authorization failed. Check the ork error above; normally leave SN_MCP_OAUTH_ISSUER empty for discovery, then rerun L3/L4.');
+}
+
 /** The remembered resource, or null if it was never created, deleted, or archived. */
 async function live<T extends { archived_at: string | null }>(
   retrieve: (id: string) => Promise<T>,
@@ -348,7 +391,7 @@ export async function askHuman(toolUse: ToolUse, { ask = prompt, out = console.l
   return ['y', 'yes'].includes((await ask('Allow it? [y/N] ')).trim().toLowerCase());
 }
 
-/** Remove what the scripts created. Agents cannot be deleted, only archived. */
+/** Archive the agent and environment, and delete the vault. Sessions reserve the environment. */
 export async function cleanup(
   client: Pick<Client, 'agents' | 'environments' | 'vaults'>,
   state: State,
@@ -357,7 +400,7 @@ export async function cleanup(
   const steps: Array<[string, string, (id: string) => Promise<unknown>]> = [
     ['agent', 'agent_id', (id) => client.agents.archive(id)],
     ['vault', 'vault_id', (id) => client.vaults.delete(id)],
-    ['environment', 'environment_id', (id) => client.environments.delete(id)],
+    ['environment', 'environment_id', (id) => client.environments.archive(id)],
   ];
   for (const [label, key, remove] of steps) {
     const resourceId = state.get(key);
diff --git a/typescript/src/doctor.ts b/typescript/src/doctor.ts
index 0472be8..f31195d 100644
--- a/typescript/src/doctor.ts
+++ b/typescript/src/doctor.ts
@@ -3,6 +3,7 @@
  *
  *     npm run doctor                 # laptop + .env + Agent Engine + Kafka + Schema Registry + MCP
  *     npm run doctor -- --offline    # laptop only (run this before the event)
+ *     npm run doctor -- --agent-only # laptop + Agent Engine (for ork local / L1)
  *
  * Every failed check prints the fix.
  */
@@ -11,12 +12,13 @@ import { existsSync } from 'node:fs';
 import { delimiter, dirname, join, resolve } from 'node:path';
 import { fileURLToPath, pathToFileURL } from 'node:url';
 
-import type { Config } from './common.js';
+import type Orca from '@runorca/orca-sdk';
+import type { Config, State } from './common.js';
 
 const TS_ROOT = resolve(dirname(fileURLToPath(import.meta.url)), '..');
 const CARD = ['SN_API_KEY', 'SN_SERVICE_ACCOUNT', 'ORCA_BASE_URL', 'KAFKA_BOOTSTRAP_SERVERS', 'SCHEMA_REGISTRY_URL', 'SN_MCP_URL', 'LOGIN_TOPIC', 'ORCA_MODEL'];
 const SQL_FIELDS = ['account_id', 'event_time', 'ip_address', 'result', 'failure_reason'];
-const MCP_TOOLS = ['sql_workspace_list_databases', 'sql_workspace_query', 'sql_workspace_insert_rows'];
+const MCP_TOOLS = ['sql_workspace_list_databases', 'sql_workspace_query', 'sql_workspace_describe_table', 'sql_workspace_insert_rows'];
 const PACKAGES = ['@runorca/orca-sdk', 'kafkajs', '@kafkajs/confluent-schema-registry', 'dotenv'];
 
 export interface Check {
@@ -45,10 +47,11 @@ export function checkOrcaBaseUrl(url: string): Check {
   } catch {
     // not a URL at all
   }
-  if (!parsed || parsed.protocol !== 'https:' || !parsed.host) {
-    return check('ORCA_BASE_URL', false, url, 'Use the full https:// registry endpoint from your team card.');
+  const localHttp = parsed?.protocol === 'http:' && ['localhost', '127.0.0.1', '[::1]'].includes(parsed.hostname);
+  if (!parsed || !parsed.host || (parsed.protocol !== 'https:' && !localHttp)) {
+    return check('ORCA_BASE_URL', false, url, 'Use the https:// registry endpoint from your team card, or http://127.0.0.1:8080 for ork local.');
   }
-  const root = `https://${parsed.host}`;
+  const root = `${parsed.protocol}//${parsed.host}`;
   if (parsed.pathname.replace(/\/+$/, '')) {
     return check('ORCA_BASE_URL', false, url, `Use the host root only: ORCA_BASE_URL=${root}`);
   }
@@ -116,28 +119,29 @@ export function kafkaHint(error: string): string {
 
 // ---------------------------------------------------------------- probes --
 
-function checkPackages(): Check[] {
-  return PACKAGES.map((pkg) => {
+function checkPackages(agentOnly = false): Check[] {
+  return PACKAGES.filter((pkg) => !agentOnly || ['@runorca/orca-sdk', 'dotenv'].includes(pkg)).map((pkg) => {
     const found = existsSync(join(TS_ROOT, 'node_modules', pkg, 'package.json'));
     return check(`package ${pkg}`, found, '', found ? '' : 'Run: npm install');
   });
 }
 
-/** Only the CLI path needs these, so a missing tool is reported but never fails. */
+/** ork is needed for first OAuth login; jq is only needed by the CLI path. */
 function checkCliTools(): Check[] {
   const onPath = (tool: string) =>
     (process.env.PATH ?? '')
       .split(delimiter)
       .some((dir) => dir && ['', '.exe', '.cmd'].some((ext) => existsSync(join(dir, tool + ext))));
   return ['ork', 'jq'].map((tool) =>
-    check(`${tool} (CLI path only)`, true, onPath(tool) ? 'found' : 'not found: fine unless you take the CLI path'),
+    check(tool, true, onPath(tool) ? 'found' : (tool === 'ork' ? 'not found: install before first OAuth MCP login' : 'not found: CLI path only')),
   );
 }
 
 async function probeOrca(config: Config): Promise<Check> {
-  const { default: Orca, APIConnectionError, APIError } = await import('@runorca/orca-sdk');
+  const { APIConnectionError, APIError } = await import('@runorca/orca-sdk');
+  const { orcaClient } = await import('./common.js');
   try {
-    await new Orca({ baseURL: config.get('ORCA_BASE_URL'), apiKey: config.get('SN_API_KEY'), timeout: 30_000 }).agents.list({ limit: 1 });
+    await orcaClient(config).agents.list({ limit: 1 }, { timeout: 30_000 });
   } catch (err) {
     // A connection error is also an APIError, so check it first.
     if (err instanceof APIConnectionError) {
@@ -146,7 +150,7 @@ async function probeOrca(config: Config): Promise<Check> {
     if (err instanceof APIError && err.status) {
       let fix = 'Ask a facilitator.';
       if (err.status === 401 || err.status === 403) {
-        fix = 'The Agent Engine rejected SN_API_KEY. A key created before its rolebinding must be re-created: ask a facilitator.';
+        fix = 'The Agent Engine rejected the key. For ork local use its generated workspace key as ORCA_API_KEY; for a team card check SN_API_KEY and its rolebinding.';
       } else if (err.status === 404) {
         fix = 'ORCA_BASE_URL is not an Agent Engine registry: copy the registry endpoint from your team card.';
       }
@@ -176,7 +180,8 @@ async function probeKafka(config: Config): Promise<Check> {
   }).admin();
   try {
     await admin.connect();
-    const { topics } = await admin.fetchTopicMetadata({ topics: [topicName] });
+    // Listing existing topics avoids auto-creating a missing topic during preflight.
+    const { topics } = await admin.fetchTopicMetadata();
     const topic = topics.find((t) => t.name === topicName);
     if (!topic) return check('Kafka', false, `${topicName}: not found`, kafkaHint('not found'));
     return check('Kafka', true, `${topicName} has ${topic.partitions.length} partition(s)`);
@@ -244,25 +249,43 @@ async function mcpToolNames(url: string, token: string): Promise<string[]> {
   return names;
 }
 
-async function probeMcp(config: Config): Promise<Check> {
+export async function probeMcp(config: Config, client?: Pick<Orca, 'vaults'>, state?: State): Promise<Check> {
+  const { mcpAuthType, orcaClient, stateFor } = await import('./common.js');
   let names: string[];
   try {
+    if (mcpAuthType(config) === 'mcp_oauth') {
+      client ??= orcaClient(config);
+      state ??= stateFor(config);
+      const vaultId = state.get('vault_id');
+      let fix = 'Run L3 to create the MCP OAuth credential with ork, then rerun doctor.';
+      if (!vaultId) return check('MCP OAuth', false, 'no tutorial vault yet', fix);
+      const { data } = await client.vaults.credentials.list(vaultId);
+      const credential = data.find((c) => c.auth.type === 'mcp_oauth' && c.auth.mcp_server_url === config.get('SN_MCP_URL') && !c.archived_at);
+      if (!credential) return check('MCP OAuth', false, 'no matching OAuth credential', fix);
+      const result = await client.vaults.credentials.validate(vaultId, credential.id);
+      const ok = result.status === 'valid';
+      if (result.status === 'unknown') fix = 'The MCP probe was inconclusive. Check Registry/MCP connectivity and rerun doctor; keep the existing credential.';
+      else if (result.status === 'invalid') fix = `Reauthorize: ork agent vaults credentials archive ${credential.id} --vault ${vaultId}, then rerun L3/L4.`;
+      return check('MCP OAuth', ok, `${result.status}: MCP initialization; SQL tools checked in L3/L4`, ok ? '' : fix);
+    }
     names = await mcpToolNames(config.get('SN_MCP_URL'), config.get('SN_API_KEY'));
   } catch (err) {
-    return check('MCP server', false, errorText(err), 'Check SN_MCP_URL, and that MCP is enabled for your key: ask a facilitator.');
+    return check('MCP server', false, errorText(err), 'Check SN_MCP_URL and SN_MCP_AUTH; for OAuth, finish the L3 browser login and verify the vault credential.');
   }
   return checkMcpTools(names);
 }
 
 // ------------------------------------------------------------------ main --
 
-export async function runChecks(offline: boolean): Promise<Check[]> {
-  const checks = [checkNode(process.versions.node), ...checkPackages(), ...checkCliTools()];
+export async function runChecks(offline: boolean, agentOnly = false): Promise<Check[]> {
+  const checks = [checkNode(process.versions.node), ...checkPackages(agentOnly), ...checkCliTools()];
   if (offline || !checks.every((c) => c.ok)) return checks;
 
   const { loadConfig } = await import('./common.js');
   const config = loadConfig([]);
-  const missing = CARD.filter((name) => !config.has(name));
+  const required = agentOnly ? ['ORCA_BASE_URL', 'ORCA_MODEL'] : CARD;
+  const missing = required.filter((name) => !config.has(name));
+  if (agentOnly && !config.has('ORCA_API_KEY') && !config.has('SN_API_KEY')) missing.push('ORCA_API_KEY or SN_API_KEY');
   if (missing.length > 0) {
     const fix = 'Copy .env.example to .env in the repo root and paste your team card.';
     return [...checks, check('team card (.env)', false, `missing ${missing.join(', ')}`, fix)];
@@ -272,6 +295,7 @@ export async function runChecks(offline: boolean): Promise<Check[]> {
   const baseUrl = checkOrcaBaseUrl(config.get('ORCA_BASE_URL'));
   checks.push(baseUrl);
   if (baseUrl.ok) checks.push(await probeOrca(config));
+  if (agentOnly) return checks;
   checks.push(await probeKafka(config));
   checks.push(...(await probeSchemaRegistry(config)));
   checks.push(await probeMcp(config));
@@ -279,7 +303,8 @@ export async function runChecks(offline: boolean): Promise<Check[]> {
 }
 
 async function main(): Promise<number> {
-  const checks = await runChecks(process.argv.slice(2).includes('--offline'));
+  const args = process.argv.slice(2);
+  const checks = await runChecks(args.includes('--offline'), args.includes('--agent-only'));
   for (const c of checks) {
     console.log(`${c.ok ? 'PASS' : 'FAIL'}  ${c.name.padEnd(32)} ${c.detail}`);
     if (!c.ok && c.fix) console.log(`      fix: ${c.fix}`);
diff --git a/typescript/src/l1-hello.ts b/typescript/src/l1-hello.ts
index 4dba981..4a05446 100644
--- a/typescript/src/l1-hello.ts
+++ b/typescript/src/l1-hello.ts
@@ -12,7 +12,7 @@ import { agentParams, ensureAgent, ensureEnvironment, loadConfig, loadLayer, orc
 const QUESTION = 'Hi! What is the Data + Agent Hackathon, and what can you see right now?';
 
 async function main(): Promise<void> {
-  const config = loadConfig(['ORCA_BASE_URL', 'SN_API_KEY', 'ORCA_MODEL']);
+  const config = loadConfig(['ORCA_BASE_URL', 'ORCA_MODEL']);
   const client = orcaClient(config);
   const state = stateFor(config);
 
diff --git a/typescript/src/l3-live-context.ts b/typescript/src/l3-live-context.ts
index d5f5d23..a3e4e15 100644
--- a/typescript/src/l3-live-context.ts
+++ b/typescript/src/l3-live-context.ts
@@ -13,7 +13,7 @@ import { agentParams, chat, ensureAgent, ensureEnvironment, ensureVault, loadCon
 const QUESTION = 'Which accounts look like an account takeover right now?';
 
 async function main(): Promise<void> {
-  const config = loadConfig(['ORCA_BASE_URL', 'SN_API_KEY', 'ORCA_MODEL', 'SN_MCP_URL']);
+  const config = loadConfig(['ORCA_BASE_URL', 'ORCA_MODEL', 'SN_MCP_URL']);
   const client = orcaClient(config);
   const state = stateFor(config);
   const environmentId = await ensureEnvironment(client, state, `hello-env-${config.participant}`);
@@ -24,7 +24,7 @@ async function main(): Promise<void> {
   console.log(`${agent.name} v${agent.version}: ${layer.summary}`);
 
   // The MCP server needs a credential. It goes in a vault, never in the prompt.
-  const vaultId = await ensureVault(client, state, `hello-vault-${config.participant}`, config.get('SN_MCP_URL'), config.get('SN_API_KEY'));
+  const vaultId = await ensureVault(client, state, `hello-vault-${config.participant}`, config);
 
   const session = await client.sessions.create({
     environment_id: environmentId,
diff --git a/typescript/src/l4-act.ts b/typescript/src/l4-act.ts
index 44c45f7..8a734a6 100644
--- a/typescript/src/l4-act.ts
+++ b/typescript/src/l4-act.ts
@@ -24,7 +24,7 @@ import {
 const REQUEST = 'Flag the account most likely to be under attack right now.';
 
 async function main(): Promise<void> {
-  const config = loadConfig(['ORCA_BASE_URL', 'SN_API_KEY', 'ORCA_MODEL', 'SN_MCP_URL']);
+  const config = loadConfig(['ORCA_BASE_URL', 'ORCA_MODEL', 'SN_MCP_URL']);
   const client = orcaClient(config);
   const state = stateFor(config);
   const environmentId = await ensureEnvironment(client, state, `hello-env-${config.participant}`);
@@ -34,7 +34,7 @@ async function main(): Promise<void> {
   const agent = await ensureAgent(client, state, agentParams(layer, config));
   console.log(`${agent.name} v${agent.version}: ${layer.summary}`);
 
-  const vaultId = await ensureVault(client, state, `hello-vault-${config.participant}`, config.get('SN_MCP_URL'), config.get('SN_API_KEY'));
+  const vaultId = await ensureVault(client, state, `hello-vault-${config.participant}`, config);
   const session = await client.sessions.create({
     environment_id: environmentId,
     agent: { type: 'agent', id: agent.id, version: agent.version },
diff --git a/typescript/test/agent-defs.test.ts b/typescript/test/agent-defs.test.ts
index a1c6d1c..8868e05 100644
--- a/typescript/test/agent-defs.test.ts
+++ b/typescript/test/agent-defs.test.ts
@@ -58,7 +58,7 @@ describe('agentParams', () => {
   it.each([
     ['l1-hello', '457f86a77738415f'],
     ['l3-live-context', 'ac4d0b08aca3f336'],
-    ['l4-act', '452e3876c295ffc8'],
+    ['l4-act', '8d424c704af22671'],
   ])('%s fingerprint matches the Python and CLI paths (%s)', (layer, expected) => {
     // Same fingerprint everywhere, so switching languages does not bump the agent's version.
     expect(agentParams(loadLayer(layer), config()).metadata.definition_sha).toBe(expected);
@@ -67,10 +67,12 @@ describe('agentParams', () => {
   it.each([
     ['l3-live-context', 'sql_workspace_list_databases', 'always_allow'],
     ['l3-live-context', 'sql_workspace_query', 'always_allow'],
+    ['l3-live-context', 'sql_workspace_describe_table', 'disabled'],
     ['l3-live-context', 'sql_workspace_insert_rows', 'disabled'],
     ['l3-live-context', 'sql_workspace_delete_rows', 'disabled'],
     ['l3-live-context', 'kafka_client_produce', 'disabled'],
     ['l4-act', 'sql_workspace_query', 'always_allow'],
+    ['l4-act', 'sql_workspace_describe_table', 'always_allow'],
     ['l4-act', 'sql_workspace_insert_rows', 'always_ask'],
     ['l4-act', 'sql_workspace_delete_rows', 'disabled'],
     ['l4-act', 'sql_workspace_create_materialized_view', 'disabled'],
diff --git a/typescript/test/config.test.ts b/typescript/test/config.test.ts
index b8af2e2..fdeeb81 100644
--- a/typescript/test/config.test.ts
+++ b/typescript/test/config.test.ts
@@ -4,7 +4,7 @@ import os from 'node:os';
 
 import { afterEach, describe, expect, it, vi } from 'vitest';
 
-import { ConfigError, loadConfig } from '../src/common.js';
+import { ConfigError, loadConfig, orcaClient } from '../src/common.js';
 
 const CARD = {
   ORCA_BASE_URL: 'https://ws.example.com',
@@ -14,6 +14,7 @@ const CARD = {
 
 afterEach(() => {
   vi.restoreAllMocks();
+  vi.unstubAllEnvs();
 });
 
 describe('loadConfig', () => {
@@ -64,3 +65,28 @@ describe('loadConfig', () => {
     expect(String(config)).not.toContain('key');
   });
 });
+
+
+describe('Registry authentication', () => {
+  it.each([
+    [{ SN_API_KEY: 'team-key' }, 'Bearer team-key', null],
+    [{ ORCA_API_KEY: 'local-key' }, null, 'local-key'],
+    [{ ORCA_API_KEY: 'local-key', SN_API_KEY: 'mcp-key' }, null, 'local-key'],
+  ])('sends exactly one Registry credential', async (credentials, bearer, workspaceKey) => {
+    vi.stubEnv('ORCA_API_KEY', 'ambient-key-must-not-be-used-as-bearer');
+    const fetch = vi.spyOn(globalThis, 'fetch').mockResolvedValue(
+      new Response(JSON.stringify({ data: [], next_page: null }), { headers: { 'Content-Type': 'application/json' } }),
+    );
+    const config = loadConfig([], { ORCA_BASE_URL: 'http://127.0.0.1:8080', ...credentials });
+    await orcaClient(config).agents.list({ limit: 1 });
+
+    expect(fetch).toHaveBeenCalledTimes(1);
+    const headers = new Headers(fetch.mock.calls[0][1]?.headers);
+    expect(headers.get('Authorization')).toBe(bearer);
+    expect(headers.get('x-api-key')).toBe(workspaceKey);
+  });
+
+  it('explains how to supply a missing Registry credential', () => {
+    expect(() => orcaClient(loadConfig([], { ORCA_BASE_URL: 'http://127.0.0.1:8080' }))).toThrow(/ORCA_API_KEY.*SN_API_KEY/);
+  });
+});
diff --git a/typescript/test/doctor.test.ts b/typescript/test/doctor.test.ts
index b5c8178..30b60a5 100644
--- a/typescript/test/doctor.test.ts
+++ b/typescript/test/doctor.test.ts
@@ -57,7 +57,7 @@ describe('checkLoginSchema', () => {
 
 describe('checkMcpTools', () => {
   it('the MCP server must offer the three tools the agent uses', () => {
-    const tools = ['sql_workspace_list_databases', 'sql_workspace_query', 'sql_workspace_insert_rows', 'sncloud_context_whoami'];
+    const tools = ['sql_workspace_list_databases', 'sql_workspace_query', 'sql_workspace_describe_table', 'sql_workspace_insert_rows', 'sncloud_context_whoami'];
 
     expect(checkMcpTools(tools).ok).toBe(true);
   });
@@ -120,3 +120,18 @@ describe('errorText', () => {
     expect(errorText(new Error('HTTP 401: unauthorized'))).toBe('HTTP 401: unauthorized');
   });
 });
+
+
+describe('local Agent Engine URL', () => {
+  it.each(['127.0.0.1', 'localhost', '[::1]'])('accepts loopback HTTP on %s', (host) => {
+    expect(checkOrcaBaseUrl(`http://${host}:8080`).ok).toBe(true);
+  });
+  it('still requires HTTPS for a non-local host', () => {
+    expect(checkOrcaBaseUrl('http://ws.example.com').ok).toBe(false);
+  });
+  it('suggests the local host root when a path was added', () => {
+    const result = checkOrcaBaseUrl('http://127.0.0.1:8080/v1');
+    expect(result.ok).toBe(false);
+    expect(result.fix).toContain('http://127.0.0.1:8080');
+  });
+});
diff --git a/typescript/test/ensure.test.ts b/typescript/test/ensure.test.ts
index f1b9264..b65790f 100644
--- a/typescript/test/ensure.test.ts
+++ b/typescript/test/ensure.test.ts
@@ -6,7 +6,7 @@ import { join } from 'node:path';
 
 import { beforeEach, describe, expect, it } from 'vitest';
 
-import { State, ensureAgent, ensureEnvironment, ensureVault } from '../src/common.js';
+import { Config, State, ensureAgent, ensureEnvironment, ensureVault } from '../src/common.js';
 import { fakeClient, makeAgent } from './fakes.js';
 import { MCP_URL, params } from './helpers.js';
 
@@ -98,7 +98,7 @@ describe('ensureVault', () => {
   it('gets one bearer credential for the MCP server', async () => {
     const client = fakeClient();
 
-    const vaultId = await ensureVault(client, state, 'hello-vault-jane', MCP_URL, 'the-api-key');
+    const vaultId = await ensureVault(client, state, 'hello-vault-jane', new Config({ SN_MCP_AUTH: 'static_bearer', SN_MCP_URL: MCP_URL, SN_API_KEY: 'the-api-key' }, 'jane'));
 
     expect(client.vaults.credentials.calls.filter((c) => c[0] === 'create')).toEqual([
       ['create', vaultId, { type: 'static_bearer', mcp_server_url: MCP_URL, token: 'the-api-key' }, 'streamnative-mcp'],
@@ -107,9 +107,9 @@ describe('ensureVault', () => {
 
   it('reuses an existing credential for the same server', async () => {
     const client = fakeClient();
-    const first = await ensureVault(client, state, 'hello-vault-jane', MCP_URL, 'the-api-key');
+    const first = await ensureVault(client, state, 'hello-vault-jane', new Config({ SN_MCP_AUTH: 'static_bearer', SN_MCP_URL: MCP_URL, SN_API_KEY: 'the-api-key' }, 'jane'));
 
-    const second = await ensureVault(client, state, 'hello-vault-jane', MCP_URL, 'the-api-key');
+    const second = await ensureVault(client, state, 'hello-vault-jane', new Config({ SN_MCP_AUTH: 'static_bearer', SN_MCP_URL: MCP_URL, SN_API_KEY: 'the-api-key' }, 'jane'));
 
     expect(first).toBe(second);
     expect(client.vaults.credentials.calls.filter((c) => c[0] === 'create')).toHaveLength(1);
@@ -117,9 +117,9 @@ describe('ensureVault', () => {
 
   it('gives a new MCP URL its own credential', async () => {
     const client = fakeClient();
-    await ensureVault(client, state, 'hello-vault-jane', MCP_URL, 'the-api-key');
+    await ensureVault(client, state, 'hello-vault-jane', new Config({ SN_MCP_AUTH: 'static_bearer', SN_MCP_URL: MCP_URL, SN_API_KEY: 'the-api-key' }, 'jane'));
 
-    await ensureVault(client, state, 'hello-vault-jane', `${MCP_URL}-v2`, 'the-api-key');
+    await ensureVault(client, state, 'hello-vault-jane', new Config({ SN_MCP_AUTH: 'static_bearer', SN_MCP_URL: `${MCP_URL}-v2`, SN_API_KEY: 'the-api-key' }, 'jane'));
 
     const urls = client.vaults.credentials.calls
       .filter((c) => c[0] === 'create')
diff --git a/typescript/test/fakes.ts b/typescript/test/fakes.ts
index c2ce609..7558e00 100644
--- a/typescript/test/fakes.ts
+++ b/typescript/test/fakes.ts
@@ -205,6 +205,15 @@ export class FakeEnvironments {
     return environment;
   }
 
+  async archive(environmentId: string): Promise<Environment> {
+    this.calls.push(['archive', environmentId]);
+    const environment = this.store.get(environmentId);
+    if (!environment) throw notFound();
+    const archived = { ...environment, archived_at: NOW };
+    this.store.set(environmentId, archived);
+    return archived;
+  }
+
   async delete(environmentId: string): Promise<void> {
     this.calls.push(['delete', environmentId]);
     if (!this.store.delete(environmentId)) throw notFound();
@@ -234,6 +243,14 @@ export class FakeCredentials {
     return credential;
   }
 
+  async archive(vaultId: string, credentialId: string): Promise<VaultCredential> {
+    this.calls.push(['archive', vaultId, credentialId]);
+    const credential = this.store.get(vaultId)?.find((c) => c.id === credentialId);
+    if (!credential) throw notFound();
+    credential.archived_at = NOW;
+    return credential;
+  }
+
   async list(vaultId: string): Promise<{ data: VaultCredential[] }> {
     this.calls.push(['list', vaultId]);
     return { data: [...(this.store.get(vaultId) ?? [])] };
diff --git a/typescript/test/mcp-oauth.test.ts b/typescript/test/mcp-oauth.test.ts
new file mode 100644
index 0000000..48cb78b
--- /dev/null
+++ b/typescript/test/mcp-oauth.test.ts
@@ -0,0 +1,106 @@
+/** OAuth delegation, reuse and doctor without a live browser or credentials. */
+import { spawnSync } from 'node:child_process';
+import { mkdtempSync, readFileSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { beforeEach, expect, it, vi } from 'vitest';
+import Orca, { type VaultCredential } from '@runorca/orca-sdk';
+import { Config, ConfigError, State, authorizeMcp, ensureVault } from '../src/common.js';
+import { probeMcp } from '../src/doctor.js';
+import { fakeClient } from './fakes.js';
+
+vi.mock('node:child_process', () => ({ spawnSync: vi.fn() }));
+const URL = 'https://mcp.example.com/mcp';
+const config = (values: Record<string, string> = {}) => new Config({ ORCA_BASE_URL: 'https://registry.example.com', SN_API_KEY: 'registry-secret', SN_MCP_URL: URL, ...values }, 'jane');
+let state: State;
+beforeEach(() => {
+  vi.clearAllMocks();
+  state = new State(join(mkdtempSync(join(tmpdir(), 'hello-oauth-')), 'jane.json'));
+  vi.mocked(spawnSync).mockReturnValue({ status: 0 } as ReturnType<typeof spawnSync>);
+});
+
+async function seed(client: ReturnType<typeof fakeClient>, authType: 'mcp_oauth' | 'static_bearer' = 'mcp_oauth', url = URL, archived = false) {
+  const vault = await client.vaults.create({ display_name: 'test' });
+  state.set('vault_id', vault.id);
+  const credential: VaultCredential = { id: 'cred_oauth', vault_id: vault.id, type: 'vault_credential', display_name: 'test', metadata: {}, created_at: 'now', updated_at: 'now', archived_at: archived ? 'old' : null, auth: { type: authType, mcp_server_url: url } };
+  client.vaults.credentials.store.set(vault.id, [credential]);
+  return vault.id;
+}
+
+it.each([false, true])('delegates OAuth without secret arguments; local=%s', (local) => {
+  process.env.ORCA_API_KEY = 'stale-local';
+  process.env.ORCA_ACCESS_TOKEN = 'stale-bearer';
+  try {
+    authorizeMcp('vlt_1', config({ SN_MCP_OAUTH_ISSUER: 'https://auth.example.com/', SN_MCP_OAUTH_SCOPE: 'openid offline_access', ...(local ? { ORCA_API_KEY: 'local-secret' } : {}) }));
+    const [command, args, opts] = vi.mocked(spawnSync).mock.calls[0];
+    expect(command).toBe('ork');
+    expect(args).toEqual(['agent', 'vaults', 'credentials', 'create', '--vault', 'vlt_1', '--display-name', 'streamnative-mcp', '--mcp-server-url', URL, '-o', 'json', '--oauth-issuer', 'https://auth.example.com/', '--oauth-scope', 'openid offline_access']);
+    const options = opts as { env: NodeJS.ProcessEnv; stdio: string };
+    expect(options.stdio).toBe('inherit');
+    expect(options.env.ORCA_REGISTRY_URL).toBe('https://registry.example.com');
+    expect(options.env.ORCA_API_KEY).toBe(local ? 'local-secret' : undefined);
+    expect(options.env.ORCA_ACCESS_TOKEN).toBe(local ? undefined : 'registry-secret');
+    expect(JSON.stringify(args)).not.toContain('secret');
+  } finally {
+    delete process.env.ORCA_API_KEY;
+    delete process.env.ORCA_ACCESS_TOKEN;
+  }
+});
+
+it('creates an OAuth vault by default and reuses its credential across runs', async () => {
+  const client = fakeClient();
+  const vaultId = await ensureVault(client, state, 'test', config());
+  expect(spawnSync).toHaveBeenCalledTimes(1);
+  expect(client.vaults.credentials.calls.filter((c) => c[0] === 'create')).toEqual([]);
+  expect(readFileSync(state.path, 'utf8')).not.toContain('registry-secret');
+  // The CLI, not the SDK, populated this read shape on the server.
+  client.vaults.credentials.store.set(vaultId, [{ id: 'cred_oauth', type: 'vault_credential', vault_id: vaultId, display_name: 'test', metadata: {}, created_at: 'now', updated_at: 'now', archived_at: null, auth: { type: 'mcp_oauth', mcp_server_url: URL } }]);
+  expect(await ensureVault(client, state, 'test', config())).toBe(vaultId);
+  expect(spawnSync).toHaveBeenCalledTimes(1);
+});
+
+it.each([
+  ['static_bearer', URL, false],
+  ['mcp_oauth', URL + '-other', false],
+  ['mcp_oauth', URL, true],
+] as const)('does not reuse %s at %s archived=%s', async (authType, url, archived) => {
+  const client = fakeClient();
+  await seed(client, authType, url, archived);
+  await ensureVault(client, state, 'test', config());
+  expect(spawnSync).toHaveBeenCalledTimes(1);
+  expect(client.vaults.credentials.calls.filter((c) => c[0] === 'archive')).toHaveLength(authType === 'static_bearer' && url === URL && !archived ? 1 : 0);
+});
+
+it.each([{ status: 1 }, { status: null, error: new Error('missing ork') }])('stops OAuth errors without a static fallback', async (result) => {
+  vi.mocked(spawnSync).mockReturnValue(result as ReturnType<typeof spawnSync>);
+  const client = fakeClient();
+  await expect(ensureVault(client, state, 'test', config())).rejects.toBeInstanceOf(ConfigError);
+  expect(client.vaults.credentials.calls.filter((c) => c[0] === 'create')).toEqual([]);
+});
+
+it('rejects an invalid auth mode before creating resources', async () => {
+  const client = fakeClient();
+  await expect(ensureVault(client, state, 'test', config({ SN_MCP_AUTH: 'typo' }))).rejects.toThrow('SN_MCP_AUTH');
+  expect(client.vaults.calls).toEqual([]);
+});
+
+it.each([['valid', true], ['invalid', false], ['unknown', false]] as const)('doctor uses server-side validation: %s', async (status, ok) => {
+  const fake = fakeClient();
+  const vaultId = await seed(fake);
+  const client = new Orca({ baseURL: 'https://registry.example.com', apiKey: 'test' });
+  vi.spyOn(client.vaults.credentials, 'list').mockImplementation(() => fake.vaults.credentials.list(vaultId) as ReturnType<typeof client.vaults.credentials.list>);
+  const validate = vi.spyOn(client.vaults.credentials, 'validate').mockResolvedValue({ status } as Awaited<ReturnType<typeof client.vaults.credentials.validate>>);
+  const check = await probeMcp(config(), client, state);
+  expect(check.ok).toBe(ok);
+  expect(check.detail).toContain('initialization');
+  if (status === 'unknown') expect(check.fix).toContain('keep the existing credential');
+  if (status === 'invalid') expect(check.fix).toContain('credentials archive cred_oauth');
+  expect(validate).toHaveBeenCalledWith(vaultId, 'cred_oauth');
+});
+
+it('doctor asks for OAuth setup before first L3', async () => {
+  const client = new Orca({ baseURL: 'https://registry.example.com', apiKey: 'test' });
+  const check = await probeMcp(config(), client, state);
+  expect(check.ok).toBe(false);
+  expect(check.fix).toContain('Run L3');
+});
diff --git a/typescript/test/scripts-support.test.ts b/typescript/test/scripts-support.test.ts
index 604c4be..2ebc538 100644
--- a/typescript/test/scripts-support.test.ts
+++ b/typescript/test/scripts-support.test.ts
@@ -7,7 +7,7 @@ import { join } from 'node:path';
 import { APIConnectionError, AuthenticationError } from '@runorca/orca-sdk';
 import { afterEach, describe, expect, it, vi } from 'vitest';
 
-import { ConfigError, State, TurnError, askHuman, chat, cleanup, ensureAgent, ensureEnvironment, ensureVault, runMain } from '../src/common.js';
+import { Config, ConfigError, State, TurnError, askHuman, chat, cleanup, ensureAgent, ensureEnvironment, ensureVault, runMain } from '../src/common.js';
 import { FakeSessionEvents, clientWithEvents, fakeClient, type RawEvent } from './fakes.js';
 import { MCP_URL, params } from './helpers.js';
 
@@ -57,18 +57,19 @@ describe('askHuman', () => {
 });
 
 describe('cleanup', () => {
-  it('archives the agent, deletes the rest, and forgets the ids', async () => {
+  it('archives the agent and environment, deletes the vault, and forgets the ids', async () => {
     const client = fakeClient();
     const dir = mkdtempSync(join(tmpdir(), 'hello-state-'));
     const state = new State(join(dir, 'jane.json'));
     const agent = await ensureAgent(client, state, params('l1-hello'));
     const environmentId = await ensureEnvironment(client, state, 'hello-env-jane');
-    const vaultId = await ensureVault(client, state, 'hello-vault-jane', MCP_URL, 'key');
+    const vaultId = await ensureVault(client, state, 'hello-vault-jane', new Config({ SN_MCP_AUTH: 'static_bearer', SN_MCP_URL: MCP_URL, SN_API_KEY: 'key' }, 'jane'));
 
     await cleanup(client, state, { out: quiet });
 
     expect(client.agents.store.get(agent.id)!.archived_at).not.toBeNull();
-    expect(client.environments.store.has(environmentId)).toBe(false);
+    expect(client.environments.store.get(environmentId)!.archived_at).not.toBeNull();
+    expect(client.environments.calls).not.toContainEqual(['delete', environmentId]);
     expect(client.vaults.store.has(vaultId)).toBe(false);
     expect(existsSync(join(dir, 'jane.json'))).toBe(false);
   });