feat(agentex): forward user session cookie to agent pods via acp headers

agentex/src/domain/delegation_headers.py

@@ -1,48 +1,90 @@
 """
 Outbound runtime-delegation headers for ACP calls to agent pods (v1).
 
-Forwards the validated user API key on a dedicated header so agents can call
-downstream APIs as the user. Agent identity for SGP will eventually be a claim
-on a pod-minted delegation token (OBO), not a separate header from agentex.
+After auth, agentex may attach x-acting-user-* headers so agent pods can call
+downstream APIs as the user. Values are minimal (never the full browser Cookie).
+
+Session cookies: only configured names are forwarded on x-acting-user-cookie.
+Default name is _identityJwt. Override with AGENTEX_DELEGATION_SESSION_COOKIE_NAMES
+(comma-separated). Set to empty string to disable cookie delegation.
 """
 
+import os
+from http.cookies import CookieError, SimpleCookie
 from typing import Any
 
 HEADER_ACTING_USER_API_KEY = "x-acting-user-api-key"
+HEADER_ACTING_USER_COOKIE = "x-acting-user-cookie"
+HEADER_COOKIE = "cookie"
 HEADER_SELECTED_ACCOUNT_ID = "x-selected-account-id"
 HEADER_USER_API_KEY = "x-api-key"
 
+ENV_SESSION_COOKIE_NAMES = "AGENTEX_DELEGATION_SESSION_COOKIE_NAMES"
+DEFAULT_SESSION_COOKIE_NAMES = ("_identityJwt",)
+
 
 def _normalize_headers(headers: dict[str, str] | None) -> dict[str, str]:
     if not headers:
         return {}
     return {k.lower(): v for k, v in headers.items()}
 
 
+def session_cookie_names_to_forward() -> tuple[str, ...]:
+    """Cookie names to include in x-acting-user-cookie. Unset env uses the default."""
+    raw = os.environ.get(ENV_SESSION_COOKIE_NAMES)
+    if raw is None:
+        return DEFAULT_SESSION_COOKIE_NAMES
+    raw = raw.strip()
+    if not raw:
+        return ()
+    return tuple(name.strip() for name in raw.split(",") if name.strip())
+
+
+def _minimal_session_cookie(cookie_header: str, names: tuple[str, ...]) -> str | None:
+    if not names:
+        return None
+
+    jar = SimpleCookie()
+    try:
+        jar.load(cookie_header)
+    except CookieError:
+        return None
+
+    pairs: list[str] = []
+    for name in names:
+        morsel = jar.get(name)
+        if morsel is not None:
+            pairs.append(f"{name}={morsel.value}")
+
+    return "; ".join(pairs) if pairs else None
+
+
 def build_delegation_headers(
     principal: Any,
     inbound_headers: dict[str, str] | None,
     *,
     agent_identity: str | None = None,
 ) -> dict[str, str]:
-    """
-    Outbound ACP headers so the agent can act on behalf of the authenticated user.
-
-    Requires a validated user principal from auth; reads x-api-key from the
-    inbound request (already checked during auth). Skips delegation when the
-    request is authenticated as the agent itself (agent_identity set).
-    """
+    """Build outbound acting headers for an authenticated user invocation."""
     if agent_identity or principal is None:
         return {}
 
     normalized = _normalize_headers(inbound_headers)
     api_key = normalized.get(HEADER_USER_API_KEY)
-    if not api_key:
-        return {}
+    cookie_header = normalized.get(HEADER_COOKIE)
 
-    result = {
-        HEADER_ACTING_USER_API_KEY: api_key,
-    }
+    result: dict[str, str] = {}
+    if api_key:
+        result[HEADER_ACTING_USER_API_KEY] = api_key
+    elif cookie_header:
+        session_cookie = _minimal_session_cookie(
+            cookie_header, session_cookie_names_to_forward()
+        )
+        if not session_cookie:
+            return {}
+        result[HEADER_ACTING_USER_COOKIE] = session_cookie
+    else:
+        return {}
 
     account_id = normalized.get(HEADER_SELECTED_ACCOUNT_ID)
     if account_id:

agentex/src/domain/services/agent_acp_service.py

@@ -76,7 +76,9 @@
         "x-api-key",
         "x-agent-api-key",
         "x-acting-user-api-key",
+        "x-acting-user-cookie",
         "x-acting-as-agent",
+        "x-selected-account-id",
     }
 )
 
@@ -88,7 +90,8 @@ def filter_request_headers(headers: dict[str, str] | None) -> dict[str, str]:
     Security filtering rules:
     1. Allow only x-* prefixed headers (allowlist approach)
     2. Block hop-by-hop headers (connection, keep-alive, etc.)
-    3. Block sensitive headers (credentials, acting delegation, x-agent-api-key)
+    3. Block sensitive headers (credentials, acting delegation, x-agent-api-key,
+       x-selected-account-id)
 
     Args:
         headers: Raw request headers from client

agentex/src/utils/request_utils.py

@@ -8,6 +8,7 @@
 REQUEST_KEY_REGEXP_BLACKLIST = [
     r"api_key",
     r"api-key",
+    r"cookie",
     r"password",
     r"secret",
     r"token",

agentex/tests/unit/domain/test_delegation_headers.py

@@ -0,0 +1,112 @@
+"""Unit tests for runtime delegation header construction."""
+
+import pytest
+from src.domain.delegation_headers import (
+    ENV_SESSION_COOKIE_NAMES,
+    HEADER_ACTING_USER_API_KEY,
+    HEADER_ACTING_USER_COOKIE,
+    build_delegation_headers,
+    session_cookie_names_to_forward,
+)
+
+
+def _user_principal():
+    return type("Principal", (), {"user_id": "user-1", "account_id": "acct-1"})()
+
+
+class TestSessionCookieNames:
+    def test_default_when_env_unset(self, monkeypatch: pytest.MonkeyPatch):
+        monkeypatch.delenv(ENV_SESSION_COOKIE_NAMES, raising=False)
+        assert session_cookie_names_to_forward() == ("_identityJwt",)
+
+    def test_empty_env_disables(self, monkeypatch: pytest.MonkeyPatch):
+        monkeypatch.setenv(ENV_SESSION_COOKIE_NAMES, "")
+        assert session_cookie_names_to_forward() == ()
+
+    def test_override_env(self, monkeypatch: pytest.MonkeyPatch):
+        monkeypatch.setenv(ENV_SESSION_COOKIE_NAMES, "session, other")
+        assert session_cookie_names_to_forward() == ("session", "other")
+
+
+class TestBuildDelegationHeaders:
+    def test_api_key_delegation(self):
+        headers = build_delegation_headers(
+            _user_principal(),
+            {
+                "x-api-key": "user-key",
+                "x-selected-account-id": "acct-1",
+            },
+        )
+        assert headers == {
+            HEADER_ACTING_USER_API_KEY: "user-key",
+            "x-selected-account-id": "acct-1",
+        }
+
+    def test_cookie_delegation_forwards_only_configured_names(self):
+        cookie = "_identityJwt=eyJ.test; csrf=secret"
+        headers = build_delegation_headers(
+            _user_principal(),
+            {"Cookie": cookie, "x-selected-account-id": "acct-2"},
+        )
+        assert headers == {
+            HEADER_ACTING_USER_COOKIE: "_identityJwt=eyJ.test",
+            "x-selected-account-id": "acct-2",
+        }
+
+    def test_cookie_delegation_respects_custom_names(
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
+        monkeypatch.setenv(ENV_SESSION_COOKIE_NAMES, "session")
+        headers = build_delegation_headers(
+            _user_principal(),
+            {"cookie": "session=abc; _identityJwt=ignored"},
+        )
+        assert headers == {HEADER_ACTING_USER_COOKIE: "session=abc"}
+
+    def test_cookie_delegation_off_when_env_empty(
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
+        monkeypatch.setenv(ENV_SESSION_COOKIE_NAMES, "")
+        assert (
+            build_delegation_headers(
+                _user_principal(),
+                {"cookie": "_identityJwt=jwt"},
+            )
+            == {}
+        )
+
+    def test_cookie_delegation_skips_when_no_matching_cookie(self):
+        assert (
+            build_delegation_headers(
+                _user_principal(),
+                {"cookie": "csrf=secret"},
+            )
+            == {}
+        )
+
+    def test_prefers_api_key_when_both_present(self):
+        headers = build_delegation_headers(
+            _user_principal(),
+            {
+                "x-api-key": "user-key",
+                "cookie": "_identityJwt=jwt",
+            },
+        )
+        assert HEADER_ACTING_USER_API_KEY in headers
+        assert HEADER_ACTING_USER_COOKIE not in headers
+
+    def test_skips_when_no_principal(self):
+        assert build_delegation_headers(None, {"x-api-key": "k"}) == {}
+
+    def test_skips_when_agent_identity(self):
+        assert (
+            build_delegation_headers(
+                _user_principal(),
+                {"x-api-key": "k"},
+                agent_identity="agent-1",
+            )
+            == {}
+        )
+
+    def test_empty_when_no_credential(self):
+        assert build_delegation_headers(_user_principal(), {"x-trace-id": "t"}) == {}

agentex/tests/unit/services/test_agent_acp_service.py

@@ -338,6 +338,61 @@ async def test_send_event_delegation_not_raw_api_key_passthrough(
         assert http_headers["x-trace-id"] == "trace-456"
         assert "x-api-key" not in http_headers
 
+    async def test_send_message_includes_cookie_delegation_headers(
+        self,
+        agent_acp_service,
+        mock_http_gateway,
+        mock_request,
+        agent_repository,
+        sample_agent,
+        sample_task,
+        sample_text_content,
+    ):
+        """JWT/session users get x-acting-user-cookie on outbound ACP calls."""
+        await create_or_get_agent(agent_repository, sample_agent)
+
+        session_cookie = "_identityJwt=eyJhbGciOiJIUzI1NiJ9.test"
+        mock_request.state.principal_context = type(
+            "Principal",
+            (),
+            {"user_id": "user-1", "account_id": "acct-1"},
+        )()
+        mock_request.state.agent_identity = None
+        mock_request.headers = {
+            "cookie": f"{session_cookie}; csrf=must-not-forward",
+            "x-selected-account-id": "acct-1",
+        }
+
+        from src.domain.entities.agents_rpc import AgentRPCMethod
+
+        expected_request_id = f"{AgentRPCMethod.MESSAGE_SEND}-{sample_task.id}"
+        mock_http_gateway.async_call.return_value = {
+            "jsonrpc": "2.0",
+            "result": {
+                "type": "text",
+                "author": "agent",
+                "style": "static",
+                "format": "plain",
+                "content": "ok",
+                "attachments": None,
+            },
+            "id": expected_request_id,
+        }
+
+        await agent_acp_service.send_message(
+            agent=sample_agent,
+            task=sample_task,
+            content=sample_text_content,
+            acp_url="http://test-acp.example.com",
+        )
+
+        http_headers = mock_http_gateway.async_call.call_args[1]["default_headers"]
+        assert http_headers["x-acting-user-cookie"] == session_cookie
+        assert "csrf" not in http_headers["x-acting-user-cookie"]
+        assert http_headers["x-selected-account-id"] == "acct-1"
+        assert "cookie" not in http_headers
+        assert "x-acting-user-api-key" not in http_headers
+
     async def test_get_headers_server_request_id_wins_over_passthrough(
         self,
         agent_acp_service,
@@ -1012,7 +1067,9 @@ def test_blocks_user_api_key_and_acting_headers(self):
             {
                 "x-api-key": "user-key",
                 "x-acting-user-api-key": "spoof",
+                "x-acting-user-cookie": "spoof-cookie",
                 "x-acting-as-agent": "spoof-agent",
+                "x-selected-account-id": "spoof-acct",
                 "x-trace-id": "trace-1",
                 "authorization": "Bearer x",
             }