diff --git a/custom_components/kyber/ci_integration.py b/custom_components/kyber/ci_integration.py new file mode 100644 index 0000000..4dd8cf7 --- /dev/null +++ b/custom_components/kyber/ci_integration.py @@ -0,0 +1,309 @@ +from __future__ import annotations + +from collections import deque +import json +import uuid +from typing import Any + +from homeassistant.core import HomeAssistant +from homeassistant.util import dt as ha_dt + +from .const import CONF_AI_TASK_ENTITY_ID + +_CI_EVENT_LOG_KEY = "kyber_ci_event_log" +_CI_EVENT_LOG_MAX = 50 +_ALLOWED_REASONS = { + "all_clear", + "dinner_window", + "oven_active", + "sleep_time", + "nobody_home", + "manual_hold", + "unknown", +} +_HOME_CONTEXT_PROMPT = ( + "You are checking if it is safe to do a production software deployment right now. \n" + "Assess the current home context: is anyone having dinner (check oven, time of day 17:30-20:00, kitchen activity), \n" + "is anyone asleep (late night, no motion), is the home in a relaxed state, \n" + "are there any reasons this would be a bad time for a system deployment?\n\n" + "Return ONLY valid JSON with these exact fields:\n" + "{\n" + ' \"can_deploy\": ,\n' + ' \"reason\": \"\",\n' + ' \"details\": \"<1-2 sentence human explanation>\",\n' + ' \"estimated_resume\": \"\",\n' + ' \"local_time\": \"\"\n' + "}\n" + "Do not include any text before or after the JSON." +) + + +def _event_log(hass: HomeAssistant) -> deque[dict[str, Any]]: + log = hass.data.get(_CI_EVENT_LOG_KEY) + if not isinstance(log, deque): + log = deque(maxlen=_CI_EVENT_LOG_MAX) + hass.data[_CI_EVENT_LOG_KEY] = log + return log + + +def _append_event(hass: HomeAssistant, event: dict[str, Any]) -> None: + _event_log(hass).append(event) + + +def _time_fallback(raw_response: str = "") -> dict[str, Any]: + now = ha_dt.now() + local_time = now.strftime("%H:%M") + minutes = now.hour * 60 + now.minute + dinner = 17 * 60 + 30 <= minutes <= 20 * 60 + can_deploy = not dinner + return { + "can_deploy": can_deploy, + "reason": "all_clear" if can_deploy else "dinner_window", + "details": ( + "Fallback time check: outside the dinner window." + if can_deploy + else f"Fallback time check: {local_time} is within the dinner window (17:30-20:00)." + ), + "estimated_resume": "now" if can_deploy else "20:00", + "local_time": local_time, + "raw_kyber_response": raw_response, + } + + +def _extract_json_payload(raw_response: str) -> dict[str, Any]: + text = (raw_response or "").strip() + if not text: + raise ValueError("empty response") + try: + parsed = json.loads(text) + except json.JSONDecodeError: + start = text.find("{") + end = text.rfind("}") + if start == -1 or end == -1 or end <= start: + raise + parsed = json.loads(text[start : end + 1]) + if not isinstance(parsed, dict): + raise ValueError("response was not a JSON object") + return parsed + + +async def _run_kyber_prompt( + hass: HomeAssistant, + prompt: str, + config: dict[str, Any], + user_id: str, +) -> str: + from .http_api import ( + _build_context, + _build_prompt_sections, + _inject_knowledge_into_instructions, + _run_ai_loop, + ) + from .knowledge import get_store as get_knowledge_store + + entity_id = str(config.get(CONF_AI_TASK_ENTITY_ID, "")).strip() + if not entity_id: + raise ValueError("No AI task entity configured in Kyber settings") + + request_id = str(uuid.uuid4()) + context, _context_stats = _build_context(hass) + kstore = get_knowledge_store(hass) + await kstore.async_load() + + body_fields = { + "user_prompt": prompt, + "user_yaml": "", + "history": [], + "compacted_summary": "", + "editor_mode": "chat", + "dashboards": [], + "lovelace_resources": [], + "request_id": request_id, + } + + class _MockUser: + def __init__(self, uid: str) -> None: + self.id = uid + self.is_admin = False + self.name = "CI integration" + + class _MockRequest: + def get(self, key: str, default: Any = None) -> Any: + if key == "hass_user": + return _MockUser(user_id) + return default + + sections = _build_prompt_sections(body_fields, context, _MockRequest()) + instructions = sections["instructions"] + intent = sections["intent"] + instructions, _ = await _inject_knowledge_into_instructions( + hass, + kstore, + prompt, + instructions, + request_id, + entity_id=entity_id, + user_id=user_id or None, + is_admin=False, + ) + response_text, _tool_log, _exchange, _cache, _intent, _loop_instr, _aliases, _tokens = ( + await _run_ai_loop( + hass, + entity_id, + instructions, + kstore, + prompt, + request_id, + [], + intent, + config=config, + user_id=user_id, + is_admin=False, + ) + ) + return str(response_text or "") + + +def _fallback_briefing(events: list[dict[str, Any]]) -> str: + if not events: + return "No recent CI events were recorded. Everything has been quiet." + lines = [] + for event in events: + repo = event.get("repo") or "unknown repo" + branch = event.get("branch") or "unknown branch" + status = event.get("status") or "info" + message = event.get("message") or event.get("reason") or event.get("type") or "update recorded" + timestamp = event.get("local_time") or event.get("timestamp") or "recently" + lines.append(f"- {timestamp}: {repo} ({branch}) reported {status} — {message}") + return "Recent CI activity while you were away:\n" + "\n".join(lines) + + +async def handle_get_home_context( + hass: HomeAssistant, + params: dict[str, Any], + config: dict[str, Any], + user_id: str, +) -> dict[str, Any]: + _ = params + raw_response = "" + try: + raw_response = await _run_kyber_prompt(hass, _HOME_CONTEXT_PROMPT, config, user_id) + parsed = _extract_json_payload(raw_response) + can_deploy = bool(parsed.get("can_deploy", True)) + reason = str(parsed.get("reason", "unknown")).strip() or "unknown" + if reason not in _ALLOWED_REASONS: + reason = "unknown" + details = str(parsed.get("details", "")).strip() or "Kyber did not provide details." + estimated_resume = str(parsed.get("estimated_resume", "now")).strip() or "now" + local_time = str(parsed.get("local_time", "")).strip() or ha_dt.now().strftime("%H:%M") + return { + "can_deploy": can_deploy, + "reason": reason, + "details": details, + "estimated_resume": estimated_resume, + "local_time": local_time, + "raw_kyber_response": raw_response, + } + except Exception: + return _time_fallback(raw_response) + + +async def handle_ci_event(hass: HomeAssistant, params: dict[str, Any]) -> dict[str, Any]: + event_type = str(params.get("type", "")).strip() + repo = str(params.get("repo", "")).strip() + branch = str(params.get("branch", "")).strip() + status = str(params.get("status", "")).strip().lower() + message = str(params.get("message", "")).strip() + + if not event_type: + return {"error": "type is required"} + if not repo: + return {"error": "repo is required"} + if not branch: + return {"error": "branch is required"} + if status not in {"success", "failure", "info"}: + return {"error": "status must be one of: success, failure, info"} + if not message: + return {"error": "message is required"} + + now = ha_dt.now() + event_data = { + "type": event_type, + "repo": repo, + "branch": branch, + "status": status, + "message": message, + "timestamp": now.isoformat(), + "local_time": now.strftime("%H:%M"), + } + _append_event(hass, event_data) + hass.bus.async_fire("kyber_ci_event", event_data) + return {"fired": True, "event_type": "kyber_ci_event", "data": event_data} + + +async def handle_celebrate(hass: HomeAssistant, params: dict[str, Any]) -> dict[str, Any]: + reason = str(params.get("reason") or "Manual merge to production!").strip() or "Manual merge to production!" + now = ha_dt.now() + event_data = { + "type": "celebrate", + "repo": str(params.get("repo") or "manual"), + "branch": str(params.get("branch") or "production"), + "status": "success", + "message": reason, + "timestamp": now.isoformat(), + "local_time": now.strftime("%H:%M"), + } + _append_event(hass, event_data) + hass.bus.async_fire("kyber_ci_event", event_data) + await hass.services.async_call( + "persistent_notification", + "create", + { + "title": "Kyber CI celebration", + "message": reason, + "notification_id": "kyber_ci_celebration", + }, + blocking=True, + ) + return {"celebrated": True, "reason": reason} + + +async def handle_peace_briefing( + hass: HomeAssistant, + params: dict[str, Any], + config: dict[str, Any], + user_id: str, +) -> dict[str, Any]: + try: + max_events = int(params.get("max_events", 10)) + except (TypeError, ValueError): + max_events = 10 + max_events = max(1, min(max_events, _CI_EVENT_LOG_MAX)) + events = list(_event_log(hass))[-max_events:] + + if not events: + return { + "briefing": _fallback_briefing(events), + "events": [], + "count": 0, + } + + prompt = ( + "You are giving a calm briefing about recent CI/CD activity. " + "Summarize the key events in 2-4 sentences, mention any failures that may need attention, " + "and keep the tone reassuring.\n\n" + f"Recent events JSON:\n{json.dumps(events, ensure_ascii=False, indent=2)}" + ) + + try: + briefing = (await _run_kyber_prompt(hass, prompt, config, user_id)).strip() + if not briefing: + raise ValueError("empty briefing") + except Exception: + briefing = _fallback_briefing(events) + + return { + "briefing": briefing, + "events": events, + "count": len(events), + } diff --git a/custom_components/kyber/mcp.py b/custom_components/kyber/mcp.py index 2c7361a..3336640 100644 --- a/custom_components/kyber/mcp.py +++ b/custom_components/kyber/mcp.py @@ -37,6 +37,12 @@ DOMAIN, _sanitize_user_input, ) +from .ci_integration import ( + handle_celebrate, + handle_ci_event, + handle_get_home_context, + handle_peace_briefing, +) from .rate_limiter import _rate_limiter _LOGGER = logging.getLogger(__name__) @@ -297,6 +303,60 @@ def _err(req_id: Any, code: int, message: str) -> dict: "required": ["domain", "service"], }, }, + { + "name": "get_home_context", + "description": ( + "Ask Kyber whether the current home context makes this a good time for a production deployment. " + "Uses Kyber's AI pipeline and the home's current state instead of hardcoded entity IDs." + ), + "inputSchema": { + "type": "object", + "properties": {}, + }, + }, + { + "name": "ci_event", + "description": ( + "Record and broadcast a CI/CD event into Home Assistant so automations can react." + ), + "inputSchema": { + "type": "object", + "properties": { + "type": {"type": "string", "description": "Event type, e.g. 'deploy_success' or 'build_failed'."}, + "repo": {"type": "string", "description": "GitHub repository name, e.g. 'pgroene/ProspectPilot'."}, + "branch": {"type": "string", "description": "Branch name, e.g. 'main'."}, + "status": {"type": "string", "enum": ["success", "failure", "info"], "description": "Event status."}, + "message": {"type": "string", "description": "Human-readable event summary."}, + }, + "required": ["type", "repo", "branch", "status", "message"], + }, + }, + { + "name": "celebrate", + "description": ( + "Celebrate a successful manual merge or deployment so Home Assistant automations can reward it." + ), + "inputSchema": { + "type": "object", + "properties": { + "reason": {"type": "string", "description": "Optional celebration reason."}, + "repo": {"type": "string", "description": "Optional repository label for the event log."}, + "branch": {"type": "string", "description": "Optional branch label for the event log."}, + }, + }, + }, + { + "name": "peace_briefing", + "description": ( + "Summarize recent CI/CD events from Home Assistant into a calm briefing." + ), + "inputSchema": { + "type": "object", + "properties": { + "max_events": {"type": "integer", "description": "Maximum number of recent events to include. Defaults to 10."}, + }, + }, + }, ] _EXECUTE_PLAN_TOOL: dict = { @@ -1078,11 +1138,11 @@ async def _call_tool( elif name == "get_entity_state": result = await _handle_get_entity_state(hass, args) elif name == "search_entities": - result = await _handle_search_entities(hass, tool_args) + result = await _handle_search_entities(hass, args) elif name == "kyber_remember": - result = await _handle_kyber_remember(hass, tool_args) + result = await _handle_kyber_remember(hass, args) elif name == "kyber_recall": - result = await _handle_kyber_recall(hass, tool_args) + result = await _handle_kyber_recall(hass, args) elif name == "list_entities": result = await _handle_list_entities(hass, args) elif name == "call_service": @@ -1093,6 +1153,14 @@ async def _call_tool( result = await _handle_get_datetime(hass, args) elif name == "get_todo_items": result = await _handle_get_todo_items(hass, args) + elif name == "get_home_context": + result = await handle_get_home_context(hass, args, self._config, user_id) + elif name == "ci_event": + result = await handle_ci_event(hass, args) + elif name == "celebrate": + result = await handle_celebrate(hass, args) + elif name == "peace_briefing": + result = await handle_peace_briefing(hass, args, self._config, user_id) elif name == "kyber_execute_plan": if not self._config.get(CONF_MCP_ALLOW_STATE_CHANGES, False): return _err(req_id, -32602, "kyber_execute_plan is disabled. Enable 'MCP can change state of home' in Kyber settings.") diff --git a/tests/test_ci_integration.py b/tests/test_ci_integration.py new file mode 100644 index 0000000..d86504f --- /dev/null +++ b/tests/test_ci_integration.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +from collections import deque +from datetime import datetime, timezone +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +pytest.importorskip("pytest_homeassistant_custom_component", reason="requires pytest-homeassistant-custom-component") + +from custom_components.kyber.ci_integration import ( + _CI_EVENT_LOG_KEY, + handle_celebrate, + handle_ci_event, + handle_get_home_context, + handle_peace_briefing, +) + + +def _make_hass(): + hass = MagicMock() + hass.data = {} + hass.bus.async_fire = MagicMock() + hass.services.async_call = AsyncMock() + return hass + + +@pytest.mark.asyncio +async def test_handle_ci_event_fires_and_logs_event(): + hass = _make_hass() + + result = await handle_ci_event( + hass, + { + "type": "build_failed", + "repo": "pgroene/ProspectPilot", + "branch": "main", + "status": "failure", + "message": "Deploy failed", + }, + ) + + assert result["fired"] is True + hass.bus.async_fire.assert_called_once() + logged = list(hass.data[_CI_EVENT_LOG_KEY]) + assert len(logged) == 1 + assert logged[0]["type"] == "build_failed" + assert logged[0]["status"] == "failure" + + +@pytest.mark.asyncio +async def test_handle_celebrate_sends_notification_and_logs_event(): + hass = _make_hass() + + result = await handle_celebrate(hass, {"reason": "Merged to prod"}) + + assert result == {"celebrated": True, "reason": "Merged to prod"} + hass.bus.async_fire.assert_called_once() + hass.services.async_call.assert_awaited_once() + assert list(hass.data[_CI_EVENT_LOG_KEY])[0]["type"] == "celebrate" + + +@pytest.mark.asyncio +async def test_handle_get_home_context_falls_back_when_ai_returns_invalid_json(): + hass = _make_hass() + now = datetime(2025, 1, 1, 18, 15, tzinfo=timezone.utc) + + with patch("custom_components.kyber.ci_integration._run_kyber_prompt", new=AsyncMock(return_value="not json")), \ + patch("custom_components.kyber.ci_integration.ha_dt.now", return_value=now): + result = await handle_get_home_context(hass, {}, {"ai_task_entity_id": "conversation.mock"}, "user-1") + + assert result["can_deploy"] is False + assert result["reason"] == "dinner_window" + assert result["local_time"] == "18:15" + assert result["raw_kyber_response"] == "not json" + + +@pytest.mark.asyncio +async def test_handle_peace_briefing_uses_fallback_when_ai_fails(): + hass = _make_hass() + hass.data[_CI_EVENT_LOG_KEY] = deque( + [ + { + "type": "deploy_success", + "repo": "pgroene/ProspectPilot", + "branch": "main", + "status": "success", + "message": "Production deploy completed", + "local_time": "20:05", + } + ], + maxlen=50, + ) + + with patch("custom_components.kyber.ci_integration._run_kyber_prompt", new=AsyncMock(side_effect=RuntimeError("offline"))): + result = await handle_peace_briefing(hass, {"max_events": 5}, {"ai_task_entity_id": "conversation.mock"}, "user-1") + + assert result["count"] == 1 + assert "Production deploy completed" in result["briefing"] + assert result["events"][0]["type"] == "deploy_success"