core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

harden amr demo replay fixture

Browse source
This commit is contained in:
Shay 2026-06-02 10:22:18 -07:00
parent f3680aa302
commit f0c270fded
1 changed files with 8 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
demos/amr_decision_substrate/run_demo.py
View file

@ -126,7 +126,9 @@ def _policy_decision(simulated_input: dict[str, Any]) -> tuple[Decision, str]:
return "REFUSE", f"under_determined: missing {','.join(missing)}" return "REFUSE", f"under_determined: missing {','.join(missing)}"
if simulated_input["route_state"] != "mapped": if simulated_input["route_state"] != "mapped":
return "REFUSE", "under_determined: route is not mapped" return "REFUSE", "under_determined: route is not mapped"
if not isinstance(simulated_input["path_confidence"], (int, float)): if isinstance(simulated_input["path_confidence"], bool) or not isinstance(
simulated_input["path_confidence"], (int, float)
):
return "REFUSE", "under_determined: path_confidence is not numeric" return "REFUSE", "under_determined: path_confidence is not numeric"
if float(simulated_input["path_confidence"]) < 0.85: if float(simulated_input["path_confidence"]) < 0.85:
return "REFUSE", "under_determined: path confidence below bound" return "REFUSE", "under_determined: path confidence below bound"
@ -136,6 +138,10 @@ def _policy_decision(simulated_input: dict[str, Any]) -> tuple[Decision, str]:
return "STOP", f"path not clear: {simulated_input['obstacle_state']}" return "STOP", f"path not clear: {simulated_input['obstacle_state']}"
if simulated_input["obstacle_state"] != "clear": if simulated_input["obstacle_state"] != "clear":
return "REFUSE", f"out_of_distribution: obstacle_state={simulated_input['obstacle_state']!r}" return "REFUSE", f"out_of_distribution: obstacle_state={simulated_input['obstacle_state']!r}"
if isinstance(simulated_input["path_count"], bool) or not isinstance(
simulated_input["path_count"], int
):
return "REFUSE", "under_determined: path_count is not numeric"
if int(simulated_input["path_count"]) < 1: if int(simulated_input["path_count"]) < 1:
return "STOP", "no admissible path in simulated record" return "STOP", "no admissible path in simulated record"
return "PROCEED", "mapped route, clear path, sufficient confidence" return "PROCEED", "mapped route, clear path, sufficient confidence"
@ -180,7 +186,7 @@ def _run_once(scenarios: tuple[Scenario, ...], trace_path: Path) -> tuple[Decisi
if trace_path.exists(): if trace_path.exists():
trace_path.unlink() trace_path.unlink()
sink = JsonlEventSink(trace_path) sink = JsonlEventSink(trace_path)
pipeline = CognitiveTurnPipeline(ChatRuntime(), recognizer=_recognizer()) pipeline = CognitiveTurnPipeline(ChatRuntime(no_load_state=True), recognizer=_recognizer())
records: list[DecisionRecord] = [] records: list[DecisionRecord] = []
for sequence_base, scenario in enumerate(scenarios): for sequence_base, scenario in enumerate(scenarios):