From 1901f4e8b6ea1edd3c00b28eb8884aba8cdf5a89 Mon Sep 17 00:00:00 2001 From: Kiro Agent <244629292+kiro-agent@users.noreply.github.com> Date: Mon, 19 Oct 2026 12:52:40 +0000 Subject: [PATCH 1/3] fix: keep offline incidents AWS-independent --- .claude/skills/verify/SKILL.md | 2 +- simshift/agent.py | 28 ++++++++-- tests/test_agent.py | 94 ++++++++++++++++++++++++++++++++++ 3 files changed, 119 insertions(+), 5 deletions(-) diff --git a/.claude/skills/verify/SKILL.md b/.claude/skills/verify/SKILL.md index 2fc766c..dbd2578 100644 --- a/.claude/skills/verify/SKILL.md +++ b/.claude/skills/verify/SKILL.md @@ -23,5 +23,5 @@ README "Live" section) and, for Bedrock, `AWS_PROFILE=`. ## Gotchas -- An invalid `AWS_PROFILE` name crashes even `--no-bedrock` runs: the inner Strands agent always builds a BedrockModel. +- Live Bedrock flows require a valid AWS profile and region configuration. - Benchmark runs take ~20-40 min with both Bedrock choosers; run them in the background. diff --git a/simshift/agent.py b/simshift/agent.py index 7cb7a79..3e66e27 100644 --- a/simshift/agent.py +++ b/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, diff --git a/tests/test_agent.py b/tests/test_agent.py index f9e3738..d65c7f5 100644 --- a/tests/test_agent.py +++ b/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,59 @@ 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" + assert not any(secret in str(error.value) for secret in + ("secret-region", "secret-credentials", "secret-prompt", "secret-scenario", "secret-future")) + + +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_enabled_incident_uses_the_cached_model_factory_result(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 +180,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"))) From 49c2751587db8633ccfe6aa797b14f2e78b73f33 Mon Sep 17 00:00:00 2001 From: Kiro Agent <244629292+kiro-agent@users.noreply.github.com> Date: Mon, 19 Oct 2026 12:53:53 +0000 Subject: [PATCH 2/3] fix: reject invalid eim profile targets --- README.md | 6 ++--- docs/architecture.md | 19 +++++++++----- simshift/simulator.py | 8 ++++++ tests/test_twin_policy.py | 55 ++++++++++++++++++++++++++++++++++++--- 4 files changed, 76 insertions(+), 12 deletions(-) diff --git a/README.md b/README.md index 9c05627..5ce66c9 100644 --- a/README.md +++ b/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. diff --git a/docs/architecture.md b/docs/architecture.md index d0b4277..04d8fe7 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -96,8 +96,12 @@ What is modelled, what is simplified and what is left out, in telecom terms, is registration, data session **and** application telemetry are verified. Delivery and execution are one simulated step. A failed verification is rolled back by the eIM re-enabling the previous profile, which the original approval covers. If the new profile cannot connect and the result cannot be reported, the device rolls back on its own (this - stands in for SGP.32's IPA-triggered rollback). Commands are idempotent per attempt. No profile download, no - cryptography, no certified SGP.32 flow. + stands in for SGP.32's IPA-triggered rollback). The virtual eIM requires a string target naming an installed, + inactive profile before it attempts delivery. Exact duplicate commands are served from the approval-keyed cache + first—even when the successful target is now active—so they return the original result without another event or + state change. New invalid-target and other command-not-sent refusals remain uncached audit events; under the + simulation's existing event accounting, each `eim_enable_profile` refusal still represents 30 simulated minutes. + No profile download, no cryptography, no certified SGP.32 flow. - **Deterministic.** There is no randomness; cloning a twin and replaying the same actions gives identical state and events. ## Keeping the answer away from the models @@ -122,10 +126,13 @@ 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 serves an exact cached command duplicate before revalidating its target. +For an uncached command, it rejects a non-string target and refuses a target that is not an installed profile or is +already active before checking the proposal, capability, owner authorization, or reachability. It also 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. **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 diff --git a/simshift/simulator.py b/simshift/simulator.py index 5e3a85c..1b07f14 100644 --- a/simshift/simulator.py +++ b/simshift/simulator.py @@ -291,6 +291,14 @@ def enable_profile(self, target: str, approval: str | None) -> dict: idempotency key, so a repeated command returns the first result.""" 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"]): diff --git a/tests/test_twin_policy.py b/tests/test_twin_policy.py index ec71f09..b703ec2 100644 --- a/tests/test_twin_policy.py +++ b/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(): From 1081033fd55c9bfa2dcf69490b95b85afa6a859d Mon Sep 17 00:00:00 2001 From: Jan Sulaiman Date: Mon, 19 Oct 2026 15:12:17 +0200 Subject: [PATCH 3/3] Review fixes: eIM cache wording, docstring, verify-skill gotcha, test names The eIM cache is keyed by approval alone, so the docs no longer call cached results exact duplicates. Drop the duplicated ordering prose and em dashes, rename the model-factory test (it does not test caching) and drop an assertion that could never fail. Co-Authored-By: Claude Opus 5.5 (1M context) --- .claude/skills/verify/SKILL.md | 3 ++- docs/architecture.md | 21 ++++++++------------- simshift/simulator.py | 4 ++-- tests/test_agent.py | 4 +--- 4 files changed, 13 insertions(+), 19 deletions(-) diff --git a/.claude/skills/verify/SKILL.md b/.claude/skills/verify/SKILL.md index dbd2578..0bb82b6 100644 --- a/.claude/skills/verify/SKILL.md +++ b/.claude/skills/verify/SKILL.md @@ -23,5 +23,6 @@ README "Live" section) and, for Bedrock, `AWS_PROFILE=`. ## Gotchas -- Live Bedrock flows require a valid AWS profile and region configuration. +- `--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. diff --git a/docs/architecture.md b/docs/architecture.md index 04d8fe7..c26db4f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -96,12 +96,8 @@ What is modelled, what is simplified and what is left out, in telecom terms, is registration, data session **and** application telemetry are verified. Delivery and execution are one simulated step. A failed verification is rolled back by the eIM re-enabling the previous profile, which the original approval covers. If the new profile cannot connect and the result cannot be reported, the device rolls back on its own (this - stands in for SGP.32's IPA-triggered rollback). The virtual eIM requires a string target naming an installed, - inactive profile before it attempts delivery. Exact duplicate commands are served from the approval-keyed cache - first—even when the successful target is now active—so they return the original result without another event or - state change. New invalid-target and other command-not-sent refusals remain uncached audit events; under the - simulation's existing event accounting, each `eim_enable_profile` refusal still represents 30 simulated minutes. - No profile download, no cryptography, no certified SGP.32 flow. + stands in for SGP.32's IPA-triggered rollback). Commands are idempotent per attempt. No profile download, no + cryptography, no certified SGP.32 flow. - **Deterministic.** There is no randomness; cloning a twin and replaying the same actions gives identical state and events. ## Keeping the answer away from the models @@ -126,13 +122,12 @@ Decider's options beforehand: - the fleet owner authorizes switching; - a human approved. -Independently of the gate, the virtual eIM serves an exact cached command duplicate before revalidating its target. -For an uncached command, it rejects a non-string target and refuses a target that is not an installed profile or is -already active before checking the proposal, capability, owner authorization, or reachability. It also 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 diff --git a/simshift/simulator.py b/simshift/simulator.py index 1b07f14..005be4e 100644 --- a/simshift/simulator.py +++ b/simshift/simulator.py @@ -287,8 +287,8 @@ 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): diff --git a/tests/test_agent.py b/tests/test_agent.py index d65c7f5..48d0682 100644 --- a/tests/test_agent.py +++ b/tests/test_agent.py @@ -48,8 +48,6 @@ def test_direct_tool_model_is_a_stateless_failing_async_adapter(method): with pytest.raises(RuntimeError) as error: asyncio.run(_consume(events)) assert str(error.value) == "inference is disabled for direct-tool-only incidents" - assert not any(secret in str(error.value) for secret in - ("secret-region", "secret-credentials", "secret-prompt", "secret-scenario", "secret-future")) def test_offline_incident_skips_bedrock_factory_and_runs_a_direct_tool(monkeypatch): @@ -62,7 +60,7 @@ def test_offline_incident_skips_bedrock_factory_and_runs_a_direct_tool(monkeypat assert inc.llm.get_config() == {} -def test_enabled_incident_uses_the_cached_model_factory_result(monkeypatch): +def test_bedrock_incident_uses_the_bedrock_model_factory(monkeypatch): import simshift.agent as ag sentinel, passed_to_tools = ag._DirectToolModel(), []