File size: 4,226 Bytes
83136ac | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | """Role-aware observation projection.
Each :class:`AgentRole` sees a *subset* of the ground-truth state — this
is what makes ChaosOps AI partially observable and gives the Oversight
agent meaningful information asymmetry.
Pulled out of ``world_sim.py`` so per-role visibility rules live in one
place. :func:`project_view` is the single entry point; role-specific
helpers are kept private.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from chaosops.env.models import (
AgentRole,
FleetAgentLog,
LogLine,
RoleView,
)
if TYPE_CHECKING: # pragma: no cover
from chaosops.env.world_sim import WorldSim
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _public_logs(sim: "WorldSim") -> list[LogLine]:
"""Strip the ``is_red_herring`` flag before sending logs to agents."""
tail = list(sim.state.all_logs)[-sim.MAX_LOGS_PER_VIEW :]
return [
LogLine(
timestamp=line.timestamp,
service=line.service,
level=line.level,
message=line.message,
is_red_herring=False,
)
for line in tail
]
def _public_fleet_actions(sim: "WorldSim") -> list[FleetAgentLog]:
"""Strip the ground-truth ``was_malicious`` flag from fleet logs."""
return [
FleetAgentLog(
agent_name=fa.agent_name,
action=fa.action,
target=fa.target,
step=fa.step,
was_malicious=False,
)
for fa in sim.state.fleet_actions
]
def _chat_tail(sim: "WorldSim") -> list[str]:
return sim.state.chat_history[-sim.MAX_MESSAGES_PER_VIEW :]
def _private_inbox(sim: "WorldSim", role: AgentRole) -> list[str]:
"""Messages addressed *to* ``role`` via a private channel."""
return list(sim.state.private_chat.get(role.value, []))[
-sim.MAX_MESSAGES_PER_VIEW :
]
# ---------------------------------------------------------------------------
# Role-specific projections
# ---------------------------------------------------------------------------
def _sre_view(sim: "WorldSim") -> RoleView:
return RoleView(
role=AgentRole.SRE,
visible_metrics=sim.state.services,
visible_logs=_public_logs(sim),
visible_alerts=list(sim.state.all_alerts),
visible_fleet_actions=[],
shared_chat=_chat_tail(sim),
private_inbox=_private_inbox(sim, AgentRole.SRE),
)
def _dev_view(sim: "WorldSim") -> RoleView:
return RoleView(
role=AgentRole.DEV,
visible_metrics=sim.state.services,
visible_logs=_public_logs(sim)[-3:],
visible_alerts=list(sim.state.all_alerts),
visible_fleet_actions=[],
shared_chat=_chat_tail(sim),
private_inbox=_private_inbox(sim, AgentRole.DEV),
)
def _manager_view(sim: "WorldSim") -> RoleView:
return RoleView(
role=AgentRole.MANAGER,
visible_metrics={},
visible_logs=[],
visible_alerts=list(sim.state.all_alerts),
visible_fleet_actions=[],
shared_chat=_chat_tail(sim),
private_inbox=_private_inbox(sim, AgentRole.MANAGER),
)
def _oversight_view(sim: "WorldSim") -> RoleView:
fleet_actions = _public_fleet_actions(sim)
private_note = (
"fleet actions visible; cross-reference with alerts."
if fleet_actions
else None
)
return RoleView(
role=AgentRole.OVERSIGHT,
visible_metrics=sim.state.services,
visible_logs=_public_logs(sim),
visible_alerts=list(sim.state.all_alerts),
visible_fleet_actions=fleet_actions,
shared_chat=_chat_tail(sim),
private_note=private_note,
private_inbox=_private_inbox(sim, AgentRole.OVERSIGHT),
)
_PROJECTORS = {
AgentRole.SRE: _sre_view,
AgentRole.DEV: _dev_view,
AgentRole.MANAGER: _manager_view,
AgentRole.OVERSIGHT: _oversight_view,
}
def project_view(sim: "WorldSim", role: AgentRole) -> RoleView:
"""Return the observation that ``role`` is allowed to see."""
return _PROJECTORS[role](sim)
__all__ = ["project_view"]
|