first commit

.gitignore

@@ -0,0 +1,8 @@
+__pycache__/
+*.py[cod]
+*.egg-info/
+.pytest_cache/
+.coverage
+build/
+dist/
+.idea/

README.md

@@ -0,0 +1,134 @@
+# cockpit-agent
+
+`cockpit-agent` implements a single-pass automotive cockpit task pipeline:
+
+```text
+instruction
+  -> SemanticAction
+  -> function semantic target
+  -> function grounding
+  -> ROI crop
+  -> UIState
+  -> ActionPlan
+  -> action grounding (when needed)
+  -> MockAction
+```
+
+It owns intent parsing, current UI understanding, deterministic planning, and
+mock action generation. The existing `cockpit-ui-grounding` package owns:
+
+```text
+image + target -> bbox -> pixel (u, v)
+```
+
+The current stage is deliberately limited to **NO ROBOT**, **NO FEEDBACK**, and
+**NO CLOSED LOOP**. It does not connect to ROS, a mechanical arm, or any real
+touch API, and it does not verify, retry, or replan an action.
+
+## Setup
+
+All model paths in `configs/default.toml` are local. No network model ID is
+used.
+
+```bash
+conda activate qwen3vl
+pip install -e /data/lgv/projects/cockpit-ui-grounding
+pip install -e /data/lgv/projects/cockpit-agent
+export CUDA_VISIBLE_DEVICES=0
+```
+
+The default configuration uses one shared Qwen3.5-2B instance for intent, UI
+understanding, and grounding. `ModelRegistry` caches models by backend and
+resolved local path, so identical settings do not load duplicate weights.
+Function ROI padding is configured independently with
+`perception.roi_padding_ratio`; the default is `0.15`.
+
+## Run A Task
+
+```bash
+python scripts/run_task.py \
+  --image /data/lgv/datasets/cockpit-ui/raw/debug/ca_car.jpg \
+  --instruction "把主驾温度调到23度" \
+  --output-dir /data/lgv/runs/cockpit-agent/demo_driver_temp_23
+```
+
+Focused CLIs are also available:
+
+```bash
+python scripts/run_intent.py --instruction "打开内循环"
+
+python scripts/run_ui_understanding.py \
+  --image /data/lgv/datasets/cockpit-ui/raw/debug/ca_car.jpg \
+  --instruction "后备箱最大开度设置成80%" \
+  --output-dir /data/lgv/runs/cockpit-agent/ui_understanding
+```
+
+A successful task output directory contains:
+
+- `intent.json`
+- `function_grounding.json`
+- `roi.jpg`
+- `ui_state.json`
+- `plan.json`
+- `action_grounding.json` when a non-noop action is proposed
+- `action.json`
+- `result.jpg`
+- `run.json`
+
+The trace uses explicit stage names: `semantic_action`, `function_target`,
+`function_grounding`, `roi`, `ui_state`, `action_plan`, `action_target`,
+`action_grounding`, and `mock_action`. Reusing an output directory replaces
+the pipeline-owned artifacts so a failed run cannot retain stale results from
+an earlier run.
+
+If a stage raises an error, the pipeline writes `failure.json` and a partial
+`run.json` with `success`, `failed_stage`, `error_type`, `error_message`, and
+`raw_model_output`. It then re-raises the error so callers cannot mistake a
+failed run for a successful one.
+
+`action.json` is always a proposal. A tap contains a pixel and repeat count, a
+drag contains start/end pixels, and a no-op contains its reason. Nothing is
+executed outside the process.
+
+## Model Output Handling
+
+Intent and ROI UI model replies are parsed as JSON and validated against
+strict dataclass schemas before planning. A small deterministic normalization layer
+handles known representation-only differences such as `"100%"` versus `100`
+and Chinese unlock option labels versus `all_doors`; it does not invent missing
+visual state. UI observation fields are nullable. Slider observations separate
+endpoint values from `current_value` and expose nullable `track_visible` and
+`knob_visible` fields; an unobservable current value remains `null` rather than
+being guessed. The first implementation proposes a drag only when the model
+explicitly observes `orientation=horizontal`; unknown and vertical sliders fail
+instead of silently using horizontal geometry. Device state (`current_state`) is separate from visual choice
+state (`selection_state`). Common choice-state wording such as
+`active`/`selected` and `inactive`/`unselected` is normalized without changing
+the model's predicted control type.
+
+`SemanticTargetBuilder` creates a complete-control/group query before UI
+understanding and a specific action-control query after planning. The function
+bbox is padded by the configured ratio and cropped without resizing. ROI action
+boxes are converted back to source-camera pixels. Slider drags use the visually
+grounded knob center as the start and the grounded track plus
+`normalized_target` as the end. The adapter does not repair predictions with
+image-, vehicle-, color-, position-, or fixed-ROI rules.
+
+The benchmark integrity policy is documented in
+[`docs/generalization_rules.md`](docs/generalization_rules.md). In particular,
+unit tests validate deterministic program logic, while model benchmarks may
+fail and must report the model's real output.
+
+## Tests
+
+Planner tests are deterministic and do not load a model:
+
+```bash
+pytest -v
+```
+
+The tests also run with the standard library runner:
+
+```bash
+python -m unittest discover -s tests -v
+```

configs/default.toml

@@ -0,0 +1,18 @@
+[intent]
+backend = "qwen35"
+model_path = "/data/lgv/models/pretrained/Qwen3.5-2B"
+max_new_tokens = 256
+
+[ui_understanding]
+backend = "qwen35"
+model_path = "/data/lgv/models/pretrained/Qwen3.5-2B"
+max_new_tokens = 256
+
+[perception]
+roi_padding_ratio = 0.15
+
+[grounding]
+project_path = "/data/lgv/projects/cockpit-ui-grounding"
+backend = "qwen35"
+model_path = "/data/lgv/models/pretrained/Qwen3.5-2B"
+max_new_tokens = 128

docs/generalization_rules.md

@@ -0,0 +1,53 @@
+# Generalization and Benchmark Integrity
+
+The purpose of the automotive UI benchmark is to measure model capability.
+A model prediction is allowed to fail and must remain visible in the artifacts.
+
+## Architecture Boundary
+
+Business logic may be deterministic. Examples include calculating a stepper's
+repeat count, deciding that a toggle is already in the requested state, and
+normalizing a slider value from its observed minimum and maximum.
+
+Visual layout may not be deterministic. Planner and adapters must not infer a
+target from a vehicle brand, image filename, benchmark ID, color, neighboring
+control, screen region, fixed pixel, fixed bounding box, crop, or ROI. Planner
+operates on semantic functions and never on visual positions.
+
+`SemanticTargetBuilder` may convert structured semantic fields into natural
+language such as `driver temperature decrease control`. It must describe only
+the target's semantic identity. The grounding model is solely responsible for
+finding that target in the current image.
+
+Every perception ROI is derived from the function grounding bbox. Padding may
+be a configurable ratio of that bbox; a fixed screen crop is prohibited. UI
+Understanding receives only this dynamically generated ROI. Grounding inside
+the ROI is translated back to source-camera coordinates before a MockAction is
+created.
+
+Unobservable UI fields remain null. In particular, slider motion starts from a
+visually grounded knob, never from a guessed `current_value`.
+
+## Dataset Policy
+
+- Development data may be inspected while improving models, training data, or
+  general prompts.
+- Validation data is used for model selection.
+- Test data is frozen and must never be used to add matching rules.
+- No image-specific, vehicle-specific, or benchmark-case-specific branch is
+  permitted in production code.
+
+## Test Policy
+
+Unit tests validate deterministic schemas, parsing, planning, semantic query
+construction, coordinate conversion, and failure recording. They do not assert
+that a vision model must return a hand-observed box for a fixed image.
+
+Model benchmarks report semantic correctness separately for Intent, UI
+Understanding, Grounding, and the proposed MockAction. A schema-valid but
+semantically wrong bounding box is a benchmark failure, not a reason to add a
+postprocessing rule.
+
+Model capability problems should be addressed with representative automotive
+data, model training or fine-tuning, and evaluated model/prompt improvements.
+They must not be hidden with layout-specific code.

pyproject.toml

@@ -0,0 +1,15 @@
+[build-system]
+requires = ["setuptools>=68"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "cockpit-agent"
+version = "0.1.0"
+description = "Single-pass automotive cockpit UI agent with mock execution"
+requires-python = ">=3.11"
+
+[project.optional-dependencies]
+test = ["pytest>=8,<10"]
+
+[tool.setuptools.packages.find]
+where = ["src"]

scripts/run_intent.py

@@ -0,0 +1,28 @@
+import argparse
+import json
+from pathlib import Path
+
+from cockpit_agent.common.model_registry import ModelRegistry
+from cockpit_agent.config import load_config
+from cockpit_agent.intent.parser import ModelIntentParser
+
+
+DEFAULT_CONFIG = Path(__file__).resolve().parents[1] / "configs" / "default.toml"
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Parse a cockpit instruction")
+    parser.add_argument("--instruction", required=True)
+    parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG)
+    args = parser.parse_args()
+
+    config = load_config(args.config)
+    registry = ModelRegistry()
+    model = registry.get_text_generator(config.intent)
+    intent_parser = ModelIntentParser(model, config.intent.max_new_tokens)
+    intent = intent_parser.parse(args.instruction)
+    print(json.dumps(intent.to_dict(), indent=2, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    main()

scripts/run_task.py

@@ -0,0 +1,57 @@
+import argparse
+import json
+from pathlib import Path
+from typing import Any
+
+from cockpit_agent.common.model_registry import ModelRegistry
+from cockpit_agent.config import load_config
+from cockpit_agent.pipeline import build_pipeline
+
+
+DEFAULT_CONFIG = Path(__file__).resolve().parents[1] / "configs" / "default.toml"
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Run the single-pass cockpit agent with mock execution",
+    )
+    parser.add_argument("--image", required=True, help="Local cockpit image")
+    parser.add_argument("--instruction", required=True, help="User instruction")
+    parser.add_argument("--output-dir", required=True, help="Artifact directory")
+    parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG)
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    config = load_config(args.config)
+    registry = ModelRegistry()
+    pipeline = build_pipeline(config, registry)
+    print(f"Loaded model instances: {registry.loaded_model_count}")
+    run = pipeline.run(
+        image_path=args.image,
+        instruction=args.instruction,
+        output_dir=args.output_dir,
+    )
+    _print_section("INTENT", run["semantic_action"])
+    print(f"\nFunction target: {run['function_target']}")
+    _print_section("FUNCTION GROUNDING", run["function_grounding"])
+    _print_section("ROI", run["roi"])
+    _print_section("UI UNDERSTANDING", run["ui_state"])
+    _print_section("PLAN", run["action_plan"])
+    print(f"\nAction target: {run['action_target']}")
+    _print_section("ACTION GROUNDING", run["action_grounding"])
+    _print_section("MOCK ACTION", run["mock_action"])
+    print(f"\nResult image: {run['result_image']}")
+
+
+def _print_section(title: str, value: dict[str, Any] | None) -> None:
+    print()
+    print("=" * 50)
+    print(title)
+    print("=" * 50)
+    print(json.dumps(value, indent=2, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    main()

scripts/run_ui_understanding.py

@@ -0,0 +1,64 @@
+import argparse
+import json
+from pathlib import Path
+
+from cockpit_agent.common.model_registry import ModelRegistry
+from cockpit_agent.config import load_config
+from cockpit_agent.grounding.adapter import GroundingAdapter
+from cockpit_agent.grounding.semantic_target import SemanticTargetBuilder
+from cockpit_agent.intent.parser import ModelIntentParser
+from cockpit_agent.perception.roi import crop_grounding_roi
+from cockpit_agent.perception.ui_understanding import ModelUIUnderstanding
+
+
+DEFAULT_CONFIG = Path(__file__).resolve().parents[1] / "configs" / "default.toml"
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Understand current cockpit UI state")
+    parser.add_argument("--image", required=True)
+    parser.add_argument("--instruction", required=True)
+    parser.add_argument("--output-dir", required=True)
+    parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG)
+    args = parser.parse_args()
+
+    config = load_config(args.config)
+    registry = ModelRegistry()
+    model = registry.get_text_generator(config.intent)
+    intent = ModelIntentParser(model, config.intent.max_new_tokens).parse(
+        args.instruction
+    )
+    target = SemanticTargetBuilder().build_function_target(intent)
+    grounding_model = registry.get(config.grounding)
+    function_grounding = GroundingAdapter(
+        grounding_model,
+        config.grounding.max_new_tokens,
+    ).ground(args.image, target)
+    output_dir = Path(args.output_dir).expanduser().resolve()
+    roi = crop_grounding_roi(
+        image_path=args.image,
+        function_grounding=function_grounding,
+        output_path=str(output_dir / "roi.jpg"),
+        padding_ratio=config.perception.roi_padding_ratio,
+    )
+    ui_model = registry.get(config.ui_understanding)
+    ui_state = ModelUIUnderstanding(
+        ui_model,
+        config.ui_understanding.max_new_tokens,
+    ).understand(roi.image_path, intent, function_grounding)
+    print(
+        json.dumps(
+            {
+                "function_target": target,
+                "function_grounding": function_grounding.to_dict(),
+                "roi": roi.to_dict(),
+                "ui_state": ui_state.to_dict(),
+            },
+            indent=2,
+            ensure_ascii=False,
+        )
+    )
+
+
+if __name__ == "__main__":
+    main()

src/cockpit_agent/__init__.py

@@ -0,0 +1,5 @@
+"""Single-pass cockpit agent without robot execution or feedback."""
+
+from cockpit_agent.pipeline import TaskPipeline
+
+__all__ = ["TaskPipeline"]

src/cockpit_agent/common/__init__.py

@@ -0,0 +1,21 @@
+from cockpit_agent.common.enums import (
+    ActionStatus,
+    ControlType,
+    Domain,
+    Interaction,
+    Operation,
+    State,
+    Unit,
+    Zone,
+)
+
+__all__ = [
+    "ActionStatus",
+    "ControlType",
+    "Domain",
+    "Interaction",
+    "Operation",
+    "State",
+    "Unit",
+    "Zone",
+]

src/cockpit_agent/common/enums.py

@@ -0,0 +1,82 @@
+from enum import Enum
+
+
+class StringEnum(str, Enum):
+    def __str__(self) -> str:
+        return self.value
+
+
+class Domain(StringEnum):
+    CLIMATE = "climate"
+    SEAT = "seat"
+    BODY = "body"
+    LIGHTING = "lighting"
+    NAVIGATION = "navigation"
+    MEDIA = "media"
+    VEHICLE = "vehicle"
+
+
+class Zone(StringEnum):
+    DRIVER = "driver"
+    PASSENGER = "passenger"
+    REAR_LEFT = "rear_left"
+    REAR_RIGHT = "rear_right"
+    ALL = "all"
+
+
+class Operation(StringEnum):
+    SET = "set"
+    INCREASE = "increase"
+    DECREASE = "decrease"
+    OPEN = "open"
+    CLOSE = "close"
+    TOGGLE = "toggle"
+
+
+class State(StringEnum):
+    ON = "on"
+    OFF = "off"
+    OPEN = "open"
+    CLOSED = "closed"
+    UNKNOWN = "unknown"
+
+
+class SelectionState(StringEnum):
+    SELECTED = "selected"
+    UNSELECTED = "unselected"
+    UNKNOWN = "unknown"
+
+
+class SliderOrientation(StringEnum):
+    HORIZONTAL = "horizontal"
+    VERTICAL = "vertical"
+
+
+class Unit(StringEnum):
+    CELSIUS = "celsius"
+    LEVEL = "level"
+    PERCENT = "percent"
+
+
+class ControlType(StringEnum):
+    BUTTON = "button"
+    ICON_BUTTON = "icon_button"
+    TOGGLE = "toggle"
+    SLIDER = "slider"
+    STEPPER = "stepper"
+    SEGMENTED_CONTROL = "segmented_control"
+    RADIO = "radio"
+    CHECKBOX = "checkbox"
+    DROPDOWN = "dropdown"
+    TAB = "tab"
+    LIST_ITEM = "list_item"
+
+
+class Interaction(StringEnum):
+    TAP = "tap"
+    DRAG = "drag"
+    NOOP = "noop"
+
+
+class ActionStatus(StringEnum):
+    PROPOSED = "PROPOSED"

src/cockpit_agent/common/json_utils.py

@@ -0,0 +1,36 @@
+import json
+from pathlib import Path
+from typing import Any
+
+from cockpit_agent.common.schema import SchemaValidationError
+
+
+def parse_model_json_object(raw_output: str, output_name: str) -> dict[str, Any]:
+    if not isinstance(raw_output, str) or not raw_output.strip():
+        raise SchemaValidationError(f"{output_name} model output is empty")
+
+    decoder = json.JSONDecoder()
+    errors: list[str] = []
+    for index, character in enumerate(raw_output):
+        if character != "{":
+            continue
+        try:
+            value, _ = decoder.raw_decode(raw_output[index:])
+        except json.JSONDecodeError as exc:
+            errors.append(f"column {index + exc.colno}: {exc.msg}")
+            continue
+        if isinstance(value, dict):
+            return value
+
+    detail = errors[-1] if errors else "no JSON object found"
+    raise SchemaValidationError(
+        f"Cannot parse {output_name} JSON: {detail}. Raw output: {raw_output!r}"
+    )
+
+
+def write_json(path: str | Path, value: dict[str, Any]) -> None:
+    output = Path(path)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    with output.open("w", encoding="utf-8") as output_file:
+        json.dump(value, output_file, indent=2, ensure_ascii=False, allow_nan=False)
+        output_file.write("\n")

src/cockpit_agent/common/model_registry.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+import gc
+from pathlib import Path
+from typing import Any
+
+from cockpit_agent.config import ModelSettings
+
+
+class ModelRegistry:
+    def __init__(self) -> None:
+        self._instances: dict[tuple[str, Path], Any] = {}
+
+    def get(self, settings: ModelSettings) -> Any:
+        key = (settings.backend, settings.model_path.resolve())
+        if key not in self._instances:
+            try:
+                from cockpit_grounding.models.factory import create_grounder
+            except ModuleNotFoundError as exc:
+                raise RuntimeError(
+                    "cockpit-ui-grounding is not installed; run "
+                    "'pip install -e /data/lgv/projects/cockpit-ui-grounding'"
+                ) from exc
+            self._instances[key] = create_grounder(
+                backend=settings.backend,
+                model_path=settings.model_path,
+            )
+        return self._instances[key]
+
+    def get_text_generator(self, settings: ModelSettings) -> Any:
+        model = self.get(settings)
+        if not callable(getattr(model, "generate_text", None)):
+            raise TypeError(
+                f"Backend {settings.backend!r} does not support text-only generation"
+            )
+        return model
+
+    @property
+    def loaded_model_count(self) -> int:
+        return len(self._instances)
+
+    def release_all(self) -> None:
+        self._instances.clear()
+        gc.collect()
+        try:
+            import torch
+
+            if torch.cuda.is_available():
+                torch.cuda.empty_cache()
+        except ImportError:
+            pass

src/cockpit_agent/common/schema.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+from enum import Enum
+from typing import Any, TypeVar
+
+
+class SchemaValidationError(ValueError):
+    pass
+
+
+EnumT = TypeVar("EnumT", bound=Enum)
+
+
+def require_object(value: Any, schema_name: str) -> dict[str, Any]:
+    if not isinstance(value, dict):
+        raise SchemaValidationError(f"{schema_name} must be a JSON object")
+    return value
+
+
+def require_exact_fields(
+    data: dict[str, Any],
+    fields: set[str],
+    schema_name: str,
+) -> None:
+    missing = fields - data.keys()
+    extra = data.keys() - fields
+    if missing:
+        raise SchemaValidationError(
+            f"{schema_name} missing fields: {', '.join(sorted(missing))}"
+        )
+    if extra:
+        raise SchemaValidationError(
+            f"{schema_name} has unknown fields: {', '.join(sorted(extra))}"
+        )
+
+
+def parse_enum(
+    enum_type: type[EnumT],
+    value: Any,
+    field_name: str,
+) -> EnumT:
+    if not isinstance(value, str):
+        raise SchemaValidationError(f"{field_name} must be a string")
+    try:
+        return enum_type(value)
+    except ValueError as exc:
+        choices = ", ".join(str(item.value) for item in enum_type)
+        raise SchemaValidationError(
+            f"{field_name} must be one of: {choices}; got {value!r}"
+        ) from exc
+
+
+def optional_enum(
+    enum_type: type[EnumT],
+    value: Any,
+    field_name: str,
+) -> EnumT | None:
+    if value is None:
+        return None
+    return parse_enum(enum_type, value, field_name)
+
+
+def require_non_empty_string(value: Any, field_name: str) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise SchemaValidationError(f"{field_name} must be a non-empty string")
+    return value.strip()
+
+
+def optional_number(value: Any, field_name: str) -> int | float | None:
+    if value is None:
+        return None
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        raise SchemaValidationError(f"{field_name} must be a number or null")
+    return value

src/cockpit_agent/config.py

@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+import tomllib
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+
+@dataclass(frozen=True)
+class ModelSettings:
+    backend: str
+    model_path: Path
+    max_new_tokens: int
+
+
+@dataclass(frozen=True)
+class GroundingSettings(ModelSettings):
+    project_path: Path
+
+
+@dataclass(frozen=True)
+class PerceptionSettings:
+    roi_padding_ratio: float
+
+
+@dataclass(frozen=True)
+class AgentConfig:
+    intent: ModelSettings
+    ui_understanding: ModelSettings
+    grounding: GroundingSettings
+    perception: PerceptionSettings
+
+
+def load_config(path: str | Path) -> AgentConfig:
+    config_path = Path(path).expanduser().resolve()
+    with config_path.open("rb") as config_file:
+        data = tomllib.load(config_file)
+
+    intent = _model_settings(_section(data, "intent"), "intent")
+    ui_understanding = _model_settings(
+        _section(data, "ui_understanding"),
+        "ui_understanding",
+    )
+    grounding_data = _section(data, "grounding")
+    grounding_model = _model_settings(grounding_data, "grounding")
+    project_path = _directory(grounding_data.get("project_path"), "grounding.project_path")
+    perception_data = _section(data, "perception")
+    roi_padding_ratio = perception_data.get("roi_padding_ratio")
+    if (
+        isinstance(roi_padding_ratio, bool)
+        or not isinstance(roi_padding_ratio, (int, float))
+        or not 0.0 <= roi_padding_ratio <= 1.0
+    ):
+        raise ValueError(
+            "perception.roi_padding_ratio must be between 0 and 1"
+        )
+    return AgentConfig(
+        intent=intent,
+        ui_understanding=ui_understanding,
+        grounding=GroundingSettings(
+            backend=grounding_model.backend,
+            model_path=grounding_model.model_path,
+            max_new_tokens=grounding_model.max_new_tokens,
+            project_path=project_path,
+        ),
+        perception=PerceptionSettings(
+            roi_padding_ratio=float(roi_padding_ratio),
+        ),
+    )
+
+
+def _model_settings(data: dict[str, Any], section: str) -> ModelSettings:
+    backend = data.get("backend")
+    if not isinstance(backend, str) or not backend.strip():
+        raise ValueError(f"{section}.backend must be a non-empty string")
+    max_new_tokens = data.get("max_new_tokens")
+    if (
+        not isinstance(max_new_tokens, int)
+        or isinstance(max_new_tokens, bool)
+        or max_new_tokens <= 0
+    ):
+        raise ValueError(f"{section}.max_new_tokens must be a positive integer")
+    return ModelSettings(
+        backend=backend,
+        model_path=_directory(data.get("model_path"), f"{section}.model_path"),
+        max_new_tokens=max_new_tokens,
+    )
+
+
+def _section(data: dict[str, Any], name: str) -> dict[str, Any]:
+    value = data.get(name)
+    if not isinstance(value, dict):
+        raise ValueError(f"Missing or invalid [{name}] section")
+    return value
+
+
+def _directory(value: Any, field_name: str) -> Path:
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError(f"{field_name} must be a non-empty path")
+    path = Path(value).expanduser().resolve()
+    if not path.is_dir():
+        raise FileNotFoundError(f"{field_name} directory not found: {path}")
+    return path

src/cockpit_agent/execution/__init__.py

@@ -0,0 +1,4 @@
+from cockpit_agent.execution.mock_executor import MockExecutor
+from cockpit_agent.execution.schemas import MockAction
+
+__all__ = ["MockAction", "MockExecutor"]

src/cockpit_agent/execution/mock_executor.py

@@ -0,0 +1,188 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from PIL import Image, ImageDraw, ImageFont
+
+from cockpit_agent.common.enums import Interaction, SliderOrientation
+from cockpit_agent.execution.schemas import MockAction
+from cockpit_agent.grounding.adapter import GroundingResult
+from cockpit_agent.grounding.schemas import ActionGroundingResult
+from cockpit_agent.planning.schemas import ActionPlan
+
+
+class MockExecutor:
+    def propose(
+        self,
+        image_path: str,
+        plan: ActionPlan,
+        function_grounding: GroundingResult,
+        action_target: str | dict[str, str] | None,
+        action_grounding: ActionGroundingResult | None,
+        output_path: str,
+    ) -> MockAction:
+        if plan.interaction is Interaction.NOOP:
+            action = MockAction(
+                interaction=Interaction.NOOP,
+                reason=plan.reason,
+            )
+        elif action_grounding is None:
+            raise ValueError("Action grounding is required for tap or drag")
+        elif plan.interaction is Interaction.TAP:
+            if action_grounding.control is None:
+                raise ValueError("Tap action requires a grounded control")
+            action = MockAction(
+                interaction=Interaction.TAP,
+                pixel=action_grounding.control.center_pixel,
+                repeat=plan.repeat,
+            )
+        elif plan.interaction is Interaction.DRAG:
+            action = self._propose_drag(plan, action_grounding)
+        else:
+            raise ValueError(f"Unsupported interaction: {plan.interaction.value}")
+
+        self._draw_result(
+            image_path=image_path,
+            output_path=output_path,
+            plan=plan,
+            function_grounding=function_grounding,
+            action_target=action_target,
+            action_grounding=action_grounding,
+            action=action,
+        )
+        return action
+
+    @staticmethod
+    def _propose_drag(
+        plan: ActionPlan,
+        grounding: ActionGroundingResult,
+    ) -> MockAction:
+        if plan.normalized_target is None:
+            raise ValueError("Drag plan requires normalized target")
+        if grounding.track is None or grounding.knob is None:
+            raise ValueError("Drag action requires grounded slider track and knob")
+        if plan.orientation is not SliderOrientation.HORIZONTAL:
+            raise ValueError("Only an observed horizontal slider can be proposed")
+        x1, y1, x2, y2 = grounding.track.bbox_pixel
+        center_y = round((y1 + y2) / 2)
+        end_x = round(x1 + plan.normalized_target * (x2 - x1))
+        return MockAction(
+            interaction=Interaction.DRAG,
+            start_pixel=grounding.knob.center_pixel,
+            end_pixel=(end_x, center_y),
+        )
+
+    @staticmethod
+    def _draw_result(
+        *,
+        image_path: str,
+        output_path: str,
+        plan: ActionPlan,
+        function_grounding: GroundingResult,
+        action_target: str | dict[str, str] | None,
+        action_grounding: ActionGroundingResult | None,
+        action: MockAction,
+    ) -> None:
+        image = Image.open(image_path).convert("RGB")
+        draw = ImageDraw.Draw(image)
+        width, height = image.size
+        line_width = max(4, width // 800)
+        marker_radius = max(10, width // 300)
+        blue = (35, 100, 220)
+        red = (220, 35, 45)
+        magenta = (175, 45, 175)
+        green = (20, 160, 90)
+
+        function_x1, function_y1, function_x2, function_y2 = (
+            function_grounding.bbox_pixel
+        )
+        draw.rectangle(
+            (
+                max(0, function_x1 - line_width),
+                max(0, function_y1 - line_width),
+                min(width, function_x2 + line_width),
+                min(height, function_y2 + line_width),
+            ),
+            outline=blue,
+            width=line_width,
+        )
+        if action_grounding is not None:
+            grounded_boxes = tuple(
+                (result, color)
+                for result, color in (
+                    (action_grounding.control, red),
+                    (action_grounding.track, red),
+                    (action_grounding.knob, magenta),
+                )
+                if result is not None
+            )
+            for grounded, color in grounded_boxes:
+                draw.rectangle(grounded.bbox_pixel, outline=color, width=line_width)
+        if action.pixel is not None:
+            u, v = action.pixel
+            draw.ellipse(
+                (
+                    u - marker_radius,
+                    v - marker_radius,
+                    u + marker_radius,
+                    v + marker_radius,
+                ),
+                fill=green,
+            )
+        if action.interaction is Interaction.DRAG:
+            assert action.start_pixel is not None and action.end_pixel is not None
+            draw.line(
+                (action.start_pixel, action.end_pixel),
+                fill=green,
+                width=line_width * 2,
+            )
+            for point in (action.start_pixel, action.end_pixel):
+                x, y = point
+                draw.ellipse(
+                    (
+                        x - marker_radius,
+                        y - marker_radius,
+                        x + marker_radius,
+                        y + marker_radius,
+                    ),
+                    fill=green,
+                )
+
+        font = _load_font(max(28, width // 90))
+        if action.interaction is Interaction.TAP:
+            action_label = f"TAP x{action.repeat}"
+        elif action.interaction is Interaction.DRAG:
+            action_label = "DRAG"
+        else:
+            action_label = f"NOOP: {action.reason}"
+        if isinstance(action_target, dict):
+            target_label = " | ".join(action_target.values())
+        else:
+            target_label = action_target
+        labels = (action_label, target_label) if target_label else (action_label,)
+        boxes = [draw.textbbox((0, 0), label, font=font) for label in labels]
+        text_height = sum(box[3] - box[1] for box in boxes)
+        padding = max(16, width // 180)
+        gap = max(8, padding // 2)
+        band_height = text_height + gap + padding * 2
+        draw.rectangle((0, 0, width, band_height), fill=(20, 20, 20))
+        y = padding
+        for label, box in zip(labels, boxes):
+            draw.text((padding, y), label, font=font, fill=(255, 255, 255))
+            y += box[3] - box[1] + gap
+
+        output = Path(output_path).expanduser().resolve()
+        output.parent.mkdir(parents=True, exist_ok=True)
+        image.save(output, quality=95)
+
+
+def _load_font(size: int) -> ImageFont.FreeTypeFont | ImageFont.ImageFont:
+    candidates = (
+        "/usr/share/fonts/google-noto-cjk/NotoSansCJK-Regular.ttc",
+        "/usr/share/fonts/google-droid-sans-fonts/DroidSansFallbackFull.ttf",
+        "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
+    )
+    for candidate in candidates:
+        if Path(candidate).is_file():
+            return ImageFont.truetype(candidate, size=size)
+    return ImageFont.load_default()

src/cockpit_agent/execution/schemas.py

@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from cockpit_agent.common.enums import ActionStatus, Interaction
+from cockpit_agent.common.schema import SchemaValidationError
+
+
+@dataclass(frozen=True)
+class MockAction:
+    interaction: Interaction
+    status: ActionStatus = ActionStatus.PROPOSED
+    pixel: tuple[int, int] | None = None
+    repeat: int | None = None
+    start_pixel: tuple[int, int] | None = None
+    end_pixel: tuple[int, int] | None = None
+    reason: str | None = None
+
+    def __post_init__(self) -> None:
+        if self.interaction is Interaction.TAP:
+            if self.pixel is None or self.repeat is None or self.repeat < 1:
+                raise SchemaValidationError("tap action requires pixel and repeat")
+        elif self.interaction is Interaction.DRAG:
+            if self.start_pixel is None or self.end_pixel is None:
+                raise SchemaValidationError(
+                    "drag action requires start_pixel and end_pixel"
+                )
+        elif self.interaction is Interaction.NOOP and not self.reason:
+            raise SchemaValidationError("noop action requires a reason")
+
+    def to_dict(self) -> dict[str, Any]:
+        value: dict[str, Any] = {
+            "interaction": self.interaction.value,
+            "status": self.status.value,
+        }
+        if self.pixel is not None:
+            value["pixel"] = list(self.pixel)
+        if self.repeat is not None:
+            value["repeat"] = self.repeat
+        if self.start_pixel is not None:
+            value["start_pixel"] = list(self.start_pixel)
+        if self.end_pixel is not None:
+            value["end_pixel"] = list(self.end_pixel)
+        if self.reason is not None:
+            value["reason"] = self.reason
+        return value

src/cockpit_agent/grounding/__init__.py

@@ -0,0 +1,10 @@
+from cockpit_agent.grounding.adapter import GroundingAdapter, GroundingResult
+from cockpit_agent.grounding.schemas import ActionGroundingResult
+from cockpit_agent.grounding.semantic_target import SemanticTargetBuilder
+
+__all__ = [
+    "ActionGroundingResult",
+    "GroundingAdapter",
+    "GroundingResult",
+    "SemanticTargetBuilder",
+]

src/cockpit_agent/grounding/adapter.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Protocol
+
+from PIL import Image
+
+from cockpit_grounding.grounding.predictor import (
+    build_grounding_prompt,
+    parse_grounding_output,
+)
+from cockpit_grounding.vision.visualize import relative_bbox_to_pixels
+
+
+class GroundingModel(Protocol):
+    def generate(
+        self,
+        image_path: str,
+        prompt: str,
+        max_new_tokens: int = 128,
+    ) -> str: ...
+
+
+@dataclass(frozen=True)
+class GroundingResult:
+    semantic_query: str
+    bbox_relative: tuple[float, float, float, float]
+    bbox_pixel: tuple[int, int, int, int]
+    center_pixel: tuple[int, int]
+    image_width: int
+    image_height: int
+    raw_output: str
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "semantic_query": self.semantic_query,
+            "bbox_relative": list(self.bbox_relative),
+            "bbox_pixel": list(self.bbox_pixel),
+            "center_pixel": list(self.center_pixel),
+            "image_width": self.image_width,
+            "image_height": self.image_height,
+            "raw_output": self.raw_output,
+        }
+
+
+class GroundingAdapter:
+    def __init__(
+        self,
+        model: GroundingModel,
+        max_new_tokens: int = 128,
+    ) -> None:
+        self._model = model
+        self._max_new_tokens = max_new_tokens
+        self.last_raw_output: str | None = None
+
+    def ground(self, image_path: str, semantic_query: str) -> GroundingResult:
+        image = Path(image_path).expanduser().resolve()
+        if not image.is_file():
+            raise FileNotFoundError(f"Image not found: {image}")
+        if not isinstance(semantic_query, str) or not semantic_query.strip():
+            raise ValueError("semantic grounding query must be a non-empty string")
+
+        self.last_raw_output = None
+        self.last_raw_output = self._model.generate(
+            image_path=str(image),
+            prompt=build_grounding_prompt(semantic_query.strip()),
+            max_new_tokens=self._max_new_tokens,
+        )
+        parsed = parse_grounding_output(self.last_raw_output)
+        with Image.open(image) as image_data:
+            width, height = image_data.size
+        bbox_relative = (parsed.x1, parsed.y1, parsed.x2, parsed.y2)
+        bbox_pixel = relative_bbox_to_pixels(parsed, width, height)
+        x1, y1, x2, y2 = bbox_pixel
+        center_pixel = (round((x1 + x2) / 2), round((y1 + y2) / 2))
+        return GroundingResult(
+            semantic_query=semantic_query.strip(),
+            bbox_relative=bbox_relative,
+            bbox_pixel=bbox_pixel,
+            center_pixel=center_pixel,
+            image_width=width,
+            image_height=height,
+            raw_output=self.last_raw_output,
+        )

src/cockpit_agent/grounding/schemas.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from cockpit_agent.grounding.adapter import GroundingResult
+
+
+@dataclass(frozen=True)
+class ActionGroundingResult:
+    source: str
+    control: GroundingResult | None = None
+    track: GroundingResult | None = None
+    knob: GroundingResult | None = None
+
+    def __post_init__(self) -> None:
+        has_control = self.control is not None
+        has_slider_parts = self.track is not None or self.knob is not None
+        if has_control == has_slider_parts:
+            raise ValueError(
+                "ActionGroundingResult requires either control or track and knob"
+            )
+        if has_slider_parts and (self.track is None or self.knob is None):
+            raise ValueError("slider action grounding requires track and knob")
+
+    def to_dict(self) -> dict[str, Any]:
+        value: dict[str, Any] = {"source": self.source}
+        if self.control is not None:
+            value["control"] = self.control.to_dict()
+        if self.track is not None:
+            value["track"] = self.track.to_dict()
+        if self.knob is not None:
+            value["knob"] = self.knob.to_dict()
+        return value

src/cockpit_agent/grounding/semantic_target.py

@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+from cockpit_agent.common.enums import Interaction, Zone
+from cockpit_agent.intent.schemas import SemanticAction
+from cockpit_agent.planning.schemas import ActionPlan
+
+
+class SemanticTargetBuilder:
+    """Convert a structured action plan into a visual-layout-free query."""
+
+    def build_function_target(self, intent: SemanticAction) -> str:
+        if intent.function == "temperature":
+            return (
+                f"{_zone_label(intent.zone)}温度调节控件整体，"
+                "包含当前温度显示以及用于升高和降低温度的控制元素"
+            )
+        if intent.function == "air_recirculation":
+            return "车辆空调内循环控制"
+        if intent.function == "climate_power":
+            return f"{_zone_label(intent.zone)}空调电源控件整体"
+        if intent.function == "tailgate_max_opening":
+            return (
+                "后背门最大开度调节控件整体，"
+                "包含滑轨、滑块当前位置以及最小和最大范围标记"
+            )
+        if intent.function == "unlock_mode":
+            return "车门解锁方式选择控件整体，包含所有可选择的解锁方式选项"
+        return (
+            f"{_zone_label(intent.zone)}"
+            f"{intent.function.replace('_', ' ')}功能控件整体"
+        )
+
+    def build_action_target(self, plan: ActionPlan) -> str:
+        if plan.interaction is Interaction.NOOP:
+            raise ValueError("A noop plan does not have a grounding target")
+        assert plan.function is not None
+
+        if plan.function in {"temperature_decrease", "temperature_increase"}:
+            direction = (
+                "降低" if plan.function.endswith("_decrease") else "升高"
+            )
+            return f"{_zone_label(plan.zone)}温度{direction}控制"
+        if plan.function == "air_recirculation":
+            return "车辆空调内循环控制"
+        if plan.function == "climate_power":
+            return f"{_zone_label(plan.zone)}空调电源控制"
+        if plan.function == "tailgate_max_opening":
+            return "后背门最大开度滑块"
+        if plan.function == "unlock_mode":
+            option = {
+                "all_doors": "所有车门解锁选项",
+                "driver_door": "主驾驶车门解锁选项",
+            }.get(plan.option)
+            if option is not None:
+                return option
+            return "车门解锁模式控制"
+
+        semantic_name = plan.function.replace("_", " ")
+        if plan.option is not None:
+            return f"{semantic_name}功能的{plan.option.replace('_', ' ')}选项"
+        return f"{_zone_label(plan.zone)}{semantic_name}功能控制"
+
+    def build_slider_track_target(self, plan: ActionPlan) -> str:
+        if plan.function == "tailgate_max_opening":
+            return "后背门最大开度滑轨"
+        return f"{self.build_action_target(plan)}滑轨"
+
+    def build_slider_knob_target(self, plan: ActionPlan) -> str:
+        if plan.function == "tailgate_max_opening":
+            return "后背门最大开度当前滑块"
+        return f"{self.build_action_target(plan)}当前滑块"
+
+
+def _zone_label(zone: Zone | None) -> str:
+    return {
+        Zone.DRIVER: "主驾驶",
+        Zone.PASSENGER: "副驾驶",
+        Zone.REAR_LEFT: "后排左座",
+        Zone.REAR_RIGHT: "后排右座",
+        Zone.ALL: "全车",
+        None: "车辆",
+    }[zone]

src/cockpit_agent/intent/__init__.py

@@ -0,0 +1,4 @@
+from cockpit_agent.intent.parser import IntentParser, ModelIntentParser
+from cockpit_agent.intent.schemas import SemanticAction
+
+__all__ = ["IntentParser", "ModelIntentParser", "SemanticAction"]

src/cockpit_agent/intent/parser.py

@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+import json
+from abc import ABC, abstractmethod
+from typing import Protocol
+
+from cockpit_agent.common.json_utils import parse_model_json_object
+from cockpit_agent.intent.schemas import SemanticAction
+
+
+class TextGenerator(Protocol):
+    def generate_text(self, prompt: str, max_new_tokens: int = 256) -> str: ...
+
+
+class IntentParser(ABC):
+    @abstractmethod
+    def parse(self, instruction: str) -> SemanticAction:
+        raise NotImplementedError
+
+
+class ModelIntentParser(IntentParser):
+    def __init__(self, model: TextGenerator, max_new_tokens: int = 256) -> None:
+        self._model = model
+        self._max_new_tokens = max_new_tokens
+        self.last_raw_output: str | None = None
+
+    def parse(self, instruction: str) -> SemanticAction:
+        if not isinstance(instruction, str) or not instruction.strip():
+            raise ValueError("instruction must be a non-empty string")
+        prompt = _build_intent_prompt(instruction.strip())
+        self.last_raw_output = None
+        self.last_raw_output = self._model.generate_text(
+            prompt,
+            max_new_tokens=self._max_new_tokens,
+        )
+        data = parse_model_json_object(self.last_raw_output, "SemanticAction")
+        return SemanticAction.from_dict(data)
+
+
+def _build_intent_prompt(instruction: str) -> str:
+    examples = (
+        (
+            "把主驾温度调到23度",
+            {
+                "domain": "climate",
+                "function": "temperature",
+                "zone": "driver",
+                "operation": "set",
+                "state": None,
+                "value": 23,
+                "unit": "celsius",
+            },
+        ),
+        (
+            "打开内循环",
+            {
+                "domain": "climate",
+                "function": "air_recirculation",
+                "zone": "all",
+                "operation": "set",
+                "state": "on",
+                "value": None,
+                "unit": None,
+            },
+        ),
+        (
+            "后备箱最大开度设置成80%",
+            {
+                "domain": "body",
+                "function": "tailgate_max_opening",
+                "zone": None,
+                "operation": "set",
+                "state": None,
+                "value": 80,
+                "unit": "percent",
+            },
+        ),
+        (
+            "解锁时打开所有车门",
+            {
+                "domain": "body",
+                "function": "unlock_mode",
+                "zone": None,
+                "operation": "set",
+                "state": None,
+                "value": "all_doors",
+                "unit": None,
+            },
+        ),
+    )
+    example_text = "\n\n".join(
+        f"示例指令：{example_instruction}\n"
+        "正确回复（只回复对象本身）：\n"
+        f"{json.dumps(example_output, ensure_ascii=False)}"
+        for example_instruction, example_output in examples
+    )
+    return f"""
+你是汽车座舱用户意图解析器。你只回答用户最终希望达到的功能目标。
+不要关心 UI 控件类型、按钮位置、bbox、像素坐标或机械臂动作。
+
+严格输出一个 JSON 对象，必须且只能包含以下字段：
+domain, function, zone, operation, state, value, unit
+
+枚举：
+domain: climate, seat, body, lighting, navigation, media, vehicle
+zone: driver, passenger, rear_left, rear_right, all, null
+operation: set, increase, decrease, open, close, toggle
+state: on, off, open, closed, null
+unit: celsius, level, percent, null
+
+温度降低 N 度表示 operation=decrease, value=N, unit=celsius。
+回复的顶层第一个字段必须是 domain。
+不要使用 input/output 包装，不要输出 Markdown，不要解释，不要添加字段。
+
+示例：
+{example_text}
+
+用户指令：{instruction}
+""".strip()

src/cockpit_agent/intent/schemas.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from cockpit_agent.common.enums import Domain, Operation, State, Unit, Zone
+from cockpit_agent.common.schema import (
+    SchemaValidationError,
+    optional_enum,
+    parse_enum,
+    require_exact_fields,
+    require_non_empty_string,
+    require_object,
+)
+
+
+@dataclass(frozen=True)
+class SemanticAction:
+    domain: Domain
+    function: str
+    zone: Zone | None
+    operation: Operation
+    state: State | None
+    value: int | float | str | None
+    unit: Unit | None
+
+    @classmethod
+    def from_dict(cls, value: Any) -> "SemanticAction":
+        data = require_object(value, "SemanticAction")
+        require_exact_fields(
+            data,
+            {"domain", "function", "zone", "operation", "state", "value", "unit"},
+            "SemanticAction",
+        )
+        raw_value = data["value"]
+        if isinstance(raw_value, bool) or not isinstance(
+            raw_value,
+            (int, float, str, type(None)),
+        ):
+            raise SchemaValidationError(
+                "SemanticAction.value must be a number, string, or null"
+            )
+        if isinstance(raw_value, str) and not raw_value.strip():
+            raise SchemaValidationError("SemanticAction.value cannot be empty")
+
+        action = cls(
+            domain=parse_enum(Domain, data["domain"], "SemanticAction.domain"),
+            function=require_non_empty_string(
+                data["function"],
+                "SemanticAction.function",
+            ),
+            zone=optional_enum(Zone, data["zone"], "SemanticAction.zone"),
+            operation=parse_enum(
+                Operation,
+                data["operation"],
+                "SemanticAction.operation",
+            ),
+            state=optional_enum(State, data["state"], "SemanticAction.state"),
+            value=raw_value.strip() if isinstance(raw_value, str) else raw_value,
+            unit=optional_enum(Unit, data["unit"], "SemanticAction.unit"),
+        )
+        action._validate_semantics()
+        return action
+
+    def _validate_semantics(self) -> None:
+        if self.function == "temperature":
+            if self.domain is not Domain.CLIMATE:
+                raise SchemaValidationError("temperature must use climate domain")
+            if self.value is not None and not _is_number(self.value):
+                raise SchemaValidationError("temperature value must be numeric")
+            if self.value is not None and self.unit is not Unit.CELSIUS:
+                raise SchemaValidationError("temperature value must use celsius")
+        if self.function == "tailgate_max_opening":
+            if not _is_number(self.value) or self.unit is not Unit.PERCENT:
+                raise SchemaValidationError(
+                    "tailgate_max_opening requires a numeric percent value"
+                )
+        if self.function == "unlock_mode" and not isinstance(self.value, str):
+            raise SchemaValidationError("unlock_mode requires a string value")
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "domain": self.domain.value,
+            "function": self.function,
+            "zone": self.zone.value if self.zone is not None else None,
+            "operation": self.operation.value,
+            "state": self.state.value if self.state is not None else None,
+            "value": self.value,
+            "unit": self.unit.value if self.unit is not None else None,
+        }
+
+
+def _is_number(value: object) -> bool:
+    return isinstance(value, (int, float)) and not isinstance(value, bool)

src/cockpit_agent/perception/__init__.py

@@ -0,0 +1,15 @@
+from cockpit_agent.perception.schemas import UIState
+from cockpit_agent.perception.roi import ROIResult, crop_grounding_roi, grounding_to_source
+from cockpit_agent.perception.ui_understanding import (
+    ModelUIUnderstanding,
+    UIUnderstanding,
+)
+
+__all__ = [
+    "ModelUIUnderstanding",
+    "ROIResult",
+    "UIState",
+    "UIUnderstanding",
+    "crop_grounding_roi",
+    "grounding_to_source",
+]

src/cockpit_agent/perception/roi.py

@@ -0,0 +1,127 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, replace
+from pathlib import Path
+from typing import Any
+
+from PIL import Image
+
+from cockpit_agent.grounding.adapter import GroundingResult
+
+
+@dataclass(frozen=True)
+class ROIResult:
+    bbox_pixel: tuple[int, int, int, int]
+    image_path: str
+    offset_x: int
+    offset_y: int
+    width: int
+    height: int
+    source_image_width: int
+    source_image_height: int
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "bbox_pixel": list(self.bbox_pixel),
+            "image_path": self.image_path,
+            "offset_x": self.offset_x,
+            "offset_y": self.offset_y,
+            "width": self.width,
+            "height": self.height,
+            "source_image_width": self.source_image_width,
+            "source_image_height": self.source_image_height,
+        }
+
+
+def crop_grounding_roi(
+    *,
+    image_path: str,
+    function_grounding: GroundingResult,
+    output_path: str,
+    padding_ratio: float,
+) -> ROIResult:
+    if not 0.0 <= padding_ratio <= 1.0:
+        raise ValueError("ROI padding_ratio must be between 0 and 1")
+
+    source = Path(image_path).expanduser().resolve()
+    output = Path(output_path).expanduser().resolve()
+    with Image.open(source) as image:
+        image_width, image_height = image.size
+        x1, y1, x2, y2 = function_grounding.bbox_pixel
+        _validate_bbox((x1, y1, x2, y2), image_width, image_height)
+        padding_x = round((x2 - x1) * padding_ratio)
+        padding_y = round((y2 - y1) * padding_ratio)
+        crop_bbox = (
+            max(0, x1 - padding_x),
+            max(0, y1 - padding_y),
+            min(image_width, x2 + padding_x),
+            min(image_height, y2 + padding_y),
+        )
+        crop = image.crop(crop_bbox).convert("RGB")
+        output.parent.mkdir(parents=True, exist_ok=True)
+        crop.save(output, quality=95)
+
+    crop_x1, crop_y1, crop_x2, crop_y2 = crop_bbox
+    return ROIResult(
+        bbox_pixel=crop_bbox,
+        image_path=str(output),
+        offset_x=crop_x1,
+        offset_y=crop_y1,
+        width=crop_x2 - crop_x1,
+        height=crop_y2 - crop_y1,
+        source_image_width=image_width,
+        source_image_height=image_height,
+    )
+
+
+def grounding_to_source(
+    grounding: GroundingResult,
+    roi: ROIResult,
+) -> GroundingResult:
+    if (
+        grounding.image_width != roi.width
+        or grounding.image_height != roi.height
+    ):
+        raise ValueError("Grounding image dimensions do not match ROI dimensions")
+    x1, y1, x2, y2 = grounding.bbox_pixel
+    source_bbox = (
+        x1 + roi.offset_x,
+        y1 + roi.offset_y,
+        x2 + roi.offset_x,
+        y2 + roi.offset_y,
+    )
+    _validate_bbox(
+        source_bbox,
+        roi.source_image_width,
+        roi.source_image_height,
+    )
+    source_x1, source_y1, source_x2, source_y2 = source_bbox
+    source_relative = (
+        source_x1 / roi.source_image_width * 1000.0,
+        source_y1 / roi.source_image_height * 1000.0,
+        source_x2 / roi.source_image_width * 1000.0,
+        source_y2 / roi.source_image_height * 1000.0,
+    )
+    return replace(
+        grounding,
+        bbox_relative=source_relative,
+        bbox_pixel=source_bbox,
+        center_pixel=(
+            round((source_x1 + source_x2) / 2),
+            round((source_y1 + source_y2) / 2),
+        ),
+        image_width=roi.source_image_width,
+        image_height=roi.source_image_height,
+    )
+
+
+def _validate_bbox(
+    bbox: tuple[int, int, int, int],
+    image_width: int,
+    image_height: int,
+) -> None:
+    x1, y1, x2, y2 = bbox
+    if not (0 <= x1 < x2 <= image_width and 0 <= y1 < y2 <= image_height):
+        raise ValueError(
+            f"bbox {bbox} is outside image size {image_width}x{image_height}"
+        )

src/cockpit_agent/perception/schemas.py

@@ -0,0 +1,174 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from cockpit_agent.common.enums import (
+    ControlType,
+    SelectionState,
+    SliderOrientation,
+    State,
+    Zone,
+)
+from cockpit_agent.common.schema import (
+    SchemaValidationError,
+    optional_enum,
+    optional_number,
+    require_exact_fields,
+    require_object,
+)
+
+
+@dataclass(frozen=True)
+class UIState:
+    function: str
+    zone: Zone | None
+    control_type: ControlType | None
+    current_value: int | float | str | None
+    step_value: int | float | None
+    current_state: State | None
+    min_value: int | float | None
+    max_value: int | float | None
+    options: tuple[str, ...] | None
+    selection_state: SelectionState | None = None
+    track_visible: bool | None = None
+    knob_visible: bool | None = None
+    orientation: SliderOrientation | None = None
+
+    @classmethod
+    def from_dict(cls, value: Any) -> "UIState":
+        data = dict(require_object(value, "UIState"))
+        for nullable_field in (
+            "selection_state",
+            "track_visible",
+            "knob_visible",
+            "orientation",
+        ):
+            data.setdefault(nullable_field, None)
+        require_exact_fields(
+            data,
+            {
+                "control_type",
+                "function",
+                "zone",
+                "current_value",
+                "step_value",
+                "current_state",
+                "min_value",
+                "max_value",
+                "options",
+                "selection_state",
+                "track_visible",
+                "knob_visible",
+                "orientation",
+            },
+            "UIState",
+        )
+        function = data["function"]
+        if not isinstance(function, str) or not function.strip():
+            raise SchemaValidationError("UIState.function must be a non-empty string")
+        current_value = data["current_value"]
+        if isinstance(current_value, bool) or not isinstance(
+            current_value,
+            (int, float, str, type(None)),
+        ):
+            raise SchemaValidationError(
+                "UIState.current_value must be a number, string, or null"
+            )
+        options = data["options"]
+        parsed_options: tuple[str, ...] | None
+        if options is None:
+            parsed_options = None
+        elif isinstance(options, list) and all(
+            isinstance(option, str) and option.strip() for option in options
+        ):
+            parsed_options = tuple(option.strip() for option in options)
+        else:
+            raise SchemaValidationError(
+                "UIState.options must be a list of non-empty strings or null"
+            )
+
+        state = cls(
+            function=function.strip(),
+            zone=optional_enum(Zone, data["zone"], "UIState.zone"),
+            control_type=optional_enum(
+                ControlType, data["control_type"], "UIState.control_type"
+            ),
+            current_value=(
+                current_value.strip()
+                if isinstance(current_value, str)
+                else current_value
+            ),
+            step_value=optional_number(data["step_value"], "UIState.step_value"),
+            current_state=optional_enum(
+                State,
+                data["current_state"],
+                "UIState.current_state",
+            ),
+            min_value=optional_number(data["min_value"], "UIState.min_value"),
+            max_value=optional_number(data["max_value"], "UIState.max_value"),
+            options=parsed_options,
+            selection_state=optional_enum(
+                SelectionState,
+                data["selection_state"],
+                "UIState.selection_state",
+            ),
+            track_visible=_optional_bool(
+                data["track_visible"],
+                "UIState.track_visible",
+            ),
+            knob_visible=_optional_bool(
+                data["knob_visible"],
+                "UIState.knob_visible",
+            ),
+            orientation=optional_enum(
+                SliderOrientation,
+                data["orientation"],
+                "UIState.orientation",
+            ),
+        )
+        state._validate_consistency()
+        return state
+
+    def _validate_consistency(self) -> None:
+        if self.step_value is not None and self.step_value <= 0:
+            raise SchemaValidationError("UIState.step_value must be positive")
+        if (
+            self.min_value is not None
+            and self.max_value is not None
+            and self.min_value >= self.max_value
+        ):
+            raise SchemaValidationError("UIState.min_value must be less than max_value")
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "function": self.function,
+            "zone": self.zone.value if self.zone is not None else None,
+            "control_type": (
+                self.control_type.value if self.control_type is not None else None
+            ),
+            "current_value": self.current_value,
+            "step_value": self.step_value,
+            "current_state": (
+                self.current_state.value if self.current_state is not None else None
+            ),
+            "min_value": self.min_value,
+            "max_value": self.max_value,
+            "options": list(self.options) if self.options is not None else None,
+            "selection_state": (
+                self.selection_state.value
+                if self.selection_state is not None
+                else None
+            ),
+            "track_visible": self.track_visible,
+            "knob_visible": self.knob_visible,
+            "orientation": (
+                self.orientation.value if self.orientation is not None else None
+            ),
+        }
+
+
+def _optional_bool(value: Any, field_name: str) -> bool | None:
+    if value is None or isinstance(value, bool):
+        return value
+    raise SchemaValidationError(f"{field_name} must be a boolean or null")

src/cockpit_agent/perception/ui_understanding.py

@@ -0,0 +1,291 @@
+from __future__ import annotations
+
+import json
+from abc import ABC, abstractmethod
+from typing import Any, Protocol
+
+from cockpit_agent.common.json_utils import parse_model_json_object
+from cockpit_agent.common.schema import SchemaValidationError
+from cockpit_agent.grounding.adapter import GroundingResult
+from cockpit_agent.intent.schemas import SemanticAction
+from cockpit_agent.perception.schemas import UIState
+
+
+class VisionGenerator(Protocol):
+    def generate(
+        self,
+        image_path: str,
+        prompt: str,
+        max_new_tokens: int = 256,
+    ) -> str: ...
+
+
+class UIUnderstanding(ABC):
+    @abstractmethod
+    def understand(
+        self,
+        roi_image_path: str,
+        intent: SemanticAction,
+        function_grounding: GroundingResult,
+    ) -> UIState:
+        raise NotImplementedError
+
+
+class ModelUIUnderstanding(UIUnderstanding):
+    def __init__(self, model: VisionGenerator, max_new_tokens: int = 256) -> None:
+        self._model = model
+        self._max_new_tokens = max_new_tokens
+        self.last_raw_output: str | None = None
+
+    def understand(
+        self,
+        roi_image_path: str,
+        intent: SemanticAction,
+        function_grounding: GroundingResult,
+    ) -> UIState:
+        prompt = _build_ui_prompt(intent, function_grounding)
+        self.last_raw_output = None
+        self.last_raw_output = self._model.generate(
+            image_path=roi_image_path,
+            prompt=prompt,
+            max_new_tokens=self._max_new_tokens,
+        )
+        data = parse_model_json_object(self.last_raw_output, "UIState")
+        try:
+            state = UIState.from_dict(_canonicalize_fields(data, intent))
+            return state
+        except SchemaValidationError as exc:
+            raise SchemaValidationError(
+                f"Invalid UIState model output: {exc}. "
+                f"Raw output: {self.last_raw_output!r}"
+            ) from exc
+
+
+def _canonicalize_fields(
+    data: dict[str, Any],
+    intent: SemanticAction,
+) -> dict[str, Any]:
+    normalized = dict(data)
+    normalized["function"] = intent.function
+    normalized["zone"] = intent.zone.value if intent.zone is not None else None
+    normalized.setdefault("selection_state", None)
+    normalized.setdefault("track_visible", None)
+    normalized.setdefault("knob_visible", None)
+    normalized.setdefault("orientation", None)
+    control_type = normalized.get("control_type")
+
+    current_state = normalized.get("current_state")
+    if _is_selection_state(current_state):
+        if normalized["selection_state"] is None:
+            normalized["selection_state"] = current_state
+        normalized["current_state"] = None
+    else:
+        normalized["current_state"] = _normalize_current_state(current_state)
+    normalized["selection_state"] = _normalize_selection_state(
+        normalized["selection_state"]
+    )
+
+    null_fields = {
+        "stepper": (
+            "current_state",
+            "selection_state",
+            "min_value",
+            "max_value",
+            "options",
+            "track_visible",
+            "knob_visible",
+            "orientation",
+        ),
+        "slider": ("step_value", "current_state", "selection_state", "options"),
+        "toggle": (
+            "current_value",
+            "step_value",
+            "selection_state",
+            "min_value",
+            "max_value",
+            "options",
+            "track_visible",
+            "knob_visible",
+            "orientation",
+        ),
+        "icon_button": (
+            "current_value",
+            "step_value",
+            "selection_state",
+            "min_value",
+            "max_value",
+            "options",
+            "track_visible",
+            "knob_visible",
+            "orientation",
+        ),
+        "segmented_control": (
+            "step_value",
+            "current_state",
+            "min_value",
+            "max_value",
+            "track_visible",
+            "knob_visible",
+            "orientation",
+        ),
+        "radio": (
+            "step_value",
+            "current_state",
+            "min_value",
+            "max_value",
+            "track_visible",
+            "knob_visible",
+            "orientation",
+        ),
+        "dropdown": (
+            "step_value",
+            "current_state",
+            "min_value",
+            "max_value",
+            "track_visible",
+            "knob_visible",
+            "orientation",
+        ),
+    }.get(control_type, ())
+    for field_name in null_fields:
+        normalized[field_name] = None
+    if control_type == "slider":
+        for field_name in ("current_value", "min_value", "max_value"):
+            normalized[field_name] = _normalize_percent_number(
+                normalized.get(field_name)
+            )
+    if intent.function == "unlock_mode":
+        normalized["current_value"] = _normalize_unlock_option(
+            normalized.get("current_value")
+        )
+        options = normalized.get("options")
+        if isinstance(options, list):
+            normalized["options"] = [
+                _normalize_unlock_option(option) for option in options
+            ]
+    return normalized
+
+
+def _is_selection_state(value: Any) -> bool:
+    if not isinstance(value, str):
+        return False
+    return _normalize_token(value) in {
+        "active",
+        "inactive",
+        "selected",
+        "unselected",
+        "not_selected",
+        "chosen",
+    }
+
+
+def _normalize_selection_state(value: Any) -> Any:
+    if not isinstance(value, str):
+        return value
+    token = _normalize_token(value)
+    aliases = {
+        "active": "selected",
+        "selected": "selected",
+        "chosen": "selected",
+        "enabled": "selected",
+        "inactive": "unselected",
+        "unselected": "unselected",
+        "not_selected": "unselected",
+        "disabled": "unselected",
+        "unknown": "unknown",
+    }
+    return aliases.get(token, value.strip())
+
+
+def _normalize_current_state(value: Any) -> Any:
+    if not isinstance(value, str):
+        return value
+    token = _normalize_token(value)
+    aliases = {
+        "enabled": "on",
+        "disabled": "off",
+        "unknown": "unknown",
+    }
+    return aliases.get(token, value.strip())
+
+
+def _normalize_token(value: str) -> str:
+    return "_".join(value.strip().lower().replace("-", " ").split())
+
+
+def _normalize_percent_number(value: Any) -> Any:
+    if isinstance(value, str) and value.strip().endswith("%"):
+        number = value.strip()[:-1].strip()
+        try:
+            parsed = float(number)
+        except ValueError:
+            return value
+        return int(parsed) if parsed.is_integer() else parsed
+    return value
+
+
+def _normalize_unlock_option(value: Any) -> Any:
+    if not isinstance(value, str):
+        return value
+    aliases = {
+        "主驾车门": "driver_door",
+        "主驾驶车门": "driver_door",
+        "driver door": "driver_door",
+        "所有车门": "all_doors",
+        "全部车门": "all_doors",
+        "all doors": "all_doors",
+    }
+    return aliases.get(value.strip().lower(), value.strip())
+
+
+def _build_ui_prompt(
+    intent: SemanticAction,
+    function_grounding: GroundingResult,
+) -> str:
+    ui_query = {
+        "domain": intent.domain.value,
+        "function": intent.function,
+        "zone": intent.zone.value if intent.zone is not None else None,
+        "function_target": function_grounding.semantic_query,
+    }
+    return f"""
+你是汽车座舱 UI 状态理解器。输入图像是视觉模型先定位功能区域后裁剪出的 ROI。
+请只分析 ROI 中给定功能的控件类型和当前可观察状态。
+你只负责理解当前 UI，不生成点击目标、bbox、坐标或机械臂动作。
+
+需要在图片中理解的功能：
+{json.dumps(ui_query, ensure_ascii=False)}
+
+用户期望的 operation、value 和 state 已刻意省略，因为它们不是当前 UI 状态。
+所有 current_* 字段必须从图片读取，不能复制或猜测用户期望值。
+
+Function Grounding 已经负责定位该功能；不要再次判断该功能是否存在。
+请根据 ROI 本身判断它由什么控件实现及其当前状态。
+不同车型可能使用不同布局、文字、图标或交互形式；不要假设其视觉位置或外观。
+
+严格输出一个 JSON 对象，必须且只能包含：
+control_type, current_value, step_value, current_state, selection_state,
+min_value, max_value, options, track_visible, knob_visible, orientation
+
+control_type 只能是：
+button, icon_button, toggle, slider, stepper, segmented_control, radio, checkbox, dropdown, tab, list_item 或 null
+
+规则：
+- 无法可靠观察的字段必须为 null，绝不能为了满足 schema 猜测数值或状态。
+- 如果无法可靠判断控件类型，control_type=null；Pipeline 会如实记录失败。
+- 带独立增减操作的数值控件是 stepper；读取对应 zone 的当前值和步长。
+- 可切换开关或有激活状态的图标输出 toggle 或 icon_button，并读取 on/off。
+- 对于 slider，端点旁的数值可能是 min/max，不是 current_value。分别读取明确的范围端点；只有画面明确单独显示当前值时才填写 current_value，否则必须为 null。
+- slider 能可靠看到滑轨或当前滑块时，将 track_visible 或 knob_visible 填 true；明确不可见时填 false，无法判断时填 null。
+- slider 的 orientation 只能根据图像填写 horizontal、vertical 或 null；无法确认方向时必须为 null。
+- 互斥选项组输出 segmented_control，current_value 和 options 使用语义值；视觉选中状态写入 selection_state，不写入 current_state。
+- stepper 的 current_state、min_value、max_value、options 必须为 null。
+- slider 的 step_value、current_state、selection_state、options 必须为 null；current_value 可以为 null。
+- toggle/icon_button 只有 current_state 可以是 on/off，其余数值和 options 字段必须为 null。
+- segmented_control 使用 current_value、selection_state 和 options，设备 current_state 以及其他数值字段必须为 null。
+- current_state 是设备功能状态，只能是 on、off、open、closed、unknown 或 null，绝不能填写 active/selected 或数值。
+- selection_state 是视觉选择状态，只能是 selected、unselected、unknown 或 null。
+- 除 slider 外，track_visible、knob_visible 和 orientation 必须为 null。
+- 不适用的字段必须为 null。
+- 只根据图像回答。不要输出 Markdown，不要解释，不要添加字段。
+""".strip()

src/cockpit_agent/pipeline.py

@@ -0,0 +1,372 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from cockpit_agent.common.enums import ControlType, Interaction
+from cockpit_agent.common.json_utils import write_json
+from cockpit_agent.common.model_registry import ModelRegistry
+from cockpit_agent.config import AgentConfig
+from cockpit_agent.execution.mock_executor import MockExecutor
+from cockpit_agent.grounding.adapter import GroundingAdapter, GroundingResult
+from cockpit_agent.grounding.schemas import ActionGroundingResult
+from cockpit_agent.grounding.semantic_target import SemanticTargetBuilder
+from cockpit_agent.intent.parser import IntentParser, ModelIntentParser
+from cockpit_agent.perception.roi import (
+    ROIResult,
+    crop_grounding_roi,
+    grounding_to_source,
+)
+from cockpit_agent.perception.ui_understanding import (
+    ModelUIUnderstanding,
+    UIUnderstanding,
+)
+from cockpit_agent.planning.planner import Planner
+
+
+class TaskPipeline:
+    def __init__(
+        self,
+        *,
+        intent_parser: IntentParser,
+        ui_understanding: UIUnderstanding,
+        planner: Planner,
+        target_builder: SemanticTargetBuilder,
+        grounding: GroundingAdapter,
+        executor: MockExecutor,
+        roi_padding_ratio: float,
+    ) -> None:
+        self._intent_parser = intent_parser
+        self._ui_understanding = ui_understanding
+        self._planner = planner
+        self._target_builder = target_builder
+        self._grounding = grounding
+        self._executor = executor
+        self._roi_padding_ratio = roi_padding_ratio
+
+    def run(
+        self,
+        *,
+        image_path: str,
+        instruction: str,
+        output_dir: str,
+    ) -> dict[str, Any]:
+        image = Path(image_path).expanduser().resolve()
+        if not image.is_file():
+            raise FileNotFoundError(f"Image not found: {image}")
+        output = Path(output_dir).expanduser().resolve()
+        output.mkdir(parents=True, exist_ok=True)
+        _clear_previous_artifacts(output)
+
+        run: dict[str, Any] = {
+            "success": False,
+            "stage": "intent",
+            "instruction": instruction,
+            "image": str(image),
+            "semantic_action": None,
+            "function_target": None,
+            "function_grounding": None,
+            "roi": None,
+            "ui_state": None,
+            "action_plan": None,
+            "action_target": None,
+            "action_grounding": None,
+            "mock_action": None,
+            "result_image": None,
+            "raw_model_outputs": {},
+        }
+
+        try:
+            intent = self._intent_parser.parse(instruction)
+        except Exception as exc:
+            self._record_failure(
+                output,
+                run,
+                stage="intent",
+                error=exc,
+                raw_output=getattr(self._intent_parser, "last_raw_output", None),
+            )
+            raise
+        run["semantic_action"] = intent.to_dict()
+        run["raw_model_outputs"]["intent"] = getattr(
+            self._intent_parser,
+            "last_raw_output",
+            None,
+        )
+        write_json(output / "intent.json", intent.to_dict())
+
+        run["stage"] = "function_target"
+        try:
+            function_target = self._target_builder.build_function_target(intent)
+        except Exception as exc:
+            self._record_failure(output, run, stage="function_target", error=exc)
+            raise
+        run["function_target"] = function_target
+
+        run["stage"] = "function_grounding"
+        try:
+            function_grounding = self._grounding.ground(
+                str(image), function_target
+            )
+        except Exception as exc:
+            self._record_failure(
+                output,
+                run,
+                stage="function_grounding",
+                error=exc,
+                raw_output=getattr(self._grounding, "last_raw_output", None),
+            )
+            raise
+        run["function_grounding"] = function_grounding.to_dict()
+        run["raw_model_outputs"]["function_grounding"] = (
+            function_grounding.raw_output
+        )
+        write_json(
+            output / "function_grounding.json",
+            function_grounding.to_dict(),
+        )
+
+        run["stage"] = "roi_crop"
+        try:
+            roi = crop_grounding_roi(
+                image_path=str(image),
+                function_grounding=function_grounding,
+                output_path=str(output / "roi.jpg"),
+                padding_ratio=self._roi_padding_ratio,
+            )
+        except Exception as exc:
+            self._record_failure(output, run, stage="roi_crop", error=exc)
+            raise
+        run["roi"] = roi.to_dict()
+
+        run["stage"] = "ui_understanding"
+        try:
+            ui_state = self._ui_understanding.understand(
+                roi.image_path,
+                intent,
+                function_grounding,
+            )
+        except Exception as exc:
+            self._record_failure(
+                output,
+                run,
+                stage="ui_understanding",
+                error=exc,
+                raw_output=getattr(self._ui_understanding, "last_raw_output", None),
+            )
+            raise
+        run["ui_state"] = ui_state.to_dict()
+        run["raw_model_outputs"]["ui_understanding"] = getattr(
+            self._ui_understanding,
+            "last_raw_output",
+            None,
+        )
+        write_json(output / "ui_state.json", ui_state.to_dict())
+
+        run["stage"] = "planning"
+        try:
+            plan = self._planner.plan(intent, ui_state)
+        except Exception as exc:
+            self._record_failure(
+                output,
+                run,
+                stage="planning",
+                error=exc,
+                raw_output=getattr(self._ui_understanding, "last_raw_output", None),
+            )
+            raise
+        run["action_plan"] = plan.to_dict()
+        write_json(output / "plan.json", plan.to_dict())
+
+        action_target: str | dict[str, str] | None = None
+        action_grounding: ActionGroundingResult | None = None
+        if plan.interaction is not Interaction.NOOP:
+            run["stage"] = "action_target"
+            try:
+                if plan.interaction is Interaction.DRAG:
+                    action_target = {
+                        "track": self._target_builder.build_slider_track_target(plan),
+                        "knob": self._target_builder.build_slider_knob_target(plan),
+                    }
+                else:
+                    action_target = self._target_builder.build_action_target(plan)
+            except Exception as exc:
+                self._record_failure(output, run, stage="action_target", error=exc)
+                raise
+            run["action_target"] = action_target
+
+            run["stage"] = "action_grounding"
+            try:
+                action_grounding = self._ground_action(
+                    roi_image_path=roi.image_path,
+                    roi=roi,
+                    function_grounding=function_grounding,
+                    ui_control_type=ui_state.control_type,
+                    plan_function=plan.function,
+                    intent_function=intent.function,
+                    action_target=action_target,
+                )
+            except Exception as exc:
+                self._record_failure(
+                    output,
+                    run,
+                    stage="action_grounding",
+                    error=exc,
+                    raw_output=getattr(self._grounding, "last_raw_output", None),
+                )
+                raise
+            run["action_grounding"] = action_grounding.to_dict()
+            run["raw_model_outputs"]["action_grounding"] = {
+                name: result.raw_output
+                for name, result in (
+                    ("control", action_grounding.control),
+                    ("track", action_grounding.track),
+                    ("knob", action_grounding.knob),
+                )
+                if result is not None
+            }
+            write_json(
+                output / "action_grounding.json",
+                action_grounding.to_dict(),
+            )
+
+        result_image = output / "result.jpg"
+        run["stage"] = "mock_action"
+        try:
+            action = self._executor.propose(
+                image_path=str(image),
+                plan=plan,
+                function_grounding=function_grounding,
+                action_target=action_target,
+                action_grounding=action_grounding,
+                output_path=str(result_image),
+            )
+        except Exception as exc:
+            self._record_failure(output, run, stage="mock_action", error=exc)
+            raise
+        run["mock_action"] = action.to_dict()
+        write_json(output / "action.json", action.to_dict())
+
+        run["success"] = True
+        run["stage"] = "complete"
+        run["result_image"] = str(result_image)
+        write_json(output / "run.json", run)
+        return run
+
+    def _ground_action(
+        self,
+        *,
+        roi_image_path: str,
+        roi: ROIResult,
+        function_grounding: GroundingResult,
+        ui_control_type: ControlType | None,
+        plan_function: str | None,
+        intent_function: str,
+        action_target: str | dict[str, str],
+    ) -> ActionGroundingResult:
+        if isinstance(action_target, dict):
+            track_roi = self._grounding.ground(
+                roi_image_path,
+                action_target["track"],
+            )
+            track = grounding_to_source(track_roi, roi)
+            knob_roi = self._grounding.ground(
+                roi_image_path,
+                action_target["knob"],
+            )
+            knob = grounding_to_source(knob_roi, roi)
+            return ActionGroundingResult(
+                source="roi_action_grounding",
+                track=track,
+                knob=knob,
+            )
+
+        can_reuse_function = (
+            plan_function == intent_function
+            and ui_control_type
+            in {
+                ControlType.BUTTON,
+                ControlType.ICON_BUTTON,
+                ControlType.TOGGLE,
+                ControlType.CHECKBOX,
+            }
+        )
+        if can_reuse_function:
+            return ActionGroundingResult(
+                source="function_grounding",
+                control=function_grounding,
+            )
+
+        action_roi = self._grounding.ground(roi_image_path, action_target)
+        return ActionGroundingResult(
+            source="roi_action_grounding",
+            control=grounding_to_source(action_roi, roi),
+        )
+
+    @staticmethod
+    def _record_failure(
+        output: Path,
+        run: dict[str, Any],
+        *,
+        stage: str,
+        error: Exception,
+        raw_output: str | None = None,
+    ) -> None:
+        failure: dict[str, Any] = {
+            "success": False,
+            "failed_stage": stage,
+            "error_type": type(error).__name__,
+            "error_message": str(error),
+            "raw_model_output": raw_output,
+        }
+        run["success"] = False
+        run["stage"] = stage
+        if raw_output is not None:
+            run.setdefault("raw_model_outputs", {})[stage] = raw_output
+        run.update(failure)
+        write_json(output / "failure.json", failure)
+        write_json(output / "run.json", run)
+
+
+def _clear_previous_artifacts(output: Path) -> None:
+    for name in (
+        "intent.json",
+        "function_grounding.json",
+        "roi.jpg",
+        "ui_state.json",
+        "plan.json",
+        "action_grounding.json",
+        "action.json",
+        "failure.json",
+        "result.jpg",
+        "run.json",
+        "grounding.json",
+    ):
+        (output / name).unlink(missing_ok=True)
+
+
+def build_pipeline(
+    config: AgentConfig,
+    registry: ModelRegistry,
+) -> TaskPipeline:
+    intent_model = registry.get_text_generator(config.intent)
+    ui_model = registry.get(config.ui_understanding)
+    grounding_model = registry.get(config.grounding)
+    return TaskPipeline(
+        intent_parser=ModelIntentParser(
+            intent_model,
+            max_new_tokens=config.intent.max_new_tokens,
+        ),
+        ui_understanding=ModelUIUnderstanding(
+            ui_model,
+            max_new_tokens=config.ui_understanding.max_new_tokens,
+        ),
+        planner=Planner(),
+        target_builder=SemanticTargetBuilder(),
+        grounding=GroundingAdapter(
+            grounding_model,
+            max_new_tokens=config.grounding.max_new_tokens,
+        ),
+        executor=MockExecutor(),
+        roi_padding_ratio=config.perception.roi_padding_ratio,
+    )

src/cockpit_agent/planning/__init__.py

@@ -0,0 +1,4 @@
+from cockpit_agent.planning.planner import Planner, PlanningError
+from cockpit_agent.planning.schemas import ActionPlan
+
+__all__ = ["ActionPlan", "Planner", "PlanningError"]

src/cockpit_agent/planning/planner.py

@@ -0,0 +1,237 @@
+from __future__ import annotations
+
+import math
+
+from cockpit_agent.common.enums import (
+    ControlType,
+    Interaction,
+    Operation,
+    State,
+    SelectionState,
+    SliderOrientation,
+)
+from cockpit_agent.intent.schemas import SemanticAction
+from cockpit_agent.perception.schemas import UIState
+from cockpit_agent.planning.schemas import ActionPlan
+
+
+class PlanningError(ValueError):
+    pass
+
+
+class Planner:
+    def plan(self, intent: SemanticAction, ui_state: UIState) -> ActionPlan:
+        if ui_state.function != intent.function:
+            raise PlanningError(
+                "UIState function does not match SemanticAction: "
+                f"{ui_state.function!r} != {intent.function!r}"
+            )
+        if ui_state.zone != intent.zone:
+            raise PlanningError(
+                "UIState zone does not match SemanticAction: "
+                f"{ui_state.zone!r} != {intent.zone!r}"
+            )
+        if ui_state.control_type is None:
+            raise PlanningError("UIState requires a control_type for planning")
+        if ui_state.control_type is ControlType.STEPPER:
+            return self._plan_stepper(intent, ui_state)
+        if ui_state.control_type is ControlType.TOGGLE:
+            return self._plan_toggle(intent, ui_state)
+        if ui_state.control_type is ControlType.ICON_BUTTON:
+            return self._plan_icon_button(intent, ui_state)
+        if ui_state.control_type is ControlType.SLIDER:
+            return self._plan_slider(intent, ui_state)
+        if ui_state.control_type in {
+            ControlType.SEGMENTED_CONTROL,
+            ControlType.RADIO,
+            ControlType.DROPDOWN,
+        }:
+            return self._plan_option_selection(intent, ui_state)
+        if ui_state.control_type in {
+            ControlType.BUTTON,
+            ControlType.CHECKBOX,
+            ControlType.TAB,
+            ControlType.LIST_ITEM,
+        }:
+            _reject_unrepresentable_value_target(intent, ui_state.control_type)
+            return ActionPlan(
+                interaction=Interaction.TAP,
+                function=intent.function,
+                zone=intent.zone,
+                option=(intent.value if isinstance(intent.value, str) else None),
+            )
+        raise PlanningError(f"Unsupported control type: {ui_state.control_type.value}")
+
+    def _plan_stepper(
+        self,
+        intent: SemanticAction,
+        ui_state: UIState,
+    ) -> ActionPlan:
+        current = _number(ui_state.current_value, "stepper current_value")
+        if ui_state.step_value is None:
+            raise PlanningError("stepper requires step_value; refusing to guess")
+        step = float(ui_state.step_value)
+
+        if intent.operation is Operation.SET:
+            target = _number(intent.value, "stepper target value")
+            delta = target - current
+        elif intent.operation in {Operation.INCREASE, Operation.DECREASE}:
+            amount = 1.0 if intent.value is None else _number(
+                intent.value,
+                "stepper change amount",
+            )
+            delta = amount if intent.operation is Operation.INCREASE else -amount
+            target = current + delta
+        else:
+            raise PlanningError(
+                f"Stepper does not support operation {intent.operation.value}"
+            )
+
+        if math.isclose(delta, 0.0, abs_tol=1e-9):
+            return ActionPlan(
+                interaction=Interaction.NOOP,
+                reason="already_in_target_state",
+            )
+        repeat_float = abs(delta) / step
+        repeat = round(repeat_float)
+        if not math.isclose(repeat_float, repeat, rel_tol=1e-9, abs_tol=1e-9):
+            raise PlanningError(
+                f"Requested change {abs(delta):g} is not divisible by step {step:g}"
+            )
+        direction = "increase" if delta > 0 else "decrease"
+        return ActionPlan(
+            interaction=Interaction.TAP,
+            function=f"{intent.function}_{direction}",
+            zone=intent.zone,
+            repeat=repeat,
+            target_value=_clean_number(target),
+        )
+
+    def _plan_toggle(
+        self,
+        intent: SemanticAction,
+        ui_state: UIState,
+    ) -> ActionPlan:
+        _reject_unrepresentable_value_target(intent, ui_state.control_type)
+        desired = intent.state
+        if intent.operation is Operation.TOGGLE:
+            desired = None
+        elif desired not in {State.ON, State.OFF, State.OPEN, State.CLOSED}:
+            raise PlanningError("toggle set operation requires a target state")
+
+        if desired is not None:
+            if ui_state.current_state is desired:
+                return ActionPlan(
+                    interaction=Interaction.NOOP,
+                    reason="already_in_target_state",
+                )
+        return ActionPlan(
+            interaction=Interaction.TAP,
+            function=intent.function,
+            zone=intent.zone,
+        )
+
+    def _plan_icon_button(
+        self,
+        intent: SemanticAction,
+        ui_state: UIState,
+    ) -> ActionPlan:
+        _reject_unrepresentable_value_target(intent, ui_state.control_type)
+        desired = intent.state
+        if desired is not None and ui_state.current_state is desired:
+            return ActionPlan(
+                interaction=Interaction.NOOP,
+                reason="already_in_target_state",
+            )
+        return ActionPlan(
+            interaction=Interaction.TAP,
+            function=intent.function,
+            zone=intent.zone,
+        )
+
+    def _plan_slider(
+        self,
+        intent: SemanticAction,
+        ui_state: UIState,
+    ) -> ActionPlan:
+        target = _number(intent.value, "slider target value")
+        minimum = _number(ui_state.min_value, "slider min_value")
+        maximum = _number(ui_state.max_value, "slider max_value")
+        if not minimum <= target <= maximum:
+            raise PlanningError(
+                f"Slider target {target:g} outside range [{minimum:g}, {maximum:g}]"
+            )
+        width = maximum - minimum
+        if width <= 0:
+            raise PlanningError("slider max_value must be greater than min_value")
+        if ui_state.current_value is not None:
+            current = _number(ui_state.current_value, "slider current_value")
+            if not minimum <= current <= maximum:
+                raise PlanningError(
+                    f"Slider current value {current:g} outside configured range"
+                )
+            if math.isclose(current, target, abs_tol=1e-9):
+                return ActionPlan(
+                    interaction=Interaction.NOOP,
+                    reason="already_in_target_state",
+                )
+        if ui_state.orientation is not SliderOrientation.HORIZONTAL:
+            raise PlanningError(
+                "slider requires an explicitly observed horizontal orientation"
+            )
+        return ActionPlan(
+            interaction=Interaction.DRAG,
+            function=intent.function,
+            zone=intent.zone,
+            target_value=_clean_number(target),
+            normalized_target=(target - minimum) / width,
+            orientation=ui_state.orientation,
+        )
+
+    def _plan_option_selection(
+        self,
+        intent: SemanticAction,
+        ui_state: UIState,
+    ) -> ActionPlan:
+        desired = intent.value
+        if not isinstance(desired, str):
+            raise PlanningError("option control requires a string target value")
+        if (
+            ui_state.current_value == desired
+            and ui_state.selection_state is SelectionState.SELECTED
+        ):
+            return ActionPlan(
+                interaction=Interaction.NOOP,
+                reason="already_in_target_state",
+            )
+        if not ui_state.options or desired not in ui_state.options:
+            raise PlanningError(
+                f"Target option {desired!r} is not present in UI options"
+            )
+        return ActionPlan(
+            interaction=Interaction.TAP,
+            function=intent.function,
+            zone=intent.zone,
+            option=desired,
+        )
+
+
+def _number(value: object, field_name: str) -> float:
+    if isinstance(value, bool) or not isinstance(value, (int, float)):
+        raise PlanningError(f"{field_name} must be numeric")
+    return float(value)
+
+
+def _clean_number(value: float) -> int | float:
+    return int(value) if value.is_integer() else value
+
+
+def _reject_unrepresentable_value_target(
+    intent: SemanticAction,
+    control_type: ControlType,
+) -> None:
+    if intent.operation is Operation.SET and isinstance(intent.value, (int, float)):
+        raise PlanningError(
+            f"{control_type.value} cannot deterministically set numeric target "
+            f"{intent.value!r}"
+        )

src/cockpit_agent/planning/schemas.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+from cockpit_agent.common.enums import Interaction, SliderOrientation, Zone
+from cockpit_agent.common.schema import SchemaValidationError
+
+
+@dataclass(frozen=True)
+class ActionPlan:
+    interaction: Interaction
+    function: str | None = None
+    zone: Zone | None = None
+    option: str | None = None
+    repeat: int = 1
+    target_value: int | float | str | None = None
+    normalized_target: float | None = None
+    reason: str | None = None
+    orientation: SliderOrientation | None = None
+
+    def __post_init__(self) -> None:
+        if (
+            isinstance(self.repeat, bool)
+            or not isinstance(self.repeat, int)
+            or self.repeat < 1
+        ):
+            raise SchemaValidationError("ActionPlan.repeat must be a positive integer")
+        if self.interaction is Interaction.NOOP:
+            if not self.reason:
+                raise SchemaValidationError("noop plan requires a reason")
+            return
+        if not isinstance(self.function, str) or not self.function.strip():
+            raise SchemaValidationError("action plan requires a semantic function")
+        if self.option is not None and (
+            not isinstance(self.option, str) or not self.option.strip()
+        ):
+            raise SchemaValidationError("ActionPlan.option must be non-empty or null")
+        if self.interaction is Interaction.DRAG:
+            if (
+                self.normalized_target is None
+                or not 0.0 <= self.normalized_target <= 1.0
+            ):
+                raise SchemaValidationError(
+                    "drag normalized_target must be between 0 and 1"
+                )
+            if self.orientation is not SliderOrientation.HORIZONTAL:
+                raise SchemaValidationError(
+                    "drag requires an explicitly observed horizontal orientation"
+                )
+
+    def to_dict(self) -> dict[str, Any]:
+        if self.interaction is Interaction.NOOP:
+            return {
+                "interaction": self.interaction.value,
+                "reason": self.reason,
+            }
+        value: dict[str, Any] = {
+            "interaction": self.interaction.value,
+            "function": self.function,
+            "zone": self.zone.value if self.zone is not None else None,
+            "option": self.option,
+        }
+        if self.interaction is Interaction.TAP:
+            value["repeat"] = self.repeat
+        if self.target_value is not None:
+            value["target_value"] = self.target_value
+        if self.normalized_target is not None:
+            value["normalized_target"] = self.normalized_target
+        if self.orientation is not None:
+            value["orientation"] = self.orientation.value
+        return value

tests/test_grounding_adapter.py

@@ -0,0 +1,65 @@
+import tempfile
+import unittest
+from pathlib import Path
+
+from PIL import Image
+
+from cockpit_agent.grounding.adapter import GroundingAdapter
+
+
+class FakeGroundingModel:
+    def __init__(self, output: str) -> None:
+        self.output = output
+        self.prompt: str | None = None
+
+    def generate(
+        self,
+        image_path: str,
+        prompt: str,
+        max_new_tokens: int = 128,
+    ) -> str:
+        del image_path, max_new_tokens
+        self.prompt = prompt
+        return self.output
+
+
+class GroundingAdapterTest(unittest.TestCase):
+    def test_uses_model_bbox_without_postprocessing(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            image_path = Path(directory) / "generic.png"
+            Image.new("RGB", (200, 100), color="white").save(image_path)
+            model = FakeGroundingModel('{"bbox_2d": [100, 200, 400, 600]}')
+
+            result = GroundingAdapter(model).ground(
+                str(image_path),
+                "车辆空调内循环控制",
+            )
+
+        self.assertEqual(result.bbox_relative, (100.0, 200.0, 400.0, 600.0))
+        self.assertEqual(result.bbox_pixel, (20, 20, 80, 60))
+        self.assertEqual(result.center_pixel, (50, 40))
+        self.assertIn("车辆空调内循环控制", model.prompt or "")
+        self.assertEqual(
+            set(result.to_dict()),
+            {
+                "semantic_query",
+                "bbox_relative",
+                "bbox_pixel",
+                "center_pixel",
+                "image_width",
+                "image_height",
+                "raw_output",
+            },
+        )
+
+    def test_invalid_model_output_is_not_repaired(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            image_path = Path(directory) / "generic.png"
+            Image.new("RGB", (32, 32), color="white").save(image_path)
+            adapter = GroundingAdapter(FakeGroundingModel("not a bbox"))
+            with self.assertRaises(ValueError):
+                adapter.ground(str(image_path), "任意语义控件")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_intent.py

@@ -0,0 +1,104 @@
+import json
+import unittest
+
+from cockpit_agent.intent.parser import ModelIntentParser
+
+
+EXPECTED = {
+    "主驾23度": {
+        "domain": "climate",
+        "function": "temperature",
+        "zone": "driver",
+        "operation": "set",
+        "state": None,
+        "value": 23,
+        "unit": "celsius",
+    },
+    "把驾驶位温度调成23℃": {
+        "domain": "climate",
+        "function": "temperature",
+        "zone": "driver",
+        "operation": "set",
+        "state": None,
+        "value": 23,
+        "unit": "celsius",
+    },
+    "副驾温度降低2度": {
+        "domain": "climate",
+        "function": "temperature",
+        "zone": "passenger",
+        "operation": "decrease",
+        "state": None,
+        "value": 2,
+        "unit": "celsius",
+    },
+    "打开内循环": {
+        "domain": "climate",
+        "function": "air_recirculation",
+        "zone": "all",
+        "operation": "set",
+        "state": "on",
+        "value": None,
+        "unit": None,
+    },
+    "关闭主驾空调": {
+        "domain": "climate",
+        "function": "climate_power",
+        "zone": "driver",
+        "operation": "set",
+        "state": "off",
+        "value": None,
+        "unit": None,
+    },
+    "后备箱最大开度80%": {
+        "domain": "body",
+        "function": "tailgate_max_opening",
+        "zone": None,
+        "operation": "set",
+        "state": None,
+        "value": 80,
+        "unit": "percent",
+    },
+    "解锁全部车门": {
+        "domain": "body",
+        "function": "unlock_mode",
+        "zone": None,
+        "operation": "set",
+        "state": None,
+        "value": "all_doors",
+        "unit": None,
+    },
+}
+
+
+class FakeTextGenerator:
+    def generate_text(self, prompt: str, max_new_tokens: int = 256) -> str:
+        del max_new_tokens
+        instruction = prompt.rsplit("用户指令：", maxsplit=1)[1].strip()
+        return "```json\n" + json.dumps(EXPECTED[instruction]) + "\n```"
+
+
+class IntentParserTest(unittest.TestCase):
+    def test_required_instruction_variants(self) -> None:
+        parser = ModelIntentParser(FakeTextGenerator())
+        for instruction, expected in EXPECTED.items():
+            with self.subTest(instruction=instruction):
+                self.assertEqual(parser.parse(instruction).to_dict(), expected)
+
+    def test_empty_instruction_is_rejected(self) -> None:
+        parser = ModelIntentParser(FakeTextGenerator())
+        with self.assertRaisesRegex(ValueError, "non-empty"):
+            parser.parse("  ")
+
+    def test_invalid_model_schema_is_rejected(self) -> None:
+        class InvalidGenerator:
+            def generate_text(self, prompt: str, max_new_tokens: int = 256) -> str:
+                del prompt, max_new_tokens
+                return '{"domain": "climate"}'
+
+        with self.assertRaisesRegex(ValueError, "missing fields"):
+            ModelIntentParser(InvalidGenerator()).parse("打开内循环")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_mock_executor.py

@@ -0,0 +1,48 @@
+import unittest
+
+from cockpit_agent.common.enums import Interaction, SliderOrientation
+from cockpit_agent.execution.mock_executor import MockExecutor
+from cockpit_agent.grounding.adapter import GroundingResult
+from cockpit_agent.grounding.schemas import ActionGroundingResult
+from cockpit_agent.planning.schemas import ActionPlan
+
+
+def grounding(
+    query: str,
+    bbox: tuple[int, int, int, int],
+) -> GroundingResult:
+    x1, y1, x2, y2 = bbox
+    return GroundingResult(
+        semantic_query=query,
+        bbox_relative=(0, 0, 0, 0),
+        bbox_pixel=bbox,
+        center_pixel=(round((x1 + x2) / 2), round((y1 + y2) / 2)),
+        image_width=400,
+        image_height=200,
+        raw_output="model output",
+    )
+
+
+class MockExecutorGeometryTest(unittest.TestCase):
+    def test_slider_uses_knob_center_and_track_target(self) -> None:
+        plan = ActionPlan(
+            interaction=Interaction.DRAG,
+            function="tailgate_max_opening",
+            target_value=80,
+            normalized_target=0.6,
+            orientation=SliderOrientation.HORIZONTAL,
+        )
+        action = MockExecutor._propose_drag(
+            plan,
+            ActionGroundingResult(
+                source="roi_action_grounding",
+                track=grounding("track", (100, 80, 300, 100)),
+                knob=grounding("knob", (270, 75, 290, 105)),
+            ),
+        )
+        self.assertEqual(action.start_pixel, (280, 90))
+        self.assertEqual(action.end_pixel, (220, 90))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_pipeline.py

@@ -0,0 +1,197 @@
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+from PIL import Image
+
+from cockpit_agent.common.enums import (
+    ControlType,
+    Domain,
+    Operation,
+    Unit,
+    Zone,
+)
+from cockpit_agent.execution.mock_executor import MockExecutor
+from cockpit_agent.grounding.adapter import GroundingResult
+from cockpit_agent.grounding.semantic_target import SemanticTargetBuilder
+from cockpit_agent.intent.schemas import SemanticAction
+from cockpit_agent.perception.schemas import UIState
+from cockpit_agent.pipeline import TaskPipeline
+from cockpit_agent.planning.planner import Planner
+
+
+class FakeIntentParser:
+    last_raw_output = '{"intent": "raw"}'
+
+    def parse(self, instruction: str) -> SemanticAction:
+        del instruction
+        return SemanticAction(
+            domain=Domain.CLIMATE,
+            function="temperature",
+            zone=Zone.DRIVER,
+            operation=Operation.SET,
+            state=None,
+            value=23,
+            unit=Unit.CELSIUS,
+        )
+
+
+class FakeUIUnderstanding:
+    last_raw_output = '{"ui": "raw"}'
+
+    def __init__(self) -> None:
+        self.image_path: str | None = None
+        self.function_query: str | None = None
+
+    def understand(
+        self,
+        roi_image_path: str,
+        intent: SemanticAction,
+        function_grounding: GroundingResult,
+    ) -> UIState:
+        self.image_path = roi_image_path
+        self.function_query = function_grounding.semantic_query
+        return UIState(
+            function=intent.function,
+            zone=intent.zone,
+            control_type=ControlType.STEPPER,
+            current_value=26,
+            step_value=1,
+            current_state=None,
+            min_value=None,
+            max_value=None,
+            options=None,
+        )
+
+
+class FakeGrounding:
+    last_raw_output = '{"bbox_2d": [100, 200, 300, 400]}'
+
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, str]] = []
+
+    def ground(self, image_path: str, semantic_query: str) -> GroundingResult:
+        self.calls.append((image_path, semantic_query))
+        with Image.open(image_path) as image:
+            width, height = image.size
+        if len(self.calls) == 1:
+            bbox = (20, 20, 80, 60)
+        else:
+            bbox = (6, 8, 18, 24)
+        x1, y1, x2, y2 = bbox
+        return GroundingResult(
+            semantic_query=semantic_query,
+            bbox_relative=(
+                x1 / width * 1000,
+                y1 / height * 1000,
+                x2 / width * 1000,
+                y2 / height * 1000,
+            ),
+            bbox_pixel=bbox,
+            center_pixel=(round((x1 + x2) / 2), round((y1 + y2) / 2)),
+            image_width=width,
+            image_height=height,
+            raw_output=self.last_raw_output,
+        )
+
+
+class FailingGrounding:
+    last_raw_output = "model returned no bbox"
+
+    def ground(self, image_path: str, semantic_query: str) -> GroundingResult:
+        del image_path, semantic_query
+        raise ValueError("cannot parse bbox")
+
+
+class PipelineTest(unittest.TestCase):
+    TEMPERATURE_FUNCTION_TARGET = (
+        "主驾驶温度调节控件整体，"
+        "包含当前温度显示以及用于升高和降低温度的控制元素"
+    )
+
+    def _pipeline(self, grounding, ui=None) -> TaskPipeline:
+        return TaskPipeline(
+            intent_parser=FakeIntentParser(),
+            ui_understanding=ui or FakeUIUnderstanding(),
+            planner=Planner(),
+            target_builder=SemanticTargetBuilder(),
+            grounding=grounding,
+            executor=MockExecutor(),
+            roi_padding_ratio=0.10,
+        )
+
+    def test_function_grounding_precedes_roi_ui_and_action_grounding(self) -> None:
+        grounding = FakeGrounding()
+        ui = FakeUIUnderstanding()
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            image = root / "input.png"
+            output = root / "output"
+            Image.new("RGB", (200, 100), color="white").save(image)
+            run = self._pipeline(grounding, ui).run(
+                image_path=str(image),
+                instruction="set temperature",
+                output_dir=str(output),
+            )
+            roi_exists = (output / "roi.jpg").is_file()
+            function_file_exists = (output / "function_grounding.json").is_file()
+            action_file_exists = (output / "action_grounding.json").is_file()
+
+        self.assertTrue(run["success"])
+        self.assertEqual(run["function_target"], self.TEMPERATURE_FUNCTION_TARGET)
+        self.assertEqual(grounding.calls[0][1], self.TEMPERATURE_FUNCTION_TARGET)
+        self.assertEqual(grounding.calls[1][1], "主驾驶温度降低控制")
+        self.assertTrue(grounding.calls[1][0].endswith("roi.jpg"))
+        self.assertTrue((ui.image_path or "").endswith("roi.jpg"))
+        self.assertEqual(ui.function_query, self.TEMPERATURE_FUNCTION_TARGET)
+        self.assertEqual(run["action_plan"]["function"], "temperature_decrease")
+        self.assertEqual(run["action_target"], "主驾驶温度降低控制")
+        self.assertEqual(
+            run["action_grounding"]["source"],
+            "roi_action_grounding",
+        )
+        self.assertEqual(run["mock_action"]["status"], "PROPOSED")
+        self.assertTrue(roi_exists)
+        self.assertTrue(function_file_exists)
+        self.assertTrue(action_file_exists)
+        self.assertEqual(run["raw_model_outputs"]["intent"], '{"intent": "raw"}')
+        self.assertEqual(run["raw_model_outputs"]["ui_understanding"], '{"ui": "raw"}')
+        self.assertIn("function_grounding", run["raw_model_outputs"])
+        self.assertIn("action_grounding", run["raw_model_outputs"])
+
+    def test_function_grounding_failure_is_recorded_before_ui(self) -> None:
+        ui = FakeUIUnderstanding()
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            image = root / "input.png"
+            output = root / "output"
+            Image.new("RGB", (200, 100), color="white").save(image)
+            with self.assertRaisesRegex(ValueError, "cannot parse bbox"):
+                self._pipeline(FailingGrounding(), ui).run(
+                    image_path=str(image),
+                    instruction="set temperature",
+                    output_dir=str(output),
+                )
+            failure = json.loads((output / "failure.json").read_text())
+            run = json.loads((output / "run.json").read_text())
+
+        self.assertEqual(failure["failed_stage"], "function_grounding")
+        self.assertEqual(failure["error_type"], "ValueError")
+        self.assertEqual(
+            failure["raw_model_output"],
+            "model returned no bbox",
+        )
+        self.assertEqual(run["function_target"], self.TEMPERATURE_FUNCTION_TARGET)
+        self.assertEqual(
+            run["raw_model_outputs"]["function_grounding"],
+            "model returned no bbox",
+        )
+        self.assertIsNone(run["function_grounding"])
+        self.assertIsNone(run["roi"])
+        self.assertIsNone(run["ui_state"])
+        self.assertIsNone(ui.image_path)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_planner.py

@@ -0,0 +1,334 @@
+import unittest
+
+from cockpit_agent.common.enums import (
+    ControlType,
+    Domain,
+    Interaction,
+    Operation,
+    SelectionState,
+    SliderOrientation,
+    State,
+    Unit,
+    Zone,
+)
+from cockpit_agent.intent.schemas import SemanticAction
+from cockpit_agent.perception.schemas import UIState
+from cockpit_agent.planning.planner import Planner, PlanningError
+
+
+def temperature_action(value: int) -> SemanticAction:
+    return SemanticAction(
+        domain=Domain.CLIMATE,
+        function="temperature",
+        zone=Zone.DRIVER,
+        operation=Operation.SET,
+        state=None,
+        value=value,
+        unit=Unit.CELSIUS,
+    )
+
+
+def ui_state(
+    control_type: ControlType | None,
+    *,
+    function: str = "temperature",
+    zone: Zone | None = Zone.DRIVER,
+    current_value=None,
+    step_value=None,
+    current_state=None,
+    min_value=None,
+    max_value=None,
+    options=None,
+    selection_state=None,
+    track_visible=None,
+    knob_visible=None,
+    orientation=None,
+) -> UIState:
+    return UIState(
+        function=function,
+        zone=zone,
+        control_type=control_type,
+        current_value=current_value,
+        step_value=step_value,
+        current_state=current_state,
+        min_value=min_value,
+        max_value=max_value,
+        options=options,
+        selection_state=selection_state,
+        track_visible=track_visible,
+        knob_visible=knob_visible,
+        orientation=orientation,
+    )
+
+
+def recirculation_action() -> SemanticAction:
+    return SemanticAction(
+        domain=Domain.CLIMATE,
+        function="air_recirculation",
+        zone=Zone.ALL,
+        operation=Operation.SET,
+        state=State.ON,
+        value=None,
+        unit=None,
+    )
+
+
+class PlannerTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.planner = Planner()
+
+    def test_stepper_decreases_three_times(self) -> None:
+        plan = self.planner.plan(
+            temperature_action(23),
+            ui_state(ControlType.STEPPER, current_value=26, step_value=1),
+        )
+        self.assertEqual(plan.interaction, Interaction.TAP)
+        self.assertEqual(plan.function, "temperature_decrease")
+        self.assertEqual(plan.zone, Zone.DRIVER)
+        self.assertEqual(plan.repeat, 3)
+        self.assertEqual(plan.target_value, 23)
+        self.assertNotIn("target", plan.to_dict())
+
+    def test_stepper_increases_three_times(self) -> None:
+        plan = self.planner.plan(
+            temperature_action(23),
+            ui_state(ControlType.STEPPER, current_value=20, step_value=1),
+        )
+        self.assertEqual(plan.function, "temperature_increase")
+        self.assertEqual(plan.repeat, 3)
+
+    def test_stepper_without_step_is_rejected(self) -> None:
+        with self.assertRaisesRegex(PlanningError, "refusing to guess"):
+            self.planner.plan(
+                temperature_action(23),
+                ui_state(ControlType.STEPPER, current_value=26),
+            )
+
+    def test_toggle(self) -> None:
+        plan = self.planner.plan(
+            recirculation_action(),
+            ui_state(
+                ControlType.TOGGLE,
+                function="air_recirculation",
+                zone=Zone.ALL,
+                current_state=State.OFF,
+            ),
+        )
+        self.assertEqual(plan.interaction, Interaction.TAP)
+        self.assertEqual(plan.function, "air_recirculation")
+
+    def test_icon_button_with_unknown_state_proposes_tap(self) -> None:
+        plan = self.planner.plan(
+            recirculation_action(),
+            ui_state(
+                ControlType.ICON_BUTTON,
+                function="air_recirculation",
+                zone=Zone.ALL,
+                current_state=None,
+            ),
+        )
+        self.assertEqual(plan.interaction, Interaction.TAP)
+        self.assertEqual(plan.function, "air_recirculation")
+
+    def test_icon_button_cannot_set_numeric_target(self) -> None:
+        with self.assertRaisesRegex(PlanningError, "numeric target"):
+            self.planner.plan(
+                temperature_action(23),
+                ui_state(ControlType.ICON_BUTTON, current_state=None),
+            )
+
+    def test_slider(self) -> None:
+        intent = SemanticAction(
+            domain=Domain.BODY,
+            function="tailgate_max_opening",
+            zone=None,
+            operation=Operation.SET,
+            state=None,
+            value=80,
+            unit=Unit.PERCENT,
+        )
+        plan = self.planner.plan(
+            intent,
+            ui_state(
+                ControlType.SLIDER,
+                function="tailgate_max_opening",
+                zone=None,
+                current_value=90,
+                min_value=50,
+                max_value=100,
+                orientation=SliderOrientation.HORIZONTAL,
+            ),
+        )
+        self.assertEqual(plan.interaction, Interaction.DRAG)
+        self.assertEqual(plan.function, "tailgate_max_opening")
+        self.assertAlmostEqual(plan.normalized_target or 0, 0.6)
+
+    def test_slider_allows_unknown_current_value(self) -> None:
+        intent = SemanticAction(
+            domain=Domain.BODY,
+            function="tailgate_max_opening",
+            zone=None,
+            operation=Operation.SET,
+            state=None,
+            value=80,
+            unit=Unit.PERCENT,
+        )
+        plan = self.planner.plan(
+            intent,
+            ui_state(
+                ControlType.SLIDER,
+                function="tailgate_max_opening",
+                zone=None,
+                current_value=None,
+                min_value=50,
+                max_value=100,
+                orientation=SliderOrientation.HORIZONTAL,
+            ),
+        )
+        self.assertEqual(plan.interaction, Interaction.DRAG)
+        self.assertAlmostEqual(plan.normalized_target or 0, 0.6)
+
+    def test_slider_requires_observed_horizontal_orientation(self) -> None:
+        intent = SemanticAction(
+            domain=Domain.BODY,
+            function="tailgate_max_opening",
+            zone=None,
+            operation=Operation.SET,
+            state=None,
+            value=80,
+            unit=Unit.PERCENT,
+        )
+        with self.assertRaisesRegex(PlanningError, "horizontal orientation"):
+            self.planner.plan(
+                intent,
+                ui_state(
+                    ControlType.SLIDER,
+                    function="tailgate_max_opening",
+                    zone=None,
+                    current_value=None,
+                    min_value=50,
+                    max_value=100,
+                    orientation=None,
+                ),
+            )
+
+    def test_segmented_control(self) -> None:
+        intent = SemanticAction(
+            domain=Domain.BODY,
+            function="unlock_mode",
+            zone=None,
+            operation=Operation.SET,
+            state=None,
+            value="all_doors",
+            unit=None,
+        )
+        plan = self.planner.plan(
+            intent,
+            ui_state(
+                ControlType.SEGMENTED_CONTROL,
+                function="unlock_mode",
+                zone=None,
+                current_value="driver_door",
+                options=("driver_door", "all_doors"),
+                selection_state=SelectionState.SELECTED,
+            ),
+        )
+        self.assertEqual(plan.interaction, Interaction.TAP)
+        self.assertEqual(plan.function, "unlock_mode")
+        self.assertEqual(plan.option, "all_doors")
+
+    def test_noop(self) -> None:
+        plan = self.planner.plan(
+            recirculation_action(),
+            ui_state(
+                ControlType.TOGGLE,
+                function="air_recirculation",
+                zone=Zone.ALL,
+                current_state=State.ON,
+            ),
+        )
+        self.assertEqual(plan.interaction, Interaction.NOOP)
+        self.assertEqual(plan.reason, "already_in_target_state")
+
+    def test_radio_option_already_selected_is_noop(self) -> None:
+        intent = SemanticAction(
+            domain=Domain.BODY,
+            function="unlock_mode",
+            zone=None,
+            operation=Operation.SET,
+            state=None,
+            value="all_doors",
+            unit=None,
+        )
+        plan = self.planner.plan(
+            intent,
+            ui_state(
+                ControlType.RADIO,
+                function="unlock_mode",
+                zone=None,
+                current_value="all_doors",
+                options=("driver_door", "all_doors"),
+                selection_state=SelectionState.SELECTED,
+            ),
+        )
+        self.assertEqual(plan.interaction, Interaction.NOOP)
+        self.assertEqual(plan.reason, "already_in_target_state")
+
+    def test_matching_option_with_unknown_selection_proposes_tap(self) -> None:
+        intent = SemanticAction(
+            domain=Domain.BODY,
+            function="unlock_mode",
+            zone=None,
+            operation=Operation.SET,
+            state=None,
+            value="all_doors",
+            unit=None,
+        )
+        plan = self.planner.plan(
+            intent,
+            ui_state(
+                ControlType.SEGMENTED_CONTROL,
+                function="unlock_mode",
+                zone=None,
+                current_value="all_doors",
+                options=("driver_door", "all_doors"),
+                selection_state=None,
+            ),
+        )
+        self.assertEqual(plan.interaction, Interaction.TAP)
+        self.assertEqual(plan.option, "all_doors")
+
+    def test_selected_target_noops_even_when_options_are_partial(self) -> None:
+        intent = SemanticAction(
+            domain=Domain.BODY,
+            function="unlock_mode",
+            zone=None,
+            operation=Operation.SET,
+            state=None,
+            value="all_doors",
+            unit=None,
+        )
+        plan = self.planner.plan(
+            intent,
+            ui_state(
+                ControlType.SEGMENTED_CONTROL,
+                function="unlock_mode",
+                zone=None,
+                current_value="all_doors",
+                options=("driver_door",),
+                selection_state=SelectionState.SELECTED,
+            ),
+        )
+        self.assertEqual(plan.interaction, Interaction.NOOP)
+        self.assertEqual(plan.reason, "already_in_target_state")
+
+    def test_unknown_control_type_is_rejected(self) -> None:
+        with self.assertRaisesRegex(PlanningError, "control_type"):
+            self.planner.plan(
+                temperature_action(23),
+                ui_state(None),
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_roi.py

@@ -0,0 +1,90 @@
+import tempfile
+import unittest
+from pathlib import Path
+
+from PIL import Image
+
+from cockpit_agent.grounding.adapter import GroundingResult
+from cockpit_agent.perception.roi import crop_grounding_roi, grounding_to_source
+
+
+def grounding(
+    *,
+    bbox_pixel: tuple[int, int, int, int],
+    width: int,
+    height: int,
+    query: str = "semantic function",
+) -> GroundingResult:
+    x1, y1, x2, y2 = bbox_pixel
+    return GroundingResult(
+        semantic_query=query,
+        bbox_relative=(
+            x1 / width * 1000,
+            y1 / height * 1000,
+            x2 / width * 1000,
+            y2 / height * 1000,
+        ),
+        bbox_pixel=bbox_pixel,
+        center_pixel=(round((x1 + x2) / 2), round((y1 + y2) / 2)),
+        image_width=width,
+        image_height=height,
+        raw_output="model output",
+    )
+
+
+class ROITest(unittest.TestCase):
+    def test_function_bbox_is_padded_and_cropped(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            source = root / "source.png"
+            output = root / "roi.jpg"
+            Image.new("RGB", (200, 100), color="white").save(source)
+            result = crop_grounding_roi(
+                image_path=str(source),
+                function_grounding=grounding(
+                    bbox_pixel=(50, 20, 150, 80),
+                    width=200,
+                    height=100,
+                ),
+                output_path=str(output),
+                padding_ratio=0.10,
+            )
+
+            with Image.open(output) as roi_image:
+                roi_size = roi_image.size
+
+        self.assertEqual(result.bbox_pixel, (40, 14, 160, 86))
+        self.assertEqual((result.offset_x, result.offset_y), (40, 14))
+        self.assertEqual((result.width, result.height), (120, 72))
+        self.assertEqual(roi_size, (120, 72))
+
+    def test_roi_grounding_is_converted_to_source_coordinates(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            source = root / "source.png"
+            Image.new("RGB", (200, 100), color="white").save(source)
+            roi = crop_grounding_roi(
+                image_path=str(source),
+                function_grounding=grounding(
+                    bbox_pixel=(50, 20, 150, 80),
+                    width=200,
+                    height=100,
+                ),
+                output_path=str(root / "roi.jpg"),
+                padding_ratio=0.10,
+            )
+            roi_grounding = grounding(
+                bbox_pixel=(10, 6, 30, 26),
+                width=roi.width,
+                height=roi.height,
+                query="semantic subcontrol",
+            )
+            converted = grounding_to_source(roi_grounding, roi)
+
+        self.assertEqual(converted.bbox_pixel, (50, 20, 70, 40))
+        self.assertEqual(converted.center_pixel, (60, 30))
+        self.assertEqual((converted.image_width, converted.image_height), (200, 100))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_semantic_target.py

@@ -0,0 +1,141 @@
+import unittest
+
+from cockpit_agent.common.enums import Interaction, SliderOrientation, Zone
+from cockpit_agent.grounding.semantic_target import SemanticTargetBuilder
+from cockpit_agent.intent.schemas import SemanticAction
+from cockpit_agent.common.enums import Domain, Operation, State, Unit
+from cockpit_agent.planning.schemas import ActionPlan
+
+
+class SemanticTargetBuilderTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self.builder = SemanticTargetBuilder()
+
+    def test_temperature_query_uses_semantic_zone_and_direction(self) -> None:
+        query = self.builder.build_action_target(
+            ActionPlan(
+                interaction=Interaction.TAP,
+                function="temperature_decrease",
+                zone=Zone.DRIVER,
+                repeat=3,
+            )
+        )
+        self.assertEqual(query, "主驾驶温度降低控制")
+
+    def test_recirculation_query(self) -> None:
+        query = self.builder.build_action_target(
+            ActionPlan(
+                interaction=Interaction.TAP,
+                function="air_recirculation",
+                zone=Zone.ALL,
+            )
+        )
+        self.assertEqual(query, "车辆空调内循环控制")
+
+    def test_slider_query(self) -> None:
+        query = self.builder.build_action_target(
+            ActionPlan(
+                interaction=Interaction.DRAG,
+                function="tailgate_max_opening",
+                normalized_target=0.6,
+                orientation=SliderOrientation.HORIZONTAL,
+            )
+        )
+        self.assertEqual(query, "后背门最大开度滑块")
+
+    def test_segmented_option_query(self) -> None:
+        query = self.builder.build_action_target(
+            ActionPlan(
+                interaction=Interaction.TAP,
+                function="unlock_mode",
+                option="all_doors",
+            )
+        )
+        self.assertEqual(query, "所有车门解锁选项")
+
+    def test_noop_has_no_grounding_query(self) -> None:
+        with self.assertRaisesRegex(ValueError, "noop"):
+            self.builder.build_action_target(
+                ActionPlan(
+                    interaction=Interaction.NOOP,
+                    reason="already_in_target_state",
+                )
+            )
+
+    def test_function_targets_are_semantic_regions(self) -> None:
+        temperature = SemanticAction(
+            domain=Domain.CLIMATE,
+            function="temperature",
+            zone=Zone.DRIVER,
+            operation=Operation.SET,
+            state=None,
+            value=23,
+            unit=Unit.CELSIUS,
+        )
+        recirculation = SemanticAction(
+            domain=Domain.CLIMATE,
+            function="air_recirculation",
+            zone=Zone.ALL,
+            operation=Operation.SET,
+            state=State.ON,
+            value=None,
+            unit=None,
+        )
+        self.assertEqual(
+            self.builder.build_function_target(temperature),
+            (
+                "主驾驶温度调节控件整体，"
+                "包含当前温度显示以及用于升高和降低温度的控制元素"
+            ),
+        )
+        self.assertEqual(
+            self.builder.build_function_target(recirculation),
+            "车辆空调内循环控制",
+        )
+
+    def test_function_targets_describe_complete_control_structure(self) -> None:
+        intents = (
+            SemanticAction(
+                domain=Domain.BODY,
+                function="tailgate_max_opening",
+                zone=None,
+                operation=Operation.SET,
+                state=None,
+                value=80,
+                unit=Unit.PERCENT,
+            ),
+            SemanticAction(
+                domain=Domain.BODY,
+                function="unlock_mode",
+                zone=None,
+                operation=Operation.SET,
+                state=None,
+                value="all_doors",
+                unit=None,
+            ),
+        )
+        targets = tuple(
+            self.builder.build_function_target(intent) for intent in intents
+        )
+        self.assertIn("滑轨", targets[0])
+        self.assertIn("滑块", targets[0])
+        self.assertIn("控件整体", targets[1])
+        self.assertIn("所有可选择的解锁方式选项", targets[1])
+
+        forbidden = (
+            "AUTO",
+            "风扇",
+            "蓝色",
+            "橙色",
+            "左边",
+            "右边",
+            "固定像素",
+        )
+        for target in targets:
+            for term in forbidden:
+                with self.subTest(target=target, term=term):
+                    self.assertNotIn(term, target)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_ui_understanding.py

@@ -0,0 +1,344 @@
+import unittest
+
+from cockpit_agent.common.enums import (
+    Domain,
+    Operation,
+    SelectionState,
+    Unit,
+    Zone,
+)
+from cockpit_agent.grounding.adapter import GroundingResult
+from cockpit_agent.intent.schemas import SemanticAction
+from cockpit_agent.perception.schemas import UIState
+from cockpit_agent.perception.ui_understanding import ModelUIUnderstanding
+
+
+class FakeVisionGenerator:
+    def __init__(self, output: str) -> None:
+        self.output = output
+        self.prompt = ""
+        self.image_path = ""
+
+    def generate(
+        self,
+        image_path: str,
+        prompt: str,
+        max_new_tokens: int = 256,
+    ) -> str:
+        del max_new_tokens
+        self.image_path = image_path
+        self.prompt = prompt
+        return self.output
+
+
+def temperature_intent() -> SemanticAction:
+    return SemanticAction(
+        domain=Domain.CLIMATE,
+        function="temperature",
+        zone=Zone.DRIVER,
+        operation=Operation.SET,
+        state=None,
+        value=23,
+        unit=Unit.CELSIUS,
+    )
+
+
+def function_grounding(query: str = "主驾驶温度控制") -> GroundingResult:
+    return GroundingResult(
+        semantic_query=query,
+        bbox_relative=(100, 200, 400, 600),
+        bbox_pixel=(20, 20, 80, 60),
+        center_pixel=(50, 40),
+        image_width=200,
+        image_height=100,
+        raw_output='{"bbox_2d": [100, 200, 400, 600]}',
+    )
+
+
+def understand(model: FakeVisionGenerator, intent: SemanticAction):
+    return ModelUIUnderstanding(model).understand(
+        "/tmp/function_roi.jpg",
+        intent,
+        function_grounding(),
+    )
+
+
+class UIUnderstandingTest(unittest.TestCase):
+    def test_slider_schema_allows_nullable_observations(self) -> None:
+        state = UIState.from_dict(
+            {
+                "function": "tailgate_max_opening",
+                "zone": None,
+                "control_type": "slider",
+                "current_value": None,
+                "step_value": None,
+                "current_state": None,
+                "selection_state": None,
+                "min_value": None,
+                "max_value": None,
+                "options": None,
+                "track_visible": True,
+                "knob_visible": None,
+                "orientation": None,
+            }
+        )
+        self.assertIsNone(state.current_value)
+        self.assertIsNone(state.min_value)
+        self.assertIsNone(state.max_value)
+        self.assertTrue(state.track_visible)
+        self.assertIsNone(state.knob_visible)
+
+    def test_selection_state_schema_accepts_selected_and_unknown(self) -> None:
+        for selection_state in ("selected", "unknown"):
+            with self.subTest(selection_state=selection_state):
+                state = UIState.from_dict(
+                    {
+                        "function": "unlock_mode",
+                        "zone": None,
+                        "control_type": "segmented_control",
+                        "current_value": "all_doors",
+                        "step_value": None,
+                        "current_state": None,
+                        "selection_state": selection_state,
+                        "min_value": None,
+                        "max_value": None,
+                        "options": ["driver_door", "all_doors"],
+                        "track_visible": None,
+                        "knob_visible": None,
+                        "orientation": None,
+                    }
+                )
+                self.assertEqual(state.selection_state.value, selection_state)
+
+    def test_partial_options_do_not_invalidate_observed_selection(self) -> None:
+        state = UIState.from_dict(
+            {
+                "function": "unlock_mode",
+                "zone": None,
+                "control_type": "segmented_control",
+                "current_value": "all_doors",
+                "step_value": None,
+                "current_state": None,
+                "selection_state": "selected",
+                "min_value": None,
+                "max_value": None,
+                "options": ["driver_door"],
+                "track_visible": None,
+                "knob_visible": None,
+                "orientation": None,
+            }
+        )
+        self.assertEqual(state.current_value, "all_doors")
+        self.assertEqual(state.options, ("driver_door",))
+        self.assertEqual(state.selection_state, SelectionState.SELECTED)
+
+    def test_clears_fields_that_do_not_apply_to_stepper(self) -> None:
+        model = FakeVisionGenerator(
+            """{
+                "control_type": "stepper",
+                "current_value": 26.0,
+                "step_value": 1.0,
+                "current_state": null,
+                "min_value": 26.0,
+                "max_value": 26.0,
+                "options": null
+            }"""
+        )
+        state = understand(model, temperature_intent())
+        self.assertEqual(state.current_value, 26.0)
+        self.assertEqual(state.function, "temperature")
+        self.assertEqual(state.zone, Zone.DRIVER)
+        self.assertIsNone(state.min_value)
+        self.assertIsNone(state.max_value)
+
+    def test_slider_allows_unobservable_current_value(self) -> None:
+        intent = SemanticAction.from_dict(
+            {
+                "domain": "body",
+                "function": "tailgate_max_opening",
+                "zone": None,
+                "operation": "set",
+                "state": None,
+                "value": 80,
+                "unit": "percent",
+            }
+        )
+        model = FakeVisionGenerator(
+            """{
+                "control_type": "slider",
+                "current_value": null,
+                "step_value": null,
+                "current_state": null,
+                "min_value": "50%",
+                "max_value": "100%",
+                "options": null,
+                "selection_state": null,
+                "track_visible": true,
+                "knob_visible": true
+                ,"orientation": "horizontal"
+            }"""
+        )
+        state = understand(model, intent)
+        self.assertIsNone(state.current_value)
+        self.assertEqual(state.min_value, 50)
+        self.assertEqual(state.max_value, 100)
+        self.assertTrue(state.track_visible)
+        self.assertTrue(state.knob_visible)
+        self.assertEqual(state.orientation.value, "horizontal")
+        self.assertIsNone(state.selection_state)
+
+    def test_normalizes_unlock_option_labels(self) -> None:
+        intent = SemanticAction.from_dict(
+            {
+                "domain": "body",
+                "function": "unlock_mode",
+                "zone": None,
+                "operation": "set",
+                "state": None,
+                "value": "all_doors",
+                "unit": None,
+            }
+        )
+        model = FakeVisionGenerator(
+            """{
+                "control_type": "segmented_control",
+                "current_value": "所有车门",
+                "step_value": null,
+                "current_state": "active",
+                "min_value": "主驾车门",
+                "max_value": "所有车门",
+                "options": ["主驾车门", "所有车门"]
+            }"""
+        )
+        state = understand(model, intent)
+        self.assertEqual(state.current_value, "all_doors")
+        self.assertEqual(state.options, ("driver_door", "all_doors"))
+        self.assertIsNone(state.current_state)
+        self.assertEqual(state.selection_state, SelectionState.SELECTED)
+
+    def test_normalizes_common_selection_state_language(self) -> None:
+        intent = SemanticAction.from_dict(
+            {
+                "domain": "body",
+                "function": "unlock_mode",
+                "zone": None,
+                "operation": "set",
+                "state": None,
+                "value": "all_doors",
+                "unit": None,
+            }
+        )
+        cases = (
+            ("active", SelectionState.SELECTED),
+            ("inactive", SelectionState.UNSELECTED),
+            ("selected", SelectionState.SELECTED),
+            ("unknown", SelectionState.UNKNOWN),
+        )
+        for raw_state, expected in cases:
+            with self.subTest(raw_state=raw_state):
+                model = FakeVisionGenerator(
+                    """{
+                        "control_type": "segmented_control",
+                        "current_value": "所有车门",
+                        "step_value": null,
+                        "current_state": null,
+                        "selection_state": "%s",
+                        "min_value": null,
+                        "max_value": null,
+                        "options": ["主驾车门", "所有车门"],
+                        "track_visible": null,
+                        "knob_visible": null
+                    }""" % raw_state
+                )
+                state = understand(model, intent)
+                self.assertEqual(state.selection_state, expected)
+
+    def test_model_control_type_is_not_repaired_for_unlock_mode(self) -> None:
+        intent = SemanticAction.from_dict(
+            {
+                "domain": "body",
+                "function": "unlock_mode",
+                "zone": None,
+                "operation": "set",
+                "state": None,
+                "value": "all_doors",
+                "unit": None,
+            }
+        )
+        model = FakeVisionGenerator(
+            """{
+                "control_type": "button",
+                "current_value": "所有车门",
+                "step_value": null,
+                "current_state": "active",
+                "min_value": null,
+                "max_value": null,
+                "options": null
+            }"""
+        )
+        state = understand(model, intent)
+        self.assertEqual(state.control_type.value, "button")
+        self.assertEqual(state.current_value, "all_doors")
+        self.assertEqual(state.selection_state, SelectionState.SELECTED)
+        self.assertIsNone(state.current_state)
+
+    def test_preserves_model_control_type_without_function_specific_repair(self) -> None:
+        model = FakeVisionGenerator(
+            """{
+                "control_type": "slider",
+                "current_value": null,
+                "step_value": null,
+                "current_state": null,
+                "min_value": 0,
+                "max_value": 100,
+                "options": null
+            }"""
+        )
+        state = understand(model, temperature_intent())
+        self.assertEqual(state.control_type.value, "slider")
+
+    def test_all_observation_fields_may_be_null(self) -> None:
+        model = FakeVisionGenerator(
+            """{
+                "control_type": null,
+                "current_value": null,
+                "step_value": null,
+                "current_state": null,
+                "min_value": null,
+                "max_value": null,
+                "options": null
+            }"""
+        )
+        state = understand(model, temperature_intent())
+        self.assertIsNone(state.control_type)
+        self.assertIsNone(state.current_value)
+        self.assertIsNone(state.current_state)
+        self.assertIsNone(state.selection_state)
+        self.assertIsNone(state.track_visible)
+        self.assertIsNone(state.knob_visible)
+
+    def test_prompt_describes_roi_and_forbids_guessing(self) -> None:
+        model = FakeVisionGenerator(
+            """{
+                "control_type": "stepper",
+                "current_value": 26,
+                "step_value": 1,
+                "current_state": null,
+                "min_value": null,
+                "max_value": null,
+                "options": null
+            }"""
+        )
+        understand(model, temperature_intent())
+        self.assertEqual(model.image_path, "/tmp/function_roi.jpg")
+        self.assertIn("ROI", model.prompt)
+        self.assertIn("绝不能", model.prompt)
+        self.assertIn("端点旁的数值", model.prompt)
+        self.assertIn("selection_state", model.prompt)
+        self.assertNotIn("available", model.prompt)
+        self.assertNotIn("空调控制栏", model.prompt)
+        self.assertNotIn("页面中央", model.prompt)
+
+
+if __name__ == "__main__":
+    unittest.main()