openclaw
eb4a3adb3f
- Add agent context isolation API, RBAC, resolver, audit, dashboard modules - Deploy context isolation policy config - Fix OpenAPI workspace permissions (chown aiman:gitops) - Update ChatGPT actions API with new internals - Add CMD files for group/owner isolation and permission fix - Archive completed command files
83 lines
2.3 KiB
Python
83 lines
2.3 KiB
Python
"""Multi-user agent context resolver.
|
|
|
|
Architecture:
|
|
- shared agent kernel
|
|
- group main agent
|
|
- owner dedicated agent
|
|
- user profile overlay
|
|
- session overlay
|
|
|
|
Personal preferences never enter shared USER.md.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ContextLayer:
|
|
name: str
|
|
priority: int
|
|
data: dict[str, Any]
|
|
|
|
|
|
@dataclass
|
|
class EffectiveContext:
|
|
agent_id: str
|
|
user_id: str
|
|
group_id: str
|
|
layers: list[ContextLayer]
|
|
resolved: dict[str, Any]
|
|
resolved_at: str = field(default_factory=utc_now)
|
|
|
|
def to_dict(self):
|
|
return {
|
|
"agent_id": self.agent_id,
|
|
"user_id": self.user_id,
|
|
"group_id": self.group_id,
|
|
"layers": [asdict(x) for x in self.layers],
|
|
"resolved": self.resolved,
|
|
"resolved_at": self.resolved_at,
|
|
}
|
|
|
|
|
|
class AgentContextResolver:
|
|
"""Resolves context without shared user-state pollution."""
|
|
|
|
def resolve(
|
|
self,
|
|
user_id: str,
|
|
group_id: str,
|
|
session: dict[str, Any] | None = None,
|
|
user_profile: dict[str, Any] | None = None,
|
|
group_policy: dict[str, Any] | None = None,
|
|
agent_default: dict[str, Any] | None = None,
|
|
owner_policy: dict[str, Any] | None = None,
|
|
owner: bool = False,
|
|
) -> EffectiveContext:
|
|
layers = [
|
|
ContextLayer("agent_default", 10, agent_default or {}),
|
|
ContextLayer("group_policy", 30, group_policy or {}),
|
|
ContextLayer("user_profile", 50, user_profile or {}),
|
|
ContextLayer("session", 70, session or {}),
|
|
]
|
|
agent_id = f"group-main-agent:{group_id}"
|
|
if owner:
|
|
agent_id = f"owner-agent:{user_id}"
|
|
layers.append(ContextLayer("owner_policy", 90, owner_policy or {}))
|
|
resolved: dict[str, Any] = {}
|
|
# Higher priority overlays lower priority except safety policies handled upstream.
|
|
for layer in sorted(layers, key=lambda x: x.priority):
|
|
resolved.update(layer.data)
|
|
return EffectiveContext(agent_id, user_id, group_id, layers, resolved)
|
|
|
|
|
|
__all__ = ["ContextLayer", "EffectiveContext", "AgentContextResolver"]
|