feat(n7web): enhance control workflows and status feedback

This commit is contained in:
Miayui
2026-09-16 12:52:07 +08:00
parent 84a3cd73bb
commit 0e4328f548
45 changed files with 7317 additions and 1001 deletions
@@ -778,7 +778,7 @@ def test_preview_and_process_share_mode_target_safety_gate(
} == {"mode_target_scope_mismatch"}
def test_active_nfc_peripheral_can_execute_but_inactive_peripheral_is_rejected(
def test_durable_nfc_peripheral_can_execute_after_bound_card_is_removed(
vehicle_service: SeatDemoService,
) -> None:
learned = {
@@ -798,10 +798,10 @@ def test_active_nfc_peripheral_can_execute_but_inactive_peripheral_is_rejected(
peripheral_provider=lambda: [learned],
nfc_provider=lambda: {"card": {"uid": "a1b2c3d4"}},
)
inactive_catalog = AssistantCapabilityCatalog(
detached_catalog = AssistantCapabilityCatalog(
vehicle_service,
peripheral_provider=lambda: [learned],
nfc_provider=lambda: {"card": {"uid": "different-card"}},
nfc_provider=lambda: {"card": None},
)
active_client = _ToolCallingCodexClient()
@@ -834,27 +834,37 @@ def test_active_nfc_peripheral_can_execute_but_inactive_peripheral_is_rejected(
)
assert executed == ["打开星空投影仪"]
inactive_client = _ToolCallingCodexClient()
inactive_assistant = CodexAssistantService(inactive_client)
errors: list[str] = []
def attempt_inactive(token: str, _turn: ActiveCodexTurn) -> None:
try:
inactive_assistant.execute_cockpit_request(token)
except CodexVehicleToolError as exc:
errors.append(str(exc))
inactive_client.on_turn = attempt_inactive
_stream_with_catalog(
inactive_assistant,
inactive_catalog,
"打开星空投影仪",
execute_control=lambda _instruction, _progress: pytest.fail(
"inactive NFC peripheral must not execute"
),
conversation_id="conv-inactive-peripheral-01",
detached_client = _ToolCallingCodexClient()
detached_assistant = CodexAssistantService(detached_client)
detached_executed: list[str] = []
detached_client.on_turn = (
lambda token, _turn: detached_assistant.execute_cockpit_request(token)
)
assert errors == ["vehicle control was not authorized by the original user request"]
_stream_with_catalog(
detached_assistant,
detached_catalog,
"打开星空投影仪",
execute_control=lambda instruction, _progress: (
detached_executed.append(instruction)
or {
"status": "succeeded",
"execution_mode": "dry_run",
"utterance": instruction,
"infrared_actions": [
{
"capability_id": "star-projector-on",
"device_id": "star-projector",
"device_name": "星空投影仪",
"action": "开启",
"slot": 7,
"transport": "infrared",
}
],
}
),
conversation_id="conv-detached-peripheral-01",
)
assert detached_executed == ["打开星空投影仪"]
@pytest.mark.parametrize(