permanence/world/dynamics.py

"""
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
        )