Merge branch 'staging' into feat/nearai-mcp

# Conflicts:
#	src/app.rs

tests/e2e/CLAUDE.md

@@ -52,7 +52,7 @@ HEADED=1 pytest scenarios/
 | `test_html_injection.py` | XSS vectors injected directly via `page.evaluate("addMessage('assistant', ...)")` are sanitized by `renderMarkdown`; user messages are shown as escaped plain text |
 | `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; all triggered via `page.evaluate("showApproval(...)")` — no real tool call needed |
+| `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 |
 
 ## `helpers.py`
 
@@ -164,7 +164,7 @@ async def test_my_ui_feature(page):
 - **`asyncio_default_fixture_loop_scope = "session"`** — all async fixtures share one event loop. Do not use `asyncio.run()` inside fixtures; use `await` directly.
 - **The `page` fixture navigates with `/?token=e2e-test-token` and waits for `#auth-screen` to be hidden.** Tests receive a page that is already past the auth screen and has SSE connected.
 - **`test_skills.py` makes real network calls to ClawHub.** Tests skip (not fail) if the registry is unreachable via `pytest.skip()`.
-- **`test_html_injection.py` and `test_tool_approval.py` inject state via `page.evaluate(...)`.** They test the browser-side rendering pipeline and do not depend on the LLM or backend tool execution.
+- **`test_html_injection.py` injects state via `page.evaluate(...)`, and most of `test_tool_approval.py` does too.** The waiting-approval regression in `test_tool_approval.py` intentionally uses a real tool approval flow so it can verify backend thread-state handling.
 - **Browser is Chromium only.** `conftest.py` uses `p.chromium.launch()`; there is no Firefox or WebKit variant.
 - **Default timeout is 120 seconds** (pyproject.toml). Individual `wait_for` calls inside tests use shorter timeouts (5–20s) for faster failure messages.
 - **The libsql database is a temp directory** created fresh per `pytest` invocation; tests do not share state across runs.

tests/e2e/README.md

@@ -164,5 +164,7 @@ await page.evaluate("""
 """)
 ```
 
-This is the pattern used in `test_tool_approval.py` and parts of
-`test_extensions.py` (auth card, configure modal).
+This is the pattern used in most of `test_tool_approval.py` and parts of
+`test_extensions.py` (auth card, configure modal). The waiting-approval
+regression in `test_tool_approval.py` uses a real tool call instead so it can
+exercise backend approval state.

tests/e2e/conftest.py

@@ -15,7 +15,13 @@ from pathlib import Path
 
 import pytest
 
-from helpers import AUTH_TOKEN, wait_for_port_line, wait_for_ready
+from helpers import (
+    AUTH_TOKEN,
+    HTTP_WEBHOOK_SECRET,
+    OWNER_SCOPE_ID,
+    wait_for_port_line,
+    wait_for_ready,
+)
 
 # Project root (two levels up from tests/e2e/)
 ROOT = Path(__file__).resolve().parent.parent.parent
@@ -39,6 +45,9 @@ except Exception:
 # Temp directory for the libSQL database file (cleaned up automatically)
 _DB_TMPDIR = tempfile.TemporaryDirectory(prefix="ironclaw-e2e-")
 
+# Temp HOME so pairing/allowFrom state never touches the developer's real ~/.ironclaw
+_HOME_TMPDIR = tempfile.TemporaryDirectory(prefix="ironclaw-e2e-home-")
+
 # Temp directories for WASM extensions. These start empty and are populated by
 # the install pipeline during tests; fixtures do not pre-populate dev build
 # artifacts into them.
@@ -46,6 +55,42 @@ _WASM_TOOLS_TMPDIR = tempfile.TemporaryDirectory(prefix="ironclaw-e2e-wasm-tools
 _WASM_CHANNELS_TMPDIR = tempfile.TemporaryDirectory(prefix="ironclaw-e2e-wasm-channels-")
 
 
+def _latest_mtime(path: Path) -> float:
+    """Return the newest mtime under a file or directory."""
+    if not path.exists():
+        return 0.0
+    if path.is_file():
+        return path.stat().st_mtime
+
+    latest = path.stat().st_mtime
+    for root, dirnames, filenames in os.walk(path):
+        dirnames[:] = [dirname for dirname in dirnames if dirname != "target"]
+        for name in filenames:
+            child = Path(root) / name
+            try:
+                latest = max(latest, child.stat().st_mtime)
+            except FileNotFoundError:
+                continue
+    return latest
+
+
+def _binary_needs_rebuild(binary: Path) -> bool:
+    """Rebuild when the binary is missing or older than embedded sources."""
+    if not binary.exists():
+        return True
+
+    binary_mtime = binary.stat().st_mtime
+    inputs = [
+        ROOT / "Cargo.toml",
+        ROOT / "Cargo.lock",
+        ROOT / "build.rs",
+        ROOT / "providers.json",
+        ROOT / "src",
+        ROOT / "channels-src",
+    ]
+    return any(_latest_mtime(path) > binary_mtime for path in inputs)
+
+
 def _find_free_port() -> int:
     """Bind to port 0 and return the OS-assigned port."""
     with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
@@ -53,11 +98,26 @@ def _find_free_port() -> int:
         return s.getsockname()[1]
 
 
+def _reserve_loopback_sockets(count: int) -> list[socket.socket]:
+    """Bind loopback sockets and keep them open until the server starts."""
+    sockets: list[socket.socket] = []
+    try:
+        while len(sockets) < count:
+            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+            sock.bind(("127.0.0.1", 0))
+            sockets.append(sock)
+        return sockets
+    except Exception:
+        for sock in sockets:
+            sock.close()
+        raise
+
+
 @pytest.fixture(scope="session")
 def ironclaw_binary():
     """Ensure ironclaw binary is built. Returns the binary path."""
     binary = ROOT / "target" / "debug" / "ironclaw"
-    if not binary.exists():
+    if _binary_needs_rebuild(binary):
         print("Building ironclaw (this may take a while)...")
         subprocess.run(
             ["cargo", "build", "--no-default-features", "--features", "libsql"],
@@ -69,6 +129,21 @@ def ironclaw_binary():
     return str(binary)
 
 
+@pytest.fixture(scope="session")
+def server_ports():
+    """Reserve dynamic ports for the gateway and HTTP webhook channel."""
+    reserved = _reserve_loopback_sockets(2)
+    try:
+        yield {
+            "gateway": reserved[0].getsockname()[1],
+            "http": reserved[1].getsockname()[1],
+            "sockets": reserved,
+        }
+    finally:
+        for sock in reserved:
+            sock.close()
+
+
 @pytest.fixture(scope="session")
 async def mock_llm_server():
     """Start the mock LLM server. Yields the base URL."""
@@ -138,20 +213,35 @@ def _wasm_build_symlinks():
 
 
 @pytest.fixture(scope="session")
-async def ironclaw_server(ironclaw_binary, mock_llm_server, wasm_tools_dir):
+async def ironclaw_server(
+    ironclaw_binary,
+    mock_llm_server,
+    wasm_tools_dir,
+    server_ports,
+):
     """Start the ironclaw gateway. Yields the base URL."""
-    gateway_port = _find_free_port()
+    home_dir = _HOME_TMPDIR.name
+    gateway_port = server_ports["gateway"]
+    http_port = server_ports["http"]
+    for sock in server_ports["sockets"]:
+        if sock.fileno() != -1:
+            sock.close()
     env = {
         # Minimal env: PATH for process spawning, HOME for Rust/cargo defaults
         "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
-        "HOME": os.environ.get("HOME", "/tmp"),
+        "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": "e2e-tester",
+        "GATEWAY_USER_ID": "e2e-web-sender",
+        "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,
@@ -221,15 +311,22 @@ async def ironclaw_server(ironclaw_binary, mock_llm_server, wasm_tools_dir):
 
 
 @pytest.fixture(scope="session")
-async def ironclaw_server_with_webhook_secret(ironclaw_binary, mock_llm_server, wasm_tools_dir):
-    """Start ironclaw with HTTP_WEBHOOK_SECRET configured for webhook tests.
+async def http_channel_server(ironclaw_server, server_ports):
+    """HTTP webhook channel base URL."""
+    base_url = f"http://127.0.0.1:{server_ports['http']}"
+    await wait_for_ready(f"{base_url}/health", timeout=30)
+    return base_url
 
-    Yields a dict with:
-    - 'url': base URL of the gateway
-    - 'secret': the webhook secret value
-    """
+
+@pytest.fixture(scope="session")
+async def http_channel_server_without_secret(
+    ironclaw_binary,
+    mock_llm_server,
+    wasm_tools_dir,
+):
+    """Start the HTTP webhook channel without a configured secret."""
     gateway_port = _find_free_port()
-    webhook_secret = "test-webhook-secret-e2e-12345"
+    http_port = _find_free_port()
     env = {
         # Minimal env: PATH for process spawning, HOME for Rust/cargo defaults
         "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
@@ -241,13 +338,14 @@ async def ironclaw_server_with_webhook_secret(ironclaw_binary, mock_llm_server,
         "GATEWAY_PORT": str(gateway_port),
         "GATEWAY_AUTH_TOKEN": AUTH_TOKEN,
         "GATEWAY_USER_ID": "e2e-tester",
-        "HTTP_WEBHOOK_SECRET": webhook_secret,
+        "HTTP_HOST": "127.0.0.1",
+        "HTTP_PORT": str(http_port),
         "CLI_ENABLED": "false",
         "LLM_BACKEND": "openai_compatible",
         "LLM_BASE_URL": mock_llm_server,
         "LLM_MODEL": "mock-model",
         "DATABASE_BACKEND": "libsql",
-        "LIBSQL_PATH": os.path.join(_DB_TMPDIR.name, "e2e-webhook.db"),
+        "LIBSQL_PATH": os.path.join(_DB_TMPDIR.name, "e2e-webhook-no-secret.db"),
         "SANDBOX_ENABLED": "false",
         "SKILLS_ENABLED": "true",
         "ROUTINES_ENABLED": "false",
@@ -277,13 +375,12 @@ async def ironclaw_server_with_webhook_secret(ironclaw_binary, mock_llm_server,
         stderr=asyncio.subprocess.PIPE,
         env=env,
     )
-    base_url = f"http://127.0.0.1:{gateway_port}"
+    gateway_url = f"http://127.0.0.1:{gateway_port}"
+    http_base_url = f"http://127.0.0.1:{http_port}"
     try:
-        await wait_for_ready(f"{base_url}/api/health", timeout=60)
-        yield {
-            "url": base_url,
-            "secret": webhook_secret,
-        }
+        await wait_for_ready(f"{gateway_url}/api/health", timeout=60)
+        await wait_for_ready(f"{http_base_url}/health", timeout=30)
+        yield http_base_url
     except TimeoutError:
         # Dump stderr so CI logs show why the server failed to start
         returncode = proc.returncode
@@ -296,7 +393,8 @@ async def ironclaw_server_with_webhook_secret(ironclaw_binary, mock_llm_server,
         stderr_text = stderr_bytes.decode("utf-8", errors="replace")
         proc.kill()
         pytest.fail(
-            f"ironclaw server with webhook secret failed to start on port {gateway_port} "
+            f"ironclaw server without webhook secret failed to start on ports "
+            f"gateway={gateway_port}, http={http_port} "
             f"(returncode={returncode}).\nstderr:\n{stderr_text}"
         )
     finally:

tests/e2e/helpers.py

@@ -1,6 +1,8 @@
 """Shared helpers for E2E tests."""
 
 import asyncio
+import hashlib
+import hmac
 import re
 import time
 
@@ -95,12 +97,21 @@ SEL = {
     "toast_success":            ".toast.toast-success",
     "toast_error":              ".toast.toast-error",
     "toast_info":               ".toast.toast-info",
+    # Jobs / routines
+    "jobs_tbody":               "#jobs-tbody",
+    "job_row":                  "#jobs-tbody .job-row",
+    "jobs_empty":               "#jobs-empty",
+    "routines_tbody":           "#routines-tbody",
+    "routine_row":              "#routines-tbody .routine-row",
+    "routines_empty":           "#routines-empty",
 }
 
 TABS = ["chat", "memory", "jobs", "routines", "extensions", "skills"]
 
 # Auth token used across all tests
 AUTH_TOKEN = "e2e-test-token"
+OWNER_SCOPE_ID = "e2e-owner-scope"
+HTTP_WEBHOOK_SECRET = "e2e-http-webhook-secret"
 
 
 async def wait_for_ready(url: str, *, timeout: float = 60, interval: float = 0.5):
@@ -162,3 +173,16 @@ async def api_post(base_url: str, path: str, **kwargs) -> httpx.Response:
             timeout=kwargs.pop("timeout", 10),
             **kwargs,
         )
+
+
+def signed_http_webhook_headers(body: bytes) -> dict[str, str]:
+    """Return headers for the owner-scoped HTTP webhook channel."""
+    digest = hmac.new(
+        HTTP_WEBHOOK_SECRET.encode("utf-8"),
+        body,
+        hashlib.sha256,
+    ).hexdigest()
+    return {
+        "Content-Type": "application/json",
+        "X-Hub-Signature-256": f"sha256={digest}",
+    }

tests/e2e/ironclaw_e2e.egg-info/SOURCES.txt

@@ -12,11 +12,17 @@ scenarios/test_csp.py
 scenarios/test_extension_oauth.py
 scenarios/test_extensions.py
 scenarios/test_html_injection.py
+scenarios/test_mcp_auth_flow.py
 scenarios/test_oauth_credential_fallback.py
+scenarios/test_owner_scope.py
 scenarios/test_pairing.py
+scenarios/test_routine_event_batch.py
 scenarios/test_routine_oauth_credential_injection.py
 scenarios/test_skills.py
 scenarios/test_sse_reconnect.py
+scenarios/test_telegram_hot_activation.py
+scenarios/test_telegram_token_validation.py
 scenarios/test_tool_approval.py
 scenarios/test_tool_execution.py
-scenarios/test_wasm_lifecycle.py
+scenarios/test_wasm_lifecycle.py
+scenarios/test_webhook.py

tests/e2e/mock_llm.py

@@ -25,7 +25,69 @@ DEFAULT_RESPONSE = "I understand your request."
 
 TOOL_CALL_PATTERNS = [
     (re.compile(r"echo (.+)", re.IGNORECASE), "echo", lambda m: {"message": m.group(1)}),
+    (
+        re.compile(r"make approval post (?P<label>[a-z0-9_-]+)", re.IGNORECASE),
+        "http",
+        lambda m: {
+            "method": "POST",
+            "url": f"https://example.com/{m.group('label')}",
+            "body": {"label": m.group("label")},
+        },
+    ),
     (re.compile(r"what time|current time", re.IGNORECASE), "time", lambda _: {"operation": "now"}),
+    (
+        re.compile(
+            r"create lightweight owner routine (?P<name>[a-z0-9][a-z0-9_-]*)",
+            re.IGNORECASE,
+        ),
+        "routine_create",
+        lambda m: {
+            "name": m.group("name"),
+            "description": f"Owner-scope routine {m.group('name')}",
+            "trigger_type": "manual",
+            "prompt": f"Confirm that {m.group('name')} executed.",
+            "action_type": "lightweight",
+            "use_tools": False,
+        },
+    ),
+    (
+        re.compile(
+            r"create full[- ]job owner routine (?P<name>[a-z0-9][a-z0-9_-]*)",
+            re.IGNORECASE,
+        ),
+        "routine_create",
+        lambda m: {
+            "name": m.group("name"),
+            "description": f"Owner-scope full-job routine {m.group('name')}",
+            "trigger_type": "manual",
+            "prompt": f"Complete the routine job for {m.group('name')}.",
+            "action_type": "full_job",
+        },
+    ),
+    (
+        re.compile(
+            r"create event routine (?P<name>[a-z0-9][a-z0-9_-]*) "
+            r"channel (?P<channel>[a-z0-9_-]+) pattern (?P<pattern>[a-z0-9_|-]+)",
+            re.IGNORECASE,
+        ),
+        "routine_create",
+        lambda m: {
+            "name": m.group("name"),
+            "description": f"Event routine {m.group('name')}",
+            "trigger_type": "event",
+            "event_channel": None if m.group("channel").lower() == "any" else m.group("channel"),
+            "event_pattern": m.group("pattern"),
+            "prompt": f"Acknowledge that {m.group('name')} fired.",
+            "action_type": "lightweight",
+            "use_tools": False,
+            "cooldown_secs": 0,
+        },
+    ),
+    (
+        re.compile(r"list owner routines", re.IGNORECASE),
+        "routine_list",
+        lambda _: {},
+    ),
 ]
 
 

tests/e2e/scenarios/test_extensions.py

@@ -885,6 +885,36 @@ async def test_auth_and_configure_helpers_escape_selector_sensitive_extension_na
     assert result["configureStillPresent"] is False
 
 
+async def test_auth_required_does_not_reopen_existing_configure_modal(page):
+    """Regression: auth_required SSE should not clobber an already-open configure modal."""
+    result = await page.evaluate(
+        """() => {
+            const overlay = document.createElement('div');
+            overlay.className = 'configure-overlay';
+            overlay.setAttribute('data-extension-name', 'telegram');
+            document.body.appendChild(overlay);
+
+            const originalShowConfigureModal = window.showConfigureModal;
+            const originalSetAuthFlowPending = window.setAuthFlowPending;
+            let showCalls = 0;
+            let pendingCalls = 0;
+
+            window.showConfigureModal = () => { showCalls += 1; };
+            window.setAuthFlowPending = () => { pendingCalls += 1; };
+
+            handleAuthRequired({ extension_name: 'telegram', instructions: 'pending', auth_url: null });
+
+            window.showConfigureModal = originalShowConfigureModal;
+            window.setAuthFlowPending = originalSetAuthFlowPending;
+            overlay.remove();
+            return { showCalls, pendingCalls };
+        }"""
+    )
+
+    assert result["showCalls"] == 0
+    assert result["pendingCalls"] == 0
+
+
 async def test_auth_completed_sse_dismisses_card(page):
     """Simulating the auth_completed SSE event removes the auth card."""
     await _show_auth_card(page, extension_name="myext")

tests/e2e/scenarios/test_owner_scope.py

@@ -0,0 +1,226 @@
+"""Owner-scope end-to-end scenarios.
+
+These tests exercise the explicit owner model across:
+- the web gateway chat UI
+- the owner-scoped HTTP webhook channel
+- routine tools / routines tab
+- job creation via routine execution / jobs tab
+"""
+
+import asyncio
+import json
+import uuid
+
+import httpx
+
+from helpers import SEL, AUTH_TOKEN, signed_http_webhook_headers
+
+
+async def _send_and_get_response(
+    page,
+    message: str,
+    *,
+    expected_fragment: str,
+    timeout: int = 30000,
+) -> str:
+    """Send a chat message and return the newest assistant response text."""
+    chat_input = page.locator(SEL["chat_input"])
+    await chat_input.wait_for(state="visible", timeout=5000)
+
+    assistant_sel = SEL["message_assistant"]
+    before_count = await page.locator(assistant_sel).count()
+
+    await chat_input.fill(message)
+    await chat_input.press("Enter")
+
+    expected = before_count + 1
+    await page.wait_for_function(
+        """({ assistantSelector, expectedCount, expectedFragment }) => {
+            const messages = document.querySelectorAll(assistantSelector);
+            if (messages.length < expectedCount) return false;
+            const text = (messages[messages.length - 1].innerText || '').trim().toLowerCase();
+            return text.includes(expectedFragment.toLowerCase());
+        }""",
+        arg={
+            "assistantSelector": assistant_sel,
+            "expectedCount": expected,
+            "expectedFragment": expected_fragment,
+        },
+        timeout=timeout,
+    )
+
+    return await page.locator(assistant_sel).last.inner_text()
+
+
+async def _post_http_webhook(
+    http_channel_server: str,
+    *,
+    content: str,
+    sender_id: str,
+    thread_id: str,
+) -> str:
+    """Send a signed request to the owner-scoped HTTP webhook channel."""
+    payload = {
+        "user_id": sender_id,
+        "thread_id": thread_id,
+        "content": content,
+        "wait_for_response": True,
+    }
+    body = json.dumps(payload).encode("utf-8")
+
+    async with httpx.AsyncClient() as client:
+        response = await client.post(
+            f"{http_channel_server}/webhook",
+            content=body,
+            headers=signed_http_webhook_headers(body),
+            timeout=90,
+        )
+
+    assert response.status_code == 200, (
+        f"HTTP webhook failed: {response.status_code} {response.text[:400]}"
+    )
+    data = response.json()
+    assert data["status"] == "accepted", f"Unexpected webhook response: {data}"
+    assert data["response"], f"Expected synchronous response body, got: {data}"
+    return data["response"]
+
+
+async def _open_tab(page, tab: str) -> None:
+    btn = page.locator(SEL["tab_button"].format(tab=tab))
+    await btn.click()
+    await page.locator(SEL["tab_panel"].format(tab=tab)).wait_for(
+        state="visible",
+        timeout=5000,
+    )
+
+
+async def _wait_for_routine(base_url: str, name: str, timeout: float = 20.0) -> dict:
+    """Poll the routines API until the named routine exists."""
+    async with httpx.AsyncClient() as client:
+        for _ in range(int(timeout * 2)):
+            response = await client.get(
+                f"{base_url}/api/routines",
+                headers={"Authorization": f"Bearer {AUTH_TOKEN}"},
+                timeout=10,
+            )
+            response.raise_for_status()
+            routines = response.json()["routines"]
+            for routine in routines:
+                if routine["name"] == name:
+                    return routine
+            await _poll_sleep()
+    raise AssertionError(f"Routine '{name}' was not created within {timeout}s")
+
+
+async def _wait_for_job(base_url: str, title: str, timeout: float = 30.0) -> dict:
+    """Poll the jobs API until the named job exists."""
+    async with httpx.AsyncClient() as client:
+        for _ in range(int(timeout * 2)):
+            response = await client.get(
+                f"{base_url}/api/jobs",
+                headers={"Authorization": f"Bearer {AUTH_TOKEN}"},
+                timeout=10,
+            )
+            response.raise_for_status()
+            jobs = response.json()["jobs"]
+            for job in jobs:
+                if job["title"] == title:
+                    return job
+            await _poll_sleep()
+    raise AssertionError(f"Job '{title}' was not created within {timeout}s")
+
+
+async def _poll_sleep() -> None:
+    """Small shared backoff for API polling loops."""
+    await asyncio.sleep(0.5)
+
+
+async def test_http_channel_created_routine_is_visible_in_web_routines_tab(
+    page,
+    ironclaw_server,
+    http_channel_server,
+):
+    """A routine created from the HTTP channel is visible in the web owner UI."""
+    routine_name = f"owner-http-{uuid.uuid4().hex[:8]}"
+
+    response_text = await _post_http_webhook(
+        http_channel_server,
+        content=f"create lightweight owner routine {routine_name}",
+        sender_id="external-sender-alpha",
+        thread_id="http-owner-routine-thread",
+    )
+    assert routine_name in response_text
+
+    await _wait_for_routine(ironclaw_server, routine_name)
+
+    await _open_tab(page, "routines")
+    await page.locator(SEL["routine_row"]).filter(has_text=routine_name).first.wait_for(
+        state="visible",
+        timeout=15000,
+    )
+
+
+async def test_web_created_routine_is_listed_from_http_channel_across_senders(
+    page,
+    ironclaw_server,
+    http_channel_server,
+):
+    """Routines created in web chat remain owner-global across HTTP senders/threads."""
+    routine_name = f"owner-web-{uuid.uuid4().hex[:8]}"
+
+    assistant_text = await _send_and_get_response(
+        page,
+        f"create lightweight owner routine {routine_name}",
+        expected_fragment=routine_name,
+    )
+    assert routine_name in assistant_text
+
+    await _wait_for_routine(ironclaw_server, routine_name)
+
+    first_sender_text = await _post_http_webhook(
+        http_channel_server,
+        content="list owner routines",
+        sender_id="http-sender-one",
+        thread_id="owner-list-thread-a",
+    )
+    second_sender_text = await _post_http_webhook(
+        http_channel_server,
+        content="list owner routines",
+        sender_id="http-sender-two",
+        thread_id="owner-list-thread-b",
+    )
+
+    assert routine_name in first_sender_text, first_sender_text
+    assert routine_name in second_sender_text, second_sender_text
+
+
+async def test_http_created_full_job_routine_can_be_run_from_web_and_shows_in_jobs(
+    page,
+    ironclaw_server,
+    http_channel_server,
+):
+    """A full-job routine created via HTTP can be run from the web UI and create a job."""
+    routine_name = f"owner-job-{uuid.uuid4().hex[:8]}"
+
+    response_text = await _post_http_webhook(
+        http_channel_server,
+        content=f"create full-job owner routine {routine_name}",
+        sender_id="http-job-sender",
+        thread_id="owner-job-thread",
+    )
+    assert routine_name in response_text
+
+    await _wait_for_routine(ironclaw_server, routine_name)
+
+    await _open_tab(page, "routines")
+    routine_row = page.locator(SEL["routine_row"]).filter(has_text=routine_name).first
+    await routine_row.wait_for(state="visible", timeout=15000)
+    await routine_row.locator('button[data-action="trigger-routine"]').click()
+
+    await _wait_for_job(ironclaw_server, routine_name, timeout=45.0)
+
+    await _open_tab(page, "jobs")
+    await page.locator(SEL["job_row"]).filter(has_text=routine_name).first.wait_for(
+        state="visible",
+        timeout=20000,
+    )

tests/e2e/scenarios/test_routine_event_batch.py

@@ -1,534 +1,317 @@
-"""
-E2E tests for event-triggered routines with batch loading.
-
-These tests verify that the N+1 query fix correctly:
-1. Fires event-triggered routines on matching messages
-2. Enforces concurrent limits via batch-loaded counts
-3. Maintains performance with multiple simultaneous triggers
-4. Works correctly through the full UI and agent loop
-
-Playwright-based UI tests + SSE verification.
-"""
+"""E2E tests for event-triggered routines over the HTTP channel."""
 
 import asyncio
 import json
+import uuid
+
+import httpx
 import pytest
-from datetime import datetime, timedelta
-from typing import List, Dict, Any
 
-from playwright.async_api import async_playwright, Page, Browser, BrowserContext
+from helpers import AUTH_TOKEN, SEL, signed_http_webhook_headers
 
 
-@pytest.fixture
-async def browser_and_context():
-    """Create a Playwright browser and context for testing."""
-    async with async_playwright() as p:
-        browser = await p.chromium.launch(headless=True)
-        context = await browser.new_context()
-        yield browser, context
-        await context.close()
-        await browser.close()
+async def _send_chat_message(page, message: str) -> None:
+    """Send a chat message and wait for the assistant turn to appear."""
+    chat_input = page.locator(SEL["chat_input"])
+    await chat_input.wait_for(state="visible", timeout=5000)
+    assistant_messages = page.locator(SEL["message_assistant"])
+    before_count = await assistant_messages.count()
+
+    await chat_input.fill(message)
+    await chat_input.press("Enter")
+
+    await page.wait_for_function(
+        """({ selector, expectedCount }) => {
+            return document.querySelectorAll(selector).length >= expectedCount;
+        }""",
+        arg={
+            "selector": SEL["message_assistant"],
+            "expectedCount": before_count + 1,
+        },
+        timeout=30000,
+    )
 
 
-class EventTriggerHelper:
-    """Helper methods for event trigger testing."""
+async def _create_event_routine(
+    page,
+    base_url: str,
+    *,
+    name: str,
+    pattern: str,
+    channel: str = "http",
+) -> dict:
+    """Create an event routine through chat and return its API record."""
+    await _send_chat_message(
+        page,
+        f"create event routine {name} channel {channel} pattern {pattern}",
+    )
+    return await _wait_for_routine(base_url, name)
 
-    def __init__(self, page: Page):
-        self.page = page
 
-    async def navigate_to_routines(self):
-        """Navigate to the routines page."""
-        await self.page.goto("http://localhost:8000/routines")
-        await self.page.wait_for_load_state("networkidle")
+async def _post_http_message(
+    http_channel_server: str,
+    *,
+    content: str,
+    sender_id: str | None = None,
+    thread_id: str | None = None,
+) -> dict:
+    """Send a signed HTTP-channel message and return the JSON body."""
+    payload = {
+        "user_id": sender_id or f"sender-{uuid.uuid4().hex[:8]}",
+        "thread_id": thread_id or f"thread-{uuid.uuid4().hex[:8]}",
+        "content": content,
+        "wait_for_response": True,
+    }
+    body = json.dumps(payload).encode("utf-8")
 
-    async def create_event_routine(
-        self,
-        name: str,
-        trigger_regex: str,
-        channel: str = "slack",
-        max_concurrent: int = 1,
-    ) -> str:
-        """
-        Create an event-triggered routine via UI.
-        Returns the routine ID.
-        """
-        await self.navigate_to_routines()
-
-        # Click "New Routine" button
-        await self.page.click('button:has-text("New Routine")')
-        await self.page.wait_for_selector('input[name="routine_name"]')
-
-        # Fill routine details
-        await self.page.fill('input[name="routine_name"]', name)
-        await self.page.fill(
-            'textarea[name="routine_description"]',
-            f"Test routine: {name}",
+    async with httpx.AsyncClient() as client:
+        response = await client.post(
+            f"{http_channel_server}/webhook",
+            content=body,
+            headers=signed_http_webhook_headers(body),
+            timeout=90,
         )
 
-        # Select "Event Trigger" type
-        await self.page.click('label:has-text("Event Trigger")')
-        await self.page.wait_for_selector('input[name="trigger_regex"]')
-
-        # Fill trigger details
-        await self.page.fill('input[name="trigger_regex"]', trigger_regex)
-        await self.page.select_option('select[name="trigger_channel"]', channel)
-
-        # Set guardrails
-        await self.page.fill('input[name="max_concurrent"]', str(max_concurrent))
-
-        # Select lightweight action
-        await self.page.click('label:has-text("Lightweight")')
-        await self.page.fill(
-            'textarea[name="lightweight_prompt"]',
-            "Acknowledge the message and confirm trigger worked.",
-        )
-
-        # Save routine
-        await self.page.click('button:has-text("Save Routine")')
-        await self.page.wait_for_selector('text=Routine created successfully')
-
-        # Extract routine ID from success message or URL
-        routine_id = await self.page.locator('data-testid=routine-id').text_content()
-        return routine_id.strip() if routine_id else None
-
-    async def create_multiple_routines(
-        self, base_name: str, count: int, trigger_regex: str = None
-    ) -> List[str]:
-        """Create multiple event-triggered routines."""
-        routine_ids = []
-        for i in range(count):
-            name = f"{base_name}_{i}"
-            regex = trigger_regex or f"({i}|{base_name})"
-            routine_id = await self.create_event_routine(name, regex)
-            routine_ids.append(routine_id)
-            await asyncio.sleep(0.1)  # Small delay between creations
-        return routine_ids
-
-    async def send_chat_message(self, message: str) -> List[str]:
-        """
-        Send a chat message and return SSE events received.
-        Captures all routine firing events.
-        """
-        await self.page.goto("http://localhost:8000/chat")
-        await self.page.wait_for_selector('input[placeholder*="message"]', timeout=5000)
-
-        # Collect SSE events
-        sse_events = []
-
-        async def capture_sse(response):
-            """Intercept SSE events."""
-            if "event-stream" in response.headers.get("content-type", ""):
-                text = await response.text()
-                for line in text.split("\n"):
-                    if line.startswith("data:"):
-                        try:
-                            event = json.loads(line[5:])
-                            sse_events.append(event)
-                        except json.JSONDecodeError:
-                            pass
-
-        self.page.on("response", capture_sse)
-
-        # Send message
-        await self.page.fill('input[placeholder*="message"]', message)
-        await self.page.press('input[placeholder*="message"]', "Enter")
-
-        # Wait for response
-        await self.page.wait_for_selector('text=Message processed', timeout=10000)
-        await asyncio.sleep(0.5)  # Allow time for SSE events
-
-        self.page.remove_listener("response", capture_sse)
-        return sse_events
-
-    async def get_routine_execution_log(self, routine_id: str) -> List[Dict]:
-        """Get execution log entries for a routine."""
-        await self.page.goto(f"http://localhost:8000/routines/{routine_id}/executions")
-        await self.page.wait_for_load_state("networkidle")
-
-        # Extract log entries from table
-        rows = await self.page.locator("tbody tr").all()
-        executions = []
-
-        for row in rows:
-            cells = await row.locator("td").all()
-            if len(cells) >= 3:
-                execution = {
-                    "timestamp": await cells[0].text_content(),
-                    "status": await cells[1].text_content(),
-                    "details": await cells[2].text_content(),
-                }
-                executions.append(execution)
-
-        return executions
-
-    async def check_database_queries_in_logs(
-        self, max_queries_expected: int = 1
-    ) -> int:
-        """Check debug logs for database query count."""
-        await self.page.goto("http://localhost:8000/debug/logs?filter=database")
-        await self.page.wait_for_load_state("networkidle")
-
-        # Count batch queries
-        log_lines = await self.page.locator("tr:has-text('batch')").all()
-        batch_count = len(log_lines)
-
-        # Count individual COUNT queries (should be 0 after fix)
-        count_queries = await self.page.locator("tr:has-text('COUNT')").all()
-        count_query_count = len(count_queries)
-
-        return batch_count, count_query_count
-
-
-# =============================================================================
-# Tests
-# =============================================================================
-
-
-@pytest.mark.asyncio
-async def test_create_event_trigger_routine(browser_and_context):
-    """Test creating an event-triggered routine via UI."""
-    browser, context = browser_and_context
-    page = await context.new_page()
-    helper = EventTriggerHelper(page)
-
-    try:
-        routine_id = await helper.create_event_routine(
-            name="Test Trigger",
-            trigger_regex="test|demo",
-            channel="slack",
-            max_concurrent=1,
-        )
-
-        assert routine_id is not None, "Routine ID should be returned"
-        assert len(routine_id) > 0, "Routine ID should not be empty"
-
-    finally:
-        await page.close()
-
-
-@pytest.mark.asyncio
-async def test_event_trigger_fires_on_matching_message(browser_and_context):
-    """Test that event-triggered routine fires when message matches."""
-    browser, context = browser_and_context
-    page = await context.new_page()
-    helper = EventTriggerHelper(page)
-
-    try:
-        # Create routine
-        routine_id = await helper.create_event_routine(
-            name="Alert Handler",
-            trigger_regex="urgent|critical|alert",
-            channel="slack",
-        )
-
-        # Send matching message
-        sse_events = await helper.send_chat_message("URGENT: Server down!")
-
-        # Verify routine fired (look for event in SSE stream)
-        routine_fired = any(
-            event.get("type") == "routine_fired" and event.get("routine_id") == routine_id
-            for event in sse_events
-        )
-        assert routine_fired, "Routine should fire on matching message"
-
-        # Check execution log
-        executions = await helper.get_routine_execution_log(routine_id)
-        assert len(executions) > 0, "Execution should be logged"
-        assert "success" in executions[0]["status"].lower()
-
-    finally:
-        await page.close()
-
-
-@pytest.mark.asyncio
-async def test_event_trigger_skips_non_matching_message(browser_and_context):
-    """Test that event-triggered routine skips when message doesn't match."""
-    browser, context = browser_and_context
-    page = await context.new_page()
-    helper = EventTriggerHelper(page)
-
-    try:
-        # Create routine
-        routine_id = await helper.create_event_routine(
-            name="Alert Handler",
-            trigger_regex="urgent|critical|alert",
-            channel="slack",
-        )
-
-        # Send non-matching message
-        sse_events = await helper.send_chat_message("Hello, how are you?")
-
-        # Verify routine did NOT fire
-        routine_fired = any(
-            event.get("type") == "routine_fired" and event.get("routine_id") == routine_id
-            for event in sse_events
-        )
-        assert not routine_fired, "Routine should not fire on non-matching message"
-
-    finally:
-        await page.close()
-
-
-@pytest.mark.asyncio
-async def test_multiple_routines_fire_on_matching_message(browser_and_context):
-    """Test that multiple event-triggered routines fire on same message."""
-    browser, context = browser_and_context
-    page = await context.new_page()
-    helper = EventTriggerHelper(page)
-
-    try:
-        # Create 3 overlapping routines
-        routine_ids = await helper.create_multiple_routines(
-            base_name="Handler", count=3, trigger_regex="alert|warning|error"
-        )
-
-        # Send matching message
-        sse_events = await helper.send_chat_message("ERROR: Database connection failed")
-
-        # Verify all 3 routines fired
-        fired_count = sum(
-            1
-            for event in sse_events
-            if event.get("type") == "routine_fired" and event.get("routine_id") in routine_ids
-        )
-
-        assert (
-            fired_count >= 3
-        ), f"Expected all 3 routines to fire, got {fired_count}"
-
-    finally:
-        await page.close()
-
-
-@pytest.mark.asyncio
-async def test_concurrent_limit_prevents_additional_fires(browser_and_context):
-    """Test that concurrent limit is enforced via batch counts."""
-    browser, context = browser_and_context
-    page = await context.new_page()
-    helper = EventTriggerHelper(page)
-
-    try:
-        # Create routine with max_concurrent=1
-        routine_id = await helper.create_event_routine(
-            name="Limited Handler",
-            trigger_regex="process|task",
-            max_concurrent=1,
-        )
-
-        # Trigger first message
-        await helper.send_chat_message("Process message 1")
-        await asyncio.sleep(1)
-
-        # Check first execution logged
-        executions_1 = await helper.get_routine_execution_log(routine_id)
-        assert len(executions_1) >= 1
-
-        # Trigger second message while first is still running
-        sse_events = await helper.send_chat_message("Process message 2")
-
-        # Second routine should be skipped (concurrent limit)
-        routine_skipped = any(
-            event.get("type") == "routine_skipped"
-            and event.get("reason") == "max_concurrent_reached"
-            and event.get("routine_id") == routine_id
-            for event in sse_events
-        )
-        assert routine_skipped, "Routine should be skipped when concurrent limit reached"
-
-    finally:
-        await page.close()
-
-
-@pytest.mark.asyncio
-async def test_rapid_messages_with_multiple_triggers_efficiency(browser_and_context):
-    """Test efficiency of batch loading with multiple rapid messages."""
-    browser, context = browser_and_context
-    page = await context.new_page()
-    helper = EventTriggerHelper(page)
-
-    try:
-        # Create 5 overlapping routines
-        routine_ids = await helper.create_multiple_routines(
-            base_name="Rapid", count=5, trigger_regex="test|demo|check"
-        )
-
-        # Send 10 matching messages rapidly
-        for i in range(10):
-            message = f"test message {i}"
-            await helper.send_chat_message(message)
-            await asyncio.sleep(0.1)
-
-        # Check database logs for query efficiency
-        batch_count, count_query_count = await helper.check_database_queries_in_logs()
-
-        # After fix: should have ~10 batch queries (1 per message)
-        # Before fix: would have ~50 individual COUNT queries (5 routines × 10 messages)
-        assert (
-            count_query_count == 0
-        ), f"Should have 0 individual COUNT queries after fix, got {count_query_count}"
-        assert (
-            batch_count <= 15
-        ), f"Should have <=15 batch queries for 10 messages, got {batch_count}"
-
-    finally:
-        await page.close()
-
-
-@pytest.mark.asyncio
-async def test_channel_filter_applied_correctly(browser_and_context):
-    """Test that channel filter prevents non-matching messages."""
-    browser, context = browser_and_context
-    page = await context.new_page()
-    helper = EventTriggerHelper(page)
-
-    try:
-        # Create routine for Slack channel
-        slack_routine_id = await helper.create_event_routine(
-            name="Slack Handler",
-            trigger_regex="alert",
-            channel="slack",
-        )
-
-        # Simulate message from Telegram channel
-        # (Note: In real UI, would need to change channel context)
-        page.goto(
-            "http://localhost:8000/chat?channel=telegram"
-        )  # Switch channel
-        await helper.send_chat_message("alert: something urgent")
-
-        # Routine should not fire (different channel)
-        executions = await helper.get_routine_execution_log(slack_routine_id)
-
-        # Check if any recent execution (last 5 min) exists
-        recent = [
-            e
-            for e in executions
-            if (datetime.now() - datetime.fromisoformat(e["timestamp"])).total_seconds()
-            < 300
-        ]
-        assert (
-            len(recent) == 0
-        ), "Routine should not fire for different channel"
-
-    finally:
-        await page.close()
-
-
-@pytest.mark.asyncio
-async def test_batch_query_failure_handling(browser_and_context):
-    """Test graceful handling of batch query failures."""
-    browser, context = browser_and_context
-    page = await context.new_page()
-    helper = EventTriggerHelper(page)
-
-    try:
-        # Create routine
-        routine_id = await helper.create_event_routine(
-            name="Error Handler",
-            trigger_regex="test",
-        )
-
-        # Simulate database error in logs (if possible with test hooks)
-        # For now, just verify error handling doesn't crash UI
-        await helper.send_chat_message("test message")
-
-        # Check that UI remains responsive
-        assert await page.locator("text=Message processed").is_visible()
-
-    finally:
-        await page.close()
-
-
-@pytest.mark.asyncio
-async def test_routine_execution_history_display(browser_and_context):
-    """Test that execution history correctly displays routine firings."""
-    browser, context = browser_and_context
-    page = await context.new_page()
-    helper = EventTriggerHelper(page)
-
-    try:
-        # Create routine
-        routine_id = await helper.create_event_routine(
-            name="History Test",
-            trigger_regex="test",
-        )
-
-        # Trigger routine 3 times
-        for i in range(3):
-            await helper.send_chat_message(f"test message {i}")
-            await asyncio.sleep(0.2)
-
-        # Check execution log
-        executions = await helper.get_routine_execution_log(routine_id)
-        assert len(executions) >= 3, "Should have at least 3 executions logged"
-
-        # Verify all are recent (within last 5 minutes)
-        for execution in executions[:3]:
-            timestamp = datetime.fromisoformat(execution["timestamp"])
-            age = datetime.now() - timestamp
-            assert age < timedelta(minutes=5), "Execution should be recent"
-
-    finally:
-        await page.close()
-
-
-@pytest.mark.asyncio
-async def test_concurrent_batch_loads_independent(browser_and_context):
-    """Test that concurrent messages each get independent batch queries."""
-    browser, context = browser_and_context
-    page = await context.new_page()
-    helper = EventTriggerHelper(page)
-
-    try:
-        # Create 5 routines matching different patterns
-        r1_id = await helper.create_event_routine(
-            name="Pattern A", trigger_regex="alpha|alpha_only"
-        )
-        r2_id = await helper.create_event_routine(
-            name="Pattern B", trigger_regex="beta|beta_only"
-        )
-        r3_id = await helper.create_event_routine(
-            name="Pattern AB", trigger_regex="alpha|beta|common"
-        )
-
-        # Send overlapping messages
-        # Message 1: matches r1, r3
-        sse1 = await helper.send_chat_message("alpha common")
-        await asyncio.sleep(0.1)
-
-        # Message 2: matches r2, r3
-        sse2 = await helper.send_chat_message("beta common")
-        await asyncio.sleep(0.1)
-
-        # Verify correct routines fired
-        r1_fired_msg1 = any(
-            e.get("routine_id") == r1_id for e in sse1 if e.get("type") == "routine_fired"
-        )
-        r2_fired_msg2 = any(
-            e.get("routine_id") == r2_id for e in sse2 if e.get("type") == "routine_fired"
-        )
-        r3_fired_both = (
-            any(
-                e.get("routine_id") == r3_id for e in sse1 if e.get("type") == "routine_fired"
+    assert response.status_code == 200, (
+        f"HTTP webhook failed: {response.status_code} {response.text[:400]}"
+    )
+    return response.json()
+
+
+async def _wait_for_routine(base_url: str, name: str, timeout: float = 20.0) -> dict:
+    """Poll the routines API until the named routine exists."""
+    async with httpx.AsyncClient() as client:
+        for _ in range(int(timeout * 2)):
+            response = await client.get(
+                f"{base_url}/api/routines",
+                headers={"Authorization": f"Bearer {AUTH_TOKEN}"},
+                timeout=10,
             )
-            and any(
-                e.get("routine_id") == r3_id for e in sse2 if e.get("type") == "routine_fired"
+            response.raise_for_status()
+            for routine in response.json()["routines"]:
+                if routine["name"] == name:
+                    return routine
+            await asyncio.sleep(0.5)
+    raise AssertionError(f"Routine '{name}' was not created within {timeout}s")
+
+
+async def _get_routine_runs(base_url: str, routine_id: str) -> list[dict]:
+    """Fetch recent routine runs from the web API."""
+    async with httpx.AsyncClient() as client:
+        response = await client.get(
+            f"{base_url}/api/routines/{routine_id}/runs",
+            headers={"Authorization": f"Bearer {AUTH_TOKEN}"},
+            timeout=10,
+        )
+    response.raise_for_status()
+    return response.json()["runs"]
+
+
+async def _wait_for_run_count(
+    base_url: str,
+    routine_id: str,
+    *,
+    expected_at_least: int,
+    timeout: float = 20.0,
+) -> list[dict]:
+    """Poll until the routine has at least the expected run count."""
+    for _ in range(int(timeout * 2)):
+        runs = await _get_routine_runs(base_url, routine_id)
+        if len(runs) >= expected_at_least:
+            return runs
+        await asyncio.sleep(0.5)
+    raise AssertionError(
+        f"Routine '{routine_id}' did not reach {expected_at_least} runs within {timeout}s"
+    )
+
+
+async def _wait_for_completed_run(
+    base_url: str,
+    routine_id: str,
+    *,
+    timeout: float = 30.0,
+) -> dict:
+    """Poll until the newest run is no longer marked running."""
+    for _ in range(int(timeout * 2)):
+        runs = await _get_routine_runs(base_url, routine_id)
+        if runs and runs[0]["status"].lower() != "running":
+            return runs[0]
+        await asyncio.sleep(0.5)
+    raise AssertionError(f"Routine '{routine_id}' did not complete within {timeout}s")
+
+
+@pytest.mark.asyncio
+async def test_create_event_trigger_routine(page, ironclaw_server):
+    """Event routines can be created through the supported chat flow."""
+    name = f"evt-{uuid.uuid4().hex[:8]}"
+    routine = await _create_event_routine(
+        page,
+        ironclaw_server,
+        name=name,
+        pattern="test|demo",
+    )
+
+    assert routine["id"]
+    assert routine["trigger_type"] == "event"
+    assert "test|demo" in routine["trigger_summary"]
+
+
+@pytest.mark.asyncio
+async def test_event_trigger_fires_on_matching_message(
+    page,
+    ironclaw_server,
+    http_channel_server,
+):
+    """Matching HTTP-channel messages create routine runs."""
+    name = f"evt-{uuid.uuid4().hex[:8]}"
+    routine = await _create_event_routine(
+        page,
+        ironclaw_server,
+        name=name,
+        pattern="urgent|critical|alert",
+    )
+
+    response = await _post_http_message(
+        http_channel_server,
+        content="urgent: server down",
+    )
+    assert response["status"] == "accepted"
+
+    await _wait_for_run_count(
+        ironclaw_server,
+        routine["id"],
+        expected_at_least=1,
+    )
+    completed_run = await _wait_for_completed_run(ironclaw_server, routine["id"])
+
+    assert completed_run["status"].lower() == "attention"
+    assert completed_run["trigger_type"] == "event"
+
+
+@pytest.mark.asyncio
+async def test_event_trigger_skips_non_matching_message(
+    page,
+    ironclaw_server,
+    http_channel_server,
+):
+    """Non-matching messages do not create routine runs."""
+    name = f"evt-{uuid.uuid4().hex[:8]}"
+    routine = await _create_event_routine(
+        page,
+        ironclaw_server,
+        name=name,
+        pattern="urgent|critical|alert",
+    )
+
+    await _post_http_message(
+        http_channel_server,
+        content="hello there",
+    )
+    await asyncio.sleep(2)
+
+    assert await _get_routine_runs(ironclaw_server, routine["id"]) == []
+
+
+@pytest.mark.asyncio
+async def test_multiple_routines_fire_on_matching_message(
+    page,
+    ironclaw_server,
+    http_channel_server,
+):
+    """A single matching message can fire multiple event routines."""
+    routines = []
+    for _ in range(3):
+        name = f"evt-{uuid.uuid4().hex[:8]}"
+        routines.append(
+            await _create_event_routine(
+                page,
+                ironclaw_server,
+                name=name,
+                pattern="error|warning|alert",
             )
         )
 
-        assert r1_fired_msg1, "Routine 1 should fire on message 1"
-        assert r2_fired_msg2, "Routine 2 should fire on message 2"
-        assert r3_fired_both, "Routine 3 should fire on both messages"
+    await _post_http_message(
+        http_channel_server,
+        content="error: database connection failed",
+    )
 
-    finally:
-        await page.close()
+    for routine in routines:
+        await _wait_for_run_count(
+            ironclaw_server,
+            routine["id"],
+            expected_at_least=1,
+        )
+        completed_run = await _wait_for_completed_run(ironclaw_server, routine["id"])
+        assert completed_run["status"].lower() == "attention"
 
 
-# =============================================================================
-# Integration with existing test patterns
-# =============================================================================
+@pytest.mark.asyncio
+async def test_channel_filter_applied_correctly(
+    page,
+    ironclaw_server,
+    http_channel_server,
+):
+    """Channel filters prevent HTTP messages from firing non-HTTP routines."""
+    http_routine = await _create_event_routine(
+        page,
+        ironclaw_server,
+        name=f"evt-{uuid.uuid4().hex[:8]}",
+        pattern="alert",
+        channel="http",
+    )
+    telegram_routine = await _create_event_routine(
+        page,
+        ironclaw_server,
+        name=f"evt-{uuid.uuid4().hex[:8]}",
+        pattern="alert",
+        channel="telegram",
+    )
+
+    await _post_http_message(
+        http_channel_server,
+        content="alert from webhook",
+    )
+
+    await _wait_for_run_count(
+        ironclaw_server,
+        http_routine["id"],
+        expected_at_least=1,
+    )
+    http_run = await _wait_for_completed_run(ironclaw_server, http_routine["id"])
+    await asyncio.sleep(2)
+    telegram_runs = await _get_routine_runs(ironclaw_server, telegram_routine["id"])
+
+    assert http_run["status"].lower() == "attention"
+    assert telegram_runs == []
 
 
-if __name__ == "__main__":
-    # Run tests with: pytest tests/e2e/scenarios/test_routine_event_batch.py -v
-    pytest.main([__file__, "-v", "-s"])
+@pytest.mark.asyncio
+async def test_routine_execution_history_is_available(
+    page,
+    ironclaw_server,
+    http_channel_server,
+):
+    """Routine run history is exposed by the routines runs API."""
+    routine = await _create_event_routine(
+        page,
+        ironclaw_server,
+        name=f"evt-{uuid.uuid4().hex[:8]}",
+        pattern="history",
+    )
+
+    await _post_http_message(
+        http_channel_server,
+        content="history event",
+    )
+
+    await _wait_for_run_count(
+        ironclaw_server,
+        routine["id"],
+        expected_at_least=1,
+    )
+    completed_run = await _wait_for_completed_run(ironclaw_server, routine["id"])
+
+    assert completed_run["id"]
+    assert completed_run["started_at"]
+    assert completed_run["status"].lower() == "attention"

tests/e2e/scenarios/test_telegram_hot_activation.py

@@ -0,0 +1,243 @@
+"""Telegram hot-activation UI coverage."""
+
+import asyncio
+import json
+
+from helpers import SEL
+
+_CONFIGURE_SECRET_INPUT = "input[type='password']"
+_CONFIGURE_SAVE_BUTTON = ".configure-actions button.btn-ext.activate"
+
+
+_TELEGRAM_INSTALLED = {
+    "name": "telegram",
+    "display_name": "Telegram",
+    "kind": "wasm_channel",
+    "description": "Telegram Bot API channel",
+    "url": None,
+    "active": False,
+    "authenticated": False,
+    "has_auth": False,
+    "needs_setup": True,
+    "tools": [],
+    "activation_status": "installed",
+    "activation_error": None,
+}
+
+_TELEGRAM_ACTIVE = {
+    **_TELEGRAM_INSTALLED,
+    "active": True,
+    "authenticated": True,
+    "needs_setup": False,
+    "activation_status": "active",
+}
+
+
+async def go_to_extensions(page):
+    await page.locator(SEL["tab_button"].format(tab="extensions")).click()
+    await page.locator(SEL["tab_panel"].format(tab="extensions")).wait_for(
+        state="visible", timeout=5000
+    )
+    await page.locator(
+        f"{SEL['extensions_list']} .empty-state, {SEL['ext_card_installed']}"
+    ).first.wait_for(state="visible", timeout=8000)
+
+
+async def mock_extension_lists(page, ext_handler):
+    async def handle_ext_list(route):
+        path = route.request.url.split("?")[0]
+        if path.endswith("/api/extensions"):
+            await ext_handler(route)
+        else:
+            await route.continue_()
+
+    async def handle_tools(route):
+        await route.fulfill(
+            status=200,
+            content_type="application/json",
+            body=json.dumps({"tools": []}),
+        )
+
+    async def handle_registry(route):
+        await route.fulfill(
+            status=200,
+            content_type="application/json",
+            body=json.dumps({"entries": []}),
+        )
+
+    # Register the broad route first so the specific endpoints below win.
+    await page.route("**/api/extensions*", handle_ext_list)
+    await page.route("**/api/extensions/tools", handle_tools)
+    await page.route("**/api/extensions/registry", handle_registry)
+
+
+async def wait_for_toast(page, text: str, *, timeout: int = 5000):
+    await page.locator(SEL["toast"], has_text=text).wait_for(
+        state="visible", timeout=timeout
+    )
+
+
+async def test_telegram_setup_modal_shows_bot_token_field(page):
+    async def handle_ext_list(route):
+        await route.fulfill(
+            status=200,
+            content_type="application/json",
+            body=json.dumps({"extensions": [_TELEGRAM_INSTALLED]}),
+        )
+
+    async def handle_setup(route):
+        await route.fulfill(
+            status=200,
+            content_type="application/json",
+            body=json.dumps(
+                {
+                    "secrets": [
+                        {
+                            "name": "telegram_bot_token",
+                            "prompt": "Enter your Telegram Bot API token (from @BotFather)",
+                            "provided": False,
+                            "optional": False,
+                            "auto_generate": False,
+                        }
+                    ]
+                }
+            ),
+        )
+
+    await mock_extension_lists(page, handle_ext_list)
+    await page.route("**/api/extensions/telegram/setup", handle_setup)
+    await go_to_extensions(page)
+
+    card = page.locator(SEL["ext_card_installed"]).first
+    await card.locator(SEL["ext_configure_btn"], has_text="Setup").click()
+
+    modal = page.locator(SEL["configure_modal"])
+    await modal.wait_for(state="visible", timeout=5000)
+    assert "Telegram Bot API token" in await modal.text_content()
+    assert "IronClaw will show a one-time code" in (
+        await modal.text_content()
+    )
+    input_el = modal.locator(_CONFIGURE_SECRET_INPUT)
+    assert await input_el.count() == 1
+
+
+async def test_telegram_hot_activation_transitions_installed_to_active(page):
+    phase = {"value": "installed"}
+    captured_setup_payloads = []
+    post_count = {"value": 0}
+    second_request_started = asyncio.Event()
+    allow_second_response = asyncio.Event()
+
+    async def handle_ext_list(route):
+        extensions = {
+            "installed": [_TELEGRAM_INSTALLED],
+            "active": [_TELEGRAM_ACTIVE],
+        }[phase["value"]]
+        await route.fulfill(
+            status=200,
+            content_type="application/json",
+            body=json.dumps({"extensions": extensions}),
+        )
+
+    async def handle_setup(route):
+        if route.request.method == "GET":
+            await route.fulfill(
+                status=200,
+                content_type="application/json",
+                body=json.dumps(
+                    {
+                        "secrets": [
+                            {
+                                "name": "telegram_bot_token",
+                                "prompt": "Enter your Telegram Bot API token (from @BotFather)",
+                                "provided": False,
+                                "optional": False,
+                                "auto_generate": False,
+                            }
+                        ]
+                    }
+                ),
+            )
+            return
+
+        payload = json.loads(route.request.post_data or "{}")
+        captured_setup_payloads.append(payload)
+        post_count["value"] += 1
+        await asyncio.sleep(0.05)
+        if post_count["value"] == 1:
+            await route.fulfill(
+                status=200,
+                content_type="application/json",
+                body=json.dumps(
+                    {
+                        "success": True,
+                        "activated": False,
+                        "message": "Configuration saved for 'telegram'. Send `/start iclaw-7qk2m9` to @test_hot_bot in Telegram. IronClaw will finish setup automatically.",
+                        "verification": {
+                            "code": "iclaw-7qk2m9",
+                            "instructions": "Send `/start iclaw-7qk2m9` to @test_hot_bot in Telegram. IronClaw will finish setup automatically.",
+                            "deep_link": "https://t.me/test_hot_bot?start=iclaw-7qk2m9",
+                        },
+                    }
+                ),
+            )
+        else:
+            second_request_started.set()
+            await allow_second_response.wait()
+            await route.fulfill(
+                status=200,
+                content_type="application/json",
+                body=json.dumps(
+                    {
+                        "success": True,
+                        "activated": True,
+                        "message": "Configuration saved, Telegram owner verified, and 'telegram' activated. Hot-activated WASM channel",
+                    }
+                ),
+            )
+
+    await mock_extension_lists(page, handle_ext_list)
+    await page.route("**/api/extensions/telegram/setup", handle_setup)
+    await go_to_extensions(page)
+
+    card = page.locator(SEL["ext_card_installed"]).first
+    await card.locator(SEL["ext_configure_btn"], has_text="Setup").click()
+
+    modal = page.locator(SEL["configure_modal"])
+    await modal.wait_for(state="visible", timeout=5000)
+    await modal.locator(_CONFIGURE_SECRET_INPUT).fill("123456789:ABCdefGhI")
+    await modal.locator(_CONFIGURE_SAVE_BUTTON).click()
+    await second_request_started.wait()
+    await modal.locator(".configure-inline-status", has_text="Waiting for Telegram owner verification...").wait_for(
+        state="visible", timeout=5000
+    )
+    assert "iclaw-7qk2m9" in (await modal.text_content())
+    assert "/start iclaw-7qk2m9" in (await modal.text_content())
+    assert await modal.locator(".configure-verification-link").count() == 1
+    await modal.locator(_CONFIGURE_SAVE_BUTTON).wait_for(state="hidden", timeout=5000)
+
+    await page.locator(SEL["configure_overlay"]).click(position={"x": 1, "y": 1})
+    assert await page.locator(SEL["configure_overlay"]).is_visible()
+
+    allow_second_response.set()
+    await page.locator(SEL["configure_overlay"]).wait_for(state="hidden", timeout=5000)
+
+    phase["value"] = "active"
+    await page.evaluate(
+        """
+        handleAuthCompleted({
+          extension_name: 'telegram',
+          success: true,
+          message: "Configuration saved, Telegram owner verified, and 'telegram' activated. Hot-activated WASM channel",
+        });
+        """
+    )
+
+    await wait_for_toast(page, "Telegram owner verified")
+    await card.locator(SEL["ext_active_label"]).wait_for(state="visible", timeout=5000)
+    assert await card.locator(SEL["ext_pairing_label"]).count() == 0
+
+    assert captured_setup_payloads == [
+        {"secrets": {"telegram_bot_token": "123456789:ABCdefGhI"}},
+        {"secrets": {}},
+    ]

tests/e2e/scenarios/test_tool_approval.py

@@ -130,3 +130,59 @@ async def test_approval_params_toggle(page):
     await toggle.click()
     await page.wait_for_timeout(300)
     assert await params.is_hidden(), "Parameters should be hidden after second toggle"
+
+
+async def test_waiting_for_approval_message_no_error_prefix(page):
+    """Verify that input submitted while awaiting approval shows non-error status with tool context.
+
+    Trigger a real approval-needed tool call, then attempt to send another message while
+    approval is pending. The backend should reject the second input with a non-error
+    status that includes the pending tool context.
+    """
+    assistant_messages = page.locator(SEL["message_assistant"])
+    chat_input = page.locator(SEL["chat_input"])
+    await chat_input.wait_for(state="visible", timeout=5000)
+
+    # Trigger a real HTTP tool call that pauses for approval in the default E2E harness.
+    await chat_input.fill("make approval post approval-required")
+    await chat_input.press("Enter")
+
+    card = page.locator(SEL["approval_card"]).last
+    await card.wait_for(state="visible", timeout=10000)
+
+    tool_name = await card.locator(".approval-tool-name").text_content()
+    desc_text = await card.locator(".approval-description").text_content()
+    assert tool_name == "http"
+    assert desc_text is not None and "HTTP requests to external APIs" in desc_text
+
+    # With the thread now genuinely awaiting approval, the next message should be rejected
+    # as a non-error pending status.
+    initial_count = await assistant_messages.count()
+    await chat_input.fill("send another message now")
+    await chat_input.press("Enter")
+
+    await page.wait_for_function(
+        f"() => document.querySelectorAll('{SEL['message_assistant']}').length > {initial_count}",
+        timeout=10000,
+    )
+
+    last_msg = assistant_messages.last.locator(".message-content")
+    msg_text = await last_msg.inner_text()
+
+    # Verify no "Error:" prefix
+    assert not msg_text.lower().startswith("error:"), (
+        f"Approval rejection must NOT have 'Error:' prefix. Got: {msg_text!r}"
+    )
+
+    # Verify it contains "waiting for approval"
+    assert "waiting for approval" in msg_text.lower(), (
+        f"Expected 'Waiting for approval' text. Got: {msg_text!r}"
+    )
+
+    # Verify it contains the tool name and description
+    assert "http" in msg_text.lower(), (
+        f"Expected tool name 'http' in message. Got: {msg_text!r}"
+    )
+    assert "HTTP requests to external APIs" in msg_text, (
+        f"Expected tool description in message. Got: {msg_text!r}"
+    )

tests/e2e/scenarios/test_webhook.py

@@ -7,7 +7,7 @@ import json
 import httpx
 import pytest
 
-from helpers import AUTH_TOKEN
+from helpers import HTTP_WEBHOOK_SECRET
 
 
 def compute_signature(secret: str, body: bytes) -> str:
@@ -16,325 +16,188 @@ def compute_signature(secret: str, body: bytes) -> str:
     return f"sha256={mac.hexdigest()}"
 
 
-@pytest.mark.asyncio
-async def test_webhook_requires_http_webhook_secret_configured(ironclaw_server):
-    """
-    Webhook endpoint rejects requests when HTTP_WEBHOOK_SECRET is not configured.
-    This tests the fail-closed security posture.
-    """
-    headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
+async def _post_webhook(
+    base_url: str,
+    body_data: dict,
+    *,
+    signature: str | None = None,
+    content_type: str = "application/json",
+) -> httpx.Response:
+    """Send a raw webhook request with optional signature."""
+    body_bytes = json.dumps(body_data).encode()
+    headers = {"Content-Type": content_type}
+    if signature is not None:
+        headers["X-Hub-Signature-256"] = signature
+
     async with httpx.AsyncClient() as client:
-        # When no webhook secret is configured on the server, all requests fail
-        r = await client.post(
-            f"{ironclaw_server}/webhook",
-            json={"content": "test message"},
+        return await client.post(
+            f"{base_url}/webhook",
+            content=body_bytes,
             headers=headers,
         )
-        # Server should reject with 503 Service Unavailable (fail closed)
-        assert r.status_code in (401, 503)
 
 
 @pytest.mark.asyncio
-async def test_webhook_hmac_signature_valid(ironclaw_server_with_webhook_secret):
+async def test_webhook_requires_http_webhook_secret_configured(
+    http_channel_server_without_secret,
+):
+    """Webhook fails closed when no secret is configured."""
+    response = await _post_webhook(
+        http_channel_server_without_secret,
+        {"content": "test message"},
+    )
+
+    assert response.status_code == 503
+    data = response.json()
+    assert data["status"] == "error"
+    assert "Webhook authentication not configured" in data.get("response", "")
+
+
+@pytest.mark.asyncio
+async def test_webhook_hmac_signature_valid(http_channel_server):
     """Valid X-Hub-Signature-256 HMAC signature is accepted."""
-    secret = ironclaw_server_with_webhook_secret["secret"]
-    base_url = ironclaw_server_with_webhook_secret["url"]
+    body = {"content": "hello from webhook"}
+    signature = compute_signature(HTTP_WEBHOOK_SECRET, json.dumps(body).encode())
 
-    headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
-    body_data = {"content": "hello from webhook"}
-    body_bytes = json.dumps(body_data).encode()
-    signature = compute_signature(secret, body_bytes)
+    response = await _post_webhook(http_channel_server, body, signature=signature)
 
-    async with httpx.AsyncClient() as client:
-        r = await client.post(
-            f"{base_url}/webhook",
-            content=body_bytes,
-            headers={
-                **headers,
-                "Content-Type": "application/json",
-                "X-Hub-Signature-256": signature,
-            },
-        )
-        assert r.status_code == 200, f"Expected 200, got {r.status_code}: {r.text}"
-        resp = r.json()
-        assert resp["status"] == "ok"
+    assert response.status_code == 200, (
+        f"Expected 200, got {response.status_code}: {response.text}"
+    )
+    data = response.json()
+    assert data["status"] == "accepted"
 
 
 @pytest.mark.asyncio
-async def test_webhook_invalid_hmac_signature_rejected(
-    ironclaw_server_with_webhook_secret,
-):
+async def test_webhook_invalid_hmac_signature_rejected(http_channel_server):
     """Invalid X-Hub-Signature-256 signature is rejected with 401."""
-    base_url = ironclaw_server_with_webhook_secret["url"]
+    response = await _post_webhook(
+        http_channel_server,
+        {"content": "hello"},
+        signature="sha256=0000000000000000000000000000000000000000000000000000000000000000",
+    )
 
-    headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
-    body_data = {"content": "hello"}
-    body_bytes = json.dumps(body_data).encode()
-    invalid_signature = "sha256=0000000000000000000000000000000000000000000000000000000000000000"
-
-    async with httpx.AsyncClient() as client:
-        r = await client.post(
-            f"{base_url}/webhook",
-            content=body_bytes,
-            headers={
-                **headers,
-                "Content-Type": "application/json",
-                "X-Hub-Signature-256": invalid_signature,
-            },
-        )
-        assert r.status_code == 401, f"Expected 401, got {r.status_code}"
-        resp = r.json()
-        assert resp["status"] == "error"
-        assert "Invalid webhook signature" in resp.get("response", "")
+    assert response.status_code == 401
+    data = response.json()
+    assert data["status"] == "error"
+    assert "Invalid webhook signature" in data.get("response", "")
 
 
 @pytest.mark.asyncio
-async def test_webhook_wrong_secret_rejected(ironclaw_server_with_webhook_secret):
+async def test_webhook_wrong_secret_rejected(http_channel_server):
     """Signature computed with wrong secret is rejected."""
-    base_url = ironclaw_server_with_webhook_secret["url"]
+    body = {"content": "hello"}
+    signature = compute_signature("wrong-secret", json.dumps(body).encode())
 
-    headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
-    body_data = {"content": "hello"}
-    body_bytes = json.dumps(body_data).encode()
-    # Compute signature with wrong secret
-    wrong_signature = compute_signature("wrong-secret", body_bytes)
+    response = await _post_webhook(http_channel_server, body, signature=signature)
 
-    async with httpx.AsyncClient() as client:
-        r = await client.post(
-            f"{base_url}/webhook",
-            content=body_bytes,
-            headers={
-                **headers,
-                "Content-Type": "application/json",
-                "X-Hub-Signature-256": wrong_signature,
-            },
-        )
-        assert r.status_code == 401
-        resp = r.json()
-        assert resp["status"] == "error"
+    assert response.status_code == 401
+    assert response.json()["status"] == "error"
 
 
 @pytest.mark.asyncio
-async def test_webhook_malformed_signature_rejected(
-    ironclaw_server_with_webhook_secret,
-):
-    """Malformed X-Hub-Signature-256 header is rejected."""
-    base_url = ironclaw_server_with_webhook_secret["url"]
+async def test_webhook_missing_signature_header_rejected(http_channel_server):
+    """Missing X-Hub-Signature-256 header is rejected when no body secret is provided."""
+    response = await _post_webhook(http_channel_server, {"content": "hello"})
 
-    headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
-    body_data = {"content": "hello"}
-    body_bytes = json.dumps(body_data).encode()
-
-    async with httpx.AsyncClient() as client:
-        # Missing sha256= prefix
-        r = await client.post(
-            f"{base_url}/webhook",
-            content=body_bytes,
-            headers={
-                **headers,
-                "Content-Type": "application/json",
-                "X-Hub-Signature-256": "deadbeef",
-            },
-        )
-        assert r.status_code == 401
+    assert response.status_code == 401
+    data = response.json()
+    assert "Webhook authentication required" in data.get("response", "")
+    assert "X-Hub-Signature-256" in data.get("response", "")
 
 
 @pytest.mark.asyncio
-async def test_webhook_missing_signature_header_rejected(
-    ironclaw_server_with_webhook_secret,
-):
-    """Missing X-Hub-Signature-256 header is rejected when no body secret provided."""
-    base_url = ironclaw_server_with_webhook_secret["url"]
+async def test_webhook_deprecated_body_secret_still_works(http_channel_server):
+    """Deprecated body secret support still accepts old clients."""
+    response = await _post_webhook(
+        http_channel_server,
+        {"content": "hello", "secret": HTTP_WEBHOOK_SECRET},
+    )
 
-    headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
-    body_data = {"content": "hello"}
-    body_bytes = json.dumps(body_data).encode()
-
-    async with httpx.AsyncClient() as client:
-        # No X-Hub-Signature-256 header and no body secret
-        r = await client.post(
-            f"{base_url}/webhook",
-            content=body_bytes,
-            headers={
-                **headers,
-                "Content-Type": "application/json",
-            },
-        )
-        assert r.status_code == 401
-        resp = r.json()
-        assert "Webhook authentication required" in resp.get("response", "")
-        assert "X-Hub-Signature-256" in resp.get("response", "")
+    assert response.status_code == 200, (
+        f"Expected 200, got {response.status_code}: {response.text}"
+    )
+    assert response.json()["status"] == "accepted"
 
 
 @pytest.mark.asyncio
-async def test_webhook_deprecated_body_secret_still_works(
-    ironclaw_server_with_webhook_secret,
-):
-    """
-    Deprecated: body 'secret' field still works for backward compatibility.
-    This test ensures we don't break existing clients during the migration period.
-    """
-    secret = ironclaw_server_with_webhook_secret["secret"]
-    base_url = ironclaw_server_with_webhook_secret["url"]
+async def test_webhook_header_takes_precedence_over_body_secret(http_channel_server):
+    """Header signature wins when both header and body secret are provided."""
+    body = {"content": "hello", "secret": "wrong-secret-in-body"}
+    signature = compute_signature(HTTP_WEBHOOK_SECRET, json.dumps(body).encode())
 
-    headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
-    # Old-style request with secret in body
-    body_data = {"content": "hello", "secret": secret}
-    body_bytes = json.dumps(body_data).encode()
+    response = await _post_webhook(http_channel_server, body, signature=signature)
 
-    async with httpx.AsyncClient() as client:
-        r = await client.post(
-            f"{base_url}/webhook",
-            content=body_bytes,
-            headers={
-                **headers,
-                "Content-Type": "application/json",
-            },
-        )
-        # Should succeed (backward compatibility)
-        assert r.status_code == 200, f"Expected 200, got {r.status_code}: {r.text}"
-        resp = r.json()
-        assert resp["status"] == "ok"
+    assert response.status_code == 200
+    assert response.json()["status"] == "accepted"
 
 
 @pytest.mark.asyncio
-async def test_webhook_header_takes_precedence_over_body_secret(
-    ironclaw_server_with_webhook_secret,
-):
-    """
-    When both X-Hub-Signature-256 header and body secret are provided,
-    header takes precedence.
-    """
-    secret = ironclaw_server_with_webhook_secret["secret"]
-    base_url = ironclaw_server_with_webhook_secret["url"]
-
-    headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
-    body_data = {"content": "hello", "secret": "wrong-secret-in-body"}
-    body_bytes = json.dumps(body_data).encode()
-    # Compute signature with correct secret
-    signature = compute_signature(secret, body_bytes)
+async def test_webhook_case_insensitive_header_lookup(http_channel_server):
+    """HTTP headers are treated case-insensitively."""
+    body = {"content": "hello"}
+    body_bytes = json.dumps(body).encode()
+    signature = compute_signature(HTTP_WEBHOOK_SECRET, body_bytes)
 
     async with httpx.AsyncClient() as client:
-        r = await client.post(
-            f"{base_url}/webhook",
+        response = await client.post(
+            f"{http_channel_server}/webhook",
             content=body_bytes,
             headers={
-                **headers,
-                "Content-Type": "application/json",
-                "X-Hub-Signature-256": signature,
-            },
-        )
-        # Should succeed because header signature is valid (takes precedence)
-        assert r.status_code == 200
-        resp = r.json()
-        assert resp["status"] == "ok"
-
-
-@pytest.mark.asyncio
-async def test_webhook_case_insensitive_header_lookup(
-    ironclaw_server_with_webhook_secret,
-):
-    """HTTP headers are case-insensitive. Test with different cases."""
-    secret = ironclaw_server_with_webhook_secret["secret"]
-    base_url = ironclaw_server_with_webhook_secret["url"]
-
-    headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
-    body_data = {"content": "hello"}
-    body_bytes = json.dumps(body_data).encode()
-    signature = compute_signature(secret, body_bytes)
-
-    async with httpx.AsyncClient() as client:
-        # Try with lowercase
-        r = await client.post(
-            f"{base_url}/webhook",
-            content=body_bytes,
-            headers={
-                **headers,
                 "Content-Type": "application/json",
                 "x-hub-signature-256": signature,
             },
         )
-        assert r.status_code == 200
+
+    assert response.status_code == 200
 
 
 @pytest.mark.asyncio
-async def test_webhook_wrong_content_type_rejected(
-    ironclaw_server_with_webhook_secret,
-):
+async def test_webhook_wrong_content_type_rejected(http_channel_server):
     """Webhook only accepts application/json Content-Type."""
-    secret = ironclaw_server_with_webhook_secret["secret"]
-    base_url = ironclaw_server_with_webhook_secret["url"]
+    body = {"content": "hello"}
+    signature = compute_signature(HTTP_WEBHOOK_SECRET, json.dumps(body).encode())
 
-    headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
-    body_data = {"content": "hello"}
-    body_bytes = json.dumps(body_data).encode()
-    signature = compute_signature(secret, body_bytes)
+    response = await _post_webhook(
+        http_channel_server,
+        body,
+        signature=signature,
+        content_type="text/plain",
+    )
 
-    async with httpx.AsyncClient() as client:
-        r = await client.post(
-            f"{base_url}/webhook",
-            content=body_bytes,
-            headers={
-                **headers,
-                "Content-Type": "text/plain",
-                "X-Hub-Signature-256": signature,
-            },
-        )
-        assert r.status_code == 415  # Unsupported Media Type
-        resp = r.json()
-        assert "application/json" in resp.get("response", "")
+    assert response.status_code == 415
+    assert "application/json" in response.json().get("response", "")
 
 
 @pytest.mark.asyncio
-async def test_webhook_invalid_json_rejected(ironclaw_server_with_webhook_secret):
+async def test_webhook_invalid_json_rejected(http_channel_server):
     """Invalid JSON in body is rejected."""
-    secret = ironclaw_server_with_webhook_secret["secret"]
-    base_url = ironclaw_server_with_webhook_secret["url"]
-
-    headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
     body_bytes = b"not valid json"
-    signature = compute_signature(secret, body_bytes)
+    signature = compute_signature(HTTP_WEBHOOK_SECRET, body_bytes)
 
     async with httpx.AsyncClient() as client:
-        r = await client.post(
-            f"{base_url}/webhook",
+        response = await client.post(
+            f"{http_channel_server}/webhook",
             content=body_bytes,
             headers={
-                **headers,
                 "Content-Type": "application/json",
                 "X-Hub-Signature-256": signature,
             },
         )
-        assert r.status_code == 401 or r.status_code == 400
+
+    assert response.status_code in (400, 401)
 
 
 @pytest.mark.asyncio
-async def test_webhook_message_queued_for_processing(
-    ironclaw_server_with_webhook_secret,
-):
-    """Message via webhook is queued and can be retrieved."""
-    secret = ironclaw_server_with_webhook_secret["secret"]
-    base_url = ironclaw_server_with_webhook_secret["url"]
+async def test_webhook_message_queued_for_processing(http_channel_server):
+    """Accepted webhook requests return a real message id."""
+    body = {"content": "webhook test message 12345"}
+    signature = compute_signature(HTTP_WEBHOOK_SECRET, json.dumps(body).encode())
 
-    headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
-    test_message = "webhook test message 12345"
-    body_data = {"content": test_message}
-    body_bytes = json.dumps(body_data).encode()
-    signature = compute_signature(secret, body_bytes)
+    response = await _post_webhook(http_channel_server, body, signature=signature)
 
-    async with httpx.AsyncClient() as client:
-        r = await client.post(
-            f"{base_url}/webhook",
-            content=body_bytes,
-            headers={
-                **headers,
-                "Content-Type": "application/json",
-                "X-Hub-Signature-256": signature,
-            },
-        )
-        assert r.status_code == 200
-        resp = r.json()
-        assert resp["status"] == "ok"
-        # Message ID should be present
-        assert "message_id" in resp
-        assert resp["message_id"] != "00000000-0000-0000-0000-000000000000"
+    assert response.status_code == 200
+    data = response.json()
+    assert data["status"] == "accepted"
+    assert "message_id" in data
+    assert data["message_id"] != "00000000-0000-0000-0000-000000000000"

tests/e2e_advanced_traces.rs

@@ -442,6 +442,9 @@ mod advanced {
             other => panic!("expected event trigger, got {other:?}"),
         }
 
+        rig.clear().await;
+        let llm_calls_before = rig.llm_call_count();
+
         rig.send_incoming(IncomingMessage::new(
             "telegram",
             "test-user",
@@ -451,8 +454,18 @@ mod advanced {
 
         let runs = wait_for_routine_run(rig.database(), routine.id, TIMEOUT).await;
         assert_eq!(runs[0].trigger_type, "event");
+        assert_eq!(
+            rig.llm_call_count(),
+            llm_calls_before + 1,
+            "matching event message should only trigger the routine LLM call"
+        );
 
-        let responses = rig.wait_for_responses(3, TIMEOUT).await;
+        let responses = rig.wait_for_responses(1, TIMEOUT).await;
+        assert_eq!(
+            responses.len(),
+            1,
+            "expected only the routine notification after the matching event"
+        );
         assert!(
             responses.iter().any(|response| {
                 response
@@ -505,6 +518,9 @@ mod advanced {
             other => panic!("expected event trigger, got {other:?}"),
         }
 
+        rig.clear().await;
+        let llm_calls_before = rig.llm_call_count();
+
         rig.send_incoming(IncomingMessage::new(
             "telegram",
             "test-user",
@@ -514,6 +530,22 @@ mod advanced {
 
         let runs = wait_for_routine_run(rig.database(), routine.id, TIMEOUT).await;
         assert_eq!(runs[0].trigger_type, "event");
+        assert_eq!(
+            rig.llm_call_count(),
+            llm_calls_before + 1,
+            "matching event message should only trigger the routine LLM call"
+        );
+
+        let responses = rig.wait_for_responses(1, TIMEOUT).await;
+        assert_eq!(
+            responses.len(),
+            1,
+            "expected only the routine notification after the matching event"
+        );
+        assert!(
+            responses[0].content.contains("Bug report detected"),
+            "expected routine notification, got: {responses:?}"
+        );
 
         rig.shutdown();
     }

tests/e2e_builtin_tool_coverage.rs

@@ -155,7 +155,7 @@ mod tests {
         }
 
         assert_eq!(routine.notify.channel.as_deref(), Some("telegram"));
-        assert_eq!(routine.notify.user, "ops-team");
+        assert_eq!(routine.notify.user.as_deref(), Some("ops-team"));
         assert_eq!(routine.guardrails.cooldown.as_secs(), 600);
 
         rig.shutdown();

tests/e2e_routine_heartbeat.rs

@@ -48,6 +48,19 @@ mod tests {
         Arc::new(Workspace::new_with_db("default", db.clone()))
     }
 
+    fn make_message(
+        channel: &str,
+        user_id: &str,
+        owner_id: &str,
+        sender_id: &str,
+        content: &str,
+    ) -> IncomingMessage {
+        IncomingMessage::new(channel, user_id, content)
+            .with_owner_id(owner_id)
+            .with_sender_id(sender_id)
+            .with_metadata(serde_json::json!({}))
+    }
+
     /// Helper to insert a routine directly into the database.
     fn make_routine(name: &str, trigger: Trigger, prompt: &str) -> Routine {
         Routine {
@@ -218,7 +231,13 @@ mod tests {
         engine.refresh_event_cache().await;
 
         // Positive match: message containing "deploy to production".
-        let matching_msg = IncomingMessage::new("test", "default", "deploy to production now");
+        let matching_msg = make_message(
+            "test",
+            "default",
+            "default",
+            "default",
+            "deploy to production now",
+        );
         let fired = engine.check_event_triggers(&matching_msg).await;
         assert!(
             fired >= 1,
@@ -229,12 +248,114 @@ mod tests {
         tokio::time::sleep(Duration::from_millis(500)).await;
 
         // Negative match: message that doesn't match.
-        let non_matching_msg =
-            IncomingMessage::new("test", "default", "check the staging environment");
+        let non_matching_msg = make_message(
+            "test",
+            "default",
+            "default",
+            "default",
+            "check the staging environment",
+        );
         let fired_neg = engine.check_event_triggers(&non_matching_msg).await;
         assert_eq!(fired_neg, 0, "Expected 0 routines fired on non-match");
     }
 
+    #[tokio::test]
+    async fn event_trigger_respects_message_user_scope() {
+        let (db, _tmp) = create_test_db().await;
+        let ws = create_workspace(&db);
+
+        let trace = LlmTrace::single_turn(
+            "test-event-user-scope",
+            "deploy",
+            vec![TraceStep {
+                request_hint: None,
+                response: TraceResponse::Text {
+                    content: "Owner event handled".to_string(),
+                    input_tokens: 50,
+                    output_tokens: 8,
+                },
+                expected_tool_results: vec![],
+            }],
+        );
+        let llm = Arc::new(TraceLlm::from_trace(trace));
+        let (notify_tx, _notify_rx) = tokio::sync::mpsc::channel(16);
+
+        let tools = Arc::new(ToolRegistry::new());
+        let safety = Arc::new(SafetyLayer::new(&SafetyConfig {
+            max_output_length: 100_000,
+            injection_check_enabled: true,
+        }));
+
+        let engine = Arc::new(RoutineEngine::new(
+            RoutineConfig::default(),
+            db.clone(),
+            llm,
+            ws,
+            notify_tx,
+            None,
+            tools,
+            safety,
+        ));
+
+        let routine = make_routine(
+            "owner-deploy-watcher",
+            Trigger::Event {
+                channel: None,
+                pattern: "deploy.*production".to_string(),
+            },
+            "Report on deployment.",
+        );
+        db.create_routine(&routine).await.expect("create_routine");
+        engine.refresh_event_cache().await;
+
+        let guest_msg = make_message(
+            "telegram",
+            "guest",
+            "default",
+            "guest-sender",
+            "deploy to production now",
+        );
+        let guest_fired = engine.check_event_triggers(&guest_msg).await;
+        assert_eq!(
+            guest_fired, 0,
+            "Guest scope must not fire owner event routines"
+        );
+        tokio::time::sleep(Duration::from_millis(200)).await;
+
+        let guest_runs = db
+            .list_routine_runs(routine.id, 10)
+            .await
+            .expect("list_routine_runs after guest message");
+        assert!(
+            guest_runs.is_empty(),
+            "Guest message should not create routine runs"
+        );
+
+        let owner_msg = make_message(
+            "telegram",
+            "default",
+            "default",
+            "owner-sender",
+            "deploy to production now",
+        );
+        let owner_fired = engine.check_event_triggers(&owner_msg).await;
+        assert!(
+            owner_fired >= 1,
+            "Owner scope should fire matching owner event routine"
+        );
+        tokio::time::sleep(Duration::from_millis(500)).await;
+
+        let owner_runs = db
+            .list_routine_runs(routine.id, 10)
+            .await
+            .expect("list_routine_runs after owner message");
+        assert_eq!(
+            owner_runs.len(),
+            1,
+            "Owner message should create exactly one run"
+        );
+    }
+
     // -----------------------------------------------------------------------
     // Test 3: system_event_trigger_matches_and_filters
     // -----------------------------------------------------------------------
@@ -434,7 +555,13 @@ mod tests {
         engine.refresh_event_cache().await;
 
         // First fire should work.
-        let msg = IncomingMessage::new("test", "default", "test-cooldown trigger");
+        let msg = make_message(
+            "test",
+            "default",
+            "default",
+            "default",
+            "test-cooldown trigger",
+        );
         let fired1 = engine.check_event_triggers(&msg).await;
         assert!(fired1 >= 1, "First fire should work");
 
@@ -553,4 +680,118 @@ mod tests {
             "Expected Skipped for empty checklist, got: {result:?}"
         );
     }
+
+    /// Helper to set up a test environment for routine engine mutation tests.
+    /// Returns the engine, database, and temp directory.
+    async fn setup_routine_mutation_test()
+    -> (Arc<RoutineEngine>, Arc<dyn Database>, tempfile::TempDir) {
+        let (db, dir) = create_test_db().await;
+        let ws = create_workspace(&db);
+        let (notify_tx, _rx) = tokio::sync::mpsc::channel(16);
+        let tools = Arc::new(ToolRegistry::new());
+
+        let safety_config = SafetyConfig {
+            max_output_length: 100_000,
+            injection_check_enabled: true,
+        };
+        let safety = Arc::new(SafetyLayer::new(&safety_config));
+
+        let trace = LlmTrace::single_turn(
+            "test-routine-mutation",
+            "test",
+            vec![TraceStep {
+                request_hint: None,
+                response: TraceResponse::Text {
+                    content: "ROUTINE_OK".to_string(),
+                    input_tokens: 50,
+                    output_tokens: 5,
+                },
+                expected_tool_results: vec![],
+            }],
+        );
+        let llm = Arc::new(TraceLlm::from_trace(trace));
+
+        let engine = Arc::new(RoutineEngine::new(
+            RoutineConfig::default(),
+            Arc::clone(&db),
+            llm,
+            ws,
+            notify_tx,
+            None,
+            tools,
+            safety,
+        ));
+
+        (engine, db, dir)
+    }
+
+    /// Regression test for issue #1076: disabling an event routine via a DB mutation
+    /// followed by refresh_event_cache() (the path now taken by the web toggle handler)
+    /// must immediately stop the routine from firing.
+    #[tokio::test]
+    async fn toggle_disabling_event_routine_removes_from_cache() {
+        let (engine, db, _dir) = setup_routine_mutation_test().await;
+
+        // Create and cache an event routine.
+        let mut routine = make_routine(
+            "disable-me",
+            Trigger::Event {
+                pattern: "DISABLE_ME".to_string(),
+                channel: None,
+            },
+            "Handle DISABLE_ME event",
+        );
+        db.create_routine(&routine).await.expect("create_routine");
+        engine.refresh_event_cache().await;
+
+        let msg = IncomingMessage::new("test", "default", "DISABLE_ME");
+        let fired_before = engine.check_event_triggers(&msg).await;
+        assert!(fired_before >= 1, "Expected routine to fire before disable");
+
+        // Simulate what routines_toggle_handler now does: update DB, then refresh.
+        routine.enabled = false;
+        routine.updated_at = Utc::now();
+        db.update_routine(&routine).await.expect("update_routine");
+        engine.refresh_event_cache().await;
+
+        let fired_after = engine.check_event_triggers(&msg).await;
+        assert_eq!(
+            fired_after, 0,
+            "Disabled routine must not fire after cache refresh"
+        );
+    }
+
+    /// Regression test for issue #1076: deleting an event routine via a DB mutation
+    /// followed by refresh_event_cache() must immediately stop the routine from firing.
+    #[tokio::test]
+    async fn delete_event_routine_removes_from_cache() {
+        let (engine, db, _dir) = setup_routine_mutation_test().await;
+
+        let routine = make_routine(
+            "delete-me",
+            Trigger::Event {
+                pattern: "DELETE_ME".to_string(),
+                channel: None,
+            },
+            "Handle DELETE_ME event",
+        );
+        db.create_routine(&routine).await.expect("create_routine");
+        engine.refresh_event_cache().await;
+
+        let msg = IncomingMessage::new("test", "default", "DELETE_ME");
+        assert!(
+            engine.check_event_triggers(&msg).await >= 1,
+            "Expected routine to fire before delete"
+        );
+
+        // Simulate what routines_delete_handler now does: delete from DB, then refresh.
+        db.delete_routine(routine.id).await.expect("delete_routine");
+        engine.refresh_event_cache().await;
+
+        assert_eq!(
+            engine.check_event_triggers(&msg).await,
+            0,
+            "Deleted routine must not fire after cache refresh"
+        );
+    }
 }

tests/e2e_telegram_message_routing.rs

@@ -0,0 +1,353 @@
+//! E2E tests for Telegram message routing through the real agent + message tool.
+
+#[cfg(feature = "libsql")]
+mod support;
+
+#[cfg(feature = "libsql")]
+mod tests {
+    use std::sync::Arc;
+    use std::time::Duration;
+
+    use async_trait::async_trait;
+    use futures::StreamExt;
+    use ironclaw::agent::{Agent, AgentDeps};
+    use ironclaw::app::{AppBuilder, AppBuilderFlags};
+    use ironclaw::channels::web::log_layer::LogBroadcaster;
+    use ironclaw::channels::{
+        Channel, ChannelManager, IncomingMessage, MessageStream, OutgoingResponse, StatusUpdate,
+    };
+    use ironclaw::config::Config;
+    use ironclaw::db::{Database, libsql::LibSqlBackend};
+    use ironclaw::error::ChannelError;
+    use ironclaw::llm::{LlmProvider, SessionConfig, SessionManager};
+    use tokio::sync::{Mutex, mpsc};
+    use tokio_stream::wrappers::ReceiverStream;
+
+    use crate::support::test_channel::{TestChannel, TestChannelHandle};
+    use crate::support::trace_llm::{LlmTrace, TraceLlm, TraceResponse, TraceStep, TraceToolCall};
+
+    type TelegramCaptures = Arc<Mutex<Vec<(String, OutgoingResponse)>>>;
+
+    struct RecordingTelegramChannel {
+        captures: TelegramCaptures,
+    }
+
+    impl RecordingTelegramChannel {
+        fn new() -> (Self, TelegramCaptures) {
+            let captures = Arc::new(Mutex::new(Vec::new()));
+            (
+                Self {
+                    captures: Arc::clone(&captures),
+                },
+                captures,
+            )
+        }
+    }
+
+    #[async_trait]
+    impl Channel for RecordingTelegramChannel {
+        fn name(&self) -> &str {
+            "telegram"
+        }
+
+        async fn start(&self) -> Result<MessageStream, ChannelError> {
+            let (_tx, rx) = mpsc::channel::<IncomingMessage>(1);
+            Ok(ReceiverStream::new(rx).boxed())
+        }
+
+        async fn respond(
+            &self,
+            _msg: &IncomingMessage,
+            response: OutgoingResponse,
+        ) -> Result<(), ChannelError> {
+            self.captures
+                .lock()
+                .await
+                .push(("respond".to_string(), response));
+            Ok(())
+        }
+
+        async fn send_status(
+            &self,
+            _status: StatusUpdate,
+            _metadata: &serde_json::Value,
+        ) -> Result<(), ChannelError> {
+            Ok(())
+        }
+
+        async fn broadcast(
+            &self,
+            user_id: &str,
+            response: OutgoingResponse,
+        ) -> Result<(), ChannelError> {
+            self.captures
+                .lock()
+                .await
+                .push((user_id.to_string(), response));
+            Ok(())
+        }
+
+        async fn health_check(&self) -> Result<(), ChannelError> {
+            Ok(())
+        }
+    }
+
+    struct Harness {
+        gateway: Arc<TestChannel>,
+        telegram_captures: Arc<Mutex<Vec<(String, OutgoingResponse)>>>,
+        db: Arc<dyn Database>,
+        owner_id: String,
+        _temp_dir: tempfile::TempDir,
+        agent_handle: Option<tokio::task::JoinHandle<()>>,
+    }
+
+    impl Harness {
+        async fn store_telegram_owner_binding(&self, owner_id: i64) {
+            for scope in [&self.owner_id, "test-user"] {
+                self.db
+                    .set_setting(
+                        scope,
+                        "channels.wasm_channel_owner_ids.telegram",
+                        &serde_json::json!(owner_id),
+                    )
+                    .await
+                    .expect("failed to store telegram owner binding");
+            }
+        }
+
+        async fn wait_for_telegram_broadcasts(
+            &self,
+            expected: usize,
+            timeout: Duration,
+        ) -> Vec<(String, OutgoingResponse)> {
+            let deadline = tokio::time::Instant::now() + timeout;
+            loop {
+                let snapshot = self.telegram_captures.lock().await.clone();
+                if snapshot.len() >= expected || tokio::time::Instant::now() >= deadline {
+                    return snapshot;
+                }
+                tokio::time::sleep(Duration::from_millis(50)).await;
+            }
+        }
+    }
+
+    impl Drop for Harness {
+        fn drop(&mut self) {
+            self.gateway.signal_shutdown();
+            if let Some(handle) = self.agent_handle.take() {
+                handle.abort();
+            }
+        }
+    }
+
+    async fn build_harness(trace: LlmTrace) -> Harness {
+        let temp_dir = tempfile::tempdir().expect("failed to create temp dir");
+        let db_path = temp_dir.path().join("telegram_message_routing.db");
+        let backend = LibSqlBackend::new_local(&db_path)
+            .await
+            .expect("failed to create test LibSqlBackend");
+        backend
+            .run_migrations()
+            .await
+            .expect("failed to run migrations");
+        let db: Arc<dyn Database> = Arc::new(backend);
+
+        let skills_dir = temp_dir.path().join("skills");
+        let installed_skills_dir = temp_dir.path().join("installed_skills");
+        let _ = std::fs::create_dir_all(&skills_dir);
+        let _ = std::fs::create_dir_all(&installed_skills_dir);
+        let mut config = Config::for_testing(db_path, skills_dir, installed_skills_dir);
+        config.agent.auto_approve_tools = true;
+
+        let session = Arc::new(SessionManager::new(SessionConfig::default()));
+        let log_broadcaster = Arc::new(LogBroadcaster::new());
+        let llm: Arc<dyn LlmProvider> = Arc::new(TraceLlm::from_trace(trace));
+
+        let mut builder = AppBuilder::new(
+            config,
+            AppBuilderFlags::default(),
+            None,
+            session,
+            log_broadcaster,
+        );
+        builder.with_database(Arc::clone(&db));
+        builder.with_llm(llm);
+
+        let mut components = builder
+            .build_all()
+            .await
+            .expect("AppBuilder::build_all() failed");
+        components.config.agent.auto_approve_tools = true;
+        components.config.agent.allow_local_tools = true;
+
+        let deps = AgentDeps {
+            owner_id: components.config.owner_id.clone(),
+            store: components.db.clone(),
+            llm: components.llm.clone(),
+            cheap_llm: components.cheap_llm.clone(),
+            safety: components.safety.clone(),
+            tools: components.tools.clone(),
+            workspace: components.workspace.clone(),
+            extension_manager: components.extension_manager.clone(),
+            skill_registry: components.skill_registry.clone(),
+            skill_catalog: components.skill_catalog.clone(),
+            skills_config: components.config.skills.clone(),
+            hooks: components.hooks.clone(),
+            cost_guard: components.cost_guard.clone(),
+            sse_tx: None,
+            http_interceptor: None,
+            transcription: None,
+            document_extraction: None,
+        };
+
+        let gateway = Arc::new(TestChannel::new());
+        let gateway_handle = TestChannelHandle::new(Arc::clone(&gateway));
+        let (telegram_channel, telegram_captures) = RecordingTelegramChannel::new();
+
+        let channel_manager = ChannelManager::new();
+        channel_manager.add(Box::new(gateway_handle)).await;
+        channel_manager.add(Box::new(telegram_channel)).await;
+        let channels = Arc::new(channel_manager);
+
+        deps.tools
+            .register_message_tools(Arc::clone(&channels), deps.extension_manager.clone())
+            .await;
+
+        let agent = Agent::new(
+            components.config.agent.clone(),
+            deps,
+            channels,
+            None,
+            None,
+            None,
+            Some(Arc::clone(&components.context_manager)),
+            None,
+        );
+
+        let agent_handle = tokio::spawn(async move {
+            if let Err(err) = agent.run().await {
+                eprintln!("[telegram routing e2e] Agent exited with error: {err}");
+            }
+        });
+
+        if let Some(rx) = gateway.take_ready_rx().await {
+            let _ = tokio::time::timeout(Duration::from_secs(5), rx).await;
+        }
+
+        Harness {
+            gateway,
+            telegram_captures,
+            db,
+            owner_id: components.config.owner_id.clone(),
+            _temp_dir: temp_dir,
+            agent_handle: Some(agent_handle),
+        }
+    }
+
+    fn single_message_trace(arguments: serde_json::Value, final_text: &str) -> LlmTrace {
+        LlmTrace::single_turn(
+            "telegram-message-routing",
+            "send a reminder",
+            vec![
+                TraceStep {
+                    request_hint: None,
+                    response: TraceResponse::ToolCalls {
+                        tool_calls: vec![TraceToolCall {
+                            id: "call_message_1".to_string(),
+                            name: "message".to_string(),
+                            arguments,
+                        }],
+                        input_tokens: 32,
+                        output_tokens: 12,
+                    },
+                    expected_tool_results: Vec::new(),
+                },
+                TraceStep {
+                    request_hint: None,
+                    response: TraceResponse::Text {
+                        content: final_text.to_string(),
+                        input_tokens: 24,
+                        output_tokens: 8,
+                    },
+                    expected_tool_results: Vec::new(),
+                },
+            ],
+        )
+    }
+
+    #[tokio::test]
+    async fn telegram_message_tool_uses_bound_owner_target_when_target_omitted() {
+        let harness = build_harness(single_message_trace(
+            serde_json::json!({
+                "content": "Walk Conan",
+                "channel": "telegram",
+            }),
+            "Sent on Telegram.",
+        ))
+        .await;
+
+        harness.store_telegram_owner_binding(424242).await;
+
+        harness
+            .gateway
+            .send_message("remind me to walk conan")
+            .await;
+        let responses = harness
+            .gateway
+            .wait_for_responses(1, Duration::from_secs(10))
+            .await;
+        assert!(
+            responses
+                .iter()
+                .any(|response| response.content.contains("Sent on Telegram")),
+            "expected assistant confirmation, got: {:?}",
+            responses
+                .iter()
+                .map(|response| &response.content)
+                .collect::<Vec<_>>()
+        );
+
+        let broadcasts = harness
+            .wait_for_telegram_broadcasts(1, Duration::from_secs(10))
+            .await;
+        assert_eq!(
+            broadcasts.len(),
+            1,
+            "expected exactly one telegram broadcast"
+        );
+        assert_eq!(broadcasts[0].0, "424242");
+        assert_eq!(broadcasts[0].1.content, "Walk Conan");
+    }
+
+    #[tokio::test]
+    async fn telegram_message_tool_prefers_explicit_target_over_bound_owner_target() {
+        let harness = build_harness(single_message_trace(
+            serde_json::json!({
+                "content": "Walk Conan",
+                "channel": "telegram",
+                "target": "999999",
+            }),
+            "Sent on Telegram.",
+        ))
+        .await;
+
+        harness.store_telegram_owner_binding(424242).await;
+
+        harness.gateway.send_message("send the reminder").await;
+        let _ = harness
+            .gateway
+            .wait_for_responses(1, Duration::from_secs(10))
+            .await;
+
+        let broadcasts = harness
+            .wait_for_telegram_broadcasts(1, Duration::from_secs(10))
+            .await;
+        assert_eq!(
+            broadcasts.len(),
+            1,
+            "expected exactly one telegram broadcast"
+        );
+        assert_eq!(broadcasts[0].0, "999999");
+        assert_eq!(broadcasts[0].1.content, "Walk Conan");
+    }
+}

tests/fixtures/llm_traces/advanced/routine_event_any_channel.json

@@ -34,14 +34,6 @@
         "output_tokens": 18
       }
     },
-    {
-      "response": {
-        "type": "text",
-        "content": "I saw the Telegram message.",
-        "input_tokens": 90,
-        "output_tokens": 12
-      }
-    },
     {
       "response": {
         "type": "text",

tests/fixtures/llm_traces/advanced/routine_event_telegram.json

@@ -35,14 +35,6 @@
         "output_tokens": 20
       }
     },
-    {
-      "response": {
-        "type": "text",
-        "content": "I saw the Telegram message.",
-        "input_tokens": 90,
-        "output_tokens": 12
-      }
-    },
     {
       "response": {
         "type": "text",

tests/support/gateway_workflow_harness.rs

@@ -239,6 +239,7 @@ impl GatewayWorkflowHarness {
         let mut agent = Agent::new(
             components.config.agent.clone(),
             AgentDeps {
+                owner_id: components.config.owner_id.clone(),
                 store: components.db,
                 llm: components.llm,
                 cheap_llm: components.cheap_llm,

tests/support/test_rig.rs

@@ -612,6 +612,7 @@ impl TestRigBuilder {
 
         // 7. Construct AgentDeps from AppComponents (mirrors main.rs).
         let deps = AgentDeps {
+            owner_id: components.config.owner_id.clone(),
             store: components.db,
             llm: components.llm,
             cheap_llm: components.cheap_llm,
@@ -652,7 +653,7 @@ impl TestRigBuilder {
 
         // 7b. Register message tool so routines can send messages to channels.
         deps.tools
-            .register_message_tools(Arc::clone(&channels))
+            .register_message_tools(Arc::clone(&channels), deps.extension_manager.clone())
             .await;
 
         // 8. Create Agent.

tests/telegram_auth_integration.rs

@@ -6,17 +6,24 @@
 //! 1. When owner_id is null and dm_policy is "allowlist", unauthorized users in
 //!    group chats are dropped even if they @mention the bot
 //! 2. When owner_id is null and dm_policy is "open", all users can interact
-//! 3. When owner_id is set, only that user can interact
+//! 3. When owner_id is set, the owner gets instance-global access while
+//!    non-owner senders remain channel-scoped guests subject to authorization
 //! 4. Authorization works correctly for both private and group chats
 
 use std::collections::HashMap;
 use std::sync::Arc;
 
+#[cfg(feature = "integration")]
+use futures::StreamExt;
+#[cfg(feature = "integration")]
+use ironclaw::channels::Channel;
 use ironclaw::channels::wasm::{
     ChannelCapabilities, PreparedChannelModule, WasmChannel, WasmChannelRuntime,
     WasmChannelRuntimeConfig,
 };
 use ironclaw::pairing::PairingStore;
+#[cfg(feature = "integration")]
+use tokio::time::{Duration, timeout};
 
 /// Skip the test if the Telegram WASM module hasn't been built.
 /// In CI (detected via the `CI` env var), panic instead of skipping so a
@@ -40,8 +47,31 @@ macro_rules! require_telegram_wasm {
 
 /// Path to the built Telegram WASM module
 fn telegram_wasm_path() -> std::path::PathBuf {
-    std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
-        .join("channels-src/telegram/target/wasm32-wasip2/release/telegram_channel.wasm")
+    let local = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+        .join("channels-src/telegram/target/wasm32-wasip2/release/telegram_channel.wasm");
+    if local.exists() {
+        return local;
+    }
+
+    if let Ok(output) = std::process::Command::new("git")
+        .args(["worktree", "list", "--porcelain"])
+        .output()
+        && output.status.success()
+    {
+        let stdout = String::from_utf8_lossy(&output.stdout);
+        for line in stdout.lines() {
+            if let Some(path) = line.strip_prefix("worktree ") {
+                let candidate = std::path::PathBuf::from(path).join(
+                    "channels-src/telegram/target/wasm32-wasip2/release/telegram_channel.wasm",
+                );
+                if candidate.exists() {
+                    return candidate;
+                }
+            }
+        }
+    }
+
+    local
 }
 
 /// Create a test runtime for WASM channel operations.
@@ -74,6 +104,14 @@ async fn load_telegram_module(
 async fn create_telegram_channel(
     runtime: Arc<WasmChannelRuntime>,
     config_json: &str,
+) -> WasmChannel {
+    create_telegram_channel_with_store(runtime, config_json, Arc::new(PairingStore::new())).await
+}
+
+async fn create_telegram_channel_with_store(
+    runtime: Arc<WasmChannelRuntime>,
+    config_json: &str,
+    pairing_store: Arc<PairingStore>,
 ) -> WasmChannel {
     let module = load_telegram_module(&runtime)
         .await
@@ -83,8 +121,9 @@ async fn create_telegram_channel(
         runtime,
         module,
         ChannelCapabilities::for_channel("telegram").with_path("/webhook/telegram"),
+        "default",
         config_json.to_string(),
-        Arc::new(PairingStore::new()),
+        pairing_store,
         None,
     )
 }
@@ -222,31 +261,29 @@ async fn test_group_message_authorized_user_allowed() {
 }
 
 #[tokio::test]
-async fn test_group_message_with_owner_id_set() {
+async fn test_private_message_with_owner_id_set_uses_guest_pairing_flow() {
     require_telegram_wasm!();
     let runtime = create_test_runtime();
+    let dir = tempfile::tempdir().expect("tempdir");
+    let pairing_store = Arc::new(PairingStore::with_base_dir(dir.path().to_path_buf()));
 
-    // Config: owner_id=123 (only this user can interact)
+    // Config: owner_id=123, non-owner private DMs should enter the guest
+    // pairing flow instead of being rejected solely for not being the owner.
     let config = serde_json::json!({
-        "bot_username": "test_bot",
+        "bot_username": null,
         "owner_id": 123,
-        "dm_policy": "allowlist",
-        "allow_from": ["anyone"],  // ignored when owner_id is set
+        "dm_policy": "pairing",
+        "allow_from": [],
         "respond_to_all_group_messages": false
     })
     .to_string();
 
-    let channel = create_telegram_channel(runtime, &config).await;
+    let channel = create_telegram_channel_with_store(runtime, &config, pairing_store.clone()).await;
 
-    // Message from different user (should be dropped)
+    // Non-owner private message should produce a pairing request.
     let update = build_telegram_update(
-        3,
-        102,
-        -123456789,
-        "group",
-        999, // Not the owner
-        "Other",
-        "Hey @test_bot hello",
+        3, 102, 999, "private", 999, // Not the owner
+        "Other", "hello",
     );
 
     let response = channel
@@ -263,8 +300,68 @@ async fn test_group_message_with_owner_id_set() {
 
     assert_eq!(response.status, 200);
 
-    // REGRESSION TEST: Non-owner messages are dropped when owner_id is set
-    // This behavior is consistent and not affected by the fix
+    let pending = pairing_store
+        .list_pending("telegram")
+        .expect("pairing store should be readable");
+    assert_eq!(pending.len(), 1);
+    assert_eq!(pending[0].id, "999");
+}
+
+#[tokio::test]
+#[cfg(feature = "integration")]
+async fn test_private_messages_use_chat_id_as_thread_scope() {
+    require_telegram_wasm!();
+    let runtime = create_test_runtime();
+
+    let config = serde_json::json!({
+        "bot_username": null,
+        "owner_id": null,
+        "dm_policy": "open",
+        "allow_from": [],
+        "respond_to_all_group_messages": false
+    })
+    .to_string();
+
+    let channel = create_telegram_channel(runtime, &config).await;
+    let mut stream = channel
+        .start_message_stream_for_test()
+        .await
+        .expect("Failed to bootstrap test message stream");
+
+    for (update_id, message_id, text) in [(6, 105, "first"), (7, 106, "second")] {
+        let update = build_telegram_update(
+            update_id,
+            message_id,
+            999,
+            "private",
+            999,
+            "ThreadUser",
+            text,
+        );
+
+        let response = channel
+            .call_on_http_request(
+                "POST",
+                "/webhook/telegram",
+                &HashMap::new(),
+                &HashMap::new(),
+                &update,
+                true,
+            )
+            .await
+            .expect("HTTP callback failed");
+
+        assert_eq!(response.status, 200);
+
+        let msg = timeout(Duration::from_secs(1), stream.next())
+            .await
+            .expect("message should arrive")
+            .expect("stream should yield a message");
+        assert_eq!(msg.thread_id.as_deref(), Some("999"));
+        assert_eq!(msg.conversation_scope(), Some("999"));
+    }
+
+    channel.shutdown().await.expect("Shutdown failed");
 }
 
 #[tokio::test]

tests/wasm_channel_integration.rs

@@ -43,6 +43,7 @@ fn create_test_channel(
         runtime,
         prepared,
         capabilities,
+        "default",
         "{}".to_string(),
         Arc::new(PairingStore::new()),
         None,