@@ -103,5 +168,49 @@
Flash attention
settingsSave.disabled = false;
}
});
+
+const promptForm = document.getElementById("prompt-form");
+const promptSave = document.getElementById("prompt-save");
+const promptReset = document.getElementById("prompt-reset");
+const promptStatus = document.getElementById("prompt-status");
+const systemPrompt = document.getElementById("system-prompt");
+const userPrompt = document.getElementById("user-prompt");
+const defaultSystemPrompt = {{ default_system_prompt | tojson }};
+const defaultUserPrompt = {{ default_user_prompt | tojson }};
+
+promptReset.addEventListener("click", () => {
+ systemPrompt.value = defaultSystemPrompt;
+ userPrompt.value = defaultUserPrompt;
+ promptStatus.textContent = "Defaults restored. Select Apply to use them.";
+});
+
+promptForm.addEventListener("submit", async event => {
+ event.preventDefault();
+ promptSave.disabled = true;
+ promptStatus.textContent = "Applying…";
+ beginModelChange("Applying prompts and rebuilding model context…");
+ try {
+ const resp = await fetch("{{ url_for('settings_prompts_apply') }}", {
+ method: "POST",
+ headers: {"Content-Type": "application/json"},
+ body: JSON.stringify({
+ system_prompt: systemPrompt.value,
+ user_prompt: userPrompt.value
+ })
+ });
+ const data = await resp.json();
+ if (!resp.ok) throw new Error(data.error || ("HTTP " + resp.status));
+ systemPrompt.value = data.system_prompt;
+ userPrompt.value = data.user_prompt;
+ promptStatus.textContent = data.customized
+ ? "Saved. Context is " + data.n_ctx + " tokens."
+ : "Saved the built-in defaults. Context is " + data.n_ctx + " tokens.";
+ } catch (error) {
+ promptStatus.textContent = error.message;
+ } finally {
+ endModelChange();
+ promptSave.disabled = false;
+ }
+});
{% endblock %}
diff --git a/script-agent/src/util.py b/script-agent/src/util.py
index 9aad9b9..ae22ef7 100644
--- a/script-agent/src/util.py
+++ b/script-agent/src/util.py
@@ -4,18 +4,18 @@
class LRUCache:
- def __init__(self, maxsize: int):
+ def __init__(self, maxsize: int) -> None:
self.maxsize = maxsize
self.data: OrderedDict[str, Any] = OrderedDict()
- def get(self, key: str, default=None):
+ def get(self, key: str, default: Any = None) -> Any:
if key not in self.data:
return default
self.data.move_to_end(key)
return self.data[key]
- def set(self, key: str, value: Any):
+ def set(self, key: str, value: Any) -> None:
if key in self.data:
self.data.move_to_end(key)
diff --git a/script-agent/src/web_server.py b/script-agent/src/web_server.py
index 812b4c2..e9f1d2f 100644
--- a/script-agent/src/web_server.py
+++ b/script-agent/src/web_server.py
@@ -6,19 +6,31 @@
import threading
import time
from pathlib import Path
-from typing import Any, Dict, List, Optional, Set, Union
+from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Set, Tuple, Union
from flask import Flask, Response, jsonify, render_template, request, url_for
+from flask.typing import ResponseReturnValue
from werkzeug.middleware.proxy_fix import ProxyFix
import benchmark
import overrides
from benchmark import Fixture
from const import AppState
+from gemma4_recognizer import (
+ DEFAULT_SYSTEM_PROMPT,
+ DEFAULT_USER_PROMPT,
+ PROMPT_PLACEHOLDERS,
+ normalize_prompt,
+ validate_prompts,
+)
from hass_api import HomeAssistantInfo, SatelliteInfo, Tool
from overrides import AREA, ENTITY, FLOOR, NAME_KINDS, NameOverrides, ScriptOverrides
from tool_mapping import map_tool_call, required_fields
+if TYPE_CHECKING:
+ # WSGI types live in typeshed only, so they cannot be imported at runtime.
+ from _typeshed.wsgi import StartResponse, WSGIApplication, WSGIEnvironment
+
_LOGGER = logging.getLogger(__name__)
MAX_PASSES = 10
@@ -215,11 +227,11 @@ def issue_run_token(script_id: str, variables: Dict[str, Any]) -> str:
return token
@flask_app.context_processor
- def inject_url_for():
+ def inject_url_for() -> Dict[str, Any]:
return dict(url_for=url_for) # pylint: disable=use-dict-literal
@flask_app.route("/", methods=["GET"])
- def index():
+ def index() -> ResponseReturnValue:
"""Show every script, split by whether the model can call it."""
scripts = [
_describe_script(
@@ -240,7 +252,7 @@ def index():
)
@flask_app.route("/test", methods=["GET"])
- def test_page():
+ def test_page() -> ResponseReturnValue:
"""Show the interactive sentence tester."""
satellites = sorted(
(
@@ -255,7 +267,7 @@ def test_page():
return render_template("test.html", satellites=satellites)
@flask_app.route("/settings", methods=["GET"])
- def settings_page():
+ def settings_page() -> ResponseReturnValue:
"""Show runtime model information and editable recognition settings."""
model = state.recognizer.describe()
return render_template(
@@ -270,10 +282,94 @@ def settings_page():
can_save=state.overrides_path is not None,
min_max_tokens=MIN_MAX_TOKENS,
max_max_tokens=MAX_MAX_TOKENS,
+ system_prompt=normalize_prompt(state.recognizer.system_prompt),
+ user_prompt=normalize_prompt(state.recognizer.user_prompt),
+ default_system_prompt=normalize_prompt(DEFAULT_SYSTEM_PROMPT),
+ default_user_prompt=normalize_prompt(DEFAULT_USER_PROMPT),
+ prompts_customized=(state.overrides.system_prompt is not None)
+ or (state.overrides.user_prompt is not None),
+ placeholders=sorted(PROMPT_PLACEHOLDERS.items()),
+ )
+
+ @flask_app.route("/settings/prompts", methods=["POST"])
+ def settings_prompts_apply() -> ResponseReturnValue:
+ """Apply and persist the system and user prompts."""
+ if state.overrides_path is None:
+ return jsonify({"error": "No settings file is configured"}), 503
+ if not state.recognizer.ready:
+ return jsonify({"error": "Model is still loading"}), 503
+
+ body = request.get_json(silent=True) or {}
+ system_prompt = body.get("system_prompt")
+ user_prompt = body.get("user_prompt")
+ if not isinstance(system_prompt, str) or not isinstance(user_prompt, str):
+ return jsonify({"error": "Both prompts are required"}), 400
+
+ try:
+ system_prompt, user_prompt = validate_prompts(system_prompt, user_prompt)
+ except ValueError as err:
+ return jsonify({"error": str(err)}), 400
+
+ if not state.reload_lock.acquire(blocking=False):
+ return jsonify({"error": "Already applying a change"}), 409
+
+ old_system_prompt = state.recognizer.system_prompt
+ old_user_prompt = state.recognizer.user_prompt
+ candidate_overrides = copy.deepcopy(state.overrides)
+ # A prompt that matches the default is not recorded, so the app keeps
+ # following the default if a later version improves it.
+ candidate_overrides.system_prompt = (
+ None
+ if system_prompt == normalize_prompt(DEFAULT_SYSTEM_PROMPT)
+ else system_prompt
+ )
+ candidate_overrides.user_prompt = (
+ None
+ if user_prompt == normalize_prompt(DEFAULT_USER_PROMPT)
+ else user_prompt
+ )
+
+ state.model_rebuilding.set()
+ try:
+ future = state.llama_executor.submit(
+ state.recognizer.set_prompts, system_prompt, user_prompt
+ )
+ future.result()
+ try:
+ overrides.save(state.overrides_path, candidate_overrides)
+ except Exception:
+ # Keep the persisted and effective prompts aligned if writing fails.
+ rollback = state.llama_executor.submit(
+ state.recognizer.set_prompts, old_system_prompt, old_user_prompt
+ )
+ rollback.result()
+ raise
+ state.overrides = candidate_overrides
+ except ValueError as err:
+ return jsonify({"error": str(err)}), 400
+ except Exception as err: # pylint: disable=broad-except
+ _LOGGER.exception("Failed to apply prompts")
+ return jsonify({"error": f"Could not apply prompts: {err}"}), 500
+ finally:
+ state.model_rebuilding.clear()
+ state.reload_lock.release()
+
+ return jsonify(
+ {
+ "system_prompt": state.recognizer.system_prompt,
+ "user_prompt": state.recognizer.user_prompt,
+ "customized": (candidate_overrides.system_prompt is not None)
+ or (candidate_overrides.user_prompt is not None),
+ "n_ctx": (
+ state.recognizer.llm.n_ctx()
+ if state.recognizer.llm is not None
+ else None
+ ),
+ }
)
@flask_app.route("/settings", methods=["POST"])
- def settings_apply():
+ def settings_apply() -> ResponseReturnValue:
"""Apply and persist the maximum generation length."""
if state.overrides_path is None:
return jsonify({"error": "No settings file is configured"}), 503
@@ -341,7 +437,7 @@ def _fetch_from_hass(
"""
assert state.loop is not None, "No event loop"
- async def fetch():
+ async def fetch() -> Tuple[HomeAssistantInfo, List[Tool]]:
hass_info = await state.hass.get_home_info()
all_tools = await state.hass.get_script_tools(
hass_info, candidate_overrides.names
@@ -360,7 +456,7 @@ async def fetch():
return hass_info, all_tools
@flask_app.route("/reload", methods=["POST"])
- def reload_from_hass():
+ def reload_from_hass() -> ResponseReturnValue:
"""Re-read Home Assistant, then rebuild the tools and model prefix.
The app otherwise reads Home Assistant once, at start, so this is how a
@@ -401,7 +497,7 @@ def _requested_names(body: Dict[str, Any]) -> Optional[Set[str]]:
return {str(name) for name in names} & known
@flask_app.route("/overrides/estimate", methods=["POST"])
- def overrides_estimate():
+ def overrides_estimate() -> ResponseReturnValue:
"""Report what a prospective tool set would cost, before applying it."""
body = request.get_json(silent=True) or {}
names = _requested_names(body)
@@ -423,7 +519,7 @@ def overrides_estimate():
)
@flask_app.route("/overrides", methods=["POST"])
- def overrides_apply():
+ def overrides_apply() -> ResponseReturnValue:
"""Change which scripts are targeted, then rebuild the model prefix.
Recognition is unavailable while the prefix is rebuilt, which can take
@@ -484,13 +580,13 @@ def overrides_apply():
state.reload_lock.release()
@flask_app.route("/tools.json", methods=["GET"])
- def tools_json():
+ def tools_json() -> ResponseReturnValue:
"""The tools as given to the model (OpenAI function spec), for debugging."""
tools = [tool.tool for tool in state.tools.values()]
return Response(json.dumps(tools, indent=2), mimetype="application/json")
@flask_app.route("/test", methods=["POST"])
- def test():
+ def test() -> ResponseReturnValue:
"""Recognize one sentence and report what it would run, without running it."""
body = request.get_json(silent=True) or {}
text = str(body.get("text") or "").strip()
@@ -575,7 +671,7 @@ def test():
)
@flask_app.route("/test/run", methods=["POST"])
- def test_run():
+ def test_run() -> ResponseReturnValue:
"""Run one resolved test result exactly once."""
body = request.get_json(silent=True) or {}
run_id = body.get("run_id")
@@ -613,7 +709,9 @@ def _name_rows() -> Dict[str, List[Dict[str, Any]]]:
names = state.overrides.names
rows: Dict[str, List[Dict[str, Any]]] = {}
- def row(kind: str, target_id: str, default_names: List[str], extra: str = ""):
+ def row(
+ kind: str, target_id: str, default_names: List[str], extra: str = ""
+ ) -> Dict[str, Any]:
current = names.names_for(kind, target_id, default_names)
return {
"id": target_id,
@@ -639,7 +737,7 @@ def row(kind: str, target_id: str, default_names: List[str], extra: str = ""):
return rows
@flask_app.route("/names", methods=["GET"])
- def names_page():
+ def names_page() -> ResponseReturnValue:
"""Edit what the model may call each entity, area, and floor."""
return render_template(
"names.html",
@@ -660,7 +758,7 @@ def _default_names(kind: str, target_id: str) -> List[str]:
return [name for name in target.names if name] if target else []
@flask_app.route("/names/field", methods=["GET"])
- def names_field_page():
+ def names_field_page() -> ResponseReturnValue:
"""Edit names for one script's field only.
Reached from that field on the Scripts page, so the list is already the
@@ -725,7 +823,7 @@ def names_field_page():
)
@flask_app.route("/names/field", methods=["POST"])
- def names_field_apply():
+ def names_field_apply() -> ResponseReturnValue:
"""Replace one script field's name overrides, then rebuild."""
if not state.recognizer.ready:
return jsonify({"error": "Model is still loading"}), 503
@@ -769,7 +867,7 @@ def names_field_apply():
return jsonify({"num_overridden": len(scoped)})
@flask_app.route("/names", methods=["POST"])
- def names_apply():
+ def names_apply() -> ResponseReturnValue:
"""Replace name overrides, then rebuild the tools and model prefix.
Names feed the enums, which are built while reading Home Assistant, so
@@ -821,22 +919,22 @@ def names_apply():
)
@flask_app.route("/health")
- def health():
+ def health() -> ResponseReturnValue:
# Deliberately "ok" while the model is still loading: the first boot can
# spend many minutes downloading it, and the container should not be
# restarted for being slow. Use /status to see readiness.
return {"status": "ok"}, 200
@flask_app.route("/status")
- def status():
+ def status() -> ResponseReturnValue:
return {"model_loaded": state.recognizer.ready}, 200
@flask_app.route("/benchmark", methods=["GET"])
- def benchmark_page():
+ def benchmark_page() -> ResponseReturnValue:
return render_template("benchmark.html", max_passes=MAX_PASSES)
@flask_app.route("/benchmark/run", methods=["POST"])
- def benchmark_run():
+ def benchmark_run() -> ResponseReturnValue:
if fixture is None:
return {"error": fixture_error or "No benchmark fixture"}, 500
if not state.recognizer.ready:
@@ -863,7 +961,7 @@ def benchmark_run():
def run_web_server(flask_app: Flask, host: str, port: int) -> threading.Thread:
- def run_flask():
+ def run_flask() -> None:
logging.getLogger("werkzeug").setLevel(logging.ERROR)
flask_app.run(host=host, port=port, use_reloader=False)
@@ -875,10 +973,12 @@ def run_flask():
class IngressPrefixMiddleware:
"""Ingress fix for Home Assistant app web UI."""
- def __init__(self, app):
+ def __init__(self, app: "WSGIApplication") -> None:
self.app = app
- def __call__(self, environ, start_response):
+ def __call__(
+ self, environ: "WSGIEnvironment", start_response: "StartResponse"
+ ) -> Iterable[bytes]:
ingress_path = environ.get("HTTP_X_INGRESS_PATH", "")
if ingress_path:
environ["SCRIPT_NAME"] = ingress_path
diff --git a/script-agent/tests/test_gemma4_recognizer.py b/script-agent/tests/test_gemma4_recognizer.py
index 6d1e130..9175892 100644
--- a/script-agent/tests/test_gemma4_recognizer.py
+++ b/script-agent/tests/test_gemma4_recognizer.py
@@ -2,6 +2,7 @@
import sys
import tempfile
import unittest
+from datetime import datetime
from pathlib import Path
from types import ModuleType
from unittest.mock import patch
@@ -15,10 +16,15 @@
sys.modules["llama_cpp"] = llama_cpp_stub
from gemma4_recognizer import ( # noqa: E402
+ DEFAULT_SYSTEM_PROMPT,
+ DEFAULT_USER_PROMPT,
LLAMA_CPP_VERSION,
Gemma4Recognizer,
_get_tools_hash,
_parse_tool_calls,
+ format_prompt,
+ prompt_values,
+ validate_prompts,
)
@@ -89,6 +95,151 @@ def test_truncated_response_is_not_partially_executed(self):
self.assertIn("64-token generation limit", text)
+class PromptTests(unittest.TestCase):
+ def test_default_user_prompt_carries_the_current_date(self):
+ # Without it the model cannot turn "Saturday" into a date, and answers a
+ # datetime field with the word instead.
+ content = format_prompt(
+ DEFAULT_USER_PROMPT,
+ prompt_values("book it for saturday", "en", datetime(2026, 8, 31, 9, 5)),
+ )
+
+ self.assertIn('Sentence: "book it for saturday"', content)
+ self.assertIn("Monday", content)
+ self.assertIn("2026-08-31", content)
+
+ def test_unknown_placeholder_is_left_alone_rather_than_raising(self):
+ self.assertEqual(
+ "say {something} about hi",
+ format_prompt("say {something} about {text}", {"text": "hi"}),
+ )
+
+ def test_default_prompts_are_valid_and_already_normalized(self):
+ # Written normalized so that restoring them in the web UI leaves the
+ # saved and running prompts identical, rather than one rebuild apart.
+ system_prompt, user_prompt = validate_prompts(
+ DEFAULT_SYSTEM_PROMPT, DEFAULT_USER_PROMPT
+ )
+
+ self.assertEqual(DEFAULT_SYSTEM_PROMPT, system_prompt)
+ self.assertEqual(DEFAULT_USER_PROMPT, user_prompt)
+
+ def test_user_prompt_without_the_sentence_is_rejected(self):
+ with self.assertRaisesRegex(ValueError, r"must include \{text\}"):
+ validate_prompts(DEFAULT_SYSTEM_PROMPT, "Language: {language}")
+
+ def test_misspelled_placeholder_is_rejected(self):
+ with self.assertRaisesRegex(ValueError, r"\{weekdya\}"):
+ validate_prompts(DEFAULT_SYSTEM_PROMPT, "{text} on {weekdya}")
+
+ def test_empty_prompt_is_rejected(self):
+ with self.assertRaisesRegex(ValueError, "System prompt cannot be empty"):
+ validate_prompts(" \n ", DEFAULT_USER_PROMPT)
+
+ def test_cache_key_follows_the_date_the_prompt_carried(self):
+ recognizer = Gemma4Recognizer(state_path="unused.bin", cache_size=10)
+ recognizer.tools = []
+ recognizer.llm = _FakeLlama(
+ response={"choices": [{"message": {"content": "no tools"}}]}
+ )
+
+ with patch("gemma4_recognizer.datetime") as clock:
+ clock.now.return_value = datetime(2026, 8, 31, 9, 5)
+ recognizer.get_tool_calls("book it for saturday")
+ recognizer.get_tool_calls("book it for saturday")
+ self.assertEqual(1, recognizer.llm.create_calls)
+
+ # Same sentence, next day: the answer would be a day early.
+ clock.now.return_value = datetime(2026, 9, 1, 9, 5)
+ recognizer.get_tool_calls("book it for saturday")
+
+ self.assertEqual(2, recognizer.llm.create_calls)
+
+
+class PromptSettingsTests(unittest.TestCase):
+ def _recognizer(self):
+ recognizer = Gemma4Recognizer(state_path="unused.bin", cache_size=10)
+ recognizer.tools = [{"type": "function", "function": {"name": "demo"}}]
+ recognizer.llm = _FakeLlama(n_ctx=1024)
+ recognizer.ready = True
+ return recognizer
+
+ def test_form_round_trip_does_not_rebuild(self):
+ recognizer = self._recognizer()
+ recognizer.cache.set("en: cached", ([], "cached"))
+
+ with patch.object(recognizer, "_restore_or_build_state") as restore:
+ # What a textarea gives back: no leading newline, CRLF line endings.
+ recognizer.set_prompts(
+ DEFAULT_SYSTEM_PROMPT.strip().replace("\n", "\r\n"),
+ DEFAULT_USER_PROMPT,
+ )
+
+ restore.assert_not_called()
+ self.assertEqual(DEFAULT_SYSTEM_PROMPT, recognizer.system_prompt)
+ self.assertIsNotNone(recognizer.cache.get("en: cached"))
+
+ def test_user_prompt_change_clears_the_cache_without_rebuilding(self):
+ recognizer = self._recognizer()
+ recognizer.cache.set("en: cached", ([], "cached"))
+
+ with patch.object(recognizer, "_restore_or_build_state") as restore:
+ recognizer.set_prompts(DEFAULT_SYSTEM_PROMPT, "{text} ({language})")
+
+ restore.assert_not_called()
+ self.assertEqual("{text} ({language})", recognizer.user_prompt)
+ self.assertIsNone(recognizer.cache.get("en: cached"))
+ self.assertTrue(recognizer.ready)
+
+ def test_system_prompt_change_rebuilds_the_prefix(self):
+ recognizer = self._recognizer()
+
+ with patch.object(recognizer, "_restore_or_build_state") as restore:
+ recognizer.set_prompts("Only call one tool.", DEFAULT_USER_PROMPT)
+
+ restore.assert_called_once_with()
+ self.assertEqual("Only call one tool.", recognizer.system_prompt)
+ self.assertEqual("Only call one tool.", recognizer.system_message["content"])
+ self.assertTrue(recognizer.ready)
+
+ def test_rejected_prompt_leaves_the_running_ones_alone(self):
+ recognizer = self._recognizer()
+
+ with self.assertRaises(ValueError):
+ recognizer.set_prompts("Only call one tool.", "no sentence here")
+
+ self.assertEqual(DEFAULT_SYSTEM_PROMPT, recognizer.system_prompt)
+ self.assertEqual(DEFAULT_USER_PROMPT, recognizer.user_prompt)
+
+ def test_failed_rebuild_rolls_the_prompts_back(self):
+ recognizer = self._recognizer()
+ old_llm = recognizer.llm
+
+ with patch.object(
+ recognizer,
+ "_restore_or_build_state",
+ side_effect=RuntimeError("prefix failed"),
+ ):
+ with self.assertRaisesRegex(RuntimeError, "prefix failed"):
+ recognizer.set_prompts("Only call one tool.", DEFAULT_USER_PROMPT)
+
+ self.assertEqual(DEFAULT_SYSTEM_PROMPT, recognizer.system_prompt)
+ self.assertEqual(DEFAULT_SYSTEM_PROMPT, recognizer.system_message["content"])
+ self.assertIs(old_llm, recognizer.llm)
+ self.assertEqual([{"state": "good"}], old_llm.loaded_states)
+ self.assertTrue(recognizer.ready)
+
+ def test_fixed_context_rejects_prompts_that_will_not_fit(self):
+ recognizer = self._recognizer()
+ recognizer.n_ctx = 1024
+
+ with patch.object(recognizer, "required_n_ctx", return_value=2048):
+ with self.assertRaisesRegex(ValueError, "fixed context is 1024"):
+ recognizer.set_prompts("Only call one tool.", DEFAULT_USER_PROMPT)
+
+ self.assertEqual(DEFAULT_SYSTEM_PROMPT, recognizer.system_prompt)
+
+
class RuntimeSettingsTests(unittest.TestCase):
def test_max_tokens_grows_automatic_context(self):
recognizer = Gemma4Recognizer(
diff --git a/script-agent/tests/test_hass_api.py b/script-agent/tests/test_hass_api.py
index 67dabf5..8f86155 100644
--- a/script-agent/tests/test_hass_api.py
+++ b/script-agent/tests/test_hass_api.py
@@ -1,4 +1,7 @@
import unittest
+from pathlib import Path
+
+import yaml
from hass_api import (
HomeAssistant,
@@ -91,6 +94,71 @@ def test_unrestricted_filter_alternative_does_not_restrict_domains(self):
)
)
+ def test_shorthand_domain_restricts_domains(self):
+ # `entity: {domain: calendar}` predates `filter` and is still valid, so a
+ # field written that way must not be offered every exposed entity.
+ self.assertEqual(
+ {"calendar"},
+ _get_entity_filter_domains({"domain": "calendar"}),
+ )
+
+ def test_shorthand_domain_list_restricts_domains(self):
+ self.assertEqual(
+ {"light", "switch"},
+ _get_entity_filter_domains({"domain": ["light", "switch"]}),
+ )
+
+ def test_shorthand_without_domain_does_not_restrict_domains(self):
+ self.assertIsNone(_get_entity_filter_domains({"device_class": "outlet"}))
+ self.assertIsNone(_get_entity_filter_domains({}))
+
+ def test_filter_beats_shorthand_domain(self):
+ self.assertEqual(
+ {"calendar"},
+ _get_entity_filter_domains(
+ {"domain": "light", "filter": {"domain": "calendar"}}
+ ),
+ )
+
+
+class BlueprintTests(unittest.TestCase):
+ """The bundled examples are what people copy, so keep them parseable."""
+
+ def _blueprints(self):
+ blueprint_dir = Path(__file__).parent.parent / "blueprints"
+ paths = sorted(blueprint_dir.glob("*.yaml"))
+ self.assertTrue(paths, "No blueprints found")
+ return [(path, yaml.safe_load(path.read_text("utf-8"))) for path in paths]
+
+ def test_entity_fields_declare_a_domain(self):
+ for path, blueprint in self._blueprints():
+ for field_key, field_info in (blueprint.get("fields") or {}).items():
+ selector = field_info.get("selector") or {}
+ if "entity" not in selector:
+ continue
+
+ with self.subTest(blueprint=path.name, field=field_key):
+ self.assertIsNotNone(
+ _get_entity_filter_domains(selector["entity"] or {}),
+ f"{field_key} is offered every exposed entity",
+ )
+
+ def test_calendar_event_targets_calendars(self):
+ path = Path(__file__).parent.parent / "blueprints/create_calendar_event.yaml"
+ blueprint = yaml.safe_load(path.read_text("utf-8"))
+ fields = blueprint["fields"]
+
+ self.assertEqual(
+ {"calendar"},
+ _get_entity_filter_domains(fields["calendar"]["selector"]["entity"]),
+ )
+ self.assertEqual({"datetime": None}, fields["start"]["selector"])
+ self.assertEqual({"duration": None}, fields["duration"]["selector"])
+ self.assertTrue(
+ all(fields[key].get("required") for key in fields),
+ "Every field is needed to create an event",
+ )
+
class SelectorSchemaTests(unittest.TestCase):
def test_labelled_select_options_use_submitted_values(self):
diff --git a/script-agent/tests/test_release.py b/script-agent/tests/test_release.py
index 28698ef..0f4b214 100644
--- a/script-agent/tests/test_release.py
+++ b/script-agent/tests/test_release.py
@@ -1,4 +1,5 @@
import argparse
+import re
import unittest
from pathlib import Path
from unittest.mock import patch
@@ -24,6 +25,16 @@ def test_config_identity_matches_runtime_identity(self):
def test_config_version_matches_runtime_version(self):
self.assertEqual(APP_VERSION, self.config["version"])
+ def test_changelog_documents_the_current_version(self):
+ # The version is written in three places: config.yaml, const.py, and the
+ # changelog's newest heading. The first two are checked above; this is
+ # the copy that is easy to leave behind when the number changes.
+ changelog = (self.project_dir / "CHANGELOG.md").read_text("utf-8")
+ versions = re.findall(r"^##\s+(\S+)", changelog, flags=re.MULTILINE)
+
+ self.assertTrue(versions, "CHANGELOG.md has no version headings")
+ self.assertEqual(APP_VERSION, versions[0])
+
def test_flash_attention_is_enabled_by_default(self):
self.assertIs(True, self.config["options"]["flash_attention"])
diff --git a/script-agent/tests/test_web_server.py b/script-agent/tests/test_web_server.py
index ae5e49c..710621a 100644
--- a/script-agent/tests/test_web_server.py
+++ b/script-agent/tests/test_web_server.py
@@ -5,6 +5,11 @@
from unittest.mock import Mock, patch
from const import AppState
+from gemma4_recognizer import (
+ DEFAULT_SYSTEM_PROMPT,
+ DEFAULT_USER_PROMPT,
+ validate_prompts,
+)
from hass_api import HomeAssistantInfo, SatelliteInfo, Tool
from models import Entity
from overrides import AREA, ENTITY, FLOOR, NameOverrides, Overrides, load
@@ -33,6 +38,9 @@ def __init__(self, error=None):
self.error = error
self.reload_calls = []
self.max_token_calls = []
+ self.prompt_calls = []
+ self.system_prompt = DEFAULT_SYSTEM_PROMPT
+ self.user_prompt = DEFAULT_USER_PROMPT
self.max_tokens = 128
self.n_ctx = None
self.ready = True
@@ -53,6 +61,15 @@ def set_max_tokens(self, max_tokens):
raise self.error
self.max_tokens = max_tokens
+ def set_prompts(self, system_prompt, user_prompt):
+ self.prompt_calls.append((system_prompt, user_prompt))
+ if self.error:
+ raise self.error
+ # The real recognizer validates and normalizes before it commits.
+ self.system_prompt, self.user_prompt = validate_prompts(
+ system_prompt, user_prompt
+ )
+
def describe(self):
return {
"repo_id": self.repo_id,
@@ -63,6 +80,8 @@ def describe(self):
"temperature": 0.0,
"flash_attn": self.flash_attn,
"draft_model": None,
+ "system_prompt": self.system_prompt,
+ "user_prompt": self.user_prompt,
}
def get_tool_calls(self, _text, _language):
@@ -489,6 +508,83 @@ def test_settings_page_rolls_back_when_persistence_fails(self):
self.assertEqual(128, self.state.recognizer.max_tokens)
self.assertIsNone(self.state.overrides.max_tokens)
+ def test_settings_page_shows_the_prompts_and_their_placeholders(self):
+ response = self.client.get("/settings")
+
+ self.assertEqual(200, response.status_code)
+ self.assertIn(b"Sentence: "{text}"", response.data)
+ self.assertIn(b"{weekday}", response.data)
+ self.assertIn(b"Using the built-in defaults.", response.data)
+
+ def test_prompts_are_applied_and_persisted(self):
+ response = self.client.post(
+ "/settings/prompts",
+ json={
+ "system_prompt": "Only call one tool.\r\n",
+ "user_prompt": '"{text}" on {weekday}',
+ },
+ )
+
+ self.assertEqual(200, response.status_code)
+ self.assertTrue(response.get_json()["customized"])
+ self.assertEqual("Only call one tool.", self.state.recognizer.system_prompt)
+ self.assertEqual(
+ "Only call one tool.", load(self.state.overrides_path).system_prompt
+ )
+ self.assertEqual(
+ '"{text}" on {weekday}', load(self.state.overrides_path).user_prompt
+ )
+
+ def test_default_prompts_are_not_recorded_as_overrides(self):
+ self.state.overrides.system_prompt = "Only call one tool."
+
+ response = self.client.post(
+ "/settings/prompts",
+ json={
+ "system_prompt": DEFAULT_SYSTEM_PROMPT,
+ "user_prompt": DEFAULT_USER_PROMPT,
+ },
+ )
+
+ self.assertEqual(200, response.status_code)
+ self.assertFalse(response.get_json()["customized"])
+ self.assertIsNone(self.state.overrides.system_prompt)
+ self.assertIsNone(load(self.state.overrides_path).system_prompt)
+
+ def test_prompt_without_the_sentence_is_rejected(self):
+ response = self.client.post(
+ "/settings/prompts",
+ json={
+ "system_prompt": DEFAULT_SYSTEM_PROMPT,
+ "user_prompt": "Language: {language}",
+ },
+ )
+
+ self.assertEqual(400, response.status_code)
+ self.assertIn("{text}", response.get_json()["error"])
+ self.assertEqual([], self.state.recognizer.prompt_calls)
+
+ def test_prompts_roll_back_when_persistence_fails(self):
+ with patch("web_server.overrides.save", side_effect=OSError("disk full")):
+ response = self.client.post(
+ "/settings/prompts",
+ json={
+ "system_prompt": "Only call one tool.",
+ "user_prompt": DEFAULT_USER_PROMPT,
+ },
+ )
+
+ self.assertEqual(500, response.status_code)
+ self.assertEqual(
+ [
+ ("Only call one tool.", DEFAULT_USER_PROMPT),
+ (DEFAULT_SYSTEM_PROMPT, DEFAULT_USER_PROMPT),
+ ],
+ self.state.recognizer.prompt_calls,
+ )
+ self.assertEqual(DEFAULT_SYSTEM_PROMPT, self.state.recognizer.system_prompt)
+ self.assertIsNone(self.state.overrides.system_prompt)
+
def test_field_details_are_available_as_json_for_dialog(self):
response = self.client.get(
"/names/field?script=notify_mike&field=target&format=json"