Files
HomeAssistantVS/custom_components/maintenance_supporter/websocket/tasks_crud.py
T

1053 lines
47 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Task create / update / delete / duplicate WS handlers."""
from __future__ import annotations
import re
from copy import deepcopy
from typing import Any
from uuid import uuid4
import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
from homeassistant.helpers import issue_registry as ir
from homeassistant.util import dt as dt_util
from ..const import (
BATTERY_FLEET_TASK_FLAG,
CONF_OBJECT,
CONF_TASKS,
DEFAULT_WARNING_DAYS,
DOMAIN,
FLAT_SCHEDULE_TYPES,
MAX_ASSIGNEE_POOL,
MAX_CHECKLIST_ITEM_LENGTH,
MAX_CHECKLIST_ITEMS,
MAX_DATE_LENGTH,
MAX_ENTITY_SLUG_LENGTH,
MAX_ICON_LENGTH,
MAX_ID_LENGTH,
MAX_LABEL_LENGTH,
MAX_LABELS,
MAX_META_LENGTH,
MAX_MIRROR_TODO_LISTS,
MAX_NAME_LENGTH,
MAX_NFC_TAG_LENGTH,
MAX_READING_UNIT_LENGTH,
MAX_TEXT_LENGTH,
MAX_TYPE_LENGTH,
MAX_URL_LENGTH,
MIRROR_TODO_ENTITY_PATTERN,
NOTIFICATION_MANAGER_KEY,
HistoryEntryType,
)
from ..helpers.aggregate import get_store
from ..helpers.dates import INTERVAL_UNITS
from ..helpers.entry_tasks import write_task
from ..helpers.permissions import require_write
from ..helpers.sanitize import strip_task_runtime_state
from ..helpers.schedule import (
FLAT_RECURRENCE_KEYS,
KIND_INTERVAL,
Schedule,
)
from ..helpers.task_fields import (
EARLIEST_COMPLETION_RANGE,
INTERVAL_ANCHORS,
INTERVAL_DAYS_RANGE,
REQUIRABLE_COMPLETION_FIELDS,
ROTATION_STRATEGY_VALUES,
TASK_PRIORITIES,
WARNING_DAYS_RANGE,
)
from . import (
ID_FIELD,
_load_object_entry,
_load_object_task,
_parse_iso_date,
cleanup_group_refs,
)
from .tasks_persist import async_persist_task
from .tasks_validation import (
_check_nfc_tag_duplicate,
_is_safe_url,
_mirror_todo_error,
_validate_trigger_config,
)
_ENTITY_SLUG_RE = re.compile(r"[a-z0-9_]+")
def _validate_entity_slug(connection: websocket_api.ActiveConnection, msg: dict[str, Any]) -> bool:
"""False (after sending ``invalid_entity_slug``) when ``msg["entity_slug"]``
is present, not None and malformed — shared by create and update."""
slug = msg.get("entity_slug")
if slug is None or _ENTITY_SLUG_RE.fullmatch(slug):
return True
connection.send_error(
msg["id"],
"invalid_entity_slug",
"entity_slug must match [a-z0-9_]+ (lowercase, digits, underscores only)",
)
return False
def _normalize_nfc_tag(
hass: HomeAssistant,
msg: dict[str, Any],
warnings: list[str],
*,
exclude_task_id: str | None = None,
) -> str | None:
"""Normalise ``msg["nfc_tag_id"]`` in place (""/whitespace -> None) and
append the duplicate-tag warning when another task already carries it.
Returns the normalised value — shared by create and update."""
nfc_val = (msg.get("nfc_tag_id") or "").strip() or None
msg["nfc_tag_id"] = nfc_val
if nfc_val:
nfc_warn = _check_nfc_tag_duplicate(hass, nfc_val, exclude_task_id=exclude_task_id)
if nfc_warn:
warnings.append(nfc_warn)
return nfc_val
# ws_update_task: wire key -> storage key. Almost all are identity; the one
# rename is deliberate and load-bearing: the WS message envelope reserves
# "type" for command routing ({"type": "maintenance_supporter/task/update"}),
# so a task's own type must travel as "task_type" on the wire and is stored
# as "type". Do NOT "simplify" this to "type": "type" — it would collide with
# the routing key. The panel↔config-flow parity test (test_parity_task_fields)
# encodes the same task_type->type alias. Module-level so the CONTRACT-FIXTURE
# tripwire (tests/test_task_contract_fixture.py) can enumerate it: a field
# added here without extending the round-trip fixture fails that test.
TASK_UPDATE_FIELD_MAP = {
"name": "name",
"task_type": "type",
"enabled": "enabled",
"schedule_type": "schedule_type",
"interval_days": "interval_days",
"interval_unit": "interval_unit",
"due_date": "due_date",
"interval_anchor": "interval_anchor",
"warning_days": "warning_days",
"earliest_completion_days": "earliest_completion_days",
"last_performed": "last_performed",
"trigger_config": "trigger_config",
"notes": "notes",
"documentation_url": "documentation_url",
"responsible_user_id": "responsible_user_id",
"assignee_pool": "assignee_pool",
"required_completion_fields": "required_completion_fields",
"rotation_strategy": "rotation_strategy",
"entity_slug": "entity_slug",
"custom_icon": "custom_icon",
"nfc_tag_id": "nfc_tag_id",
"require_tag_scan": "require_tag_scan",
"allow_skip": "allow_skip",
"notify_enabled": "notify_enabled",
"reading_unit": "reading_unit",
"readings": "readings",
"consumes_parts": "consumes_parts",
"phases": "phases",
"phase_sequence": "phase_sequence",
"priority": "priority",
"checklist": "checklist",
"labels": "labels",
# D#183: external to-do lists the task is mirrored into while due.
"mirror_todo_entities": "mirror_todo_entities",
"schedule_time": "schedule_time",
# v1.3.0
"on_complete_action": "on_complete_action",
"quick_complete_defaults": "quick_complete_defaults",
}
def _apply_phase_fields(
hass: HomeAssistant,
entry: ConfigEntry,
task_data: dict[str, Any],
raw_defs: object,
raw_sequence: object,
) -> None:
"""Sanitize + store the phase config (#139) — shared by create and update.
Empty/invalid defs or sequence remove BOTH fields (a task either has a
usable cycle or none). Per-phase consumes_parts get the same part-registry
validation the task-level field gets.
"""
from ..const import CONF_PARTS
from ..helpers.parts import sanitize_consumes_parts
from ..helpers.phases import sanitize_phase_defs, sanitize_phase_sequence
from . import foreign_part_resolver
defs = sanitize_phase_defs(raw_defs)
for definition in defs.values():
if "consumes_parts" in definition:
links = sanitize_consumes_parts(
definition["consumes_parts"],
set(entry.data.get(CONF_PARTS) or {}),
foreign_part_ids=foreign_part_resolver(hass),
)
if links:
definition["consumes_parts"] = links
else:
del definition["consumes_parts"]
sequence = sanitize_phase_sequence(raw_sequence, defs)
if defs and sequence:
task_data["phases"] = defs
task_data["phase_sequence"] = sequence
else:
task_data.pop("phases", None)
task_data.pop("phase_sequence", None)
# Hoisted as a module constant so the schema/field-map parity tripwire
# (tests/test_task_schema_parity.py, drift audit 2026-08) can introspect it.
_TASK_CREATE_SCHEMA: dict[Any, Any] = {
vol.Required("type"): "maintenance_supporter/task/create",
vol.Required("entry_id"): vol.All(str, vol.Length(max=MAX_ID_LENGTH)),
vol.Required("name"): vol.All(str, vol.Length(min=1, max=MAX_NAME_LENGTH)),
vol.Optional("task_type", default="custom"): vol.All(str, vol.Length(max=MAX_TYPE_LENGTH)),
vol.Optional("schedule_type", default="time_based"): vol.All(str, vol.Length(max=MAX_TYPE_LENGTH)),
vol.Optional("interval_days"): vol.Any(
vol.All(int, vol.Range(min=INTERVAL_DAYS_RANGE[0], max=INTERVAL_DAYS_RANGE[1])), None
),
vol.Optional("interval_unit", default="days"): vol.In(INTERVAL_UNITS),
vol.Optional("due_date"): vol.Any(vol.All(str, vol.Length(max=MAX_DATE_LENGTH)), None),
vol.Optional("interval_anchor", default="completion"): vol.In(INTERVAL_ANCHORS),
# Nested recurrence (calendar kinds: weekdays / nth_weekday / day_of_month).
# Validated/canonicalized in the handler via Schedule.from_dict.
vol.Optional("schedule"): vol.Any(dict, None),
vol.Optional("warning_days", default=DEFAULT_WARNING_DAYS): vol.All(
int, vol.Range(min=WARNING_DAYS_RANGE[0], max=WARNING_DAYS_RANGE[1])
),
vol.Optional("earliest_completion_days"): vol.Any(
vol.All(int, vol.Range(min=EARLIEST_COMPLETION_RANGE[0], max=EARLIEST_COMPLETION_RANGE[1])), None
),
vol.Optional("last_performed"): vol.Any(vol.All(str, vol.Length(max=MAX_DATE_LENGTH)), None),
vol.Optional("trigger_config"): vol.Any(dict, None),
vol.Optional("notes"): vol.Any(vol.All(str, vol.Length(max=MAX_TEXT_LENGTH)), None),
vol.Optional("documentation_url"): vol.Any(vol.All(str, vol.Length(max=MAX_URL_LENGTH)), None),
vol.Optional("responsible_user_id"): vol.Any(vol.All(str, vol.Length(max=MAX_META_LENGTH)), None),
vol.Optional("assignee_pool"): vol.Any(
vol.All([vol.All(str, vol.Length(max=MAX_META_LENGTH))], vol.Length(max=MAX_ASSIGNEE_POOL)), None
),
vol.Optional("required_completion_fields"): vol.Any([vol.In(REQUIRABLE_COMPLETION_FIELDS)], None),
vol.Optional("rotation_strategy"): vol.Any(vol.In(ROTATION_STRATEGY_VALUES), None),
vol.Optional("entity_slug"): vol.Any(vol.All(str, vol.Length(max=MAX_ENTITY_SLUG_LENGTH)), None),
vol.Optional("custom_icon"): vol.Any(vol.All(str, vol.Length(max=MAX_ICON_LENGTH)), None),
vol.Optional("nfc_tag_id"): vol.Any(vol.All(str, vol.Length(max=MAX_NFC_TAG_LENGTH)), None),
# Proof of presence (#139 family): completion only via NFC/QR scan.
vol.Optional("require_tag_scan"): vol.Any(bool, None),
# #150: per-task skip lock — false hides Skip in the UIs and the
# coordinator refuses (WS + voice), so automations cannot skip either.
vol.Optional("allow_skip"): vol.Any(bool, None),
# #173: per-task notification mute — false = no reminders for this
# task (status changes, repeats, lead-time, bundles); the dashboard
# and entities still show it.
vol.Optional("notify_enabled"): vol.Any(bool, None),
# v2.20 (#83): unit for `reading`-type tasks ("kWh", "m³", ...).
vol.Optional("reading_unit"): vol.Any(vol.All(str, vol.Length(max=MAX_READING_UNIT_LENGTH)), None),
# #161 phase 2: reading slots [{id?, name, unit?}] — shape-validated
# by helpers/reading_slots.sanitize_reading_slots at both write paths.
vol.Optional("readings"): vol.Any(list, None),
# Spare parts consumed on completion: [{part_id, quantity}].
vol.Optional("consumes_parts"): vol.Any(list, None),
# Task phases (#139): cyclic content rotation on one cadence.
# Shape-validated in helpers/phases.py at both write paths.
vol.Optional("phases"): vol.Any(dict, None),
vol.Optional("phase_sequence"): vol.Any(list, None),
vol.Optional("priority"): vol.In(TASK_PRIORITIES),
vol.Optional("checklist"): vol.Any(
vol.All([vol.All(str, vol.Length(max=MAX_CHECKLIST_ITEM_LENGTH))], vol.Length(max=MAX_CHECKLIST_ITEMS)), None
),
vol.Optional("labels"): vol.Any(
vol.All([vol.All(str, vol.Length(max=MAX_LABEL_LENGTH))], vol.Length(max=MAX_LABELS)), None
),
# D#183: todo.* entity ids the due task is mirrored into ([] / None = off).
vol.Optional("mirror_todo_entities"): vol.Any(
vol.All([vol.All(str, vol.Match(MIRROR_TODO_ENTITY_PATTERN))], vol.Length(max=MAX_MIRROR_TODO_LISTS)), None
),
# HH:MM strict (00–23 : 00–59). None clears the time → midnight semantic.
vol.Optional("schedule_time"): vol.Any(
vol.All(str, vol.Match(r"^([01]\d|2[0-3]):[0-5]\d$")),
None,
),
# v1.3.0: per-task on_complete_action + quick_complete_defaults.
# Both kept loose at the schema level (vol.Any(dict, None)); strict
# field-by-field validation lives in helpers/sanitize.py so the
# config-flow path (which doesn't go through this schema) gets
# identical validation behaviour.
vol.Optional("on_complete_action"): vol.Any(dict, None),
vol.Optional("quick_complete_defaults"): vol.Any(dict, None),
vol.Optional("enabled", default=True): bool,
vol.Optional("dry_run", default=False): bool,
}
@websocket_api.websocket_command(_TASK_CREATE_SCHEMA)
@require_write
@websocket_api.async_response
async def ws_create_task(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
"""Add a new task to an existing maintenance object."""
entry = _load_object_entry(hass, connection, msg)
if entry is None:
return
task_id = uuid4().hex
name = msg["name"].strip()
if not name:
connection.send_error(msg["id"], "invalid_input", "Name must not be empty")
return
task_data: dict[str, Any] = {
"id": task_id,
"object_id": entry.data.get(CONF_OBJECT, {}).get("id", ""),
"name": name,
"type": msg.get("task_type", "custom"),
"enabled": msg.get("enabled", True),
"warning_days": msg.get("warning_days", DEFAULT_WARNING_DAYS),
# Anchor for next_due fallback when last_performed is None (issue #30).
# Use HA's timezone-aware "today" to match next_due computation.
"created_at": dt_util.now().date().isoformat(),
}
# Dynamic state (last_performed, history) for Store initialization
initial_last_performed: str | None = None
initial_history: list[dict[str, Any]] = []
# Recurrence: an explicit nested `schedule` (calendar kinds) takes
# precedence; otherwise build from the flat v2.6.x fields.
if msg.get("schedule"):
incoming = Schedule.from_dict(msg["schedule"]).to_dict()
task_data["schedule"] = incoming
# A BARE interval schedule (kind=interval, no `every`) is only the
# carrier for the season/ends extras the panel always sends for a
# time-based task — the real interval still rides the flat
# interval_days / interval_unit / interval_anchor fields. Carry them
# over so normalize_task_storage merges them onto the schedule; without
# this the interval was dropped entirely (#88 regression).
if incoming.get("kind") == KIND_INTERVAL and incoming.get("every") is None:
if msg.get("interval_days") is not None:
task_data["interval_days"] = msg["interval_days"]
if msg.get("interval_unit", "days") != "days":
task_data["interval_unit"] = msg["interval_unit"]
if msg.get("interval_anchor", "completion") != "completion":
task_data["interval_anchor"] = msg["interval_anchor"]
else:
task_data["schedule_type"] = msg.get("schedule_type", "time_based")
if msg.get("interval_days") is not None:
task_data["interval_days"] = msg["interval_days"]
if msg.get("interval_unit", "days") != "days":
task_data["interval_unit"] = msg["interval_unit"]
if msg.get("due_date") is not None:
task_data["due_date"] = msg["due_date"]
if msg.get("interval_anchor", "completion") != "completion":
task_data["interval_anchor"] = msg["interval_anchor"]
if msg.get("last_performed") is not None:
lp_date = _parse_iso_date(connection, msg["id"], msg["last_performed"], field="last_performed", code="invalid_format")
if lp_date is None:
return
initial_last_performed = msg["last_performed"]
# Add initial history entry so times_performed reflects the value.
# Use HA-TZ-aware midnight to keep interval_analyzer consistent.
from datetime import datetime, time
lp_dt = datetime.combine(lp_date, time.min, tzinfo=dt_util.DEFAULT_TIME_ZONE)
initial_history.append(
{
"timestamp": lp_dt.isoformat(),
"type": HistoryEntryType.COMPLETED,
"notes": "Initial value set during task creation",
}
)
trigger_config = msg.get("trigger_config")
tc_errors: list[str] = []
tc_warnings: list[str] = []
if trigger_config is not None:
tc_errors, tc_warnings = _validate_trigger_config(hass, trigger_config)
if tc_errors:
connection.send_error(
msg["id"],
"invalid_trigger_config",
"; ".join(tc_errors),
)
return
task_data["trigger_config"] = trigger_config
if msg.get("notes") is not None:
task_data["notes"] = msg["notes"]
if msg.get("documentation_url") is not None:
if not _is_safe_url(msg["documentation_url"]):
connection.send_error(msg["id"], "invalid_url", "Only http/https URLs are allowed")
return
task_data["documentation_url"] = msg["documentation_url"]
if msg.get("responsible_user_id") is not None:
task_data["responsible_user_id"] = msg["responsible_user_id"]
if msg.get("earliest_completion_days") is not None:
task_data["earliest_completion_days"] = msg["earliest_completion_days"]
if msg.get("assignee_pool"):
from ..helpers.sanitize import sanitize_assignee_pool
task_data["assignee_pool"] = sanitize_assignee_pool(msg["assignee_pool"])
if msg.get("rotation_strategy"):
task_data["rotation_strategy"] = msg["rotation_strategy"]
if msg.get("required_completion_fields"):
from ..helpers.completion_requirements import sanitize_required_completion_fields
task_data["required_completion_fields"] = sanitize_required_completion_fields(
msg["required_completion_fields"]
)
from ..helpers.sanitize import seed_rotation_assignee
seed_rotation_assignee(task_data)
if not _validate_entity_slug(connection, msg):
return
if msg.get("entity_slug") is not None:
task_data["entity_slug"] = msg["entity_slug"]
if msg.get("custom_icon") is not None:
task_data["custom_icon"] = msg["custom_icon"]
if msg.get("priority") is not None:
task_data["priority"] = msg["priority"]
if msg.get("nfc_tag_id") is not None:
task_data["nfc_tag_id"] = _normalize_nfc_tag(hass, msg, tc_warnings)
if msg.get("require_tag_scan") is not None:
task_data["require_tag_scan"] = bool(msg["require_tag_scan"])
# #150: stored only when False — absence means skipping is allowed.
if msg.get("allow_skip") is False:
task_data["allow_skip"] = False
# #173: same shape — stored only when False (absence = notifications on).
if msg.get("notify_enabled") is False:
task_data["notify_enabled"] = False
# v2.20 (#83): unit for `reading`-type tasks.
if msg.get("reading_unit") is not None:
task_data["reading_unit"] = (msg["reading_unit"] or "").strip() or None
if msg.get("readings"):
from ..helpers.reading_slots import sanitize_reading_slots
slots = sanitize_reading_slots(msg["readings"])
if slots:
task_data["readings"] = slots
if msg.get("consumes_parts") is not None:
from ..const import CONF_PARTS
from ..helpers.parts import sanitize_consumes_parts
from . import foreign_part_resolver
links = sanitize_consumes_parts(
msg["consumes_parts"],
set(entry.data.get(CONF_PARTS) or {}),
foreign_part_ids=foreign_part_resolver(hass),
)
if links:
task_data["consumes_parts"] = links
if msg.get("phases") is not None or msg.get("phase_sequence") is not None:
_apply_phase_fields(hass, entry, task_data, msg.get("phases"), msg.get("phase_sequence"))
if msg.get("checklist"):
task_data["checklist"] = msg["checklist"]
if msg.get("labels"):
from ..helpers.sanitize import sanitize_labels
task_data["labels"] = sanitize_labels(msg["labels"])
# D#183: mirror targets — shape-sanitized, and our own to-do platform is
# refused (it cannot take rows, and mirroring into ourselves is circular).
if msg.get("mirror_todo_entities"):
from ..helpers.sanitize import sanitize_mirror_todo_entities
mirrors = sanitize_mirror_todo_entities(msg["mirror_todo_entities"])
mirror_err = _mirror_todo_error(hass, mirrors)
if mirror_err:
connection.send_error(msg["id"], "invalid_mirror_todo", mirror_err)
return
if mirrors:
task_data["mirror_todo_entities"] = mirrors
if msg.get("schedule_time"):
task_data["schedule_time"] = msg["schedule_time"]
# v1.3.0: optional completion-action + quick-defaults. Strict shape
# validated by sanitize.cap_action_field / cap_quick_complete_defaults_field
# below — accepted loosely here, dropped if malformed.
if msg.get("on_complete_action"):
task_data["on_complete_action"] = msg["on_complete_action"]
if msg.get("quick_complete_defaults"):
task_data["quick_complete_defaults"] = msg["quick_complete_defaults"]
from ..helpers.sanitize import cap_action_field, cap_quick_complete_defaults_field
cap_action_field(task_data)
cap_quick_complete_defaults_field(task_data)
# Dry-run mode: validate only, do not persist
if msg.get("dry_run"):
result: dict[str, Any] = {"valid": True, "task_id": None}
if tc_warnings:
result["warnings"] = tc_warnings
connection.send_result(msg["id"], result)
return
try:
await async_persist_task(
hass,
entry,
task_data,
last_performed=initial_last_performed,
history=initial_history,
)
except ValueError as err:
connection.send_error(msg["id"], "limit_reached", str(err))
return
result = {"task_id": task_id}
if tc_warnings:
result["warnings"] = tc_warnings
connection.send_result(msg["id"], result)
# Hoisted as a module constant so the schema/field-map parity tripwire
# (tests/test_task_schema_parity.py, drift audit 2026-08) can introspect it.
_TASK_UPDATE_SCHEMA: dict[Any, Any] = {
vol.Required("type"): "maintenance_supporter/task/update",
vol.Required("entry_id"): vol.All(str, vol.Length(max=MAX_ID_LENGTH)),
vol.Required("task_id"): vol.All(str, vol.Length(max=MAX_ID_LENGTH)),
vol.Optional("name"): vol.All(str, vol.Length(min=1, max=MAX_NAME_LENGTH)),
vol.Optional("task_type"): vol.All(str, vol.Length(max=MAX_TYPE_LENGTH)),
vol.Optional("enabled"): bool,
vol.Optional("schedule_type"): vol.All(str, vol.Length(max=MAX_TYPE_LENGTH)),
vol.Optional("interval_days"): vol.Any(
vol.All(int, vol.Range(min=INTERVAL_DAYS_RANGE[0], max=INTERVAL_DAYS_RANGE[1])), None
),
vol.Optional("interval_unit"): vol.In(INTERVAL_UNITS),
vol.Optional("due_date"): vol.Any(vol.All(str, vol.Length(max=MAX_DATE_LENGTH)), None),
vol.Optional("interval_anchor"): vol.In(INTERVAL_ANCHORS),
# Nested recurrence (calendar kinds); see create schema.
vol.Optional("schedule"): vol.Any(dict, None),
vol.Optional("warning_days"): vol.All(int, vol.Range(min=WARNING_DAYS_RANGE[0], max=WARNING_DAYS_RANGE[1])),
vol.Optional("earliest_completion_days"): vol.Any(
vol.All(int, vol.Range(min=EARLIEST_COMPLETION_RANGE[0], max=EARLIEST_COMPLETION_RANGE[1])), None
),
vol.Optional("last_performed"): vol.Any(vol.All(str, vol.Length(max=MAX_DATE_LENGTH)), None),
vol.Optional("trigger_config"): vol.Any(dict, None),
vol.Optional("notes"): vol.Any(vol.All(str, vol.Length(max=MAX_TEXT_LENGTH)), None),
vol.Optional("documentation_url"): vol.Any(vol.All(str, vol.Length(max=MAX_URL_LENGTH)), None),
vol.Optional("responsible_user_id"): vol.Any(vol.All(str, vol.Length(max=MAX_META_LENGTH)), None),
vol.Optional("assignee_pool"): vol.Any(
vol.All([vol.All(str, vol.Length(max=MAX_META_LENGTH))], vol.Length(max=MAX_ASSIGNEE_POOL)), None
),
vol.Optional("required_completion_fields"): vol.Any([vol.In(REQUIRABLE_COMPLETION_FIELDS)], None),
vol.Optional("rotation_strategy"): vol.Any(vol.In(ROTATION_STRATEGY_VALUES), None),
vol.Optional("entity_slug"): vol.Any(vol.All(str, vol.Length(max=MAX_ENTITY_SLUG_LENGTH)), None),
vol.Optional("custom_icon"): vol.Any(vol.All(str, vol.Length(max=MAX_ICON_LENGTH)), None),
vol.Optional("nfc_tag_id"): vol.Any(vol.All(str, vol.Length(max=MAX_NFC_TAG_LENGTH)), None),
# Proof of presence (#139 family): completion only via NFC/QR scan.
vol.Optional("require_tag_scan"): vol.Any(bool, None),
# #150: per-task skip lock — false hides Skip in the UIs and the
# coordinator refuses (WS + voice), so automations cannot skip either.
vol.Optional("allow_skip"): vol.Any(bool, None),
# #173: per-task notification mute — false = no reminders for this
# task (status changes, repeats, lead-time, bundles); the dashboard
# and entities still show it.
vol.Optional("notify_enabled"): vol.Any(bool, None),
# v2.20 (#83): unit for `reading`-type tasks ("kWh", "m³", ...).
vol.Optional("reading_unit"): vol.Any(vol.All(str, vol.Length(max=MAX_READING_UNIT_LENGTH)), None),
# #161 phase 2: reading slots [{id?, name, unit?}] — shape-validated
# by helpers/reading_slots.sanitize_reading_slots at both write paths.
vol.Optional("readings"): vol.Any(list, None),
# Spare parts consumed on completion: [{part_id, quantity}].
vol.Optional("consumes_parts"): vol.Any(list, None),
# Task phases (#139): cyclic content rotation on one cadence.
# Shape-validated in helpers/phases.py at both write paths.
vol.Optional("phases"): vol.Any(dict, None),
vol.Optional("phase_sequence"): vol.Any(list, None),
vol.Optional("priority"): vol.In(TASK_PRIORITIES),
vol.Optional("checklist"): vol.Any(
vol.All([vol.All(str, vol.Length(max=MAX_CHECKLIST_ITEM_LENGTH))], vol.Length(max=MAX_CHECKLIST_ITEMS)), None
),
vol.Optional("labels"): vol.Any(
vol.All([vol.All(str, vol.Length(max=MAX_LABEL_LENGTH))], vol.Length(max=MAX_LABELS)), None
),
# D#183: see create schema.
vol.Optional("mirror_todo_entities"): vol.Any(
vol.All([vol.All(str, vol.Match(MIRROR_TODO_ENTITY_PATTERN))], vol.Length(max=MAX_MIRROR_TODO_LISTS)), None
),
vol.Optional("schedule_time"): vol.Any(
vol.All(str, vol.Match(r"^([01]\d|2[0-3]):[0-5]\d$")),
None,
),
# v1.3.0: same loose schema as create. Sanitize layer enforces shape.
vol.Optional("on_complete_action"): vol.Any(dict, None),
vol.Optional("quick_complete_defaults"): vol.Any(dict, None),
}
@websocket_api.websocket_command(_TASK_UPDATE_SCHEMA)
@require_write
@websocket_api.async_response
async def ws_update_task(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
"""Update an existing task."""
ctx = _load_object_task(hass, connection, msg)
if ctx is None:
return
entry, _rd, stored_task = ctx
task_id = msg["task_id"]
task = dict(stored_task)
# Strip and validate name if provided
if "name" in msg:
msg["name"] = msg["name"].strip()
if not msg["name"]:
connection.send_error(msg["id"], "invalid_input", "Name must not be empty")
return
# Battery Fleet guard (issue #106): the single fleet task without its
# threshold trigger is definitionally broken — it never fires and never
# auto-completes. No fleet UI offers removing the trigger, so an incoming
# null can only be a client (possibly stale/cached bundle) failing to
# round-trip a trigger it did not hydrate. Ignore the null; every other
# field in the message still applies.
if task.get(BATTERY_FLEET_TASK_FLAG) and "trigger_config" in msg and msg["trigger_config"] is None:
del msg["trigger_config"]
# Validate trigger_config if provided
tc_warnings: list[str] = []
if "trigger_config" in msg and msg["trigger_config"] is not None:
tc_errors, tc_warnings = _validate_trigger_config(hass, msg["trigger_config"])
if tc_errors:
connection.send_error(
msg["id"],
"invalid_trigger_config",
"; ".join(tc_errors),
)
return
# Validate entity_slug if provided
if not _validate_entity_slug(connection, msg):
return
# Normalise empty NFC tag to None and check uniqueness
if "nfc_tag_id" in msg:
_normalize_nfc_tag(hass, msg, tc_warnings, exclude_task_id=task_id)
# Validate last_performed date format if provided
if (
msg.get("last_performed") is not None
and _parse_iso_date(connection, msg["id"], msg["last_performed"], field="last_performed", code="invalid_format")
is None
):
return
# Validate documentation_url if provided
if "documentation_url" in msg and not _is_safe_url(msg["documentation_url"]):
connection.send_error(msg["id"], "invalid_url", "Only http/https URLs are allowed")
return
# D#183: mirror targets — same sanitize + own-platform refusal as create;
# an empty list clears the field (the raw-field-map lesson).
if "mirror_todo_entities" in msg:
from ..helpers.sanitize import sanitize_mirror_todo_entities
msg["mirror_todo_entities"] = sanitize_mirror_todo_entities(msg["mirror_todo_entities"])
mirror_err = _mirror_todo_error(hass, msg["mirror_todo_entities"])
if mirror_err:
connection.send_error(msg["id"], "invalid_mirror_todo", mirror_err)
return
for msg_key, data_key in TASK_UPDATE_FIELD_MAP.items():
if msg_key in msg:
task[data_key] = msg[msg_key]
# last_performed is a DYNAMIC field: the Store is its system of record
# and merge_task_data overlays the Store's value whenever the key exists
# there (every task that was ever completed / reset / created with a
# date). Writing the edit into entry.data alone was masked for exactly
# those tasks — the dialog's "last performed" edit never took effect.
# Write the Store too; the static copy above stays as the legacy /
# no-Store fallback (bug audit 2026-09-12). A MOVED anchor starts a new
# cycle, so a postpone (due_override) of the old one goes with it; the
# dialog re-sends an unchanged date on every save, which must not.
if "last_performed" in msg:
store_lp = get_store(hass, entry.entry_id)
if store_lp is not None:
new_anchor = msg["last_performed"] or None
store_lp.set_anchor(task_id, new_anchor, clear_modifiers=new_anchor != store_lp.get_last_performed(task_id))
await store_lp.async_save()
# #150: allow_skip is stored only when False (absence = allowed) — the
# verbatim copy above would persist True/None literals.
if "allow_skip" in msg:
if msg["allow_skip"] is False:
task["allow_skip"] = False
else:
task.pop("allow_skip", None)
# #173: notify_enabled follows the same only-when-False rule.
if "notify_enabled" in msg:
if msg["notify_enabled"] is False:
task["notify_enabled"] = False
else:
task.pop("notify_enabled", None)
# D#183: an empty mirror list means "off" — drop the key, don't store [].
if "mirror_todo_entities" in msg and not msg["mirror_todo_entities"]:
task.pop("mirror_todo_entities", None)
# The loop above copies values verbatim, which is wrong for part links:
# `task/create` validates them and `task/update` did not, so an edit could
# persist a link to a part — or, since #111, to an OBJECT — that does not
# exist. Nothing complained: the consume path simply skipped it. Sanitize
# here so both write paths agree.
if "consumes_parts" in msg:
from ..const import CONF_PARTS
from ..helpers.parts import sanitize_consumes_parts
from . import foreign_part_resolver
task["consumes_parts"] = sanitize_consumes_parts(
msg["consumes_parts"],
set(entry.data.get(CONF_PARTS) or {}),
foreign_part_ids=foreign_part_resolver(hass),
)
# #161 phase 2: same raw-field-map lesson — validate the slot list here so
# both write paths persist the identical shape; an empty list clears.
if "readings" in msg:
from ..helpers.reading_slots import sanitize_reading_slots
slots = sanitize_reading_slots(msg["readings"])
if slots:
task["readings"] = slots
else:
task.pop("readings", None)
# Phases (#139): the raw field-map copy above skipped validation (the
# raw-field-map lesson) — sanitize here, and clamp the Store cursor
# against the possibly shortened sequence (the Store wins on restore, so
# an out-of-range cursor would otherwise survive the edit forever).
if "phases" in msg or "phase_sequence" in msg:
_apply_phase_fields(hass, entry, task, msg.get("phases", task.get("phases")), msg.get("phase_sequence", task.get("phase_sequence")))
from ..helpers.phases import clamp_phase_cursor
store_phase = get_store(hass, entry.entry_id)
if store_phase is not None:
seq = task.get("phase_sequence") or []
if seq:
cur = store_phase.get_task_state(task_id).get("phase_cursor", 0)
store_phase.set_phase_cursor(task_id, clamp_phase_cursor(cur, len(seq)))
else:
store_phase.get_task_state(task_id).pop("phase_cursor", None)
store_phase.async_delay_save()
# Recurrence resolution: an explicit nested `schedule` wins (calendar kinds
# and kind-switches). Otherwise rebuild from the flat view ONLY when a real
# legacy recurrence signal is present — a flat interval/due field, or a legacy
# schedule_type. A bare calendar-kind schedule_type (e.g. a client echoing
# the derived "nth_weekday" the payload exposed) is NOT a flat rebuild signal
# and must not collapse the calendar schedule to manual (the #58/#42 class).
_flat_recurrence_edit = (
any(k in msg for k in ("interval_days", "interval_unit", "interval_anchor", "due_date"))
or msg.get("schedule_type") in FLAT_SCHEDULE_TYPES
)
if msg.get("schedule"):
incoming = Schedule.from_dict(msg["schedule"]).to_dict()
task["schedule"] = incoming
# A COMPLETE nested schedule is authoritative — drop the flat recurrence
# keys so it wins. But a BARE interval schedule (kind=interval with no
# `every`) is only the carrier for the season/ends extras the panel
# always sends for a time-based task; the real interval still rides the
# flat interval_days / interval_unit / interval_anchor fields, which
# normalize_task_storage merges onto it. Popping them for the bare
# interval dropped the interval entirely (#88 regression from the
# season/ends work). Keep them in that one case.
bare_interval = incoming.get("kind") == KIND_INTERVAL and incoming.get("every") is None
if not bare_interval:
for key in FLAT_RECURRENCE_KEYS:
task.pop(key, None)
elif _flat_recurrence_edit:
task.pop("schedule", None)
elif "schedule_type" in msg:
# Calendar-kind schedule_type echo with no schedule → keep the nested
# schedule; drop the stray flat schedule_type so normalize stays clean.
task.pop("schedule_type", None)
# Validate/cap the newly-applied fields. NOTE: this deliberately does NOT
# call cap_task_fields() — the `task/update` schema above already
# length/range-caps every scalar field the sanitiser covers (name,
# task_type, schedule_type, interval_days, warning_days, notes,
# documentation_url, priority, checklist, labels, …), which is the
# WS-boundary contract documented in helpers/sanitize. What the schema
# *can't* express is the shape of the loose dicts and the dedup/trim/seed
# rules on the list fields — exactly the sub-helpers called below. Adding
# cap_task_fields here would be a no-op on everything else and would swap
# the WS layer's reject-at-boundary semantics for silent clamping.
# (The service path in tasks_persist.py DOES call cap_task_fields: those
# functions are reachable directly from Python, not only behind a schema.)
from ..helpers.sanitize import (
cap_action_field,
cap_quick_complete_defaults_field,
sanitize_assignee_pool,
sanitize_labels,
seed_rotation_assignee,
)
cap_action_field(task)
cap_quick_complete_defaults_field(task)
if "labels" in task:
task["labels"] = sanitize_labels(task["labels"])
if "assignee_pool" in task:
task["assignee_pool"] = sanitize_assignee_pool(task["assignee_pool"])
if "required_completion_fields" in task:
from ..helpers.completion_requirements import sanitize_required_completion_fields
task["required_completion_fields"] = sanitize_required_completion_fields(task["required_completion_fields"])
seed_rotation_assignee(task)
# Clear stale trigger runtime in Store only when trigger fundamentally changes
if "trigger_config" in msg:
old_tc = stored_task.get("trigger_config") or {}
new_tc = msg["trigger_config"] or {}
# An edited baseline counts as fundamental: the Store baseline wins
# over the config on restore (#102 restart fix), so without clearing
# it a user-entered start value would silently never take effect.
if (
old_tc.get("type") != new_tc.get("type")
or old_tc.get("entity_id") != new_tc.get("entity_id")
or old_tc.get("entity_ids") != new_tc.get("entity_ids")
or old_tc.get("trigger_baseline_value") != new_tc.get("trigger_baseline_value")
):
store_tc = get_store(hass, entry.entry_id)
if store_tc is not None:
store_tc.clear_trigger_runtime(task_id)
store_tc.async_delay_save()
# Patch only this task's key onto a fresh read (normalized) — never write
# back a whole-map snapshot taken before the store awaits above.
write_task(hass, entry, task_id, task)
# Reload entry to pick up changed task config (triggers, schedule, etc.)
await hass.config_entries.async_reload(entry.entry_id)
result: dict[str, Any] = {"success": True}
if tc_warnings:
result["warnings"] = tc_warnings
connection.send_result(msg["id"], result)
@websocket_api.websocket_command(
{
vol.Required("type"): "maintenance_supporter/task/delete",
vol.Required("entry_id"): vol.All(str, vol.Length(max=MAX_ID_LENGTH)),
vol.Required("task_id"): vol.All(str, vol.Length(max=MAX_ID_LENGTH)),
}
)
@require_write
@websocket_api.async_response
async def ws_delete_task(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
"""Delete a task from a maintenance object."""
ctx = _load_object_task(hass, connection, msg)
if ctx is None:
return
entry, _rd, _task = ctx
await async_delete_task(hass, entry, msg["task_id"])
# Reload to re-create remaining entities
await hass.config_entries.async_reload(entry.entry_id)
connection.send_result(msg["id"], {"success": True})
async def async_delete_task(
hass: HomeAssistant,
entry: ConfigEntry,
task_id: str,
) -> bool:
"""Remove a task and all its side-state from an object entry.
Shared by the ``task/delete`` WS command and the retention auto-delete sweep
(helpers/retention). Does the ConfigEntry write plus the Store / notification
/ entity-registry / group-ref / repair-issue cleanup, but NOT the entry
reload and NOT any WS reply — the caller reloads (once, even for a batch) and
replies. Returns False when ``task_id`` isn't in the entry.
"""
new_data = dict(entry.data)
new_tasks = dict(new_data.get(CONF_TASKS, {}))
if task_id not in new_tasks:
return False
old_trigger_config = new_tasks[task_id].get("trigger_config")
# Adopted problem-sensor task? Preserve its notes + one-time setup
# (responsible user, priority, labels, part link) for a later re-adopt
# (no-op for everything else).
from ..helpers.problem_sensors import stash_task_config_for_readopt
stash_task_config_for_readopt(hass, new_tasks[task_id])
del new_tasks[task_id]
new_data[CONF_TASKS] = new_tasks
# Remove from task_ids
obj = dict(new_data.get(CONF_OBJECT, {}))
obj["task_ids"] = [tid for tid in obj.get("task_ids", []) if tid != task_id]
new_data[CONF_OBJECT] = obj
hass.config_entries.async_update_entry(entry, data=new_data)
# Clean up Store (a mirrored task takes its to-do rows with it, D#183)
store = get_store(hass, entry.entry_id)
if store is not None:
mirror = hass.data.get(DOMAIN, {}).get("todo_mirror")
if mirror is not None and hasattr(mirror, "async_forget_task"):
await mirror.async_forget_task(store, task_id)
store.remove_task(task_id)
await store.async_save()
# Clean up notification state for deleted task
nm = hass.data.get(DOMAIN, {}).get(NOTIFICATION_MANAGER_KEY)
if nm is not None:
nm.clear_task_state(entry.entry_id, task_id)
# Remove orphaned entity registry entries for the deleted task. Match any
# per-task entity — sensor (`_{task_id}`), binary_sensor (`_{task_id}_overdue`),
# action buttons (`_{task_id}_complete/skip/reset`) and any future platform.
# task_id is a UUID, so the contained-segment check is unambiguous.
ent_reg = er.async_get(hass)
for ent_entry in er.async_entries_for_config_entry(ent_reg, entry.entry_id):
if ent_entry.unique_id and f"_{task_id}" in ent_entry.unique_id:
ent_reg.async_remove(ent_entry.entity_id)
# Clean up group references
cleanup_group_refs(hass, task_id=task_id)
# Drop the task's document links (task_ids / task_pages in the global
# DocumentStore) — persistent, task-id keyed, and otherwise never pruned.
from ..const import DOCUMENT_STORE_KEY
doc_store = hass.data.get(DOMAIN, {}).get(DOCUMENT_STORE_KEY)
if doc_store is not None:
await doc_store.async_unlink_task(task_id)
# Clean up the global vacation exempt list (journey L1): the list is
# persistent and task-id keyed — without this, deleted ids accumulate
# there forever and confuse the vacation preview UI.
from ..const import CONF_VACATION_EXEMPT_TASK_IDS
from ..helpers.global_options import get_global_entry
ge = get_global_entry(hass)
if ge is not None:
exempt = ge.options.get(CONF_VACATION_EXEMPT_TASK_IDS) or []
if isinstance(exempt, list) and task_id in exempt:
hass.config_entries.async_update_entry(
ge,
options={
**dict(ge.options),
CONF_VACATION_EXEMPT_TASK_IDS: [t for t in exempt if t != task_id],
},
)
# Clean up any repair issues referencing this task
if old_trigger_config:
from ..entity.triggers import normalize_entity_ids
for eid in normalize_entity_ids(old_trigger_config):
ir.async_delete_issue(
hass,
DOMAIN,
f"missing_trigger_{entry.entry_id}_{task_id}_{eid}",
)
return True
@websocket_api.websocket_command(
{
vol.Required("type"): "maintenance_supporter/task/duplicate",
vol.Required("entry_id"): vol.All(str, vol.Length(max=MAX_ID_LENGTH)),
vol.Required("task_id"): vol.All(str, vol.Length(max=MAX_ID_LENGTH)),
}
)
@require_write
@websocket_api.async_response
async def ws_duplicate_task(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
"""Clone a task within the same object as a fresh, un-started copy.
Copies the source task's *configuration* (schedule, trigger, checklist,
completion actions, …) — which lives in ConfigEntry.data — into a new task
with a new id and a "… (copy)" name. Dynamic state (history, last_performed,
trigger runtime) is not copied: the copy starts clean. Fields that must be
unique per task (entity_slug, nfc_tag_id) are dropped so the copy gets its
own auto-generated slug and no colliding tag.
"""
ctx = _load_object_task(hass, connection, msg)
if ctx is None:
return
entry, _rd, source = ctx
new_task = deepcopy(dict(source))
new_task["id"] = uuid4().hex
base_name = str(source.get("name", "")).strip() or "Task"
new_task["name"] = f"{base_name} (copy)"[:MAX_NAME_LENGTH]
new_task["created_at"] = dt_util.now().date().isoformat()
# Never carry over per-task-unique keys or any stray dynamic state.
strip_task_runtime_state(new_task)
await async_persist_task(hass, entry, new_task)
connection.send_result(msg["id"], {"task_id": new_task["id"]})
@websocket_api.websocket_command(
{
vol.Required("type"): f"{DOMAIN}/task/move",
vol.Required("entry_id"): ID_FIELD,
vol.Required("task_id"): ID_FIELD,
vol.Required("target_entry_id"): ID_FIELD,
}
)
@require_write
@websocket_api.async_response
async def ws_move_task(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
"""Move a task to another object — config, history, readings and trigger
state travel with it; the task gets a new reference number and its
entities are recreated under the target object (forum #23)."""
ctx = _load_object_task(hass, connection, msg)
if ctx is None:
return
entry, _rd, _task = ctx
target = _load_object_entry(
hass, connection, {**msg, "entry_id": msg["target_entry_id"]}, not_found_message="Target object not found"
)
if target is None:
return
if target.entry_id == entry.entry_id:
connection.send_error(msg["id"], "invalid_target", "The task already belongs to that object")
return
if (target.data.get(CONF_OBJECT) or {}).get("archived_at"):
connection.send_error(msg["id"], "invalid_target", "The target object is archived")
return
from .tasks_persist import TaskMoveRefused, async_move_task
try:
await async_move_task(hass, entry, target, msg["task_id"])
except TaskMoveRefused as err:
# task_not_movable / object_not_loaded — raised before anything changed.
connection.send_error(msg["id"], err.code, str(err))
return
except ValueError as err:
connection.send_error(msg["id"], "limit_reached", str(err))
return
await hass.config_entries.async_reload(entry.entry_id)
await hass.config_entries.async_reload(target.entry_id)
connection.send_result(msg["id"], {"task_id": msg["task_id"], "entry_id": target.entry_id})