vee
671 lines
34 KiB
Python
671 lines
34 KiB
Python
"""Dynamic scene creation through the real assistant, compiler and saved-mode APIs."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import queue
|
|
from copy import deepcopy
|
|
from threading import Event
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
from jsonschema import Draft202012Validator
|
|
|
|
from cognitive_agent.application.assistant_capabilities import AssistantCapabilityCatalog
|
|
from cognitive_agent.application.codex_assistant import ActiveCodexTurn, CodexAssistantService
|
|
from cognitive_agent.application.mode_profile import ModeProfileRegistry
|
|
from cognitive_agent.application.mode_schedule import ModeSchedule
|
|
from cognitive_agent.application.mode_schedule_service import public_schedule_status
|
|
from cognitive_agent.application.scene_composer import (
|
|
SceneCompositionError,
|
|
build_scene_output_schema,
|
|
normalize_scene_output,
|
|
)
|
|
from cognitive_agent.application.schedule_intent import needs_mode_schedule
|
|
from cognitive_agent.application.seat_service import REPOSITORY_ROOT, SeatDemoService
|
|
from cognitive_agent.infrastructure.can.access import HardwareAccessRequiredError
|
|
|
|
|
|
@pytest.fixture
|
|
def schedule_service(tmp_path, monkeypatch):
|
|
for variable in (
|
|
"COGNITIVE_AGENT_HARDWARE_EXECUTION_ENABLED", "COGNITIVE_AGENT_VEHICLE_CAN_ENABLED",
|
|
"COGNITIVE_AGENT_WIRED_PASSENGER_CAN_ENABLED", "COGNITIVE_AGENT_VEHICLE_MQTT_ENABLED",
|
|
):
|
|
monkeypatch.setenv(variable, "false")
|
|
service = SeatDemoService(REPOSITORY_ROOT)
|
|
service.artifact_root = tmp_path / "artifacts"
|
|
user_root = tmp_path / "user-modes"
|
|
service.mode_profiles = ModeProfileRegistry(REPOSITORY_ROOT / "config" / "modes", user_root=user_root)
|
|
with (
|
|
patch.object(service.hardware_execution, "reserve", side_effect=AssertionError("physical vehicle access forbidden")) as reserve,
|
|
patch.object(service._vehicle_mqtt_transport, "bus", side_effect=AssertionError("physical MQTT access forbidden")),
|
|
):
|
|
try:
|
|
yield service, AssistantCapabilityCatalog(service), user_root
|
|
finally:
|
|
service.close()
|
|
reserve.assert_not_called()
|
|
|
|
|
|
def assistant_preview(service, catalog, message, conversation="conv-schedule-flow", client=None):
|
|
if client is None:
|
|
client = MagicMock()
|
|
client.available = False
|
|
assistant = CodexAssistantService(client)
|
|
snapshot = catalog.snapshot(vehicle_id="denza_n7")
|
|
fast = catalog.fast_answer(message, snapshot)
|
|
results, events, passed_plans = [], [], []
|
|
|
|
def execute(instruction, progress, assistant_plan=None):
|
|
passed_plans.append(deepcopy(assistant_plan))
|
|
result = service.preview_native_can(
|
|
instruction, provider="local", conversation_id=conversation,
|
|
execution_mode="hardware", vehicle_id="denza_n7",
|
|
on_commands_validated=progress, assistant_plan=assistant_plan,
|
|
)
|
|
results.append(result)
|
|
return result
|
|
|
|
response = assistant.stream(
|
|
message, conversation_id=conversation, principal="schedule-test", history=[],
|
|
authoritative_answer=fast.answer if fast else None,
|
|
emit=lambda event: events.append(event) or True,
|
|
execute_control=execute, capability_snapshot=snapshot,
|
|
capability_context=catalog.compact_context(snapshot),
|
|
scene_capability_context=catalog.scene_composition_context(snapshot),
|
|
control_terms=catalog.control_terms(snapshot),
|
|
)
|
|
assert len(results) == 1, (response, results)
|
|
return response, results[0], events, passed_plans, client
|
|
|
|
|
|
def create_saved(service, catalog, message="构建一个上车预热的功能模式"):
|
|
response, preview, _, _, _ = assistant_preview(service, catalog, message)
|
|
assert response["success"] is True, (response, preview)
|
|
draft = preview["mode_profile_draft"]
|
|
saved = service.create_mode_profile_card(
|
|
draft["name"], draft["instruction"], draft["source_request_id"], draft["vehicle_id"],
|
|
)
|
|
assert saved["status"] == "succeeded"
|
|
return preview, saved["mode_profile"]
|
|
|
|
|
|
@pytest.mark.parametrize(("message", "template_id", "count", "repeats", "elapsed"), [
|
|
("构建一个座椅的动态模式", "seat-dynamic", 4, 5, 300000),
|
|
("构建一个上车预热的功能模式", "boarding-warmup", 7, 1, 420000),
|
|
("帮我创建一个座椅动态模式吧", "seat-dynamic", 4, 5, 300000),
|
|
("我想构建一个上车预热模式并预览", "boarding-warmup", 7, 1, 420000),
|
|
("构建一个上车暖舱模式", "boarding-warmup", 7, 1, 420000),
|
|
("构建一个空调预热模式", "boarding-warmup", 7, 1, 420000),
|
|
("构建一个动态座椅模式", "seat-dynamic", 4, 5, 300000),
|
|
("构建一个座椅交替放松模式", "seat-dynamic", 4, 5, 300000),
|
|
])
|
|
def test_user_temporal_requests_create_complete_multistage_template_without_model(
|
|
schedule_service, message, template_id, count, repeats, elapsed,
|
|
):
|
|
service, catalog, user_root = schedule_service
|
|
response, preview, events, plans, client = assistant_preview(service, catalog, message)
|
|
assert response["success"] is True, (response, preview)
|
|
assert response["route"] == "scene_template"
|
|
assert preview["status"] == "succeeded"
|
|
assert preview["execution_mode"] == "dry_run"
|
|
assert preview["hardware_execution_allowed"] is False
|
|
assert preview["orchestration_route"] == "mode_profile_create_preview"
|
|
draft = preview["mode_profile_draft"]
|
|
assert draft["source_template_id"] == template_id
|
|
assert draft["generation_source"] == "scene_template_copy"
|
|
assert draft["operation_count"] == count
|
|
expected_schedule = deepcopy(plans[0]["schedule"])
|
|
for phase in expected_schedule["phases"]:
|
|
phase["operations"] = [
|
|
op for op in phase["operations"]
|
|
if op["domain"] != "seat" or op["target_scope"] == "passenger"
|
|
]
|
|
assert draft["schedule"] == preview["schedule"] == expected_schedule
|
|
schedule = ModeSchedule.from_dict(preview["schedule"])
|
|
assert schedule.repeat_count == repeats
|
|
assert schedule.total_duration_ms == elapsed
|
|
actions = preview["intent_ir"]["spec"]["actions"]
|
|
assert len(actions) == count
|
|
assert len({action["id"] for action in actions}) == count
|
|
assert {item["phase_id"] for item in preview["schedule_phase_previews"]} == {phase.id for phase in schedule.phases}
|
|
assert len(preview["schedule_phase_previews"]) == 2
|
|
assert all(item["status"] == "succeeded" for item in preview["schedule_phase_previews"])
|
|
assert all(any(key.startswith(phase.id + "-") for key in preview["command_plans"]) for phase in schedule.phases)
|
|
assert {
|
|
op["target_scope"] for phase in schedule.phases for op in phase.operations
|
|
if op["domain"] == "seat"
|
|
} == {"passenger"}
|
|
assert events
|
|
client.start_thread.assert_not_called()
|
|
client.start_turn.assert_not_called()
|
|
client.start_structured_turn.assert_not_called()
|
|
assert not list(user_root.rglob("*.yaml"))
|
|
assert service.mode_schedule_runner.status("denza_n7")["status"] == "idle"
|
|
|
|
|
|
@pytest.mark.parametrize("message", ["构建一个座椅的动态模式", "构建一个上车预热的功能模式"])
|
|
def test_draft_trial_preserves_all_stages_repeats_and_virtual_time_without_thread(schedule_service, message):
|
|
service, catalog, _ = schedule_service
|
|
_, preview, _, _, _ = assistant_preview(service, catalog, message)
|
|
draft = preview["mode_profile_draft"]
|
|
with patch("cognitive_agent.application.mode_schedule.Thread", side_effect=AssertionError("dry-run spawned worker")):
|
|
trial = service.trial_mode_profile_draft(
|
|
draft["name"], draft["instruction"], draft["source_request_id"], draft["vehicle_id"],
|
|
)
|
|
assert trial["status"] == "succeeded"
|
|
assert trial["hardware_execution_allowed"] is False
|
|
assert trial["schedule"] == preview["schedule"]
|
|
execution = trial["schedule_execution"]
|
|
schedule = ModeSchedule.from_dict(preview["schedule"])
|
|
assert execution["completed"] is True
|
|
assert execution["virtual_elapsed_ms"] == schedule.total_duration_ms
|
|
assert execution["executed_phases"] == len(schedule.phases) * schedule.repeat_count
|
|
assert [step["phase_id"] for step in execution["phase_results"]] == [phase.id for phase in schedule.phases] * schedule.repeat_count
|
|
assert len(trial["intent_ir"]["spec"]["actions"]) == schedule.operation_count
|
|
|
|
|
|
@pytest.mark.parametrize("message", ["构建一个座椅的动态模式", "构建一个上车预热的功能模式"])
|
|
def test_save_reload_and_direct_execute_keep_temporal_semantics(schedule_service, message):
|
|
service, catalog, user_root = schedule_service
|
|
preview, public_profile = create_saved(service, catalog, message)
|
|
assert public_profile["schedule"] == preview["schedule"]
|
|
assert public_profile["behavior"] == "cyclic"
|
|
mode_id = public_profile["id"]
|
|
service.mode_profiles = ModeProfileRegistry(REPOSITORY_ROOT / "config" / "modes", user_root=user_root)
|
|
reloaded = service.mode_profiles.get_profile(mode_id, "denza_n7")
|
|
assert reloaded.schedule.to_dict() == preview["schedule"]
|
|
with patch("cognitive_agent.application.mode_schedule.Thread", side_effect=AssertionError("dry-run spawned worker")):
|
|
executed = service.execute_mode_profile_card(
|
|
mode_id, conversation_id="conv-schedule-execute", execution_mode="dry_run", vehicle_id="denza_n7",
|
|
)
|
|
assert executed["status"] == "succeeded", executed
|
|
assert executed["orchestration_route"] == "mode_schedule_execute"
|
|
assert executed["execution_mode"] == "dry_run"
|
|
assert executed["hardware_execution_allowed"] is False
|
|
assert "mode_profile_draft" not in executed
|
|
execution = executed["schedule_execution"]
|
|
assert execution["completed"] is True
|
|
assert execution["virtual_elapsed_ms"] == reloaded.schedule.total_duration_ms
|
|
assert len(execution["phase_results"]) == len(reloaded.schedule.phases) * reloaded.schedule.repeat_count
|
|
expected_phases = list(reloaded.schedule.phases) * reloaded.schedule.repeat_count
|
|
for expected, actual in zip(expected_phases, execution["phase_results"], strict=True):
|
|
phase_result = actual["result"]
|
|
assert phase_result["status"] == "succeeded"
|
|
assert phase_result["execution_mode"] == "dry_run"
|
|
actual_actions = phase_result["intent_ir"]["spec"]["actions"]
|
|
assert len(actual_actions) == len(expected.operations)
|
|
expected_params = sorted(json.dumps(op["parameters"], sort_keys=True) for op in expected.operations)
|
|
actual_params = sorted(json.dumps(op["parameters"], sort_keys=True) for op in actual_actions)
|
|
assert actual_params == expected_params
|
|
assert service.mode_schedule_runner.status("denza_n7")["status"] == "idle"
|
|
|
|
|
|
def test_natural_language_enter_saved_schedule_runs_timeline_not_flat_actions(schedule_service):
|
|
service, catalog, _ = schedule_service
|
|
preview, saved = create_saved(service, catalog)
|
|
result = service.process(
|
|
f"进入{saved['name']}", provider="local", conversation_id="conv-schedule-natural-enter",
|
|
execution_mode="dry_run", vehicle_id="denza_n7",
|
|
)
|
|
assert result["status"] == "succeeded", result
|
|
assert result["orchestration_route"] == "mode_schedule_execute"
|
|
assert result["schedule"] == preview["schedule"]
|
|
temperatures = []
|
|
for phase in result["schedule_execution"]["phase_results"]:
|
|
temperatures.extend(action["parameters"]["temperature_c"] for action in phase["result"]["intent_ir"]["spec"]["actions"] if action["capability"] == "climate.set_temperature")
|
|
assert temperatures == [28, 24]
|
|
|
|
|
|
def test_missing_hardware_lease_rejects_before_any_worker_or_physical_access(schedule_service):
|
|
service, catalog, _ = schedule_service
|
|
_, saved = create_saved(service, catalog)
|
|
with (
|
|
patch("cognitive_agent.application.mode_schedule.Thread", side_effect=AssertionError("unauthorized worker started")),
|
|
pytest.raises(HardwareAccessRequiredError),
|
|
):
|
|
service.execute_mode_profile_card(
|
|
saved["id"], execution_mode="hardware", vehicle_id="denza_n7",
|
|
conversation_id="conv-schedule-unauthorized", hardware_access_bypass=False,
|
|
hardware_client_id="hwc-schedule-unauthorized",
|
|
)
|
|
assert service.mode_schedule_runner.status("denza_n7")["status"] == "idle"
|
|
|
|
|
|
@pytest.mark.parametrize(("field", "value"), [
|
|
("name", "其他模式"), ("instruction", "直接执行危险指令"),
|
|
("vehicle_id", "other_vehicle"), ("source_request_id", "../../result"),
|
|
])
|
|
def test_client_cannot_rebind_schedule_draft_trial(schedule_service, field, value):
|
|
service, catalog, user_root = schedule_service
|
|
_, preview, _, _, _ = assistant_preview(service, catalog, "构建一个上车预热的功能模式")
|
|
draft = preview["mode_profile_draft"]
|
|
arguments = {key: draft[key] for key in ("name", "instruction", "source_request_id", "vehicle_id")}
|
|
arguments[field] = value
|
|
with pytest.raises(ValueError):
|
|
service.trial_mode_profile_draft(**arguments)
|
|
assert not list(user_root.rglob("*.yaml"))
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ["capability", "duration", "temperature", "target"])
|
|
def test_tampered_template_schedule_never_creates_or_executes_draft(schedule_service, mutation):
|
|
service, catalog, user_root = schedule_service
|
|
template = next(item for item in catalog.snapshot(vehicle_id="denza_n7")["scene_templates"] if item["id"] == "boarding-warmup")
|
|
plan = deepcopy(template["plan"])
|
|
phase = plan["schedule"]["phases"][0]
|
|
if mutation == "capability":
|
|
phase["operations"][0]["capability"] = "door.set_vehicle_lock"
|
|
elif mutation == "duration":
|
|
phase["duration_ms"] = 1
|
|
elif mutation == "temperature":
|
|
phase["operations"][1]["parameters"]["temperature_c"] = 99
|
|
else:
|
|
phase["operations"][3]["target_scope"] = "passenger"
|
|
try:
|
|
result = service.preview_native_can(
|
|
"构建一个上车预热的功能模式", provider="local", conversation_id="conv-schedule-tamper",
|
|
execution_mode="hardware", vehicle_id="denza_n7", assistant_plan=plan,
|
|
)
|
|
except ValueError:
|
|
result = None
|
|
if result is not None:
|
|
assert result["status"] != "succeeded"
|
|
assert "mode_profile_draft" not in result
|
|
assert result.get("hardware_execution_allowed") is not True
|
|
assert not list(user_root.rglob("*.yaml"))
|
|
assert service.mode_schedule_runner.status("denza_n7")["status"] == "idle"
|
|
|
|
|
|
def test_invalid_later_saved_phase_is_revalidated_before_any_phase_executes(schedule_service):
|
|
service, catalog, _ = schedule_service
|
|
_, saved = create_saved(service, catalog)
|
|
profile = service.mode_profiles.get_profile(saved["id"], "denza_n7")
|
|
# Represents a stale or externally edited user profile, not a trusted runtime
|
|
# capability change. Its structure remains valid, but the target is unsafe.
|
|
profile.schedule.phases[1].operations[0]["parameters"]["temperature_c"] = 99
|
|
with patch.object(service.mode_schedule_runner, "simulate", side_effect=AssertionError("invalid later phase reached executor")):
|
|
result = service.execute_mode_profile_card(
|
|
saved["id"], conversation_id="conv-schedule-stale-capability",
|
|
execution_mode="dry_run", vehicle_id="denza_n7",
|
|
)
|
|
assert result["status"] == "validation_failed", result
|
|
assert result["mode_creation_failure"]["code"] == "MODE_SCHEDULE_PHASE_INVALID"
|
|
assert "mode_profile_draft" not in result
|
|
assert service.mode_schedule_runner.status("denza_n7")["status"] == "idle"
|
|
|
|
|
|
def test_execution_failure_after_validated_first_phase_prevents_later_repeats(schedule_service):
|
|
service, catalog, _ = schedule_service
|
|
_, saved = create_saved(service, catalog, "构建一个主驾和副驾的座椅动态模式")
|
|
preview_method = service._preview_mode_schedule
|
|
virtual_method = service.native_can_runtime.execute_virtual
|
|
execution_started = False
|
|
executed_massage_modes = []
|
|
|
|
def validate_then_arm(*args, **kwargs):
|
|
nonlocal execution_started
|
|
result = preview_method(*args, **kwargs)
|
|
execution_started = True
|
|
return result
|
|
|
|
def fault_on_second_phase(command_plan, function_config):
|
|
if execution_started:
|
|
for op in function_config["spec"]["operations"]:
|
|
if op["capability"] == "seat.set_driver_massage_native":
|
|
executed_massage_modes.append(op["parameters"]["mode"])
|
|
if op["parameters"]["mode"] == "slack":
|
|
raise RuntimeError("injected lower-level actuator failure")
|
|
return virtual_method(command_plan, function_config)
|
|
|
|
with (
|
|
patch.object(service, "_preview_mode_schedule", side_effect=validate_then_arm),
|
|
patch.object(service.native_can_runtime, "execute_virtual", side_effect=fault_on_second_phase),
|
|
):
|
|
result = service.execute_mode_profile_card(
|
|
saved["id"], conversation_id="conv-schedule-injected-failure",
|
|
execution_mode="dry_run", vehicle_id="denza_n7",
|
|
)
|
|
execution = result["schedule_execution"]
|
|
assert result["status"] == "failed", result
|
|
assert execution["completed"] is False
|
|
assert execution["error"]
|
|
assert execution["executed_phases"] == 1
|
|
assert execution["virtual_elapsed_ms"] == 30000
|
|
assert execution["repeat_index"] == 0
|
|
assert len(execution["phase_results"]) == 2
|
|
assert executed_massage_modes == ["relax", "slack"]
|
|
assert result["hardware_execution_allowed"] is False
|
|
|
|
|
|
def structured_schedule():
|
|
return {
|
|
"requests": [], "ambiguities": [], "confidence": 0.96,
|
|
"schedule": {
|
|
"repeat_count": 2,
|
|
"phases": [
|
|
{"id": "warm", "name": "暖舱", "duration_ms": 30000,
|
|
"requests": [{"capability": "climate.set_temperature", "target_scope": "vehicle", "parameters": {"temperature_c": 28}}]},
|
|
{"id": "settle", "name": "恒温", "duration_ms": 60000,
|
|
"requests": [{"capability": "climate.set_temperature", "target_scope": "vehicle", "parameters": {"temperature_c": 24}}]},
|
|
],
|
|
},
|
|
}
|
|
|
|
|
|
def structured_client(output):
|
|
client = MagicMock()
|
|
client.available = True
|
|
client.status = "ready"
|
|
client.start_thread.return_value = "thread-schedule-test"
|
|
client.start_turn.side_effect = AssertionError("dynamic creation fell back to unstructured agent")
|
|
|
|
def start(thread_id, text, *, application_context, output_schema, timeout=None):
|
|
Draft202012Validator(output_schema).validate(output)
|
|
turn = ActiveCodexTurn(thread_id=thread_id, turn_id="turn-schedule-test", events=queue.Queue())
|
|
turn.events.put({"method": "item/completed", "params": {"item": {"id": "answer", "type": "agentMessage", "text": json.dumps(output, ensure_ascii=False)}}})
|
|
turn.events.put({"method": "turn/completed", "params": {"turn": {"id": turn.turn_id, "status": "completed", "items": [], "error": None}}})
|
|
return turn
|
|
|
|
client.start_structured_turn.side_effect = start
|
|
return client
|
|
|
|
|
|
@pytest.mark.parametrize("message", [
|
|
"构建一个星云动态模式，空调先28度持续30秒，然后24度持续60秒，循环2次",
|
|
"构建一个阅读模式，空调先28度持续30秒，然后24度持续60秒，循环2次",
|
|
"构建一个上车预热的功能模式，空调先28度持续30秒，然后24度持续60秒，循环2次",
|
|
])
|
|
def test_explicit_temporal_parameters_use_structured_composition_not_static_template(schedule_service, message):
|
|
service, catalog, user_root = schedule_service
|
|
output = structured_schedule()
|
|
client = structured_client(output)
|
|
response, preview, _, plans, _ = assistant_preview(service, catalog, message, client=client)
|
|
assert response["route"] == "scene_composer", response
|
|
assert response["success"] is True, (response, preview)
|
|
assert preview["mode_profile_draft"]["generation_source"] != "scene_template_copy"
|
|
client.start_structured_turn.assert_called_once()
|
|
client.start_turn.assert_not_called()
|
|
assert "schedule" in client.start_structured_turn.call_args.kwargs["output_schema"]["properties"]
|
|
assert preview["schedule"] == plans[0]["schedule"]
|
|
assert preview["schedule"]["repeat_count"] == 2
|
|
assert [phase["duration_ms"] for phase in preview["schedule"]["phases"]] == [30000, 60000]
|
|
assert [phase["operations"][0]["parameters"]["temperature_c"] for phase in preview["schedule"]["phases"]] == [28, 24]
|
|
assert ModeSchedule.from_dict(preview["schedule"]).total_duration_ms == 180000
|
|
assert not list(user_root.rglob("*.yaml"))
|
|
|
|
|
|
def test_dynamic_schema_normalization_preserves_complete_phase_contract(schedule_service):
|
|
_, catalog, _ = schedule_service
|
|
context = catalog.scene_composition_context(catalog.snapshot(vehicle_id="denza_n7"))
|
|
output = structured_schedule()
|
|
schema = build_scene_output_schema(context, dynamic=True)
|
|
Draft202012Validator(schema).validate(output)
|
|
plan = normalize_scene_output(output, context, dynamic=True)
|
|
assert plan["ambiguities"] == []
|
|
schedule = ModeSchedule.from_dict(plan["schedule"])
|
|
assert schedule.total_duration_ms == 180000
|
|
assert schedule.operation_count == 2
|
|
assert schedule.phases[0].operations[0]["domain"] == "climate"
|
|
assert schedule.phases[1].operations[0]["parameters"] == {"temperature_c": 24}
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", [
|
|
"static-root", "unknown-root", "unknown-schedule", "unknown-phase", "unknown-capability",
|
|
"duplicate-phase", "duplicate-operation", "no-phases", "empty-phase", "bool-duration",
|
|
"bool-repeat", "long-duration", "long-total", "nan-parameter", "motion-capability",
|
|
])
|
|
def test_invalid_structured_dynamic_plans_fail_closed_during_normalization(schedule_service, mutation):
|
|
_, catalog, _ = schedule_service
|
|
context = catalog.scene_composition_context(catalog.snapshot(vehicle_id="denza_n7"))
|
|
output = structured_schedule()
|
|
schedule = output["schedule"]
|
|
phase = schedule["phases"][0]
|
|
if mutation == "static-root":
|
|
output["requests"] = deepcopy(phase["requests"])
|
|
elif mutation == "unknown-root":
|
|
output["execute_now"] = True
|
|
elif mutation == "unknown-schedule":
|
|
schedule["unbounded"] = True
|
|
elif mutation == "unknown-phase":
|
|
phase["skip_safety"] = True
|
|
elif mutation == "unknown-capability":
|
|
phase["requests"][0]["capability"] = "unknown.actuate"
|
|
elif mutation == "motion-capability":
|
|
phase["requests"][0] = {"capability": "seat.set_driver_axis_native", "target_scope": "driver", "parameters": {"duration_ms": 1000, "axis": "backrest", "direction": "backward"}}
|
|
elif mutation == "duplicate-phase":
|
|
schedule["phases"][1]["id"] = phase["id"]
|
|
elif mutation == "duplicate-operation":
|
|
phase["requests"].append(deepcopy(phase["requests"][0]))
|
|
elif mutation == "no-phases":
|
|
schedule["phases"] = []
|
|
elif mutation == "empty-phase":
|
|
phase["requests"] = []
|
|
elif mutation == "bool-duration":
|
|
phase["duration_ms"] = True
|
|
elif mutation == "bool-repeat":
|
|
schedule["repeat_count"] = True
|
|
elif mutation == "long-duration":
|
|
phase["duration_ms"] = 900001
|
|
elif mutation == "long-total":
|
|
schedule["repeat_count"] = 20
|
|
phase["duration_ms"] = 900000
|
|
elif mutation == "nan-parameter":
|
|
phase["requests"][0]["parameters"]["temperature_c"] = float("nan")
|
|
with pytest.raises(SceneCompositionError):
|
|
normalize_scene_output(output, context, dynamic=True)
|
|
|
|
|
|
def test_dynamic_ambiguity_produces_no_schedule_or_executable_actions(schedule_service):
|
|
_, catalog, _ = schedule_service
|
|
context = catalog.scene_composition_context(catalog.snapshot(vehicle_id="denza_n7"))
|
|
output = {"requests": [], "ambiguities": ["当前未开放该动态动作，请指定其他设备。"], "confidence": 0.2, "schedule": None}
|
|
Draft202012Validator(build_scene_output_schema(context, dynamic=True)).validate(output)
|
|
plan = normalize_scene_output(output, context, dynamic=True)
|
|
assert not plan["requests"]
|
|
assert "schedule" not in plan
|
|
assert plan["ambiguities"]
|
|
|
|
|
|
def test_model_schedule_with_unresolved_requirement_never_generates_successful_draft(schedule_service):
|
|
service, catalog, user_root = schedule_service
|
|
context = catalog.scene_composition_context(catalog.snapshot(vehicle_id="denza_n7"))
|
|
output = structured_schedule()
|
|
output["ambiguities"] = ["不能完成要求的主驾座椅绝对位置控制。"]
|
|
plan = normalize_scene_output(output, context, dynamic=True)
|
|
assert "metadata" not in plan
|
|
with pytest.raises(ValueError, match="未解决"):
|
|
service.preview_native_can(
|
|
"构建一个流光动态模式", provider="local", conversation_id="conv-schedule-ambiguous",
|
|
execution_mode="hardware", vehicle_id="denza_n7", assistant_plan=plan,
|
|
)
|
|
assert not list(user_root.rglob("*.yaml"))
|
|
assert service.mode_schedule_runner.status("denza_n7")["status"] == "idle"
|
|
|
|
|
|
@pytest.mark.parametrize(("seat", "allowed_targets"), [
|
|
("主驾", {"driver", "front_left"}), ("副驾", {"passenger", "front_right"}),
|
|
])
|
|
def test_single_seat_dynamic_execution_cannot_expand_to_both_seats(schedule_service, seat, allowed_targets):
|
|
service, catalog, _ = schedule_service
|
|
_, saved = create_saved(service, catalog, "构建一个主驾和副驾的座椅动态模式")
|
|
result = service.process(
|
|
f"只让{seat}进入{saved['name']}", provider="local", conversation_id="conv-schedule-single-seat",
|
|
execution_mode="dry_run", vehicle_id="denza_n7",
|
|
)
|
|
if result["status"] == "succeeded":
|
|
assert result["orchestration_route"] == "mode_schedule_execute"
|
|
execution = result["schedule_execution"]
|
|
for phase in execution["phase_results"]:
|
|
actions = phase["result"]["intent_ir"]["spec"]["actions"]
|
|
seat_targets = {action["target"].get("instance") or action["target"].get("zone") for action in actions if action["domain"] == "seat"}
|
|
assert seat_targets and seat_targets <= allowed_targets
|
|
else:
|
|
assert result["status"] in {"validation_failed", "needs_clarification", "needs_confirmation", "failed"}
|
|
assert not result.get("schedule_execution", {}).get("phase_results")
|
|
assert result.get("hardware_execution_allowed") is not True
|
|
|
|
|
|
def start_waiting_schedule(service):
|
|
"""Run only a no-device test callback, holding until the tested stop request."""
|
|
calls = []
|
|
entered = Event()
|
|
schedule = ModeSchedule.from_dict({
|
|
"schema": "cognitive-agent-mode-schedule/v1", "repeat_count": 1,
|
|
"phases": [
|
|
{"id": phase_id, "name": phase_id, "duration_ms": 900000,
|
|
"operations": [{"domain": "climate", "capability": "climate.set_temperature",
|
|
"target_scope": "vehicle", "operation": "set",
|
|
"parameters": {"temperature_c": temperature}, "reason": "test only"}]}
|
|
for phase_id, temperature in (("first", 28), ("second", 24))
|
|
],
|
|
})
|
|
|
|
def execute(phase):
|
|
calls.append(phase.id)
|
|
entered.set()
|
|
return {"status": "succeeded"}
|
|
|
|
service.mode_schedule_runner.start("denza_n7", schedule, execute, session_id="conv-stop-status")
|
|
assert entered.wait(2)
|
|
assert service.mode_schedule_runner.status("denza_n7")["status"] == "running"
|
|
return calls
|
|
|
|
|
|
def assistant_control(catalog, message, execute):
|
|
client = MagicMock()
|
|
client.available = False
|
|
assistant = CodexAssistantService(client)
|
|
snapshot = catalog.snapshot(vehicle_id="denza_n7")
|
|
events = []
|
|
response = assistant.stream(
|
|
message, conversation_id="conv-stop-status", principal="schedule-status-test", history=[],
|
|
emit=lambda event: events.append(event) or True,
|
|
execute_control=execute, capability_snapshot=snapshot,
|
|
capability_context=catalog.compact_context(snapshot),
|
|
scene_capability_context=catalog.scene_composition_context(snapshot),
|
|
control_terms=catalog.control_terms(snapshot),
|
|
)
|
|
client.start_thread.assert_not_called()
|
|
client.start_structured_turn.assert_not_called()
|
|
client.start_turn.assert_not_called()
|
|
return response, events
|
|
|
|
|
|
def test_dry_run_stop_does_not_cancel_a_running_vehicle_schedule(schedule_service):
|
|
service, _, _ = schedule_service
|
|
calls = start_waiting_schedule(service)
|
|
result = service.process(
|
|
"停止动态模式", provider="local", conversation_id="conv-stop-status",
|
|
execution_mode="dry_run", vehicle_id="denza_n7",
|
|
)
|
|
assert result["status"] == "succeeded"
|
|
assert result["schedule_execution"]["status"] == "simulated"
|
|
assert result["hardware_execution_allowed"] is False
|
|
assert service.mode_schedule_runner.status("denza_n7")["status"] == "running"
|
|
assert calls == ["first"]
|
|
persisted = json.loads((service.artifact_root / result["request_id"] / "result.json").read_text(encoding="utf-8"))
|
|
assert persisted["orchestration_route"] == "mode_schedule_stop"
|
|
assert persisted["trace_events"] == []
|
|
|
|
|
|
def test_hardware_stop_cancels_future_stages_and_persists_the_result(schedule_service):
|
|
service, _, _ = schedule_service
|
|
calls = start_waiting_schedule(service)
|
|
result = service.process(
|
|
"停止动态模式", provider="local", conversation_id="conv-stop-status",
|
|
execution_mode="hardware", vehicle_id="denza_n7",
|
|
)
|
|
assert result["status"] == "succeeded"
|
|
assert result["schedule_execution"]["status"] == "cancelled"
|
|
assert result["schedule_execution"]["completed"] is False
|
|
assert service.mode_schedule_runner.status("denza_n7")["status"] == "cancelled"
|
|
assert service.mode_schedule_runner.close(timeout=1)
|
|
assert calls == ["first"]
|
|
persisted = json.loads((service.artifact_root / result["request_id"] / "result.json").read_text(encoding="utf-8"))
|
|
assert persisted["orchestration_route"] == "mode_schedule_stop"
|
|
assert persisted["schedule_execution"]["status"] == "cancelled"
|
|
assert persisted["trace_events"] == []
|
|
|
|
|
|
@pytest.mark.parametrize(("execution_mode", "action_state", "runner_state"), [
|
|
("hardware", "cancelled", "cancelled"), ("dry_run", "simulated", "running"),
|
|
])
|
|
def test_assistant_stop_is_zero_model_success_and_preserves_execution_policy(
|
|
schedule_service, execution_mode, action_state, runner_state,
|
|
):
|
|
service, catalog, _ = schedule_service
|
|
calls = start_waiting_schedule(service)
|
|
|
|
def execute(message, progress):
|
|
return service.process(
|
|
message, provider="local", conversation_id="conv-stop-status",
|
|
execution_mode=execution_mode, vehicle_id="denza_n7",
|
|
)
|
|
|
|
response, events = assistant_control(catalog, "停止动态模式", execute)
|
|
assert response["success"] is True, response
|
|
assert response["route"] == "vehicle_control"
|
|
action = next(event for event in events if event["type"] == "action.completed")
|
|
assert action["success"] is True
|
|
assert action["state"] == action_state
|
|
assert action["payload"]["orchestration_route"] == "mode_schedule_stop"
|
|
assert service.mode_schedule_runner.status("denza_n7")["status"] == runner_state
|
|
assert calls == ["first"]
|
|
assert "已完成" not in response["answer"]
|
|
assert "未改变正在运行" in response["answer"] if execution_mode == "dry_run" else "已停止" in response["answer"]
|
|
|
|
|
|
def test_assistant_running_schedule_is_accepted_without_claiming_all_phases_completed(schedule_service):
|
|
service, catalog, _ = schedule_service
|
|
preview, profile = create_saved(service, catalog)
|
|
start_waiting_schedule(service)
|
|
state = service.mode_schedule_runner.status("denza_n7")
|
|
state["phase_results"] = [{"command_frames": ["sensitive-can-frame"]}]
|
|
|
|
def execute(message, progress):
|
|
return {
|
|
"status": "running", "execution_mode": "hardware", "hardware_execution_allowed": True,
|
|
"orchestration_route": "mode_schedule_execute", "schedule_execution": state,
|
|
"mode_profile": profile, "matched_modes": [profile], "schedule": preview["schedule"],
|
|
}
|
|
|
|
response, events = assistant_control(catalog, f"进入{profile['name']}", execute)
|
|
assert response["success"] is True, response
|
|
action = next(event for event in events if event["type"] == "action.completed")
|
|
assert action["success"] is True
|
|
assert action["state"] == "running"
|
|
assert action["payload"]["status"] == "running"
|
|
assert action["payload"]["schedule_execution"]["completed"] is False
|
|
assert "phase_results" not in action["payload"]["schedule_execution"]
|
|
assert "sensitive-can-frame" not in json.dumps(events, ensure_ascii=False)
|
|
assert "尚未完成所有阶段" in action["display_text"]
|
|
assert "尚未完成所有阶段" in response["answer"]
|
|
completed = next(event for event in events if event["type"] == "work.completed")
|
|
assert completed["state"] == "running"
|
|
assert "已完成" not in completed["display_text"]
|
|
|
|
|
|
def test_public_schedule_status_exposes_only_lifecycle_and_redacts_raw_error():
|
|
state = {
|
|
"run_id": "run-public", "vehicle_id": "denza_n7", "status": "failed",
|
|
"phase_index": 1, "repeat_index": 0, "completed": False,
|
|
"error": "secret-token /private/path/driver.dll CAN payload 123456",
|
|
"phase_results": [{"command_plan": {"data_hex": "deadbeef"}}],
|
|
"raw_error": {"token": "secret-token"}, "command_plans": {"internal": "data"},
|
|
"phase_count": {"nested": "unexpected"},
|
|
}
|
|
public = public_schedule_status(state)
|
|
assert public["status"] == "failed"
|
|
assert public["phase_index"] == 1
|
|
assert public["completed"] is False
|
|
assert public["error"] and public["error"] != state["error"]
|
|
assert set(public) == {"run_id", "vehicle_id", "status", "phase_index", "repeat_index", "completed", "error"}
|
|
for sensitive in ("secret-token", "driver.dll", "deadbeef", "123456"):
|
|
assert sensitive not in json.dumps(public, ensure_ascii=False)
|
|
|
|
|
|
@pytest.mark.parametrize("message", ["外循环通风模式", "内循环模式"])
|
|
def test_hvac_circulation_state_is_not_a_temporal_schedule(message):
|
|
assert needs_mode_schedule(message) is False
|