Harden offline incidents and virtual eIM transitions

.claude/skills/verify/SKILL.md

@@ -23,5 +23,6 @@ README "Live" section) and, for Bedrock, `AWS_PROFILE=<your-profile>`.
 
 ## Gotchas
 
-- An invalid `AWS_PROFILE` name crashes even `--no-bedrock` runs: the inner Strands agent always builds a BedrockModel.
+- `--no-bedrock` runs, replay and validate ignore `AWS_PROFILE`. `make test` still needs it unset or valid: the pattern
+  tests and the Bedrock-mocked agent tests construct a `BedrockModel`.
 - Benchmark runs take ~20-40 min with both Bedrock choosers; run them in the background.

README.md

@@ -115,7 +115,7 @@ Decider ran on a laptop GPU and Bedrock in eu-central-1, and each chooser ran on
 ```bash
 git clone https://github.com/HDGoldi/simshift-agent && cd simshift-agent
 make install                     # or: uv venv && uv pip install -e . pytest
-make test                        # 138 tests + trace validation; no Decider server or AWS needed
+make test                        # 147 tests + trace validation; no Decider server or AWS needed
 make ui                          # Recovery Decision Inspector on recorded real Decider answers: http://127.0.0.1:8090
 make replay                      # re-run all six recorded traces against the current code
 ```
@@ -153,8 +153,8 @@ twin, and Bedrock's plan review and explanation.
 
 ## Tests
 
-`make test` runs 138 tests and validates every published trace and the benchmark file. Without AWS credentials and a
-Decider server, 137 pass and the real-Decider test skips (set `SIMSHIFT_REQUIRE_DECIDER=1` to make that a failure).
+`make test` runs 147 tests and validates every published trace and the benchmark file. Without AWS credentials and a
+Decider server, 146 pass and the real-Decider test skips (set `SIMSHIFT_REQUIRE_DECIDER=1` to make that a failure).
 Bedrock is never called in tests. The acceptance properties covered:
 - Decider's choice controls which Strands tool runs; Bedrock in orchestrator mode cannot change it.
 - Excluded options are never offered and are denied on any call path; an independent audit counts violations.

docs/architecture.md

@@ -122,10 +122,12 @@ Decider's options beforehand:
 - the fleet owner authorizes switching;
 - a human approved.
 
-Independently of the gate, the virtual eIM refuses an enable command whose approval does not match the proposal, one
-for an eUICC that cannot enable profiles or a fleet whose owner forbids switching, and one it cannot deliver because
-the management path is down. It does not repeat the gate's evidence checks (reject cause, reselection, coverage). A test
-replaces the gate with one that always passes and checks that the eIM still refuses an unapproved switch.
+Independently of the gate, the virtual eIM refuses an enable command whose target is not an installed, inactive
+profile, whose approval does not match the proposal, for an eUICC that cannot enable profiles or a fleet whose owner
+forbids switching, and one it cannot deliver because the management path is down. A command that reuses an approval
+returns the first result for that approval, whatever its target. It does not repeat the gate's evidence checks (reject
+cause, reselection, coverage). A test replaces the gate with one that always passes and checks that the eIM still
+refuses an unapproved switch.
 
 **A profile change that does not restore service is handed off.** If the gate blocks it for any reason other than a
 pending approval, the eIM cannot deliver it, verification fails and it is rolled back, rollback itself fails, or the

simshift/agent.py

@@ -19,7 +19,7 @@
 from pydantic import BaseModel
 from strands import Agent, tool
 from strands.hooks import AfterToolCallEvent, BeforeToolCallEvent
-from strands.models import BedrockModel
+from strands.models import BedrockModel, Model
 
 from patterns.capability_aware_router.router import CapabilityAwareRouter, DeciderClient, ReplayDecider
 
@@ -48,19 +48,39 @@ class ProfilePlan(BaseModel):
     risks: list[str]
 
 
+class _DirectToolModel(Model):
+    """Provider-neutral model for incidents that only invoke tools directly."""
+
+    def get_config(self) -> dict:
+        return {}
+
+    def update_config(self, **model_config) -> None:
+        pass
+
+    async def stream(self, messages, tool_specs=None, system_prompt=None, **kwargs):
+        if False:
+            yield {}
+        raise RuntimeError("inference is disabled for direct-tool-only incidents")
+
+    async def structured_output(self, output_model, prompt, system_prompt=None, **kwargs):
+        if False:
+            yield {}
+        raise RuntimeError("inference is disabled for direct-tool-only incidents")
+
+
 @functools.cache
 def bedrock_model() -> BedrockModel:
     """One Bedrock client per process; creating the boto3 session per incident cost about 0.1 s each."""
     return BedrockModel(model_id=BEDROCK_MODEL, region_name=REGION, max_tokens=2048)
 
 
-def _bedrock(llm: BedrockModel, prompt: str, schema=None):
+def _bedrock(llm: Model, prompt: str, schema=None):
     agent = Agent(model=llm, tools=[], callback_handler=None)
     res = agent(prompt, structured_output_model=schema) if schema else agent(prompt)
     return res.structured_output.model_dump() if schema else str(res).strip()
 
 
-def build_tools(twin: FleetTwin, approver: Callable[[str], bool], llm: BedrockModel | None) -> list:
+def build_tools(twin: FleetTwin, approver: Callable[[str], bool], llm: Model | None) -> list:
     """The six routed tools. `approver(proposal_id)` says whether a human approved that exact proposal."""
     plans: dict[str, dict | None] = {}  # one Bedrock review per proposal, even if it is re-submitted after approval
 
@@ -148,7 +168,7 @@ def __init__(self, scenario_id: str, approve: bool = False, decider=None, use_be
         self.scenario_id, self.pre_approved, self.use_bedrock = scenario_id, approve, use_bedrock
         self.twin = twin = FleetTwin(scenario_id)
         self.router = make_router(decider or DeciderClient(os.environ.get("DECIDER_URL", "http://127.0.0.1:8099")))
-        self.llm = bedrock_model()
+        self.llm: Model = bedrock_model() if use_bedrock else _DirectToolModel()
         self.approved: set[str] = set()  # proposal ids a human approved during the session
         self.answers: dict[str, bool] = {}  # every human answer: proposal id -> approved?
         self.agent = Agent(model=self.llm, tools=build_tools(twin, lambda pid: approve or pid in self.approved,

simshift/simulator.py

@@ -287,10 +287,18 @@ def proposal_id(self, target: str) -> str:
 
     def enable_profile(self, target: str, approval: str | None) -> dict:
         """Simulator-only. Independent of the policy gate, the virtual eIM compares the approval with its own id for
-        this proposal, re-checks eUICC capability and owner authorization, and refuses a command it cannot deliver. The approval doubles as the
-        idempotency key, so a repeated command returns the first result."""
+        this proposal, re-checks the target, eUICC capability and owner authorization, and refuses a command it cannot deliver.
+        The approval doubles as the idempotency key, so a repeated approval returns the first result whatever the target."""
         if approval in self.commands:
             return self.commands[approval]
+        if not isinstance(target, str):
+            raise TypeError("target must be a string")
+        if target not in self.s["profiles"]:
+            return self._log("eim_enable_profile", ok=False,
+                             reason="target profile is not installed; command not sent")
+        if target == self.s["active_profile"]:
+            return self._log("eim_enable_profile", ok=False,
+                             reason="target profile is already active; command not sent")
         if approval != self.proposal_id(target):  # not cached: a later approved attempt must really execute
             return self._log("eim_enable_profile", ok=False, reason="no operator approval for this proposal; command not sent")
         if not (self.s["euicc"]["supports_profile_enable"] and self.s["owner_allows_profile_switch"]):

tests/test_agent.py

@@ -1,5 +1,8 @@
+import asyncio
 import json
 import os
+import subprocess
+import sys
 import urllib.request
 
 import pytest
@@ -22,6 +25,57 @@ def choose(self, state, instructions, options, name="next_tool"):
         return {"choice": c, "probabilities": {o: float(o == c) for o in options}, "confidence": 1.0}
 
 
+async def _consume(events):
+    async for _ in events:
+        pass
+
+
+@pytest.mark.parametrize("method", ["stream", "structured_output"])
+def test_direct_tool_model_is_a_stateless_failing_async_adapter(method):
+    from simshift.agent import ProfilePlan, _DirectToolModel
+
+    model = _DirectToolModel()
+    model.update_config(region_name="secret-region", credentials="secret-credentials")
+    assert model.get_config() == {}
+    if method == "stream":
+        events = model.stream([], tool_specs=[], system_prompt="secret-prompt", tool_choice={"auto": {}},
+                              system_prompt_content=[], invocation_state={"scenario": "secret-scenario"},
+                              cancel_signal=None, agent_metadata=None, future_sdk_kwarg="secret-future")
+    else:
+        events = model.structured_output(ProfilePlan, [], system_prompt="secret-prompt",
+                                         invocation_state={"scenario": "secret-scenario"},
+                                         future_sdk_kwarg="secret-future")
+    with pytest.raises(RuntimeError) as error:
+        asyncio.run(_consume(events))
+    assert str(error.value) == "inference is disabled for direct-tool-only incidents"
+
+
+def test_offline_incident_skips_bedrock_factory_and_runs_a_direct_tool(monkeypatch):
+    import simshift.agent as ag
+
+    monkeypatch.setattr(ag, "bedrock_model", lambda: pytest.fail("offline incident called bedrock_model"))
+    inc = ag.Incident("BR_PROFILE", decider=ScriptedDecider(["investigate_device"]), use_bedrock=False)
+    step = inc.step()
+    assert step["tool"] == TOOLS["investigate_device"] and step["tool_status"] == "success"
+    assert inc.llm.get_config() == {}
+
+
+def test_bedrock_incident_uses_the_bedrock_model_factory(monkeypatch):
+    import simshift.agent as ag
+
+    sentinel, passed_to_tools = ag._DirectToolModel(), []
+    real_build_tools = ag.build_tools
+
+    def capture_tools(twin, approver, llm):
+        passed_to_tools.append(llm)
+        return real_build_tools(twin, approver, llm)
+
+    monkeypatch.setattr(ag, "bedrock_model", lambda: sentinel)
+    monkeypatch.setattr(ag, "build_tools", capture_tools)
+    inc = ag.Incident("TRANSIENT", decider=ScriptedDecider(["retry_connection"]), use_bedrock=True)
+    assert inc.llm is sentinel and passed_to_tools == [sentinel]
+
+
 def test_decider_controls_tool():
     a = run("BR_PROFILE", decider=ScriptedDecider(["investigate_device"]), use_bedrock=False)
     b = run("BR_PROFILE", decider=ScriptedDecider(["investigate_data_path"]), use_bedrock=False)
@@ -124,6 +178,44 @@ def test_off_menu_choice_never_reaches_the_eim():
 SAMPLE_RUNS = os.path.join(os.path.dirname(__file__), "..", "docs", "sample-runs")
 
 
+def test_all_offline_surfaces_ignore_a_hostile_aws_profile(monkeypatch):
+    import simshift.agent as ag
+    from simshift.benchmark import RandomChooser, RulesChooser, make, run_one
+    from simshift.validate import problems
+    from simshift.web import App
+
+    monkeypatch.setenv("AWS_PROFILE", "simshift-profile-that-does-not-exist")
+    monkeypatch.setattr(ag, "bedrock_model", lambda: pytest.fail("offline surface called bedrock_model"))
+
+    recording = json.load(open(os.path.join(SAMPLE_RUNS, "TRANSIENT.json")))
+    assert problems(recording) == []
+
+    sid = "generated-hostile-profile-transient"
+    monkeypatch.setitem(SCENARIOS, sid, make("transient", 9007))
+    for chooser in (RulesChooser(), RandomChooser(7)):
+        row = run_one(sid, "transient", chooser)
+        assert row["incident"] == sid and row["trace"]
+
+    app = App("http://127.0.0.1:1", replay_dir=os.path.join(SAMPLE_RUNS, "orchestrated"))
+    snapshot = app.create("TRANSIENT")
+    while not snapshot["done"]:
+        snapshot = app.act(snapshot["id"], "step", {})
+    assert snapshot["outcome"]["restored"]
+
+
+def test_no_bedrock_cli_replay_ignores_a_hostile_aws_profile():
+    command = [sys.executable, "-m", "simshift.agent", "TRANSIENT", "--approve", "--no-bedrock",
+               "--replay", "docs/sample-runs/TRANSIENT.json"]
+    env = {**os.environ, "AWS_PROFILE": "simshift-profile-that-does-not-exist",
+           "DECIDER_URL": "http://127.0.0.1:1"}
+    completed = subprocess.run(command, cwd=os.path.dirname(os.path.dirname(__file__)), env=env,
+                               capture_output=True, text=True, timeout=30)
+    output = completed.stdout + completed.stderr
+    assert completed.returncode == 0, output
+    assert "no Bedrock" in completed.stdout and "restored=True" in completed.stdout
+    assert "ProfileNotFound" not in output
+
+
 @pytest.mark.parametrize("name", sorted(f for d in ("", "orchestrated") for f in
                                          (os.path.join(d, x) for x in os.listdir(os.path.join(SAMPLE_RUNS, d)))
                                          if f.endswith(".json")))

tests/test_twin_policy.py

@@ -1,6 +1,8 @@
-import pytest
+import copy
 import json
 
+import pytest
+
 from simshift.policy import (counterfactual, exclusions, guarded_profile_transition, profile_transition_gate,
                              proposal_id)
 from simshift.simulator import SCENARIOS, FleetTwin
@@ -66,11 +68,58 @@ def test_germany_operator_reselection():
     assert ev["restored"] and twin.s["active_profile"] == "profile_A"
 
 
+def _twin_snapshot(twin):
+    return {
+        "state": copy.deepcopy(twin.s),
+        "diagnostics": copy.deepcopy(twin.diagnostics),
+        "rates": twin.rates(),
+        "profile_changes": twin.profile_changes,
+        "commands": copy.deepcopy(twin.commands),
+        "event_count": len(twin.events),
+        "simulated_minutes": twin.simulated_minutes(),
+    }
+
+
 def test_idempotent_enable():
     twin = FleetTwin("BR_PROFILE")
     pid = twin.proposal_id("profile_B")
-    a = twin.enable_profile("profile_B", pid)
-    assert twin.enable_profile("profile_B", pid) is a and len(twin.events) == 1
+    first = twin.enable_profile("profile_B", pid)
+    before_retry = _twin_snapshot(twin)
+    repeated = twin.enable_profile("profile_B", pid)
+    assert repeated is first
+    assert twin.s["active_profile"] == "profile_B"
+    assert _twin_snapshot(twin) == before_retry
+
+
+@pytest.mark.parametrize(("target", "reason"), [
+    ("profile_C", "target profile is not installed; command not sent"),
+    ("profile_A", "target profile is already active; command not sent"),
+])
+def test_eim_refuses_invalid_profile_targets_atomically(target, reason):
+    twin = FleetTwin("BR_PROFILE")
+    approval = twin.proposal_id(target)
+    before = _twin_snapshot(twin)
+
+    result = twin.enable_profile(target, approval)
+
+    after = _twin_snapshot(twin)
+    assert result is twin.events[-1]
+    assert result == {"seq": 1, "kind": "eim_enable_profile", "ok": False, "reason": reason}
+    for field in ("state", "diagnostics", "rates", "profile_changes", "commands"):
+        assert after[field] == before[field]
+    assert after["event_count"] == before["event_count"] + 1
+    assert after["simulated_minutes"] == before["simulated_minutes"] + 30
+
+
+def test_eim_rejects_non_string_target_without_mutation():
+    twin = FleetTwin("BR_PROFILE")
+    before = _twin_snapshot(twin)
+
+    with pytest.raises(TypeError) as error:
+        twin.enable_profile([], "uncached-approval")
+
+    assert str(error.value) == "target must be a string"
+    assert _twin_snapshot(twin) == before
 
 
 def test_seed_reproducibility():