Spaces:
Sleeping
Sleeping
File size: 5,853 Bytes
796da7c | 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 147 148 149 150 | """
PERMANENCE — latent (background) world dynamics.
Applied AFTER every step, BEFORE the success/catastrophe check. These are
the "things that happen while you're deciding" — the world does not sit
still. Combined with the deterministic action consequences, this turns
the environment from "response to agent" into "live system where decisions
also have a ticking cost."
All dynamics are deterministic given the (scenario_id, step) pair, so
episodes remain reproducible when rerun with the same seed. No torch /
numpy — we use Python's `random` seeded from the scenario id for speed
and portability.
Three dynamics families:
1. Trust decay — employee trust score drifts toward their "natural
baseline" (a function of role) unless actively maintained. Mimics
real-world relationship erosion when a leader never checks in.
2. Deadline pressure — projects under time pressure accumulate
momentum loss. Momentum below 0.2 triggers the project becoming
a blocker for certain actions.
3. Board expectation drift — if the public record grows fast without
follow-through, expectation level climbs (board has heard your
plans and will judge you on them).
These dynamics are lightweight and additive. They give the agent a real
reason to time its actions carefully — waiting has a cost.
"""
from __future__ import annotations
import hashlib
import random
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .state import WorldState
# ---------------------------------------------------------------------------
# Tuning knobs
# ---------------------------------------------------------------------------
TRUST_DECAY_PER_STEP = 0.012 # trust drifts ~1.2% toward baseline per step
TRUST_MAINTENANCE_RADIUS = 2 # recent action with employee resets decay timer
DEADLINE_MOMENTUM_DECAY = 0.02 # projects with >0.7 pressure lose 2% momentum / step
BOARD_EXPECTATION_DRIFT_PER_COMMITMENT = 0.015 # per unanswered public record entry
# Role-based "natural" trust baseline — drift is towards this value
ROLE_TRUST_BASELINE = {
"report_owner": 0.60,
"reviewer": 0.55,
"distributor": 0.55,
"team_lead": 0.58,
"engineer": 0.52,
"manager": 0.65,
"product_lead": 0.62,
"qa_lead": 0.60,
"sales_ops": 0.55,
"communications": 0.60,
"legal": 0.70,
"executive": 0.62,
"contract_owner": 0.62,
"legal_counsel": 0.72,
"client_manager": 0.58,
"sre_lead": 0.65,
"platform_engineer": 0.60,
"incident_commander": 0.65,
"database_administrator": 0.66,
"backend_engineer": 0.58,
"sre": 0.65,
}
STOCHASTIC_NOISE_MAGNITUDE = 0.005 # +/- up to 0.5% noise per step on trust scores
def _seeded_rng(scenario_id: str, step: int) -> random.Random:
"""Deterministic RNG keyed on (scenario, step) — same seed → same noise."""
digest = hashlib.sha256(f"{scenario_id}:{step}".encode("utf-8")).hexdigest()
return random.Random(int(digest[:16], 16))
def _recent_interaction_set(world_state: "WorldState") -> set[str]:
"""Set of employee_ids touched within TRUST_MAINTENANCE_RADIUS steps."""
touched: set[str] = set()
recent = world_state.action_history[-TRUST_MAINTENANCE_RADIUS:]
for record in recent:
for key, value in record.parameters.items():
if "employee" in key or "recipient" in key or "participant" in key:
if isinstance(value, str):
for piece in value.split(","):
piece = piece.strip()
if piece.startswith("emp_"):
touched.add(piece)
return touched
def apply_latent_dynamics(world_state: "WorldState", step_index: int) -> None:
"""
Apply all latent dynamics in place. Called from PermanenceEnv.step()
AFTER the action's own consequences are applied.
"""
rng = _seeded_rng(world_state.scenario_id, step_index)
touched = _recent_interaction_set(world_state)
# 1. Trust decay + stochastic noise
for employee_id, employee in world_state.employees.items():
if employee.availability != "active":
continue
baseline = ROLE_TRUST_BASELINE.get(employee.role, 0.55)
current = employee.trust_score
# Drift toward baseline when not recently touched
if employee_id not in touched:
drift = TRUST_DECAY_PER_STEP * (baseline - current)
current = current + drift
# Small zero-mean noise
current += rng.uniform(-STOCHASTIC_NOISE_MAGNITUDE, STOCHASTIC_NOISE_MAGNITUDE)
employee.trust_score = max(0.0, min(1.0, current))
# 2. Deadline pressure erodes momentum on high-pressure projects
for project in world_state.projects.values():
if project.deadline_pressure > 0.7 and project.status == "active":
loss = DEADLINE_MOMENTUM_DECAY * project.deadline_pressure
project.momentum = max(0.0, project.momentum - loss)
# 3. Board expectation drifts with public commitments that haven't been
# addressed by a follow-up "RESOLUTION" or "POSTMORTEM" record.
commitments = [
entry
for entry in world_state.external.public_record
if entry.startswith("COMMITMENT:") or entry.startswith("LAUNCH:") or entry.startswith("PUBLIC_STATEMENT:")
]
resolutions = [
entry
for entry in world_state.external.public_record
if entry.startswith("RESOLUTION:") or entry.startswith("POSTMORTEM:") or entry.startswith("ROLLBACK:")
]
unanswered = max(0, len(commitments) - len(resolutions))
if unanswered > 0:
drift = BOARD_EXPECTATION_DRIFT_PER_COMMITMENT * unanswered
world_state.external.board_expectation_level = min(
1.0, world_state.external.board_expectation_level + drift
)
|