chore: resolve conflicts

tests/e2e/CLAUDE.md

@@ -53,6 +53,7 @@ HEADED=1 pytest scenarios/
 | `test_skills.py` | Skills tab UI visibility, ClawHub search (skipped if registry unreachable), install + remove lifecycle |
 | `test_sse_reconnect.py` | SSE reconnects after programmatic `eventSource.close()` + `connectSSE()`; history is reloaded after reconnect |
 | `test_tool_approval.py` | Approval card appears, buttons disable on approve/deny, parameters toggle via `page.evaluate("showApproval(...)")`; the waiting-approval regression uses a real HTTP tool call |
+| `test_oauth_refresh.py` | Hosted Gmail OAuth regression: complete setup via `/oauth/callback`, expire the stored access token in libSQL, trigger a real `gmail` tool call through `/api/chat/send`, and verify refresh goes through the mock `/oauth/refresh` proxy without forwarding `client_secret` |
 
 ## `helpers.py`
 
@@ -75,6 +76,7 @@ All fixtures are defined in `tests/e2e/conftest.py`. Running `pytest scenarios/`
 | `ironclaw_binary` | Checks `target/debug/ironclaw`; if absent, runs `cargo build --no-default-features --features libsql` (timeout 600s). |
 | `mock_llm_server` | Starts `mock_llm.py --port 0`, reads the assigned port from stdout, waits for `/v1/models` to return 200. Yields the base URL. |
 | `ironclaw_server` | Starts the ironclaw binary with a minimal env (see below), waits for `/api/health` (timeout 60s). Yields the base URL. On teardown sends **SIGINT** (not SIGTERM) so the tokio ctrl_c handler triggers a graceful shutdown and LLVM coverage data is flushed. |
+| `hosted_oauth_refresh_server` | Starts a second ironclaw instance with a dedicated libSQL DB and `GOOGLE_OAUTH_CLIENT_ID=hosted-google-client-id`, while still pointing `IRONCLAW_OAUTH_EXCHANGE_URL` at `mock_llm.py`. Yields a dict with `base_url`, `db_path`, `gateway_user_id`, and `mock_llm_url` for the hosted refresh regression scenario. |
 | `browser` | Launches a single Chromium instance (headless by default; set `HEADED=1` for headed). Shared across all tests. |
 
 ### Function-scoped fixtures
@@ -100,6 +102,8 @@ EMBEDDING_ENABLED=false, SKILLS_ENABLED=true
 ONBOARD_COMPLETED=true   # prevents setup wizard
 ```
 
+The `hosted_oauth_refresh_server` fixture uses the same baseline, but with its own DB/home tempdirs and `GOOGLE_OAUTH_CLIENT_ID=hosted-google-client-id` so hosted OAuth flows exercise proxy credential injection instead of the baked-in desktop Google app.
+
 The binary is also started with `--no-onboard`. Coverage env vars (`CARGO_LLVM_COV*`, `LLVM_*`, `CARGO_ENCODED_RUSTFLAGS`, `CARGO_INCREMENTAL`) are forwarded from the outer environment when present.
 
 ## Mock LLM (`mock_llm.py`)
@@ -113,6 +117,11 @@ python mock_llm.py --port 0
 
 It serves `POST /v1/chat/completions` (streaming + non-streaming) and `GET /v1/models`. Responses are pattern-matched from `CANNED_RESPONSES` against the last user message. Unmatched messages return `"I understand your request."`. The model name reported is always `"mock-model"`.
 
+It also hosts OAuth test endpoints:
+- `POST /oauth/exchange` for hosted auth-code exchange
+- `POST /oauth/refresh` for hosted refresh-token exchange
+- `GET /__mock/oauth/state` and `POST /__mock/oauth/reset` so HTTP E2E scenarios can assert exact proxy payloads and reset counters between setup and refresh assertions
+
 To add a new canned response:
 ```python
 # In mock_llm.py

tests/e2e/conftest.py

@@ -112,6 +112,39 @@ def _reserve_loopback_sockets(count: int) -> list[socket.socket]:
             sock.close()
         raise
 
+async def _stop_process(
+    proc: asyncio.subprocess.Process, *, sig: int | None = None, timeout: float
+) -> None:
+    """Signal a subprocess and wait briefly without masking exit races."""
+    if proc.returncode is not None:
+        return
+
+    try:
+        if sig is None:
+            proc.kill()
+        else:
+            proc.send_signal(sig)
+    except ProcessLookupError:
+        try:
+            await asyncio.wait_for(proc.wait(), timeout=timeout)
+        except asyncio.TimeoutError:
+            pass
+        return
+
+    try:
+        await asyncio.wait_for(proc.wait(), timeout=timeout)
+    except asyncio.TimeoutError:
+        pass
+
+
+def _forward_coverage_env(env: dict[str, str]) -> None:
+    """Forward cargo-llvm-cov env vars into child processes when present."""
+    cov_env_prefixes = ("CARGO_LLVM_COV", "LLVM_")
+    cov_env_extras = ("CARGO_ENCODED_RUSTFLAGS", "CARGO_INCREMENTAL")
+    for key, val in os.environ.items():
+        if key.startswith(cov_env_prefixes) or key in cov_env_extras:
+            env[key] = val
+
 
 @pytest.fixture(scope="session")
 def ironclaw_binary():
@@ -264,14 +297,7 @@ async def ironclaw_server(
         "IRONCLAW_OAUTH_CALLBACK_URL": "https://oauth.test.example/oauth/callback",
         "IRONCLAW_OAUTH_EXCHANGE_URL": mock_llm_server,
     }
-    # Forward LLVM coverage instrumentation env vars when present
-    # (allows cargo-llvm-cov to collect profraw data from E2E runs).
-    # Use prefix matching to stay resilient to cargo-llvm-cov changes.
-    COV_ENV_PREFIXES = ("CARGO_LLVM_COV", "LLVM_")
-    COV_ENV_EXTRAS = ("CARGO_ENCODED_RUSTFLAGS", "CARGO_INCREMENTAL")
-    for key, val in os.environ.items():
-        if key.startswith(COV_ENV_PREFIXES) or key in COV_ENV_EXTRAS:
-            env[key] = val
+    _forward_coverage_env(env)
     proc = await asyncio.create_subprocess_exec(
         ironclaw_binary, "--no-onboard",
         stdin=asyncio.subprocess.DEVNULL,
@@ -279,35 +305,145 @@ async def ironclaw_server(
         stderr=asyncio.subprocess.PIPE,
         env=env,
     )
+    startup_kill_attempted = False
     base_url = f"http://127.0.0.1:{gateway_port}"
     try:
         await wait_for_ready(f"{base_url}/api/health", timeout=60)
         yield base_url
     except TimeoutError:
         # Dump stderr so CI logs show why the server failed to start
+        if proc.returncode is None:
+            startup_kill_attempted = True
+            await _stop_process(proc, timeout=2)
         returncode = proc.returncode
         stderr_bytes = b""
         if proc.stderr:
             try:
                 stderr_bytes = await asyncio.wait_for(proc.stderr.read(8192), timeout=2)
-            except (asyncio.TimeoutError, Exception):
+            except asyncio.TimeoutError:
                 pass
         stderr_text = stderr_bytes.decode("utf-8", errors="replace")
-        proc.kill()
         pytest.fail(
             f"ironclaw server failed to start on port {gateway_port} "
             f"(returncode={returncode}).\nstderr:\n{stderr_text}"
         )
     finally:
         if proc.returncode is None:
-            # Use SIGINT (not SIGTERM) so tokio's ctrl_c handler triggers a
-            # graceful shutdown.  This lets the LLVM coverage runtime run its
-            # atexit handler and flush .profraw files for cargo-llvm-cov.
-            proc.send_signal(signal.SIGINT)
-            try:
-                await asyncio.wait_for(proc.wait(), timeout=10)
-            except asyncio.TimeoutError:
-                proc.kill()
+            if startup_kill_attempted:
+                await _stop_process(proc, timeout=2)
+            else:
+                # Use SIGINT (not SIGTERM) so tokio's ctrl_c handler triggers a
+                # graceful shutdown.  This lets the LLVM coverage runtime run its
+                # atexit handler and flush .profraw files for cargo-llvm-cov.
+                await _stop_process(proc, sig=signal.SIGINT, timeout=10)
+                if proc.returncode is None:
+                    await _stop_process(proc, timeout=2)
+
+
+@pytest.fixture(scope="session")
+async def hosted_oauth_refresh_server(
+    ironclaw_binary,
+    mock_llm_server,
+    wasm_tools_dir,
+):
+    """Start a hosted-mode ironclaw instance for OAuth refresh regression tests."""
+    reserved = _reserve_loopback_sockets(2)
+    db_tmpdir = tempfile.TemporaryDirectory(prefix="ironclaw-e2e-hosted-oauth-db-")
+    home_tmpdir = tempfile.TemporaryDirectory(prefix="ironclaw-e2e-hosted-oauth-home-")
+
+    try:
+        gateway_port = reserved[0].getsockname()[1]
+        http_port = reserved[1].getsockname()[1]
+        for sock in reserved:
+            if sock.fileno() != -1:
+                sock.close()
+
+        db_path = os.path.join(db_tmpdir.name, "hosted-oauth-refresh.db")
+        home_dir = home_tmpdir.name
+        env = {
+            "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
+            "HOME": home_dir,
+            "IRONCLAW_BASE_DIR": os.path.join(home_dir, ".ironclaw"),
+            "RUST_LOG": "ironclaw=info",
+            "RUST_BACKTRACE": "1",
+            "IRONCLAW_OWNER_ID": OWNER_SCOPE_ID,
+            "GATEWAY_ENABLED": "true",
+            "GATEWAY_HOST": "127.0.0.1",
+            "GATEWAY_PORT": str(gateway_port),
+            "GATEWAY_AUTH_TOKEN": AUTH_TOKEN,
+            "GATEWAY_USER_ID": OWNER_SCOPE_ID,
+            "HTTP_HOST": "127.0.0.1",
+            "HTTP_PORT": str(http_port),
+            "HTTP_WEBHOOK_SECRET": HTTP_WEBHOOK_SECRET,
+            "CLI_ENABLED": "false",
+            "LLM_BACKEND": "openai_compatible",
+            "LLM_BASE_URL": mock_llm_server,
+            "LLM_MODEL": "mock-model",
+            "DATABASE_BACKEND": "libsql",
+            "LIBSQL_PATH": db_path,
+            "SECRETS_MASTER_KEY": "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
+            "SANDBOX_ENABLED": "false",
+            "SKILLS_ENABLED": "true",
+            "ROUTINES_ENABLED": "true",
+            "HEARTBEAT_ENABLED": "false",
+            "EMBEDDING_ENABLED": "false",
+            "WASM_ENABLED": "true",
+            "WASM_TOOLS_DIR": wasm_tools_dir,
+            "WASM_CHANNELS_DIR": _WASM_CHANNELS_TMPDIR.name,
+            "ONBOARD_COMPLETED": "true",
+            "IRONCLAW_OAUTH_CALLBACK_URL": "https://oauth.test.example/oauth/callback",
+            "IRONCLAW_OAUTH_EXCHANGE_URL": mock_llm_server,
+            "GOOGLE_OAUTH_CLIENT_ID": "hosted-google-client-id",
+        }
+        _forward_coverage_env(env)
+
+        proc = await asyncio.create_subprocess_exec(
+            ironclaw_binary, "--no-onboard",
+            stdin=asyncio.subprocess.DEVNULL,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+            env=env,
+        )
+        startup_kill_attempted = False
+        base_url = f"http://127.0.0.1:{gateway_port}"
+        try:
+            await wait_for_ready(f"{base_url}/api/health", timeout=60)
+            yield {
+                "base_url": base_url,
+                "db_path": db_path,
+                "gateway_user_id": OWNER_SCOPE_ID,
+                "mock_llm_url": mock_llm_server,
+            }
+        except TimeoutError:
+            if proc.returncode is None:
+                startup_kill_attempted = True
+                await _stop_process(proc, timeout=2)
+            returncode = proc.returncode
+            stderr_bytes = b""
+            if proc.stderr:
+                try:
+                    stderr_bytes = await asyncio.wait_for(proc.stderr.read(8192), timeout=2)
+                except asyncio.TimeoutError:
+                    pass
+            stderr_text = stderr_bytes.decode("utf-8", errors="replace")
+            pytest.fail(
+                f"hosted oauth refresh server failed to start on port {gateway_port} "
+                f"(returncode={returncode}).\nstderr:\n{stderr_text}"
+            )
+        finally:
+            if proc.returncode is None:
+                if startup_kill_attempted:
+                    await _stop_process(proc, timeout=2)
+                else:
+                    await _stop_process(proc, sig=signal.SIGINT, timeout=10)
+                    if proc.returncode is None:
+                        await _stop_process(proc, timeout=2)
+    finally:
+        for sock in reserved:
+            if sock.fileno() != -1:
+                sock.close()
+        db_tmpdir.cleanup()
+        home_tmpdir.cleanup()
 
 
 @pytest.fixture(scope="session")
@@ -362,12 +498,7 @@ async def http_channel_server_without_secret(
         "IRONCLAW_OAUTH_CALLBACK_URL": "https://oauth.test.example/oauth/callback",
         "IRONCLAW_OAUTH_EXCHANGE_URL": mock_llm_server,
     }
-    # Forward LLVM coverage instrumentation env vars when present
-    COV_ENV_PREFIXES = ("CARGO_LLVM_COV", "LLVM_")
-    COV_ENV_EXTRAS = ("CARGO_ENCODED_RUSTFLAGS", "CARGO_INCREMENTAL")
-    for key, val in os.environ.items():
-        if key.startswith(COV_ENV_PREFIXES) or key in COV_ENV_EXTRAS:
-            env[key] = val
+    _forward_coverage_env(env)
     proc = await asyncio.create_subprocess_exec(
         ironclaw_binary, "--no-onboard",
         stdin=asyncio.subprocess.DEVNULL,
@@ -375,6 +506,7 @@ async def http_channel_server_without_secret(
         stderr=asyncio.subprocess.PIPE,
         env=env,
     )
+    startup_kill_attempted = False
     gateway_url = f"http://127.0.0.1:{gateway_port}"
     http_base_url = f"http://127.0.0.1:{http_port}"
     try:
@@ -383,15 +515,17 @@ async def http_channel_server_without_secret(
         yield http_base_url
     except TimeoutError:
         # Dump stderr so CI logs show why the server failed to start
+        if proc.returncode is None:
+            startup_kill_attempted = True
+            await _stop_process(proc, timeout=2)
         returncode = proc.returncode
         stderr_bytes = b""
         if proc.stderr:
             try:
                 stderr_bytes = await asyncio.wait_for(proc.stderr.read(8192), timeout=2)
-            except (asyncio.TimeoutError, Exception):
+            except asyncio.TimeoutError:
                 pass
         stderr_text = stderr_bytes.decode("utf-8", errors="replace")
-        proc.kill()
         pytest.fail(
             f"ironclaw server without webhook secret failed to start on ports "
             f"gateway={gateway_port}, http={http_port} "
@@ -399,14 +533,15 @@ async def http_channel_server_without_secret(
         )
     finally:
         if proc.returncode is None:
-            # Use SIGINT (not SIGTERM) so tokio's ctrl_c handler triggers a
-            # graceful shutdown.  This lets the LLVM coverage runtime run its
-            # atexit handler and flush .profraw files for cargo-llvm-cov.
-            proc.send_signal(signal.SIGINT)
-            try:
-                await asyncio.wait_for(proc.wait(), timeout=10)
-            except asyncio.TimeoutError:
-                proc.kill()
+            if startup_kill_attempted:
+                await _stop_process(proc, timeout=2)
+            else:
+                # Use SIGINT (not SIGTERM) so tokio's ctrl_c handler triggers a
+                # graceful shutdown.  This lets the LLVM coverage runtime run its
+                # atexit handler and flush .profraw files for cargo-llvm-cov.
+                await _stop_process(proc, sig=signal.SIGINT, timeout=10)
+                if proc.returncode is None:
+                    await _stop_process(proc, timeout=2)
 
 
 @pytest.fixture(scope="session")

tests/e2e/mock_llm.py

@@ -34,6 +34,15 @@ TOOL_CALL_PATTERNS = [
             "body": {"label": m.group("label")},
         },
     ),
+    (
+        re.compile(r"check gmail unread|gmail unread", re.IGNORECASE),
+        "gmail",
+        lambda _: {
+            "action": "list_messages",
+            "query": "is:unread",
+            "max_results": 1,
+        },
+    ),
     (re.compile(r"what time|current time", re.IGNORECASE), "time", lambda _: {"operation": "now"}),
     (
         re.compile(
@@ -91,6 +100,15 @@ TOOL_CALL_PATTERNS = [
 ]
 
 
+def _new_oauth_state() -> dict:
+    return {
+        "exchange_count": 0,
+        "refresh_count": 0,
+        "last_exchange": None,
+        "last_refresh": None,
+    }
+
+
 def _last_user_content(messages: list[dict]) -> str:
     for msg in reversed(messages):
         if msg.get("role") == "user":
@@ -272,6 +290,12 @@ async def oauth_exchange(request: web.Request) -> web.Response:
     specific token params such as RFC 8707 `resource` are forwarded here.
     """
     data = await request.post()
+    oauth_state = request.app["oauth_state"]
+    oauth_state["exchange_count"] += 1
+    oauth_state["last_exchange"] = {
+        "authorization": request.headers.get("Authorization"),
+        "form": dict(data),
+    }
     code = data.get("code", "")
     access_token_field = data.get("access_token_field", "access_token")
 
@@ -290,6 +314,39 @@ async def oauth_exchange(request: web.Request) -> web.Response:
     })
 
 
+async def oauth_refresh(request: web.Request) -> web.Response:
+    """Mock OAuth token refresh proxy for hosted refresh E2E tests."""
+    data = await request.post()
+    oauth_state = request.app["oauth_state"]
+    oauth_state["refresh_count"] += 1
+    oauth_state["last_refresh"] = {
+        "authorization": request.headers.get("Authorization"),
+        "form": dict(data),
+    }
+
+    if request.headers.get("Authorization") != "Bearer e2e-test-token":
+        return web.json_response({"error": "invalid_gateway_auth"}, status=401)
+    if data.get("client_id") != "hosted-google-client-id":
+        return web.json_response({"error": "invalid_client_id"}, status=400)
+    if "client_secret" in data:
+        return web.json_response({"error": "unexpected_client_secret"}, status=400)
+
+    return web.json_response({
+        "access_token": "mock-refreshed-access-token",
+        "refresh_token": "mock-rotated-refresh-token",
+        "expires_in": 3600,
+    })
+
+
+async def oauth_state_handler(request: web.Request) -> web.Response:
+    return web.json_response(request.app["oauth_state"])
+
+
+async def oauth_reset(request: web.Request) -> web.Response:
+    request.app["oauth_state"] = _new_oauth_state()
+    return web.json_response({"ok": True})
+
+
 async def models(_request: web.Request) -> web.Response:
     return web.json_response({
         "object": "list",
@@ -424,12 +481,16 @@ def main():
     parser.add_argument("--port", type=int, default=0)
     args = parser.parse_args()
     app = web.Application()
+    app["oauth_state"] = _new_oauth_state()
     # Register both /v1/ and non-/v1/ paths (rig-core omits the /v1/ prefix)
     app.router.add_post("/v1/chat/completions", chat_completions)
     app.router.add_post("/chat/completions", chat_completions)
     app.router.add_get("/v1/models", models)
     app.router.add_get("/models", models)
     app.router.add_post("/oauth/exchange", oauth_exchange)
+    app.router.add_post("/oauth/refresh", oauth_refresh)
+    app.router.add_get("/__mock/oauth/state", oauth_state_handler)
+    app.router.add_post("/__mock/oauth/reset", oauth_reset)
     # Mock MCP server endpoints
     app.router.add_post("/mcp", mcp_endpoint)
     app.router.add_post("/mcp-400", mcp_endpoint_400)

tests/e2e/scenarios/test_oauth_refresh.py

@@ -0,0 +1,227 @@
+"""Hosted OAuth refresh HTTP regression test.
+
+Runs a real ironclaw binary in hosted mode, expires a stored Gmail access
+token in the libSQL database, triggers a real gmail tool call through the
+chat API, and verifies that refresh uses the hosted proxy endpoint.
+"""
+
+import asyncio
+import sqlite3
+from datetime import datetime, timezone
+from urllib.parse import parse_qs, urlparse
+
+import httpx
+
+from helpers import api_get, api_post
+
+
+def _extract_state(auth_url: str) -> str:
+    parsed = urlparse(auth_url)
+    state = parse_qs(parsed.query).get("state", [None])[0]
+    assert state, f"auth_url should include state: {auth_url}"
+    return state
+
+
+def _parse_timestamp(value: str | None) -> datetime | None:
+    if value is None:
+        return None
+    return datetime.fromisoformat(value.replace("Z", "+00:00"))
+
+
+def _expire_access_token(db_path: str, user_id: str, secret_name: str) -> None:
+    with sqlite3.connect(db_path) as conn:
+        cursor = conn.execute(
+            """
+            UPDATE secrets
+            SET expires_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now', '-1 hour')
+            WHERE user_id = ?1 AND name = ?2
+            """,
+            (user_id, secret_name),
+        )
+        conn.commit()
+    assert cursor.rowcount == 1, f"Expected one secret row for {user_id}/{secret_name}"
+
+
+def _find_secret_row(
+    db_path: str,
+    secret_name: str,
+) -> tuple[str, str | None, str | None]:
+    with sqlite3.connect(db_path) as conn:
+        row = conn.execute(
+            """
+            SELECT user_id, expires_at, updated_at
+            FROM secrets
+            WHERE name = ?1
+            ORDER BY updated_at DESC
+            LIMIT 1
+            """,
+            (secret_name,),
+        ).fetchone()
+    assert row is not None, f"Missing secret row for {secret_name}"
+    return row[0], row[1], row[2]
+
+
+async def _get_extension(base_url: str, name: str) -> dict | None:
+    response = await api_get(base_url, "/api/extensions", timeout=15)
+    response.raise_for_status()
+    for extension in response.json().get("extensions", []):
+        if extension["name"] == name:
+            return extension
+    return None
+
+
+async def _reset_mock_oauth_state(mock_base_url: str) -> None:
+    async with httpx.AsyncClient() as client:
+        response = await client.post(f"{mock_base_url}/__mock/oauth/reset", timeout=10)
+    response.raise_for_status()
+
+
+async def _get_mock_oauth_state(mock_base_url: str) -> dict:
+    async with httpx.AsyncClient() as client:
+        response = await client.get(f"{mock_base_url}/__mock/oauth/state", timeout=10)
+    response.raise_for_status()
+    return response.json()
+
+
+async def _approve_pending_request(base_url: str, thread_id: str, request_id: str) -> None:
+    response = await api_post(
+        base_url,
+        "/api/chat/approval",
+        json={"request_id": request_id, "action": "approve", "thread_id": thread_id},
+        timeout=15,
+    )
+    assert response.status_code == 202, (
+        f"Approval submission failed: {response.status_code} {response.text[:400]}"
+    )
+
+
+async def _wait_for_gmail_tool_call(base_url: str, thread_id: str, timeout: float = 30.0) -> dict:
+    approved_request_ids = set()
+    for _ in range(int(timeout * 2)):
+        response = await api_get(
+            base_url,
+            f"/api/chat/history?thread_id={thread_id}",
+            timeout=15,
+        )
+        response.raise_for_status()
+        history = response.json()
+
+        pending = history.get("pending_approval")
+        if pending and pending["request_id"] not in approved_request_ids:
+            await _approve_pending_request(base_url, thread_id, pending["request_id"])
+            approved_request_ids.add(pending["request_id"])
+
+        for turn in history.get("turns", []):
+            for tool_call in turn.get("tool_calls", []):
+                if tool_call.get("name") == "gmail":
+                    return history
+
+        await asyncio.sleep(0.5)
+
+    raise AssertionError(f"Timed out waiting for gmail tool call in thread {thread_id}")
+
+
+async def _wait_for_refresh_request(mock_base_url: str, timeout: float = 20.0) -> dict:
+    for _ in range(int(timeout * 2)):
+        state = await _get_mock_oauth_state(mock_base_url)
+        if state.get("refresh_count") == 1:
+            return state
+        await asyncio.sleep(0.5)
+    raise AssertionError("Timed out waiting for exactly one OAuth refresh request")
+
+
+async def test_hosted_gmail_oauth_refresh_uses_proxy(hosted_oauth_refresh_server):
+    server = hosted_oauth_refresh_server["base_url"]
+    db_path = hosted_oauth_refresh_server["db_path"]
+    mock_base_url = hosted_oauth_refresh_server["mock_llm_url"]
+
+    install_response = await api_post(
+        server,
+        "/api/extensions/install",
+        json={"name": "gmail"},
+        timeout=180,
+    )
+    assert install_response.status_code == 200, install_response.text
+    assert install_response.json().get("success") is True
+
+    setup_response = await api_post(
+        server,
+        "/api/extensions/gmail/setup",
+        json={"secrets": {}},
+        timeout=30,
+    )
+    assert setup_response.status_code == 200, setup_response.text
+    setup_data = setup_response.json()
+    assert setup_data.get("success") is True, setup_data
+    auth_url = setup_data.get("auth_url")
+    assert auth_url, setup_data
+    auth_params = parse_qs(urlparse(auth_url).query)
+    assert auth_params.get("client_id") == ["hosted-google-client-id"]
+
+    async with httpx.AsyncClient() as client:
+        callback_response = await client.get(
+            f"{server}/oauth/callback",
+            params={"code": "mock_auth_code", "state": _extract_state(auth_url)},
+            timeout=30,
+            follow_redirects=True,
+        )
+
+    assert callback_response.status_code == 200, callback_response.text[:400]
+    callback_body = callback_response.text.lower()
+    assert "connected" in callback_body or "success" in callback_body
+
+    gmail = await _get_extension(server, "gmail")
+    assert gmail is not None, "gmail should be installed"
+    assert gmail["authenticated"] is True, gmail
+    assert "gmail" in gmail.get("tools", []), gmail
+
+    await _reset_mock_oauth_state(mock_base_url)
+
+    stored_user_id, expires_before, updated_before = _find_secret_row(
+        db_path, "google_oauth_token"
+    )
+    assert _parse_timestamp(expires_before) is not None
+    assert _parse_timestamp(updated_before) is not None
+
+    await asyncio.sleep(0.1)
+    _expire_access_token(db_path, stored_user_id, "google_oauth_token")
+
+    thread_response = await api_post(server, "/api/chat/thread/new", timeout=15)
+    assert thread_response.status_code == 200, thread_response.text
+    thread_id = thread_response.json()["id"]
+
+    send_response = await api_post(
+        server,
+        "/api/chat/send",
+        json={"content": "check gmail unread", "thread_id": thread_id},
+        timeout=30,
+    )
+    assert send_response.status_code == 202, send_response.text
+
+    history = await _wait_for_gmail_tool_call(server, thread_id)
+    assert any(
+        tool_call.get("name") == "gmail"
+        for turn in history.get("turns", [])
+        for tool_call in turn.get("tool_calls", [])
+    ), history
+
+    oauth_state = await _wait_for_refresh_request(mock_base_url)
+    assert oauth_state["refresh_count"] == 1, oauth_state
+    last_refresh = oauth_state["last_refresh"]
+    assert last_refresh is not None, oauth_state
+    assert last_refresh["authorization"] == "Bearer e2e-test-token"
+    assert last_refresh["form"]["client_id"] == "hosted-google-client-id"
+    assert "client_secret" not in last_refresh["form"], last_refresh
+
+    refreshed_user_id, expires_after, updated_after = _find_secret_row(
+        db_path, "google_oauth_token"
+    )
+    assert refreshed_user_id == stored_user_id
+    expires_after_dt = _parse_timestamp(expires_after)
+    updated_after_dt = _parse_timestamp(updated_after)
+    updated_before_dt = _parse_timestamp(updated_before)
+    assert expires_after_dt is not None
+    assert updated_after_dt is not None
+    assert updated_before_dt is not None
+    assert expires_after_dt > datetime.now(timezone.utc)
+    assert updated_after_dt > updated_before_dt

tests/multi_tenant_integration.rs

@@ -19,10 +19,13 @@ use axum::middleware;
 use axum::routing::{get, post};
 use tower::ServiceExt;
 
+use ironclaw::channels::IncomingMessage;
 use ironclaw::channels::web::auth::{
     AuthenticatedUser, MultiAuthState, UserIdentity, auth_middleware,
 };
-use ironclaw::channels::web::server::{GatewayState, PerUserRateLimiter, RateLimiter};
+use ironclaw::channels::web::server::{
+    GatewayState, PerUserRateLimiter, RateLimiter, start_server,
+};
 use ironclaw::channels::web::sse::SseManager;
 use ironclaw::channels::web::test_helpers::TestGatewayBuilder;
 use ironclaw::channels::web::ws::WsConnectionTracker;
@@ -37,6 +40,9 @@ const ALICE_TOKEN: &str = "tok-alice-secret";
 const BOB_TOKEN: &str = "tok-bob-secret";
 const ALICE_USER_ID: &str = "alice";
 const BOB_USER_ID: &str = "bob";
+const OWNER_TOKEN: &str = "tok-owner-secret";
+const OWNER_SCOPE_ID: &str = "owner-scope";
+const GATEWAY_SENDER_ID: &str = "gateway-sender";
 
 /// Build a MultiAuthState with two users.
 fn two_user_auth() -> MultiAuthState {
@@ -301,7 +307,7 @@ fn per_user_rate_limiter_single_user_mode() {
 
 #[tokio::test]
 async fn sse_scoped_event_only_delivered_to_target_user() {
-    use ironclaw::channels::web::types::SseEvent;
+    use ironclaw_common::AppEvent;
     use tokio_stream::StreamExt;
 
     let manager = SseManager::new();
@@ -319,34 +325,34 @@ async fn sse_scoped_event_only_delivered_to_target_user() {
     // Send event scoped to alice
     manager.broadcast_for_user(
         ALICE_USER_ID,
-        SseEvent::Status {
+        AppEvent::Status {
             message: "alice's event".to_string(),
             thread_id: None,
         },
     );
 
     // Send global heartbeat (both should get it)
-    manager.broadcast(SseEvent::Heartbeat);
+    manager.broadcast(AppEvent::Heartbeat);
 
     // Alice gets her scoped event first
     let e = alice_stream.next().await.unwrap();
     match &e {
-        SseEvent::Status { message, .. } => assert_eq!(message, "alice's event"),
+        AppEvent::Status { message, .. } => assert_eq!(message, "alice's event"),
         _ => panic!("Expected Status, got {:?}", e),
     }
 
     // Alice also gets heartbeat
     let e = alice_stream.next().await.unwrap();
-    assert!(matches!(e, SseEvent::Heartbeat));
+    assert!(matches!(e, AppEvent::Heartbeat));
 
     // Bob only gets the heartbeat (alice's event was filtered)
     let e = bob_stream.next().await.unwrap();
-    assert!(matches!(e, SseEvent::Heartbeat));
+    assert!(matches!(e, AppEvent::Heartbeat));
 }
 
 #[tokio::test]
 async fn sse_global_event_delivered_to_all_users() {
-    use ironclaw::channels::web::types::SseEvent;
+    use ironclaw_common::AppEvent;
     use tokio_stream::StreamExt;
 
     let manager = SseManager::new();
@@ -361,7 +367,7 @@ async fn sse_global_event_delivered_to_all_users() {
             .expect("subscribe"),
     );
 
-    manager.broadcast(SseEvent::Status {
+    manager.broadcast(AppEvent::Status {
         message: "global announcement".to_string(),
         thread_id: None,
     });
@@ -369,7 +375,7 @@ async fn sse_global_event_delivered_to_all_users() {
     let ea = alice.next().await.unwrap();
     let eb = bob.next().await.unwrap();
     match (&ea, &eb) {
-        (SseEvent::Status { message: a, .. }, SseEvent::Status { message: b, .. }) => {
+        (AppEvent::Status { message: a, .. }, AppEvent::Status { message: b, .. }) => {
             assert_eq!(a, "global announcement");
             assert_eq!(b, "global announcement");
         }
@@ -379,7 +385,7 @@ async fn sse_global_event_delivered_to_all_users() {
 
 #[tokio::test]
 async fn sse_user_b_event_not_visible_to_user_a() {
-    use ironclaw::channels::web::types::SseEvent;
+    use ironclaw_common::AppEvent;
     use tokio_stream::StreamExt;
 
     let manager = SseManager::new();
@@ -392,19 +398,19 @@ async fn sse_user_b_event_not_visible_to_user_a() {
     // Send event for bob only
     manager.broadcast_for_user(
         BOB_USER_ID,
-        SseEvent::Response {
+        AppEvent::Response {
             content: "bob's secret".to_string(),
             thread_id: "t1".to_string(),
         },
     );
 
     // Send heartbeat so alice has something to receive
-    manager.broadcast(SseEvent::Heartbeat);
+    manager.broadcast(AppEvent::Heartbeat);
 
     // Alice should only get heartbeat, not bob's response
     let e = alice.next().await.unwrap();
     assert!(
-        matches!(e, SseEvent::Heartbeat),
+        matches!(e, AppEvent::Heartbeat),
         "Expected Heartbeat, got {:?}",
         e
     );
@@ -412,7 +418,7 @@ async fn sse_user_b_event_not_visible_to_user_a() {
 
 #[tokio::test]
 async fn sse_unscoped_subscriber_receives_all_events() {
-    use ironclaw::channels::web::types::SseEvent;
+    use ironclaw_common::AppEvent;
     use tokio_stream::StreamExt;
 
     let manager = SseManager::new();
@@ -421,19 +427,19 @@ async fn sse_unscoped_subscriber_receives_all_events() {
 
     manager.broadcast_for_user(
         ALICE_USER_ID,
-        SseEvent::Status {
+        AppEvent::Status {
             message: "alice only".to_string(),
             thread_id: None,
         },
     );
     manager.broadcast_for_user(
         BOB_USER_ID,
-        SseEvent::Status {
+        AppEvent::Status {
             message: "bob only".to_string(),
             thread_id: None,
         },
     );
-    manager.broadcast(SseEvent::Heartbeat);
+    manager.broadcast(AppEvent::Heartbeat);
 
     // Unscoped subscriber gets ALL three events
     let e1 = stream.next().await.unwrap();
@@ -441,14 +447,14 @@ async fn sse_unscoped_subscriber_receives_all_events() {
     let e3 = stream.next().await.unwrap();
 
     match &e1 {
-        SseEvent::Status { message, .. } => assert_eq!(message, "alice only"),
+        AppEvent::Status { message, .. } => assert_eq!(message, "alice only"),
         _ => panic!("Expected alice's Status"),
     }
     match &e2 {
-        SseEvent::Status { message, .. } => assert_eq!(message, "bob only"),
+        AppEvent::Status { message, .. } => assert_eq!(message, "bob only"),
         _ => panic!("Expected bob's Status"),
     }
-    assert!(matches!(e3, SseEvent::Heartbeat));
+    assert!(matches!(e3, AppEvent::Heartbeat));
 }
 
 // ===========================================================================
@@ -537,7 +543,8 @@ fn gateway_state_has_multi_tenant_fields() {
         job_manager: None,
         prompt_queue: None,
         scheduler: None,
-        default_user_id: "fallback".to_string(), // Multi-tenant: renamed from user_id
+        owner_id: "fallback".to_string(),
+        default_sender_id: "fallback".to_string(),
         shutdown_tx: tokio::sync::RwLock::new(None),
         ws_tracker: Some(Arc::new(WsConnectionTracker::new())),
         llm_provider: None,
@@ -554,7 +561,8 @@ fn gateway_state_has_multi_tenant_fields() {
         secrets_store: None,
     };
 
-    assert_eq!(state.default_user_id, "fallback");
+    assert_eq!(state.owner_id, "fallback");
+    assert_eq!(state.default_sender_id, "fallback");
     assert!(state.workspace_pool.is_none());
 }
 
@@ -573,6 +581,70 @@ async fn start_multi_user_server() -> (SocketAddr, Arc<GatewayState>) {
         .expect("Failed to start multi-user test server")
 }
 
+async fn start_owner_scoped_sender_server() -> (
+    SocketAddr,
+    Arc<GatewayState>,
+    tokio::sync::mpsc::Receiver<IncomingMessage>,
+) {
+    let (agent_tx, agent_rx) = tokio::sync::mpsc::channel(64);
+
+    let mut tokens = HashMap::new();
+    tokens.insert(
+        OWNER_TOKEN.to_string(),
+        UserIdentity {
+            user_id: OWNER_SCOPE_ID.to_string(),
+            workspace_read_scopes: Vec::new(),
+        },
+    );
+    tokens.insert(
+        BOB_TOKEN.to_string(),
+        UserIdentity {
+            user_id: BOB_USER_ID.to_string(),
+            workspace_read_scopes: Vec::new(),
+        },
+    );
+
+    let state = Arc::new(GatewayState {
+        msg_tx: tokio::sync::RwLock::new(Some(agent_tx)),
+        sse: Arc::new(SseManager::new()),
+        workspace: None,
+        workspace_pool: None,
+        session_manager: None,
+        log_broadcaster: None,
+        log_level_handle: None,
+        extension_manager: None,
+        tool_registry: None,
+        store: None,
+        job_manager: None,
+        prompt_queue: None,
+        scheduler: None,
+        owner_id: OWNER_SCOPE_ID.to_string(),
+        default_sender_id: GATEWAY_SENDER_ID.to_string(),
+        shutdown_tx: tokio::sync::RwLock::new(None),
+        ws_tracker: Some(Arc::new(WsConnectionTracker::new())),
+        llm_provider: None,
+        skill_registry: None,
+        skill_catalog: None,
+        chat_rate_limiter: PerUserRateLimiter::new(30, 60),
+        oauth_rate_limiter: RateLimiter::new(10, 60),
+        webhook_rate_limiter: RateLimiter::new(10, 60),
+        registry_entries: Vec::new(),
+        cost_guard: None,
+        routine_engine: Arc::new(tokio::sync::RwLock::new(None)),
+        startup_time: std::time::Instant::now(),
+        active_config: Default::default(),
+        secrets_store: None,
+    });
+
+    let auth = MultiAuthState::multi(tokens);
+    let addr: SocketAddr = "127.0.0.1:0".parse().unwrap();
+    let bound = start_server(addr, state.clone(), auth)
+        .await
+        .expect("Failed to start owner-scoped sender test server");
+
+    (bound, state, agent_rx)
+}
+
 #[tokio::test]
 async fn full_server_alice_can_access_protected_endpoint() {
     let (addr, _state) = start_multi_user_server().await;
@@ -678,6 +750,49 @@ async fn full_server_chat_send_accepted_for_alice() {
     assert_eq!(msg.channel, "gateway");
 }
 
+#[tokio::test]
+async fn full_server_chat_send_rewrites_sender_only_for_owner_scope_rebind() {
+    let (addr, _state, mut agent_rx) = start_owner_scoped_sender_server().await;
+
+    let client = reqwest::Client::new();
+
+    let owner_resp = client
+        .post(format!("http://{}/api/chat/send", addr))
+        .header("Authorization", format!("Bearer {}", OWNER_TOKEN))
+        .header("Content-Type", "application/json")
+        .body(r#"{"content":"hello from owner"}"#)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(owner_resp.status(), 202);
+
+    let owner_msg = tokio::time::timeout(Duration::from_secs(2), agent_rx.recv())
+        .await
+        .expect("Timed out waiting for owner message")
+        .expect("Agent channel closed");
+    assert_eq!(owner_msg.user_id, OWNER_SCOPE_ID);
+    assert_eq!(owner_msg.sender_id, GATEWAY_SENDER_ID);
+    assert_eq!(owner_msg.content, "hello from owner");
+
+    let other_resp = client
+        .post(format!("http://{}/api/chat/send", addr))
+        .header("Authorization", format!("Bearer {}", BOB_TOKEN))
+        .header("Content-Type", "application/json")
+        .body(r#"{"content":"hello from bob"}"#)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(other_resp.status(), 202);
+
+    let other_msg = tokio::time::timeout(Duration::from_secs(2), agent_rx.recv())
+        .await
+        .expect("Timed out waiting for non-owner message")
+        .expect("Agent channel closed");
+    assert_eq!(other_msg.user_id, BOB_USER_ID);
+    assert_eq!(other_msg.sender_id, BOB_USER_ID);
+    assert_eq!(other_msg.content, "hello from bob");
+}
+
 #[tokio::test]
 async fn full_server_chat_send_rejected_without_auth() {
     let (addr, _state) = start_multi_user_server().await;
@@ -768,7 +883,7 @@ async fn full_server_jobs_endpoint_rejected_without_auth() {
 #[tokio::test]
 async fn full_server_ws_multi_user_event_isolation() {
     use futures::StreamExt;
-    use ironclaw::channels::web::types::SseEvent;
+    use ironclaw_common::AppEvent;
     use tokio_tungstenite::tungstenite::Message;
     use tokio_tungstenite::tungstenite::client::IntoClientRequest;
 
@@ -801,14 +916,14 @@ async fn full_server_ws_multi_user_event_isolation() {
     // Broadcast an event scoped to Alice only
     state.sse.broadcast_for_user(
         ALICE_USER_ID,
-        SseEvent::Status {
+        AppEvent::Status {
             message: "alice-only-event".to_string(),
             thread_id: None,
         },
     );
 
     // Broadcast a global heartbeat so Bob has something to receive
-    state.sse.broadcast(SseEvent::Heartbeat);
+    state.sse.broadcast(AppEvent::Heartbeat);
 
     // Alice should get her scoped event
     let alice_msg = tokio::time::timeout(Duration::from_secs(2), alice_ws.next())
@@ -889,7 +1004,8 @@ async fn start_multi_user_server_with_db() -> (
         job_manager: None,
         prompt_queue: None,
         scheduler: None,
-        default_user_id: ALICE_USER_ID.to_string(),
+        owner_id: ALICE_USER_ID.to_string(),
+        default_sender_id: ALICE_USER_ID.to_string(),
         shutdown_tx: tokio::sync::RwLock::new(None),
         ws_tracker: Some(Arc::new(WsConnectionTracker::new())),
         llm_provider: None,

tests/openai_compat_integration.rs

@@ -203,7 +203,8 @@ async fn start_test_server_with_provider(
         job_manager: None,
         prompt_queue: None,
         scheduler: None,
-        default_user_id: "test-user".to_string(),
+        owner_id: "test-user".to_string(),
+        default_sender_id: "test-user".to_string(),
         shutdown_tx: tokio::sync::RwLock::new(None),
         ws_tracker: Some(Arc::new(WsConnectionTracker::new())),
         llm_provider: Some(llm_provider),
@@ -702,7 +703,8 @@ async fn test_no_llm_provider_returns_503() {
         job_manager: None,
         prompt_queue: None,
         scheduler: None,
-        default_user_id: "test-user".to_string(),
+        owner_id: "test-user".to_string(),
+        default_sender_id: "test-user".to_string(),
         shutdown_tx: tokio::sync::RwLock::new(None),
         ws_tracker: Some(Arc::new(WsConnectionTracker::new())),
         llm_provider: None, // No LLM!

tests/support/gateway_workflow_harness.rs

@@ -226,7 +226,8 @@ impl GatewayWorkflowHarness {
             job_manager: None,
             prompt_queue: None,
             scheduler: Some(scheduler_slot.clone()),
-            default_user_id: user_id.clone(),
+            owner_id: user_id.clone(),
+            default_sender_id: user_id.clone(),
             shutdown_tx: tokio::sync::RwLock::new(None),
             ws_tracker: Some(Arc::new(WsConnectionTracker::new())),
             llm_provider: Some(Arc::clone(&components.llm)),

tests/ws_gateway_integration.rs

@@ -5,7 +5,7 @@
 //! - WebSocket upgrade with auth
 //! - Ping/pong
 //! - Client message → agent msg_tx
-//! - Broadcast SSE event → WebSocket client
+//! - Broadcast AppEvent → WebSocket client
 //! - Connection tracking (counter increment/decrement)
 //! - Gateway status endpoint
 
@@ -22,8 +22,8 @@ use tokio_tungstenite::tungstenite::client::IntoClientRequest;
 use ironclaw::channels::IncomingMessage;
 use ironclaw::channels::web::server::{GatewayState, start_server};
 use ironclaw::channels::web::sse::SseManager;
-use ironclaw::channels::web::types::SseEvent;
 use ironclaw::channels::web::ws::WsConnectionTracker;
+use ironclaw_common::AppEvent;
 
 const AUTH_TOKEN: &str = "test-token-12345";
 const TIMEOUT: Duration = Duration::from_secs(5);
@@ -51,7 +51,8 @@ async fn start_test_server() -> (
         job_manager: None,
         prompt_queue: None,
         scheduler: None,
-        default_user_id: "test-user".to_string(),
+        owner_id: "test-user".to_string(),
+        default_sender_id: "test-user".to_string(),
         shutdown_tx: tokio::sync::RwLock::new(None),
         ws_tracker: Some(Arc::new(WsConnectionTracker::new())),
         llm_provider: None,
@@ -164,8 +165,8 @@ async fn test_ws_broadcast_event_received() {
     // Give the connection a moment to fully establish
     tokio::time::sleep(Duration::from_millis(50)).await;
 
-    // Broadcast an SSE event (simulates agent sending a response)
-    state.sse.broadcast(SseEvent::Response {
+    // Broadcast an event (simulates agent sending a response)
+    state.sse.broadcast(AppEvent::Response {
         content: "agent says hi".to_string(),
         thread_id: "t1".to_string(),
     });
@@ -186,7 +187,7 @@ async fn test_ws_thinking_event() {
     let mut ws = connect_ws(addr).await;
     tokio::time::sleep(Duration::from_millis(50)).await;
 
-    state.sse.broadcast(SseEvent::Thinking {
+    state.sse.broadcast(AppEvent::Thinking {
         message: "analyzing...".to_string(),
         thread_id: None,
     });
@@ -311,22 +312,22 @@ async fn test_ws_multiple_events_in_sequence() {
     tokio::time::sleep(Duration::from_millis(50)).await;
 
     // Broadcast multiple events rapidly
-    state.sse.broadcast(SseEvent::Thinking {
+    state.sse.broadcast(AppEvent::Thinking {
         message: "step 1".to_string(),
         thread_id: None,
     });
-    state.sse.broadcast(SseEvent::ToolStarted {
+    state.sse.broadcast(AppEvent::ToolStarted {
         name: "shell".to_string(),
         thread_id: None,
     });
-    state.sse.broadcast(SseEvent::ToolCompleted {
+    state.sse.broadcast(AppEvent::ToolCompleted {
         name: "shell".to_string(),
         success: true,
         error: None,
         parameters: None,
         thread_id: None,
     });
-    state.sse.broadcast(SseEvent::Response {
+    state.sse.broadcast(AppEvent::Response {
         content: "done".to_string(),
         thread_id: "t1".to_string(),
     });