Spaces:
Paused
Paused
| """Stateful hill-climbing simulation: multi-objective EM decision exercise. | |
| State vector: system_performance, safety, reliability, cost, velocity, | |
| maintainability, team_morale, customer_impact, risk, schedule (0-100; cost and | |
| risk are "higher is worse"). A seeded scenario applies per-turn pressure; each | |
| turn the user writes a hypothesis, picks an intervention (with probabilistic- | |
| but-seeded effects, delays and second-order consequences), measures the result, | |
| and keeps or rejects the change. The objective is a balanced multi-objective | |
| score with HARD FLOORS (safety below threshold triggers an incident event). | |
| Fully deterministic given the seed and the same choice sequence. | |
| """ | |
| from __future__ import annotations | |
| import random | |
| from typing import Dict, List, Optional | |
| from pydantic import BaseModel, Field | |
| from sensorflow.hillclimb.models import Evidence, Store, get_store, new_id, now_iso | |
| METRICS = ["system_performance", "safety", "reliability", "cost", "velocity", | |
| "maintainability", "team_morale", "customer_impact", "risk", "schedule"] | |
| INVERTED = {"cost", "risk"} # higher is worse | |
| OBJECTIVE_WEIGHTS = { | |
| "system_performance": 0.14, "safety": 0.20, "reliability": 0.10, "cost": 0.06, | |
| "velocity": 0.10, "maintainability": 0.08, "team_morale": 0.12, | |
| "customer_impact": 0.10, "risk": 0.05, "schedule": 0.05, | |
| } | |
| SAFETY_FLOOR = 30.0 | |
| MORALE_FLOOR = 25.0 | |
| class InterventionSpec(BaseModel): | |
| id: str | |
| label: str | |
| description: str | |
| immediate: Dict[str, float] # mean effect; jitter applied per-turn | |
| delayed: List[Dict] = Field(default_factory=list) # {delay: int, effects: {metric: delta}, note} | |
| second_order: str = "" | |
| INTERVENTIONS: List[InterventionSpec] = [ | |
| InterventionSpec( | |
| id="add_regression_gate", label="Add a release regression gate", | |
| description="Block releases whose eval deltas regress safety-critical slices.", | |
| immediate={"safety": 7, "reliability": 4, "velocity": -6, "schedule": -4}, | |
| delayed=[{"delay": 2, "effects": {"system_performance": 5, "customer_impact": 3}, | |
| "note": "Gate catches a bad build before it ships"}], | |
| second_order="Slower releases short-term; fewer production regressions later."), | |
| InterventionSpec( | |
| id="pause_launches", label="Pause launches for stabilization", | |
| description="Freeze feature launches for a sprint; all hands on regressions.", | |
| immediate={"safety": 10, "reliability": 6, "schedule": -10, "customer_impact": -6, "risk": -8}, | |
| delayed=[{"delay": 1, "effects": {"team_morale": -4}, | |
| "note": "Exec pressure builds during the freeze"}], | |
| second_order="Buys stability at the price of roadmap credibility."), | |
| InterventionSpec( | |
| id="hire_senior", label="Hire a senior ML-infra engineer", | |
| description="Open and close a senior req to deepen the bench.", | |
| immediate={"cost": 8, "velocity": -2}, | |
| delayed=[{"delay": 3, "effects": {"velocity": 8, "system_performance": 5, "team_morale": 4}, | |
| "note": "New senior hire reaches productivity"}], | |
| second_order="Expensive and slow to pay off, then compounding."), | |
| InterventionSpec( | |
| id="invest_test_infra", label="Invest in test & eval infrastructure", | |
| description="Two engineers build slice-level eval + CI hardening for a sprint.", | |
| immediate={"velocity": -5, "schedule": -3, "cost": 3}, | |
| delayed=[{"delay": 2, "effects": {"reliability": 7, "maintainability": 8, "velocity": 6, "safety": 4}, | |
| "note": "Test infra lands: regressions caught pre-merge"}], | |
| second_order="Classic J-curve: worse before durably better."), | |
| InterventionSpec( | |
| id="ship_feature_fast", label="Ship the exec-requested feature now", | |
| description="Cut the corner: ship the feature leadership is pressing for.", | |
| immediate={"schedule": 10, "customer_impact": 7, "velocity": 3, | |
| "safety": -7, "maintainability": -6, "risk": 10}, | |
| delayed=[{"delay": 2, "effects": {"reliability": -6, "team_morale": -3}, | |
| "note": "Corner-cutting surfaces as flaky behavior"}], | |
| second_order="Relief now; debt and risk arrive on a delay."), | |
| InterventionSpec( | |
| id="team_1on1s", label="Deep-dive 1:1s across the team", | |
| description="Spend the week understanding people: load, friction, morale.", | |
| immediate={"team_morale": 7, "velocity": -2}, | |
| delayed=[{"delay": 1, "effects": {"velocity": 4, "risk": -3}, | |
| "note": "Friction removed after 1:1 follow-ups"}], | |
| second_order="Costs a week; surfaces problems before they explode."), | |
| InterventionSpec( | |
| id="exec_expectation_reset", label="Reset expectations with leadership", | |
| description="Present the regression data and renegotiate the ship date.", | |
| immediate={"schedule": 6, "team_morale": 5, "risk": -4, "customer_impact": -3}, | |
| delayed=[], | |
| second_order="Spends political capital to buy engineering room."), | |
| InterventionSpec( | |
| id="incident_review_process", label="Institute blameless incident reviews", | |
| description="Every escaped regression gets a review with tracked actions.", | |
| immediate={"velocity": -3, "maintainability": 3}, | |
| delayed=[{"delay": 2, "effects": {"safety": 6, "reliability": 5}, | |
| "note": "Incident-review actions close systemic holes"}], | |
| second_order="Culture change: slow, then decisive."), | |
| InterventionSpec( | |
| id="add_monitoring", label="Add production model monitoring", | |
| description="Slice-level drift + regression monitoring on live traffic.", | |
| immediate={"reliability": 5, "cost": 4, "velocity": -2}, | |
| delayed=[{"delay": 1, "effects": {"safety": 6, "system_performance": 4}, | |
| "note": "Monitoring catches a drift you didn't know about"}], | |
| second_order="Visibility first; fixes follow the visibility."), | |
| InterventionSpec( | |
| id="refactor_pipeline", label="Refactor the training/eval pipeline", | |
| description="Pay down the pipeline debt that makes every change slow.", | |
| immediate={"velocity": -7, "schedule": -5, "cost": 4}, | |
| delayed=[{"delay": 3, "effects": {"maintainability": 10, "velocity": 9, "system_performance": 4}, | |
| "note": "Refactor lands: iteration speed jumps"}], | |
| second_order="The biggest J-curve on the menu."), | |
| InterventionSpec( | |
| id="cut_scope", label="Cut scope to protect the date", | |
| description="Descope the release to the safety-critical core.", | |
| immediate={"schedule": 8, "risk": -5, "customer_impact": -5, "velocity": 2}, | |
| delayed=[], | |
| second_order="Trades breadth for certainty."), | |
| ] | |
| INTERVENTION_INDEX = {i.id: i for i in INTERVENTIONS} | |
| class Scenario(BaseModel): | |
| scenario_id: str | |
| title: str | |
| narrative: str | |
| initial: Dict[str, float] | |
| drift: Dict[str, float] # per-turn pressure applied before interventions | |
| SCENARIOS: List[Scenario] = [ | |
| Scenario( | |
| scenario_id="inherited_perception_team", | |
| title="Inherited Perception Team Under Pressure", | |
| narrative=("You just inherited a 14-engineer perception team. The regression rate on " | |
| "safety-critical slices has risen for three consecutive releases, on-call is " | |
| "burning people out, and an exec has publicly committed your team to shipping " | |
| "a major feature in 8 weeks. Every objective is in tension: climb carefully."), | |
| initial={"system_performance": 55, "safety": 48, "reliability": 50, "cost": 45, | |
| "velocity": 52, "maintainability": 40, "team_morale": 44, | |
| "customer_impact": 55, "risk": 55, "schedule": 50}, | |
| drift={"safety": -2.0, "team_morale": -1.5, "maintainability": -1.0, | |
| "risk": 1.5, "schedule": -1.0}, | |
| ), | |
| ] | |
| class TurnRecord(BaseModel): | |
| turn: int | |
| hypothesis: str | |
| hypothesis_assessment: Dict | |
| intervention_id: str | |
| reverted_previous: bool = False | |
| applied_effects: Dict[str, float] = Field(default_factory=dict) | |
| delayed_landed: List[str] = Field(default_factory=list) | |
| events: List[str] = Field(default_factory=list) | |
| metrics_after: Dict[str, float] = Field(default_factory=dict) | |
| objective_after: float = 0.0 | |
| objective_delta: float = 0.0 | |
| verdict: str = "" # keep_recommended | reject_recommended | |
| timestamp: str = Field(default_factory=now_iso) | |
| class SimulationState(BaseModel): | |
| sim_id: str = Field(default_factory=lambda: new_id("sim")) | |
| user_id: str = "default" | |
| scenario_id: str = "inherited_perception_team" | |
| seed: int = 42 | |
| max_turns: int = 8 | |
| turn: int = 0 | |
| status: str = "active" # active | complete | |
| metrics: Dict[str, float] = Field(default_factory=dict) | |
| pending: List[Dict] = Field(default_factory=list) # {due_turn, effects, note} | |
| history: List[TurnRecord] = Field(default_factory=list) | |
| events: List[Dict] = Field(default_factory=list) | |
| objective_history: List[float] = Field(default_factory=list) | |
| debrief: Optional[Dict] = None | |
| def objective_score(metrics: Dict[str, float]) -> float: | |
| total = 0.0 | |
| for m, w in OBJECTIVE_WEIGHTS.items(): | |
| v = metrics.get(m, 50.0) | |
| total += w * ((100.0 - v) if m in INVERTED else v) | |
| return round(total, 2) | |
| def create_simulation(user_id: str = "default", scenario_id: str = "inherited_perception_team", | |
| seed: int = 42, max_turns: int = 8, | |
| store: Optional[Store] = None) -> SimulationState: | |
| store = store or get_store() | |
| scenario = next((s for s in SCENARIOS if s.scenario_id == scenario_id), SCENARIOS[0]) | |
| sim = SimulationState(user_id=user_id, scenario_id=scenario.scenario_id, seed=seed, | |
| max_turns=max_turns, metrics=dict(scenario.initial)) | |
| sim.objective_history.append(objective_score(sim.metrics)) | |
| store.put("simulations", sim.sim_id, sim) | |
| return sim | |
| def _assess_hypothesis(hypothesis: str, spec: InterventionSpec) -> Dict: | |
| """Rule-based check: is the hypothesis falsifiable and targeted?""" | |
| low = hypothesis.lower() | |
| touched = set(spec.immediate) | {m for d in spec.delayed for m in d["effects"]} | |
| mentions_metric = [m for m in METRICS if m.replace("_", " ") in low or m in low] | |
| targeted = any(m in touched for m in mentions_metric) | |
| directional = bool(any(w in low for w in ["increase", "decrease", "improve", "reduce", "raise", | |
| "lower", "recover", "drop", "+", "-"])) | |
| falsifiable = directional and (bool(mentions_metric) or any(ch.isdigit() for ch in low)) | |
| quality = "strong" if (targeted and falsifiable) else ("directional" if directional else "vague") | |
| return {"mentions_metrics": mentions_metric, "targets_intervention_effects": targeted, | |
| "directional": directional, "falsifiable": falsifiable, "quality": quality} | |
| def step_simulation(sim_id: str, hypothesis: str, intervention_id: str, | |
| revert_previous: bool = False, | |
| store: Optional[Store] = None) -> SimulationState: | |
| store = store or get_store() | |
| raw = store.get("simulations", sim_id) | |
| if raw is None: | |
| raise ValueError(f"unknown simulation '{sim_id}'") | |
| sim = SimulationState(**raw) | |
| if sim.status != "active": | |
| raise ValueError("simulation already complete") | |
| spec = INTERVENTION_INDEX.get(intervention_id) | |
| if spec is None: | |
| raise ValueError(f"unknown intervention '{intervention_id}'") | |
| scenario = next(s for s in SCENARIOS if s.scenario_id == sim.scenario_id) | |
| sim.turn += 1 | |
| rng = random.Random(sim.seed * 1000 + sim.turn) | |
| record = TurnRecord(turn=sim.turn, hypothesis=hypothesis, | |
| hypothesis_assessment=_assess_hypothesis(hypothesis, spec), | |
| intervention_id=intervention_id) | |
| # keep/reject mechanic: reject the previous turn's intervention (undo its | |
| # immediate effects) before applying the new one. | |
| if revert_previous and sim.history: | |
| prev = sim.history[-1] | |
| for m, dv in prev.applied_effects.items(): | |
| sim.metrics[m] = sim.metrics.get(m, 50.0) - dv | |
| record.reverted_previous = True | |
| record.events.append(f"Rejected previous intervention '{prev.intervention_id}' — its immediate effects were rolled back.") | |
| # scenario pressure (drift) | |
| for m, dv in scenario.drift.items(): | |
| sim.metrics[m] = sim.metrics.get(m, 50.0) + dv | |
| # intervention immediate effects with seeded jitter (±30%) | |
| applied: Dict[str, float] = {} | |
| for m, mean in spec.immediate.items(): | |
| jitter = 1.0 + (rng.random() - 0.5) * 0.6 | |
| dv = round(mean * jitter, 2) | |
| sim.metrics[m] = sim.metrics.get(m, 50.0) + dv | |
| applied[m] = dv | |
| record.applied_effects = applied | |
| # enqueue delayed / second-order effects | |
| for d in spec.delayed: | |
| sim.pending.append({"due_turn": sim.turn + int(d["delay"]), | |
| "effects": d["effects"], "note": d.get("note", "")}) | |
| # deliver due delayed effects | |
| still_pending: List[Dict] = [] | |
| for p in sim.pending: | |
| if p["due_turn"] <= sim.turn: | |
| for m, dv in p["effects"].items(): | |
| jitter = 1.0 + (rng.random() - 0.5) * 0.4 | |
| sim.metrics[m] = sim.metrics.get(m, 50.0) + round(dv * jitter, 2) | |
| record.delayed_landed.append(p["note"] or "Delayed effect landed") | |
| else: | |
| still_pending.append(p) | |
| sim.pending = still_pending | |
| # clamp | |
| for m in METRICS: | |
| sim.metrics[m] = round(max(0.0, min(100.0, sim.metrics.get(m, 50.0))), 2) | |
| # hard floors → events | |
| if sim.metrics["safety"] < SAFETY_FLOOR: | |
| incident = { | |
| "turn": sim.turn, "type": "SAFETY_INCIDENT", | |
| "detail": (f"Safety fell to {sim.metrics['safety']:.0f} (< floor {SAFETY_FLOOR:.0f}): a " | |
| "regression escaped to the fleet. Forced remediation: launches frozen, " | |
| "customer trust and morale take the hit."), | |
| } | |
| sim.events.append(incident) | |
| record.events.append(incident["detail"]) | |
| sim.metrics["customer_impact"] = max(0.0, sim.metrics["customer_impact"] - 12) | |
| sim.metrics["team_morale"] = max(0.0, sim.metrics["team_morale"] - 8) | |
| sim.metrics["risk"] = min(100.0, sim.metrics["risk"] + 10) | |
| sim.metrics["safety"] = min(100.0, sim.metrics["safety"] + 12) # forced remediation | |
| if sim.metrics["team_morale"] < MORALE_FLOOR: | |
| attrition = { | |
| "turn": sim.turn, "type": "ATTRITION", | |
| "detail": (f"Morale fell to {sim.metrics['team_morale']:.0f} (< floor {MORALE_FLOOR:.0f}): " | |
| "a senior engineer resigns; velocity drops."), | |
| } | |
| sim.events.append(attrition) | |
| record.events.append(attrition["detail"]) | |
| sim.metrics["velocity"] = max(0.0, sim.metrics["velocity"] - 10) | |
| sim.metrics["team_morale"] = min(100.0, sim.metrics["team_morale"] + 6) | |
| # measure & evaluate | |
| obj = objective_score(sim.metrics) | |
| prev_obj = sim.objective_history[-1] | |
| record.objective_after = obj | |
| record.objective_delta = round(obj - prev_obj, 2) | |
| record.verdict = "keep_recommended" if record.objective_delta >= 0 else "reject_recommended" | |
| record.metrics_after = dict(sim.metrics) | |
| sim.objective_history.append(obj) | |
| sim.history.append(record) | |
| if sim.turn >= sim.max_turns: | |
| sim.status = "complete" | |
| sim.debrief = build_debrief(sim, store) | |
| store.put("simulations", sim.sim_id, sim) | |
| return sim | |
| def build_debrief(sim: SimulationState, store: Optional[Store] = None) -> Dict: | |
| """Map the decision history onto competency evidence.""" | |
| store = store or get_store() | |
| mappings: List[Dict] = [] | |
| strong_hyps = [t for t in sim.history if t.hypothesis_assessment.get("quality") == "strong"] | |
| if len(strong_hyps) >= max(1, len(sim.history) // 2): | |
| mappings.append({ | |
| "competency_id": "p4.hill_climbing", | |
| "verdict": "evidenced", | |
| "reason": (f"{len(strong_hyps)}/{len(sim.history)} turns used falsifiable, targeted " | |
| "hypotheses before intervening."), | |
| "quotes": [t.hypothesis for t in strong_hyps[:3]], | |
| }) | |
| else: | |
| mappings.append({ | |
| "competency_id": "p4.hill_climbing", | |
| "verdict": "gap", | |
| "reason": "Most hypotheses were vague (no metric, no direction) — hill climbing without a hypothesis is guessing.", | |
| "quotes": [t.hypothesis for t in sim.history[:2]], | |
| }) | |
| rejects = [t for t in sim.history if t.reverted_previous] | |
| negative_turns = [t for t in sim.history if t.objective_delta < 0] | |
| if rejects: | |
| mappings.append({ | |
| "competency_id": "p3.closed_loop_execution", | |
| "verdict": "evidenced", | |
| "reason": f"Rejected {len(rejects)} intervention(s) after measurement showed regression — real keep/reject discipline.", | |
| "quotes": [f"turn {t.turn}: rejected previous after Δobjective {sim.history[t.turn-2].objective_delta:+.1f}" | |
| for t in rejects[:2] if t.turn >= 2], | |
| }) | |
| elif len(negative_turns) >= 2: | |
| mappings.append({ | |
| "competency_id": "p3.closed_loop_execution", | |
| "verdict": "gap", | |
| "reason": f"{len(negative_turns)} turns measured negative but nothing was ever rejected/rolled back.", | |
| "quotes": [], | |
| }) | |
| incidents = [e for e in sim.events if e["type"] == "SAFETY_INCIDENT"] | |
| safety_interventions = {"add_regression_gate", "pause_launches", "incident_review_process", "add_monitoring"} | |
| used_safety = [t for t in sim.history if t.intervention_id in safety_interventions] | |
| if incidents and any(t.turn > incidents[0]["turn"] for t in used_safety): | |
| mappings.append({"competency_id": "p3.safety_culture", "verdict": "evidenced", | |
| "reason": "Responded to the safety incident with systemic safety interventions.", | |
| "quotes": [t.intervention_id for t in used_safety[:3]]}) | |
| elif incidents: | |
| mappings.append({"competency_id": "p3.safety_culture", "verdict": "gap", | |
| "reason": "A safety incident occurred and no safety-focused intervention followed.", | |
| "quotes": []}) | |
| elif used_safety: | |
| mappings.append({"competency_id": "p3.safety_culture", "verdict": "evidenced", | |
| "reason": "Kept safety above the floor all game via proactive safety investment.", | |
| "quotes": [t.intervention_id for t in used_safety[:3]]}) | |
| final = sim.metrics | |
| balanced = all((100 - final[m] if m in INVERTED else final[m]) >= 35 for m in METRICS) | |
| mappings.append({ | |
| "competency_id": "p2.reliability_tradeoffs", | |
| "verdict": "evidenced" if balanced else "gap", | |
| "reason": ("Ended with every objective above the balance threshold — no metric was sacrificed." | |
| if balanced else | |
| "At least one objective ended critically low — the climb over-optimized a subset."), | |
| "quotes": [], | |
| }) | |
| start_obj, end_obj = sim.objective_history[0], sim.objective_history[-1] | |
| summary = { | |
| "objective_start": start_obj, | |
| "objective_end": end_obj, | |
| "objective_delta": round(end_obj - start_obj, 2), | |
| "incidents": len(incidents), | |
| "turns": sim.turn, | |
| "balanced_finish": balanced, | |
| "competency_mappings": mappings, | |
| } | |
| # persist evidence for the evidenced mappings | |
| for m in mappings: | |
| if m["verdict"] != "evidenced": | |
| continue | |
| ev = Evidence( | |
| user_id=sim.user_id, | |
| competency_ids=[m["competency_id"]], | |
| artifact_type="simulation_debrief", | |
| source=f"Simulation {sim.sim_id} ({sim.scenario_id})", | |
| summary=m["reason"], | |
| quotes=[q for q in m["quotes"] if isinstance(q, str)][:3], | |
| score=4.0 if balanced else 3.0, | |
| confidence=0.6, | |
| payload={"sim_id": sim.sim_id, "objective_delta": summary["objective_delta"]}, | |
| ) | |
| store.put("evidence", ev.evidence_id, ev) | |
| m["evidence_id"] = ev.evidence_id | |
| return summary | |
| def get_simulation(sim_id: str, store: Optional[Store] = None) -> Optional[SimulationState]: | |
| raw = (store or get_store()).get("simulations", sim_id) | |
| return SimulationState(**raw) if raw else None | |