Cross-Platform Session Runtime integration

docs/cross-platform-session-integration.md

@@ -0,0 +1,14 @@
+# Cross-Platform Session 통합 브랜치
+
+- 기준: `main` (`35b7ec422ae85234c98e12c89794b74fdcc658cf`)
+- 상태: Draft. 서버 계약과 격리·회귀 검증이 끝날 때까지 `main`에 병합하지 않는다.
+
+## Runtime 작업 경계
+
+1. Runtime 내부 실행 `session_id`와 서버가 발급하는 Agent Session ID를 구분하고, 실행·재시작·도구 호출 경계에서 사용자 및 Agent Session 문맥을 명시적으로 전달한다.
+2. 개인 credential은 사용자·실행별 참조로만 전달한다. 로그, trace, prompt, 도구 결과와 subprocess 환경으로 원문이 새지 않는지 검증한다.
+3. 도구 실행의 capability 대상, 승인과 취소 신호를 서버 계약에 맞춰 처리하고 다른 사용자·기기 문맥이 섞이지 않도록 격리 테스트를 추가한다.
+
+첫 하위 작업은 검증된 호스트가 제공하는 `CanonicalSessionIdentity`를 `ToolContext`에 보관하고 Stage 10의 호출별 문맥 재구성에서 유지한다. `PipelineState.session_id`는 로컬 실행 ID로 남으며, canonical Agent Session ID와 혼동하지 않는다. 기존 호스트의 기본값은 `None`이다.
+
+이 타입은 전달 용도이며 인증을 수행하지 않는다. Workflow는 Gateway principal과 Agent Session 소유권을 검증한 뒤에만 값을 주입해야 한다. 개인 credential, secret lease, capability 권한 판단과 실제 Workflow 주입은 후속 작업으로 남아 있다.

src/xgen_agent_runtime/session/__init__.py

@@ -4,11 +4,14 @@
 from xgen_agent_runtime.session.manager import SessionManager
 from xgen_agent_runtime.session.freshness import FreshnessPolicy, FreshnessStatus
 from xgen_agent_runtime.session.persistence import FileSessionPersistence
+from xgen_agent_runtime.session.identity import CanonicalSessionIdentity, PlatformType
 
 __all__ = [
     "Session",
     "SessionManager",
     "FreshnessPolicy",
     "FreshnessStatus",
     "FileSessionPersistence",
+    "CanonicalSessionIdentity",
+    "PlatformType",
 ]

src/xgen_agent_runtime/session/identity.py

@@ -0,0 +1,41 @@
+"""Server-issued identity carried through a single runtime execution.
+
+The runtime does not authenticate this identity. A host must construct it only
+from a verified Gateway principal and an authorized Agent Session lookup.
+"""
+
+from dataclasses import dataclass
+from typing import Literal
+from uuid import UUID
+
+
+PlatformType = Literal["web", "desktop", "mobile", "cli", "vscode"]
+_PLATFORMS = frozenset(("web", "desktop", "mobile", "cli", "vscode"))
+
+
+@dataclass(frozen=True, slots=True)
+class CanonicalSessionIdentity:
+    """Account, platform session, device and Agent Session for tool routing.
+
+    ``platform_session_id`` is the Gateway ``sid``. ``agent_session_id`` is
+    the canonical server-issued conversation ID, distinct from the runtime's
+    local ``PipelineState.session_id`` execution identifier.
+    """
+
+    tenant_id: str
+    user_id: int
+    platform_session_id: UUID
+    device_id: UUID
+    agent_session_id: UUID
+    platform_type: PlatformType
+
+    def __post_init__(self) -> None:
+        if not isinstance(self.tenant_id, str) or not self.tenant_id.strip():
+            raise ValueError("tenant_id must be a non-empty string")
+        if type(self.user_id) is not int or self.user_id <= 0:
+            raise ValueError("user_id must be a positive integer")
+        for name in ("platform_session_id", "device_id", "agent_session_id"):
+            if not isinstance(getattr(self, name), UUID):
+                raise ValueError(f"{name} must be a UUID")
+        if not isinstance(self.platform_type, str) or self.platform_type not in _PLATFORMS:
+            raise ValueError("platform_type is unsupported")

src/xgen_agent_runtime/stages/s10_tool/artifact/default/stage.py

@@ -197,6 +197,7 @@ def _state_apply(mutations: Dict[str, Any], tool_name: str) -> Dict[str, Any]:
 
         ctx = ToolContext(
             session_id=state.session_id,
+            canonical_identity=getattr(self._context, "canonical_identity", None),
             working_dir=self._context.working_dir,
             storage_path=self._context.storage_path,
             env_vars=self._context.env_vars,

src/xgen_agent_runtime/tools/base.py

@@ -15,6 +15,8 @@
 from dataclasses import dataclass, field
 from typing import Any, Awaitable, Callable, Dict, List, Optional, Tuple
 
+from xgen_agent_runtime.session.identity import CanonicalSessionIdentity
+
 
 # ─────────────────────────────────────────────────────────────────
 # Capability + Permission primitives (new)
@@ -117,6 +119,8 @@ class ToolContext:
 
     Attributes:
         session_id: Unique session identifier.
+        canonical_identity: Server-issued account and Agent Session identity,
+            supplied only after host-side principal and ownership checks.
         working_dir: Working directory for file operations. Tools should
             resolve relative paths against this directory.
         storage_path: Session-specific storage directory (e.g. for logs,
@@ -199,6 +203,11 @@ class ToolContext:
     # original result is used. ``None`` (default) = no filtering. Set by the
     # host through ``HostServices.tool_result_filter`` (see host/host.py).
     result_filter: Optional[Callable[["Tool", "ToolResult"], Awaitable["ToolResult"]]] = None
+    # Keyword-only so upstream result_filter retains its positional field.
+    # Keep this separate from the local execution ID and model-facing bags.
+    canonical_identity: Optional[CanonicalSessionIdentity] = field(
+        default=None, repr=False, compare=False, kw_only=True
+    )
 
 
 # ─────────────────────────────────────────────────────────────────

tests/unit/test_stage10_canonical_identity.py

@@ -0,0 +1,136 @@
+"""The server-issued identity survives real tool dispatch without leaking to model state."""
+
+from dataclasses import FrozenInstanceError
+from uuid import uuid4
+
+import pytest
+
+from xgen_agent_runtime.core.state import PipelineState
+from xgen_agent_runtime.session import CanonicalSessionIdentity
+from xgen_agent_runtime.stages.s10_tool.artifact.default.stage import ToolStage
+from xgen_agent_runtime.tools.base import Tool, ToolContext, ToolResult
+from xgen_agent_runtime.tools.registry import ToolRegistry
+from xgen_agent_runtime.tools.sandbox import SandboxConfig, ToolSandbox
+
+
+def _identity() -> CanonicalSessionIdentity:
+    return CanonicalSessionIdentity(
+        tenant_id="system",
+        user_id=42,
+        platform_session_id=uuid4(),
+        device_id=uuid4(),
+        agent_session_id=uuid4(),
+        platform_type="web",
+    )
+
+
+class _CaptureTool(Tool):
+    def __init__(self) -> None:
+        self.contexts: list[ToolContext] = []
+
+    @property
+    def name(self) -> str:
+        return "capture_identity"
+
+    @property
+    def description(self) -> str:
+        return "Capture dispatch context"
+
+    @property
+    def input_schema(self) -> dict:
+        return {"type": "object"}
+
+    async def execute(self, input: dict, context: ToolContext) -> ToolResult:
+        self.contexts.append(context)
+        return ToolResult(content="ok")
+
+
+def test_identity_requires_typed_server_ids_and_is_immutable() -> None:
+    identity = _identity()
+    with pytest.raises(FrozenInstanceError):
+        identity.user_id = 99
+    with pytest.raises(ValueError, match="platform_session_id"):
+        CanonicalSessionIdentity(
+            tenant_id="system", user_id=42, platform_session_id="unverified",
+            device_id=uuid4(), agent_session_id=uuid4(), platform_type="web",
+        )
+    with pytest.raises(ValueError, match="user_id"):
+        CanonicalSessionIdentity(
+            tenant_id="system", user_id=True, platform_session_id=uuid4(),
+            device_id=uuid4(), agent_session_id=uuid4(), platform_type="web",
+        )
+
+
+@pytest.mark.asyncio
+async def test_identity_reaches_each_tool_call_without_changing_local_session_or_model_state() -> None:
+    identity = _identity()
+    tool = _CaptureTool()
+    registry = ToolRegistry()
+    registry.register(tool)
+    stage = ToolStage(
+        registry=registry,
+        context=ToolContext(session_id="host-local-id", canonical_identity=identity),
+    )
+    state = PipelineState(session_id="execution-123")
+    state.pending_tool_calls = [
+        {"tool_use_id": f"tu_{index}", "tool_name": tool.name, "tool_input": {}}
+        for index in range(2)
+    ]
+
+    await stage.execute(None, state)
+
+    assert len(tool.contexts) == 2
+    for context in tool.contexts:
+        assert context.canonical_identity is identity
+        assert context.session_id == "execution-123"
+        assert context.metadata == {}
+        assert context.extras == {}
+        assert context.env_vars is None
+    assert identity.agent_session_id.hex not in repr(state.messages)
+    assert identity.platform_session_id.hex not in repr(state.messages)
+    assert "canonical_identity" not in repr(tool.contexts[0])
+
+
+@pytest.mark.asyncio
+async def test_sandbox_environment_enrichment_keeps_identity() -> None:
+    identity = _identity()
+    tool = _CaptureTool()
+    wrapper = ToolSandbox(SandboxConfig(env_vars={"INJECTED": "1"}))
+    context = ToolContext(session_id="execution-123", canonical_identity=identity)
+
+    await wrapper.execute_tool(tool, {}, context)
+
+    assert tool.contexts[0].canonical_identity is identity
+    assert tool.contexts[0].env_vars == {"INJECTED": "1"}
+    assert "canonical_identity" not in tool.contexts[0].env_vars
+    assert ToolContext().canonical_identity is None
+
+
+@pytest.mark.asyncio
+async def test_upstream_result_filter_keeps_the_verified_identity_private() -> None:
+    identity = _identity()
+    tool = _CaptureTool()
+    registry = ToolRegistry()
+    registry.register(tool)
+    observed = []
+
+    async def filter_result(actual_tool, result):
+        observed.append(actual_tool)
+        assert result.content == "ok"
+        return ToolResult(content="filtered")
+
+    stage = ToolStage(
+        registry=registry,
+        context=ToolContext(canonical_identity=identity, result_filter=filter_result),
+    )
+    state = PipelineState(session_id="execution-filter")
+    results = await stage.dispatch_calls(
+        [{"tool_use_id": "filter-1", "tool_name": tool.name, "tool_input": {}}], state
+    )
+
+    assert observed == [tool]
+    assert tool.contexts[0].canonical_identity is identity
+    assert tool.contexts[0].result_filter is filter_result
+    assert "filtered" in str(results)
+    assert identity.agent_session_id.hex not in str(results)
+    assert "canonical_identity" not in repr(tool.contexts[0])