From 6c1bf4d7c37b7c15fa0c1a660796e01c9597701e Mon Sep 17 00:00:00 2001 From: AndresMpa Date: Thu, 2 Jul 2026 01:24:12 +0000 Subject: [PATCH 1/3] feat(cli): expand interactive terminal flows Add native interactive menus for Dailybot actions and autocomplete support so terminal users can complete chat-style workflows without Slack buttons. Co-authored-by: Cursor --- dailybot_cli/api_client.py | 154 ++- dailybot_cli/tui/app.py | 1853 +++++++++++++++++++++++++++++++++- dailybot_cli/tui/commands.py | 71 +- dailybot_cli/tui/intents.py | 205 ++++ tests/api_client_test.py | 138 ++- tests/tui_app_flows_test.py | 359 +++++++ tests/tui_intents_test.py | 75 ++ 7 files changed, 2830 insertions(+), 25 deletions(-) create mode 100644 dailybot_cli/tui/intents.py create mode 100644 tests/tui_app_flows_test.py create mode 100644 tests/tui_intents_test.py diff --git a/dailybot_cli/api_client.py b/dailybot_cli/api_client.py index 736d420..fedf9ac 100644 --- a/dailybot_cli/api_client.py +++ b/dailybot_cli/api_client.py @@ -220,6 +220,150 @@ def complete_checkin( ) return self._handle_response(response) + def list_checkins(self) -> list[dict[str, Any]]: + """GET /v1/checkins/ — fetch visible check-ins.""" + results: list[dict[str, Any]] = [] + url: str | None = f"{self.api_url}/v1/checkins/" + pages_fetched: int = 0 + while url is not None and pages_fetched < _MAX_LIST_PAGES: + response: httpx.Response = httpx.get( + url, + headers=self._headers(), + timeout=self.timeout, + ) + if response.status_code >= 400: + self._handle_response(response) + body: Any = response.json() + if isinstance(body, dict) and "results" in body: + results.extend(body.get("results", [])) + url = body.get("next") + elif isinstance(body, list): + results.extend(body) + url = None + else: + url = None + pages_fetched += 1 + return results + + def get_checkin(self, followup_uuid: str) -> dict[str, Any]: + """GET /v1/checkins//.""" + response: httpx.Response = httpx.get( + f"{self.api_url}/v1/checkins/{followup_uuid}/", + headers=self._headers(), + timeout=self.timeout, + ) + return self._handle_response(response) + + def get_template( + self, + template_uuid: str, + *, + followup_uuid: str | None = None, + ) -> dict[str, Any]: + """GET /v1/templates// — template question definitions.""" + params: dict[str, str] = {} + if followup_uuid: + params = {"render_special_vars": "true", "followup_id": followup_uuid} + response: httpx.Response = httpx.get( + f"{self.api_url}/v1/templates/{template_uuid}/", + headers=self._headers(), + params=params, + timeout=self.timeout, + ) + return self._handle_response(response) + + def list_checkin_responses( + self, + followup_uuid: str, + *, + date_start: str | None = None, + date_end: str | None = None, + ) -> list[dict[str, Any]]: + """GET /v1/checkins//responses/.""" + params: dict[str, str] = {} + if date_start: + params["date_start"] = date_start + if date_end: + params["date_end"] = date_end + response: httpx.Response = httpx.get( + f"{self.api_url}/v1/checkins/{followup_uuid}/responses/", + headers=self._headers(), + params=params, + timeout=self.timeout, + ) + if response.status_code >= 400: + self._handle_response(response) + body: Any = response.json() + if isinstance(body, dict) and "results" in body: + return list(body.get("results", [])) + if isinstance(body, list): + return body + return [] + + def update_checkin_response( + self, + followup_uuid: str, + responses: list[dict[str, Any]], + last_question_index: int | None = None, + ) -> dict[str, Any]: + """PUT /v1/checkins//responses/ — update today's response.""" + payload: dict[str, Any] = {"responses": responses} + if last_question_index is not None: + payload["last_question_index"] = last_question_index + response: httpx.Response = httpx.put( + f"{self.api_url}/v1/checkins/{followup_uuid}/responses/", + json=payload, + headers=self._headers(), + timeout=self.timeout, + ) + return self._handle_response(response) + + def delete_checkin_response( + self, + followup_uuid: str, + *, + response_date: str | None = None, + ) -> dict[str, Any]: + """DELETE /v1/checkins//responses/ — reset a submitted response.""" + params: dict[str, str] = {} + if response_date: + params["date_start"] = response_date + params["date_end"] = response_date + response: httpx.Response = httpx.request( + "DELETE", + f"{self.api_url}/v1/checkins/{followup_uuid}/responses/", + headers=self._headers(), + params=params, + timeout=self.timeout, + ) + return self._handle_response(response) + + def get_mood(self, mood_date: str | None = None) -> dict[str, Any]: + """GET /v1/mood/track/ — fetch today's mood response.""" + params: dict[str, str] = {} + if mood_date: + params["date"] = mood_date + response: httpx.Response = httpx.get( + f"{self.api_url}/v1/mood/track/", + headers=self._headers(), + params=params, + timeout=self.timeout, + ) + return self._handle_response(response) + + def track_mood(self, score: int, mood_date: str | None = None) -> dict[str, Any]: + """POST /v1/mood/track/ — record a mood score.""" + payload: dict[str, Any] = {"score": score} + if mood_date: + payload["date"] = mood_date + response: httpx.Response = httpx.post( + f"{self.api_url}/v1/mood/track/", + json=payload, + headers=self._headers(), + timeout=self.timeout, + ) + return self._handle_response(response) + def list_forms(self, *, include_questions: bool = False) -> list[dict[str, Any]]: """GET /v1/forms/ — optionally expand question definitions per form.""" params: dict[str, str] = {"include": "questions"} if include_questions else {} @@ -378,13 +522,17 @@ def give_kudos( """POST /v1/kudos/ At least one of ``user_uuid_receivers`` or ``team_uuid_receivers`` must - be non-empty — the backend rejects an empty receiver set. + be non-empty. The public API requires the legacy ``receivers`` list for + validation and the more specific receiver lists for team expansion. """ payload: dict[str, Any] = {"content": content} + receivers: list[str] = [*(user_uuid_receivers or []), *(team_uuid_receivers or [])] + if receivers: + payload["receivers"] = receivers if user_uuid_receivers: - payload["user_uuid_receivers"] = user_uuid_receivers + payload["users_receivers"] = user_uuid_receivers if team_uuid_receivers: - payload["team_uuid_receivers"] = team_uuid_receivers + payload["teams_receivers"] = team_uuid_receivers if company_value: payload["company_value"] = company_value response: httpx.Response = httpx.post( diff --git a/dailybot_cli/tui/app.py b/dailybot_cli/tui/app.py index d8ba868..f73b3d4 100644 --- a/dailybot_cli/tui/app.py +++ b/dailybot_cli/tui/app.py @@ -1,6 +1,7 @@ """Textual application for `dailybot interactive`.""" import asyncio +from datetime import date from typing import Any, ClassVar import httpx @@ -18,15 +19,38 @@ from dailybot_cli.tui.commands import ( COMMAND_CHECKINS, COMMAND_CLEAR, + COMMAND_COMPLETIONS, + COMMAND_DASHBOARD, + COMMAND_FORM, + COMMAND_FORMS, COMMAND_HELP, + COMMAND_KUDOS, + COMMAND_MOOD, COMMAND_REPORT, COMMAND_STATUS, + COMMAND_TEAM, + COMMAND_TEAMS, + COMMAND_TIMEOFF, + COMMAND_USERS, EXIT_COMMANDS, HELP_TEXT, KNOWN_COMMANDS, + TERMINAL_COMMANDS, parse_command, + parse_command_args, ) from dailybot_cli.tui.conversation import ConversationSession +from dailybot_cli.tui.intents import ( + KudosIntent, + TerminalActionIntent, + TerminalCheckinIntent, + is_checkins_intent, + matching_teams, + matching_users, + parse_kudos_intent, + parse_terminal_action_intent, + parse_terminal_checkin_intent, +) INPUT_HISTORY_LIMIT: int = 100 @@ -120,6 +144,9 @@ def __init__(self, client: DailyBotClient) -> None: self.input_history: list[str] = [] self.input_history_index: int | None = None self.input_history_draft: str = "" + self.pending_flow: dict[str, Any] | None = None + self.completion_state: dict[str, Any] | None = None + self.mention_completion_items: list[str] | None = None def compose(self) -> ComposeResult: with Container(id="shell"): @@ -157,12 +184,31 @@ def on_key(self, event: Key) -> None: event.prevent_default() event.stop() self._show_next_input() + return + if event.key == "tab": + event.prevent_default() + event.stop() + self.run_worker( + self._complete_prompt(reverse=bool(getattr(event, "shift", False))), + exclusive=True, + ) + return + if event.key in {"shift+tab", "shift_tab", "backtab"}: + event.prevent_default() + event.stop() + self.run_worker(self._complete_prompt(reverse=True), exclusive=True) async def on_input_submitted(self, event: Input.Submitted) -> None: prompt = self.query_one("#prompt", Input) raw_value: str = event.value.strip() prompt.value = "" - if not raw_value: + if not raw_value and self.pending_flow is None: + return + + if self.pending_flow is not None: + self._remember_input(raw_value) + self._write_user(raw_value) + await self._handle_pending_flow(raw_value) return if self.report_mode: @@ -171,15 +217,40 @@ async def on_input_submitted(self, event: Input.Submitted) -> None: return self._remember_input(raw_value) + checkin_intent: TerminalCheckinIntent | None = parse_terminal_checkin_intent(raw_value) + if checkin_intent is not None: + self._write_user(raw_value) + await self._handle_terminal_checkin_intent(checkin_intent) + return + command: str | None = parse_command(raw_value) if command is not None: - await self._handle_command(command) + await self._handle_command(command, raw_value) + return + + if is_checkins_intent(raw_value): + self._write_user(raw_value) + await self._start_checkins_flow() + return + + kudos_intent: KudosIntent | None = parse_kudos_intent(raw_value) + if kudos_intent is not None: + self._write_user(raw_value) + await self._start_kudos_flow(kudos_intent) + return + + action_intent: TerminalActionIntent | None = parse_terminal_action_intent(raw_value) + if action_intent is not None: + self._write_user(raw_value) + await self._handle_terminal_action_intent(action_intent) return await self._send_turn(raw_value) def action_clear(self) -> None: self.session.clear() + self.pending_flow = None + self.report_mode = False log = self.query_one("#transcript", RichLog) log.clear() self._write_intro() @@ -200,7 +271,8 @@ def _write_intro(self) -> None: ) log.write(intro) - async def _handle_command(self, command: str) -> None: + async def _handle_command(self, command: str, raw_value: str = "") -> None: + args: list[str] = parse_command_args(raw_value) if command in EXIT_COMMANDS: self.exit() return @@ -214,7 +286,34 @@ async def _handle_command(self, command: str) -> None: await self._show_status() return if command == COMMAND_CHECKINS: - await self._show_checkins() + await self._start_checkins_flow() + return + if command == COMMAND_KUDOS: + await self._start_kudos_menu() + return + if command in {COMMAND_FORMS, COMMAND_FORM}: + await self._handle_form_command(args) + return + if command == COMMAND_USERS: + await self._show_users() + return + if command == COMMAND_TEAMS: + await self._show_teams() + return + if command == COMMAND_TEAM: + await self._start_team_detail_flow(" ".join(args)) + return + if command == COMMAND_DASHBOARD: + self._show_dashboard() + return + if command == COMMAND_MOOD: + await self._start_mood_flow() + return + if command == COMMAND_TIMEOFF: + self._write_dailybot( + "The native time-off terminal flow is not available yet. " + "Use Dailybot chat or the web app for time off until that API is exposed to the CLI." + ) return if command == COMMAND_REPORT: self.report_mode = True @@ -223,6 +322,65 @@ async def _handle_command(self, command: str) -> None: if command not in KNOWN_COMMANDS: self._write_error(f"Unknown command `/{command}`. Type `/help` for commands.") + async def _handle_terminal_checkin_intent(self, intent: TerminalCheckinIntent) -> None: + if intent.action == "complete": + await self._start_checkins_flow() + return + if intent.action == "edit": + await self._start_checkin_edit_flow() + return + if intent.action in {"reset", "delete"}: + await self._start_checkin_reset_flow() + return + self._write_error("Unsupported check-in action. Type `/checkins` to complete one.") + + async def _handle_pending_flow(self, raw_value: str) -> None: + if raw_value.lower() in {"cancel", "q", "quit", "exit"}: + self.pending_flow = None + self._write_system("Cancelled.") + self._set_prompt_hint() + return + command: str | None = parse_command(raw_value) + if command == COMMAND_CLEAR: + self.action_clear() + return + + assert self.pending_flow is not None + flow_type: str = str(self.pending_flow.get("type")) + handlers: dict[str, Any] = { + "checkin_select": self._select_checkin, + "checkin_answer": self._answer_checkin_question, + "checkin_edit_select": self._select_checkin_to_edit, + "checkin_edit_answer": self._answer_checkin_edit_question, + "checkin_reset_select": self._select_checkin_to_reset, + "checkin_reset_confirm": self._confirm_checkin_reset, + "kudos_receiver_kind": self._select_kudos_receiver_kind, + "kudos_user_select": self._select_kudos_user, + "kudos_team_select": self._select_kudos_team, + "kudos_message": self._send_kudos_message, + "kudos_value": self._set_kudos_value, + "kudos_confirm": self._confirm_kudos, + "form_select": self._select_form_action, + "form_action": self._select_form_action_choice, + "form_submit_answer": self._answer_form_submit_question, + "form_response_select": self._select_form_response, + "form_update_answer": self._answer_form_update_question, + "form_transition_select": self._select_form_transition, + "form_transition_note": self._set_form_transition_note, + "form_delete_confirm": self._confirm_form_delete, + "team_detail_select": self._select_team_detail, + "mood_select": self._select_mood, + "terminal_action_confirm": self._confirm_terminal_action, + } + handler: Any = handlers.get(flow_type) + if handler is not None: + await handler(raw_value) + return + + self.pending_flow = None + self._write_error("That terminal menu expired. Try the command again.") + self._set_prompt_hint() + async def _send_turn(self, message: str) -> None: self._write_user(message) self._set_loading(True) @@ -232,6 +390,7 @@ async def _send_turn(self, message: str) -> None: message=message, history=self.session.recent_history(), session_id=self.session.session_id, + available_commands=list(TERMINAL_COMMANDS), ) except httpx.TimeoutException: self._write_error("Dailybot took longer than expected to answer. Please try again.") @@ -253,13 +412,15 @@ async def _send_turn(self, message: str) -> None: actions: list[dict[str, Any]] = response.get("actions") or [] if actions: - action_names: str = ", ".join(str(action.get("name", "action")) for action in actions) - self._write_system(f"Suggested action: {action_names}") + await self._handle_chat_actions(actions) async def _show_status(self) -> None: self._set_loading(True) try: - data = await asyncio.to_thread(self.client.auth_status) + data, status_data = await asyncio.gather( + asyncio.to_thread(self.client.auth_status), + asyncio.to_thread(self.client.get_status), + ) except APIError as exc: self._write_error(self._format_api_error(exc)) return @@ -273,11 +434,22 @@ async def _show_status(self) -> None: ) org_raw: Any = data.get("organization", "") org_name: str = org_raw.get("name", "") if isinstance(org_raw, dict) else str(org_raw) + pending_checkins: list[dict[str, Any]] = status_data.get("pending_checkins", []) + checkin_lines: list[str] = [ + f"- {self._checkin_label(checkin)}" for checkin in pending_checkins[:5] + ] + if len(pending_checkins) > 5: + checkin_lines.append(f"- ...and {len(pending_checkins) - 5} more") + checkins_text: str = "\n".join(checkin_lines) if checkin_lines else "- None pending" self._write_dailybot( - f"**Session status**\n\n- User: {email or 'Unknown'}\n- Org: {org_name}" + "**Session status**\n\n" + f"- User: {email or 'Unknown'}\n" + f"- Org: {org_name}\n" + f"- Pending check-ins: {len(pending_checkins)}\n\n" + f"**Pending check-ins**\n{checkins_text}" ) - async def _show_checkins(self) -> None: + async def _start_checkins_flow(self) -> None: self._set_loading(True) try: data = await asyncio.to_thread(self.client.get_status) @@ -290,15 +462,1228 @@ async def _show_checkins(self) -> None: if not checkins: self._write_dailybot("No pending check-ins for today.") return - lines: list[str] = ["**Pending check-ins**"] - for checkin in checkins: - name = str(checkin.get("followup_name") or "Check-in") - question_count = len(checkin.get("template_questions", [])) - lines.append( - f"- {name} ({question_count} question{'s' if question_count != 1 else ''})" + self.pending_flow = {"type": "checkin_select", "items": checkins} + self._write_dailybot( + self._format_numbered_menu( + "Select a check-in to complete:", + checkins, + self._checkin_label, + ) + ) + self._set_prompt_hint("Type a number, or `cancel`.") + + async def _start_checkin_edit_flow(self) -> None: + self._set_loading(True) + try: + candidates = await asyncio.to_thread(self._load_editable_checkin_candidates) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + + if not candidates: + self._write_dailybot("No submitted check-ins found for today.") + return + self.pending_flow = {"type": "checkin_edit_select", "items": candidates} + self._write_dailybot( + self._format_numbered_menu( + "Select a submitted check-in to edit:", + candidates, + self._checkin_edit_label, + ) + ) + self._set_prompt_hint("Type a number, or `cancel`.") + + def _load_editable_checkin_candidates(self) -> list[dict[str, Any]]: + today: str = date.today().isoformat() + candidates: list[dict[str, Any]] = [] + for checkin in self.client.list_checkins(): + followup_uuid: str = str(checkin.get("id") or checkin.get("followup_uuid") or "") + if not followup_uuid: + continue + responses = self.client.list_checkin_responses( + followup_uuid, + date_start=today, + date_end=today, + ) + if responses: + candidates.append({"checkin": checkin, "response": responses[0]}) + return candidates + + async def _select_checkin_to_edit(self, raw_value: str) -> None: + selected: dict[str, Any] | None = self._select_numbered_item(raw_value) + if selected is None: + return + + self._set_loading(True) + try: + edit_context = await asyncio.to_thread(self._load_checkin_edit_context, selected) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + + questions: list[dict[str, Any]] = edit_context["questions"] + if not questions: + self.pending_flow = None + self._set_prompt_hint() + self._write_error("Selected check-in has no editable questions.") + return + self.pending_flow = { + "type": "checkin_edit_answer", + "checkin": edit_context["checkin"], + "questions": questions, + "existing_responses": edit_context["existing_responses"], + "answers": [], + "index": 0, + } + self._ask_current_checkin_edit_question() + + def _load_checkin_edit_context(self, selected: dict[str, Any]) -> dict[str, Any]: + checkin: dict[str, Any] = selected["checkin"] + followup_uuid: str = str(checkin.get("id") or checkin.get("followup_uuid") or "") + checkin_detail = self.client.get_checkin(followup_uuid) + template_uuid: str = str(checkin_detail.get("template") or "") + if not template_uuid: + return {"checkin": checkin_detail, "questions": [], "existing_responses": []} + template = self.client.get_template(template_uuid, followup_uuid=followup_uuid) + questions: list[dict[str, Any]] = (template.get("questions") or {}).get("fields") or [] + existing_responses: list[dict[str, Any]] = selected.get("response", {}).get("responses") or [] + return { + "checkin": checkin_detail, + "questions": questions, + "existing_responses": existing_responses, + } + + async def _select_checkin(self, raw_value: str) -> None: + selected: dict[str, Any] | None = self._select_numbered_item(raw_value) + if selected is None: + return + self._set_loading(True) + try: + questions = await asyncio.to_thread(self._load_checkin_questions, selected) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + if not questions: + self.pending_flow = None + self._set_prompt_hint() + self._write_error("Selected check-in has no questions.") + return + self.pending_flow = { + "type": "checkin_answer", + "checkin": selected, + "questions": questions, + "answers": [], + "index": 0, + } + self._ask_current_checkin_question() + + def _load_checkin_questions(self, checkin: dict[str, Any]) -> list[dict[str, Any]]: + followup_uuid: str = str(checkin.get("followup_uuid") or checkin.get("id") or "") + if not followup_uuid: + return checkin.get("template_questions") or [] + checkin_detail = self.client.get_checkin(followup_uuid) + template_uuid: str = str(checkin_detail.get("template") or "") + if not template_uuid: + return checkin.get("template_questions") or [] + template = self.client.get_template(template_uuid, followup_uuid=followup_uuid) + rendered_questions: list[dict[str, Any]] = (template.get("questions") or {}).get("fields") or [] + return rendered_questions or checkin.get("template_questions") or [] + + async def _answer_checkin_question(self, raw_value: str) -> None: + assert self.pending_flow is not None + questions: list[dict[str, Any]] = self.pending_flow["questions"] + index: int = int(self.pending_flow["index"]) + question: dict[str, Any] = questions[index] + answer: Any = self._parse_question_answer(raw_value, question) + if answer is None: + return + self.pending_flow["answers"].append(answer) + self.pending_flow["index"] = index + 1 + if self.pending_flow["index"] < len(questions): + self._ask_current_checkin_question() + return + + checkin: dict[str, Any] = self.pending_flow["checkin"] + responses: list[dict[str, Any]] = self._build_checkin_responses( + questions, + self.pending_flow["answers"], + ) + self.pending_flow = None + self._set_prompt_hint() + self._set_loading(True) + try: + await asyncio.to_thread( + self.client.complete_checkin, + str(checkin.get("followup_uuid") or ""), + responses, + len(responses) - 1, + ) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + self._write_dailybot(f"Done — submitted **{self._checkin_name(checkin)}**.") + + async def _answer_checkin_edit_question(self, raw_value: str) -> None: + assert self.pending_flow is not None + questions: list[dict[str, Any]] = self.pending_flow["questions"] + index: int = int(self.pending_flow["index"]) + question: dict[str, Any] = questions[index] + existing_value: Any = self._existing_response_value(index) + answer: Any + if raw_value == "": + answer = existing_value + else: + answer = self._parse_question_answer(raw_value, question) + if answer is None: + return + + self.pending_flow["answers"].append(answer) + self.pending_flow["index"] = index + 1 + if self.pending_flow["index"] < len(questions): + self._ask_current_checkin_edit_question() + return + + checkin: dict[str, Any] = self.pending_flow["checkin"] + responses: list[dict[str, Any]] = self._build_checkin_responses( + questions, + self.pending_flow["answers"], + ) + self.pending_flow = None + self._set_prompt_hint() + self._set_loading(True) + try: + await asyncio.to_thread( + self.client.update_checkin_response, + str(checkin.get("id") or checkin.get("followup_uuid") or ""), + responses, + len(responses) - 1, + ) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + self._write_dailybot(f"Done — updated **{self._checkin_name(checkin)}**.") + + async def _start_kudos_menu(self) -> None: + self._set_loading(True) + try: + users, teams, current_uuid = await asyncio.gather( + asyncio.to_thread(self.client.list_users), + asyncio.to_thread(self.client.list_teams), + asyncio.to_thread(self._get_current_user_uuid), + ) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + + self.pending_flow = { + "type": "kudos_receiver_kind", + "users": users, + "teams": teams, + "current_uuid": current_uuid, + } + self._write_dailybot("Who should receive kudos?\n1. One or more users\n2. One or more teams") + self._set_prompt_hint("Type 1 or 2, or `cancel`.") + + async def _start_kudos_flow(self, intent: KudosIntent) -> None: + self._set_loading(True) + try: + users, teams, current_uuid = await asyncio.gather( + asyncio.to_thread(self.client.list_users), + asyncio.to_thread(self.client.list_teams), + asyncio.to_thread(self._get_current_user_uuid), + ) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + + user_candidates: list[dict[str, Any]] = [ + user + for user in matching_users(users, intent.receiver_query) + if str(user.get("uuid") or "") != str(current_uuid or "") + ] + team_candidates: list[dict[str, Any]] = matching_teams(teams, intent.receiver_query) + if intent.receiver_kind == "team" or (not user_candidates and team_candidates): + if not team_candidates: + self._write_error(f"I couldn't find a team matching `{intent.receiver_query}`.") + return + self._prompt_kudos_team_selection(team_candidates, intent.message) + return + + if not user_candidates: + self._write_error(f"I couldn't find a teammate matching `{intent.receiver_query}`.") + return + self._prompt_kudos_user_selection(user_candidates, intent.message, current_uuid) + + async def _select_kudos_receiver_kind(self, raw_value: str) -> None: + assert self.pending_flow is not None + value: str = raw_value.strip().lower() + if value not in {"1", "2", "user", "users", "team", "teams"}: + self._write_error("Type 1 for users or 2 for teams.") + return + if value in {"1", "user", "users"}: + users: list[dict[str, Any]] = [ + user + for user in self.pending_flow.get("users", []) + if str(user.get("uuid") or "") != str(self.pending_flow.get("current_uuid") or "") + ] + self._prompt_kudos_user_selection( + users, + "", + self.pending_flow.get("current_uuid"), + ) + return + self._prompt_kudos_team_selection(self.pending_flow.get("teams", []), "") + + def _prompt_kudos_user_selection( + self, + candidates: list[dict[str, Any]], + message: str, + current_uuid: str | None, + ) -> None: + if not candidates: + self.pending_flow = None + self._set_prompt_hint() + self._write_error("No teammates are available for kudos.") + return + self.pending_flow = { + "type": "kudos_user_select", + "items": candidates, + "message": message, + "current_uuid": current_uuid, + } + self._write_dailybot( + self._format_numbered_menu( + "Who should receive kudos? You can type multiple numbers like `1,3`.", + candidates, + self._user_label, + ) + ) + self._set_prompt_hint("Type one or more numbers, or `cancel`.") + + def _prompt_kudos_team_selection( + self, + candidates: list[dict[str, Any]], + message: str, + ) -> None: + if not candidates: + self.pending_flow = None + self._set_prompt_hint() + self._write_error("No teams are visible to you.") + return + self.pending_flow = { + "type": "kudos_team_select", + "items": candidates, + "message": message, + } + self._write_dailybot( + self._format_numbered_menu( + "Which team should receive kudos? You can type multiple numbers like `1,3`.", + candidates, + self._team_label, + ) + ) + self._set_prompt_hint("Type one or more numbers, or `cancel`.") + + async def _select_kudos_user(self, raw_value: str) -> None: + selected: list[dict[str, Any]] | None = self._select_numbered_items(raw_value) + if selected is None: + return + assert self.pending_flow is not None + message: str = str(self.pending_flow.get("message") or "").strip() + current_uuid: str | None = self.pending_flow.get("current_uuid") + selected = [ + user for user in selected if str(user.get("uuid") or "") != str(current_uuid or "") + ] + if not selected: + self._write_error("You cannot give kudos to yourself.") + return + if not message: + self.pending_flow = { + "type": "kudos_message", + "users": selected, + "teams": [], + "current_uuid": current_uuid, + } + self._write_dailybot( + f"What should the kudos message say for **{self._join_labels(selected, self._user_label)}**?" + ) + self._set_prompt_hint("Type the kudos message, or `cancel`.") + return + self._ask_kudos_value(selected, [], message, current_uuid=current_uuid) + + async def _select_kudos_team(self, raw_value: str) -> None: + selected: list[dict[str, Any]] | None = self._select_numbered_items(raw_value) + if selected is None: + return + assert self.pending_flow is not None + message: str = str(self.pending_flow.get("message") or "").strip() + if not message: + self.pending_flow = { + "type": "kudos_message", + "users": [], + "teams": selected, + } + self._write_dailybot( + f"What should the kudos message say for **{self._join_labels(selected, self._team_label)}**?" + ) + self._set_prompt_hint("Type the kudos message, or `cancel`.") + return + self._ask_kudos_value([], selected, message, current_uuid=None) + + async def _send_kudos_message(self, raw_value: str) -> None: + message: str = raw_value.strip() + if not message: + self._write_error("Empty kudos message. Type a message or `cancel`.") + return + assert self.pending_flow is not None + self._ask_kudos_value( + self.pending_flow.get("users") or [], + self.pending_flow.get("teams") or [], + message, + current_uuid=self.pending_flow.get("current_uuid"), + ) + + def _ask_kudos_value( + self, + users: list[dict[str, Any]], + teams: list[dict[str, Any]], + message: str, + *, + current_uuid: str | None = None, + ) -> None: + self.pending_flow = { + "type": "kudos_value", + "users": users, + "teams": teams, + "message": message, + "current_uuid": current_uuid, + } + self._write_dailybot("Optional: type a company value for this kudos, or press Enter to skip.") + self._set_prompt_hint("Type a value, press Enter to skip, or `cancel`.") + + async def _set_kudos_value(self, raw_value: str) -> None: + assert self.pending_flow is not None + self.pending_flow["company_value"] = raw_value.strip() or None + self.pending_flow["type"] = "kudos_confirm" + self._write_dailybot(self._kudos_confirmation_text(self.pending_flow)) + self._set_prompt_hint("Type 1 to send, 2 to cancel.") + + async def _confirm_kudos(self, raw_value: str) -> None: + assert self.pending_flow is not None + normalized: str = raw_value.strip().lower() + if normalized not in {"1", "yes", "y", "send", "2", "no", "n"}: + self._write_error("Type 1 to send or 2 to cancel.") + return + if normalized in {"2", "no", "n"}: + self.pending_flow = None + self._set_prompt_hint() + self._write_system("Kudos cancelled.") + return + users: list[dict[str, Any]] = self.pending_flow.get("users") or [] + teams: list[dict[str, Any]] = self.pending_flow.get("teams") or [] + message: str = str(self.pending_flow.get("message") or "") + company_value: str | None = self.pending_flow.get("company_value") + user_uuids: list[str] = [str(user.get("uuid") or "") for user in users if user.get("uuid")] + team_uuids: list[str] = [str(team.get("uuid") or team.get("id") or "") for team in teams if team.get("uuid") or team.get("id")] + self.pending_flow = None + self._set_prompt_hint() + self._set_loading(True) + try: + await asyncio.to_thread( + self.client.give_kudos, + content=message, + user_uuid_receivers=user_uuids or None, + team_uuid_receivers=team_uuids or None, + company_value=company_value, + ) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + receiver_labels: list[str] = [ + *(self._user_label(user) for user in users), + *(self._team_label(team) for team in teams), + ] + self._write_dailybot( + f"Done — kudos sent to **{', '.join(receiver_labels)}** for {message}." + ) + + async def _handle_form_command(self, args: list[str]) -> None: + action: str | None = args[0].lower() if args else None + aliases: dict[str, str] = { + "list": "list", + "submit": "submit", + "responses": "responses", + "response": "responses", + "update": "update", + "edit": "update", + "transition": "transition", + "delete": "delete", + "remove": "delete", + } + if action is not None and action not in aliases: + self._write_error("Unknown form action. Type `/help` for form commands.") + return + await self._start_forms_flow(aliases.get(action) if action else None) + + async def _start_forms_flow(self, action: str | None = None) -> None: + self._set_loading(True) + try: + forms = await asyncio.to_thread(self.client.list_forms, include_questions=True) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + + if not forms: + self._write_dailybot("No forms are visible to you.") + return + if action == "list": + self._write_dailybot( + self._format_numbered_menu("Visible forms:", forms, self._form_label) + ) + return + self.pending_flow = {"type": "form_select", "items": forms, "action": action} + heading: str = "Select a form:" + if action: + heading = f"Select a form to {action}:" + self._write_dailybot(self._format_numbered_menu(heading, forms, self._form_label)) + self._set_prompt_hint("Type a number, or `cancel`.") + + async def _select_form_action(self, raw_value: str) -> None: + selected: dict[str, Any] | None = self._select_numbered_item(raw_value) + if selected is None: + return + assert self.pending_flow is not None + action: str | None = self.pending_flow.get("action") + if action: + await self._prepare_form_action(selected, action) + return + self.pending_flow = { + "type": "form_action", + "form": selected, + "items": [ + {"action": "submit", "label": "Submit a response"}, + {"action": "responses", "label": "Browse responses"}, + {"action": "update", "label": "Update a response"}, + {"action": "transition", "label": "Transition a workflow response"}, + {"action": "delete", "label": "Delete a response"}, + ], + } + self._write_dailybot( + self._format_numbered_menu( + f"What do you want to do with **{self._form_label(selected)}**?", + self.pending_flow["items"], + lambda item: str(item["label"]), + ) + ) + self._set_prompt_hint("Type a number, or `cancel`.") + + async def _select_form_action_choice(self, raw_value: str) -> None: + selected: dict[str, Any] | None = self._select_numbered_item(raw_value) + if selected is None: + return + assert self.pending_flow is not None + await self._prepare_form_action(self.pending_flow["form"], str(selected["action"])) + + async def _prepare_form_action(self, form: dict[str, Any], action: str) -> None: + if action == "submit": + await self._start_form_submit_flow(form) + return + if action == "responses": + await self._start_form_response_picker(form, purpose="view") + return + if action == "update": + await self._start_form_response_picker(form, purpose="update") + return + if action == "transition": + await self._start_form_response_picker(form, purpose="transition") + return + if action == "delete": + await self._start_form_response_picker(form, purpose="delete") + return + self._write_error("Unsupported form action.") + + async def _start_form_submit_flow(self, form: dict[str, Any]) -> None: + form_data: dict[str, Any] | None = await self._load_form_detail(form) + if form_data is None: + return + questions: list[dict[str, Any]] = self._form_questions(form_data) + if not questions: + self._write_error("Selected form has no answerable questions.") + return + self.pending_flow = { + "type": "form_submit_answer", + "form": form_data, + "questions": questions, + "answers": [], + "index": 0, + } + self._ask_current_form_question() + + async def _answer_form_submit_question(self, raw_value: str) -> None: + assert self.pending_flow is not None + questions: list[dict[str, Any]] = self.pending_flow["questions"] + index: int = int(self.pending_flow["index"]) + question: dict[str, Any] = questions[index] + answer: Any = self._parse_question_answer(raw_value, question) + if answer is None: + return + self.pending_flow["answers"].append(answer) + self.pending_flow["index"] = index + 1 + if self.pending_flow["index"] < len(questions): + self._ask_current_form_question() + return + + form: dict[str, Any] = self.pending_flow["form"] + content: dict[str, Any] = self._build_form_content(questions, self.pending_flow["answers"]) + self.pending_flow = None + self._set_prompt_hint() + self._set_loading(True) + try: + await asyncio.to_thread( + self.client.submit_form_response, + str(form.get("id") or form.get("uuid") or ""), + content, ) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + self._write_dailybot(f"Done — submitted **{self._form_label(form)}**.") + + async def _start_form_response_picker(self, form: dict[str, Any], *, purpose: str) -> None: + form_uuid: str = str(form.get("id") or form.get("uuid") or "") + self._set_loading(True) + try: + form_data, responses = await asyncio.gather( + asyncio.to_thread(self.client.get_form, form_uuid), + asyncio.to_thread(self.client.list_form_responses, form_uuid), + ) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + if not responses: + self._write_dailybot(f"No responses found for **{self._form_label(form_data)}**.") + return + self.pending_flow = { + "type": "form_response_select", + "form": form_data, + "items": responses, + "purpose": purpose, + } + self._write_dailybot( + self._format_numbered_menu( + f"Select a response to {purpose}:", + responses, + self._form_response_label, + ) + ) + self._set_prompt_hint("Type a number, or `cancel`.") + + async def _select_form_response(self, raw_value: str) -> None: + selected: dict[str, Any] | None = self._select_numbered_item(raw_value) + if selected is None: + return + assert self.pending_flow is not None + form: dict[str, Any] = self.pending_flow["form"] + purpose: str = str(self.pending_flow.get("purpose") or "view") + response_uuid: str = str(selected.get("id") or selected.get("uuid") or "") + form_uuid: str = str(form.get("id") or form.get("uuid") or "") + self._set_loading(True) + try: + response = await asyncio.to_thread( + self.client.get_form_response, + form_uuid, + response_uuid, + ) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + + if purpose == "view": + self.pending_flow = None + self._set_prompt_hint() + self._write_dailybot(self._form_response_detail_text(response, form)) + return + if purpose == "update": + await self._start_form_update_flow(form, response) + return + if purpose == "transition": + self._start_form_transition_flow(form, response) + return + if purpose == "delete": + self._start_form_delete_flow(form, response) + return + + async def _start_form_update_flow( + self, + form: dict[str, Any], + response: dict[str, Any], + ) -> None: + questions: list[dict[str, Any]] = self._form_questions(form) + if not questions: + self.pending_flow = None + self._set_prompt_hint() + self._write_error("Selected form has no editable questions.") + return + self.pending_flow = { + "type": "form_update_answer", + "form": form, + "response": response, + "questions": questions, + "answers": [], + "index": 0, + } + self._ask_current_form_update_question() + + async def _answer_form_update_question(self, raw_value: str) -> None: + assert self.pending_flow is not None + questions: list[dict[str, Any]] = self.pending_flow["questions"] + index: int = int(self.pending_flow["index"]) + question: dict[str, Any] = questions[index] + existing_value: Any = self._existing_form_response_value(question) + answer: Any = existing_value if raw_value == "" else self._parse_question_answer(raw_value, question) + if answer is None: + return + self.pending_flow["answers"].append(answer) + self.pending_flow["index"] = index + 1 + if self.pending_flow["index"] < len(questions): + self._ask_current_form_update_question() + return + + form: dict[str, Any] = self.pending_flow["form"] + response: dict[str, Any] = self.pending_flow["response"] + content: dict[str, Any] = self._build_form_content(questions, self.pending_flow["answers"]) + self.pending_flow = None + self._set_prompt_hint() + self._set_loading(True) + try: + await asyncio.to_thread( + self.client.update_form_response, + str(form.get("id") or form.get("uuid") or ""), + str(response.get("id") or response.get("uuid") or ""), + content, + ) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + self._write_dailybot(f"Done — updated response **{self._form_response_id(response)}**.") + + def _start_form_transition_flow(self, form: dict[str, Any], response: dict[str, Any]) -> None: + transitions: list[dict[str, Any]] = list(response.get("allowed_transitions") or []) + if not transitions: + self.pending_flow = None + self._set_prompt_hint() + self._write_dailybot("This response has no allowed workflow transitions.") + return + self.pending_flow = { + "type": "form_transition_select", + "form": form, + "response": response, + "items": transitions, + } + self._write_dailybot( + self._format_numbered_menu( + f"Select the next state for response **{self._form_response_id(response)}**:", + transitions, + self._transition_label, + ) + ) + self._set_prompt_hint("Type a number, or `cancel`.") + + async def _select_form_transition(self, raw_value: str) -> None: + selected: dict[str, Any] | None = self._select_numbered_item(raw_value) + if selected is None: + return + assert self.pending_flow is not None + self.pending_flow["transition"] = selected + self.pending_flow["type"] = "form_transition_note" + self._write_dailybot("Optional: type a transition note, or press Enter to skip.") + self._set_prompt_hint("Type a note, press Enter to skip, or `cancel`.") + + async def _set_form_transition_note(self, raw_value: str) -> None: + assert self.pending_flow is not None + form: dict[str, Any] = self.pending_flow["form"] + response: dict[str, Any] = self.pending_flow["response"] + transition: dict[str, Any] = self.pending_flow["transition"] + note: str | None = raw_value.strip() or None + to_state: str = str(transition.get("to_state") or transition.get("key") or "") + self.pending_flow = None + self._set_prompt_hint() + self._set_loading(True) + try: + await asyncio.to_thread( + self.client.transition_form_response, + str(form.get("id") or form.get("uuid") or ""), + str(response.get("id") or response.get("uuid") or ""), + to_state, + note, + ) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + self._write_dailybot( + f"Done — moved response **{self._form_response_id(response)}** to **{self._transition_label(transition)}**." + ) + + def _start_form_delete_flow(self, form: dict[str, Any], response: dict[str, Any]) -> None: + self.pending_flow = { + "type": "form_delete_confirm", + "form": form, + "response": response, + } + self._write_dailybot( + f"Delete response **{self._form_response_id(response)}** from **{self._form_label(form)}**?\n" + "1. Delete\n2. Cancel" + ) + self._set_prompt_hint("Type 1 to delete, 2 to cancel.") + + async def _confirm_form_delete(self, raw_value: str) -> None: + assert self.pending_flow is not None + normalized: str = raw_value.strip().lower() + if normalized not in {"1", "yes", "y", "delete", "2", "no", "n"}: + self._write_error("Type 1 to delete or 2 to cancel.") + return + if normalized in {"2", "no", "n"}: + self.pending_flow = None + self._set_prompt_hint() + self._write_system("Delete cancelled.") + return + form: dict[str, Any] = self.pending_flow["form"] + response: dict[str, Any] = self.pending_flow["response"] + form_uuid: str = str(form.get("id") or form.get("uuid") or "") + response_uuid: str = str(response.get("id") or response.get("uuid") or "") + self.pending_flow = None + self._set_prompt_hint() + self._set_loading(True) + try: + await asyncio.to_thread(self.client.delete_form_response, form_uuid, response_uuid) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + self._write_dailybot(f"Done — deleted response **{response_uuid}**.") + + async def _load_form_detail(self, form: dict[str, Any]) -> dict[str, Any] | None: + form_uuid: str = str(form.get("id") or form.get("uuid") or "") + if not form_uuid: + self._write_error("Selected form has no UUID.") + return None + if self._form_questions(form): + return form + self._set_loading(True) + try: + return await asyncio.to_thread(self.client.get_form, form_uuid) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return None + finally: + self._set_loading(False) + + def _ask_current_form_question(self) -> None: + assert self.pending_flow is not None + questions: list[dict[str, Any]] = self.pending_flow["questions"] + index: int = int(self.pending_flow["index"]) + question: dict[str, Any] = questions[index] + lines: list[str] = [f"Question {index + 1}/{len(questions)}: {self._question_label(question, index)}"] + choices: list[str] = self._question_choices(question) + if self._question_type(question) == "boolean": + lines.extend(["1. Yes", "2. No"]) + self._set_prompt_hint("Type 1 for yes, 2 for no, or `cancel`.") + elif choices: + for choice_index, choice in enumerate(choices, start=1): + lines.append(f"{choice_index}. {choice}") + self._set_prompt_hint("Type a number, or `cancel`.") + else: + self._set_prompt_hint("Type your answer, or `cancel`.") + self._write_dailybot("\n".join(lines)) + + def _ask_current_form_update_question(self) -> None: + assert self.pending_flow is not None + questions: list[dict[str, Any]] = self.pending_flow["questions"] + index: int = int(self.pending_flow["index"]) + question: dict[str, Any] = questions[index] + existing_value: Any = self._existing_form_response_value(question) + lines: list[str] = [ + f"Question {index + 1}/{len(questions)}: {self._question_label(question, index)}", + f"Current answer: {self._format_answer(existing_value)}", + ] + choices: list[str] = self._question_choices(question) + if self._question_type(question) == "boolean": + lines.extend(["1. Yes", "2. No"]) + self._set_prompt_hint("Type 1/2, Enter to keep, or `cancel`.") + elif choices: + for choice_index, choice in enumerate(choices, start=1): + lines.append(f"{choice_index}. {choice}") + self._set_prompt_hint("Type a number, Enter to keep, or `cancel`.") + else: + self._set_prompt_hint("Type a replacement, Enter to keep, or `cancel`.") self._write_dailybot("\n".join(lines)) + def _existing_form_response_value(self, question: dict[str, Any]) -> Any: + assert self.pending_flow is not None + response: dict[str, Any] = self.pending_flow.get("response") or {} + content: dict[str, Any] = response.get("content") if isinstance(response.get("content"), dict) else {} + question_uuid: str = str(question.get("uuid") or question.get("id") or "") + return content.get(question_uuid, "") + + @staticmethod + def _form_questions(form: dict[str, Any]) -> list[dict[str, Any]]: + questions: Any = form.get("questions") or [] + if isinstance(questions, dict): + fields: Any = questions.get("fields") or [] + return list(fields) if isinstance(fields, list) else [] + return list(questions) if isinstance(questions, list) else [] + + @staticmethod + def _build_form_content( + questions: list[dict[str, Any]], + answers: list[Any], + ) -> dict[str, Any]: + content: dict[str, Any] = {} + for question, answer in zip(questions, answers, strict=True): + question_uuid: str = str(question.get("uuid") or question.get("id") or "") + if question_uuid: + content[question_uuid] = answer + return content + + @classmethod + def _form_label(cls, form: dict[str, Any]) -> str: + questions_count: int = len(cls._form_questions(form)) + suffix: str = f" ({questions_count} question{'s' if questions_count != 1 else ''})" + return str(form.get("name") or form.get("title") or form.get("uuid") or form.get("id") or "Form") + suffix + + @classmethod + def _form_response_label(cls, response: dict[str, Any]) -> str: + response_id: str = cls._form_response_id(response) + state: str = str(response.get("current_state") or "no state") + created_at: str = str(response.get("created_at") or response.get("updated_at") or "") + suffix: str = f" — {created_at[:10]}" if created_at else "" + return f"{response_id} ({state}){suffix}" + + @staticmethod + def _form_response_id(response: dict[str, Any]) -> str: + return str(response.get("id") or response.get("uuid") or "response") + + @staticmethod + def _transition_label(transition: dict[str, Any]) -> str: + return str( + transition.get("label") + or transition.get("to_state") + or transition.get("key") + or "Next state" + ) + + def _form_response_detail_text( + self, + response: dict[str, Any], + form: dict[str, Any], + ) -> str: + lines: list[str] = [ + f"**Response {self._form_response_id(response)}**", + f"- Form: {self._form_label(form)}", + f"- Current state: {response.get('current_state') or 'None'}", + ] + content: Any = response.get("content") + if isinstance(content, dict) and content: + lines.append("\n**Answers**") + for question_uuid, answer in content.items(): + lines.append(f"- {question_uuid}: {self._format_answer(answer)}") + transitions: list[dict[str, Any]] = list(response.get("allowed_transitions") or []) + if transitions: + lines.append("\n**Allowed transitions**") + for transition in transitions: + lines.append(f"- {self._transition_label(transition)}") + return "\n".join(lines) + + async def _show_users(self) -> None: + self._set_loading(True) + try: + users = await asyncio.to_thread(self.client.list_users) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + if not users: + self._write_dailybot("No users are visible to you.") + return + self._write_dailybot( + self._format_numbered_menu( + f"Organization members ({len(users)}):", + users[:25], + self._user_label, + ) + ) + + async def _show_teams(self) -> None: + self._set_loading(True) + try: + teams = await asyncio.to_thread(self.client.list_teams) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + if not teams: + self._write_dailybot("No teams are visible to you.") + return + self._write_dailybot( + self._format_numbered_menu(f"Teams ({len(teams)}):", teams[:25], self._team_label) + ) + + async def _start_team_detail_flow(self, query: str = "") -> None: + self._set_loading(True) + try: + teams = await asyncio.to_thread(self.client.list_teams) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + candidates: list[dict[str, Any]] = matching_teams(teams, query) if query else teams + if not candidates: + self._write_error(f"No team matching `{query}` is visible to you.") + return + self.pending_flow = {"type": "team_detail_select", "items": candidates} + self._write_dailybot( + self._format_numbered_menu("Select a team to inspect:", candidates, self._team_label) + ) + self._set_prompt_hint("Type a number, or `cancel`.") + + async def _select_team_detail(self, raw_value: str) -> None: + selected: dict[str, Any] | None = self._select_numbered_item(raw_value) + if selected is None: + return + team_uuid: str = str(selected.get("uuid") or selected.get("id") or "") + self.pending_flow = None + self._set_prompt_hint() + self._set_loading(True) + try: + team, members = await asyncio.gather( + asyncio.to_thread(self.client.get_team, team_uuid), + asyncio.to_thread(self.client.list_team_members, team_uuid), + ) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + member_lines: list[str] = [f"- {self._user_label(member)}" for member in members[:25]] + members_text: str = "\n".join(member_lines) if member_lines else "- No members returned" + self._write_dailybot( + f"**{self._team_label(team)}**\n\n" + f"- UUID: {team_uuid}\n" + f"- Members: {len(members)}\n\n" + f"**Members**\n{members_text}" + ) + + def _show_dashboard(self) -> None: + self._write_dailybot( + f"Open your Dailybot dashboard: {self.client.api_url.replace('/api', '').rstrip('/')}/home" + ) + + async def _start_mood_flow(self) -> None: + self.pending_flow = {"type": "mood_select"} + self._write_dailybot( + "How is your mood today?\n" + "1. Very low\n" + "2. Low\n" + "3. Okay\n" + "4. Good\n" + "5. Great" + ) + self._set_prompt_hint("Type a number from 1 to 5, or `cancel`.") + + async def _select_mood(self, raw_value: str) -> None: + try: + score: int = int(raw_value.strip()) + except ValueError: + self._write_error("Type a number from 1 to 5.") + return + if score < 1 or score > 5: + self._write_error("Type a number from 1 to 5.") + return + self.pending_flow = None + self._set_prompt_hint() + self._set_loading(True) + try: + await asyncio.to_thread(self.client.track_mood, score) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + self._write_dailybot("Done — your mood was tracked for today.") + + async def _start_checkin_reset_flow(self) -> None: + self._set_loading(True) + try: + candidates = await asyncio.to_thread(self._load_editable_checkin_candidates) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + if not candidates: + self._write_dailybot("No submitted check-ins found for today.") + return + self.pending_flow = {"type": "checkin_reset_select", "items": candidates} + self._write_dailybot( + self._format_numbered_menu( + "Select a submitted check-in to delete:", + candidates, + self._checkin_edit_label, + ) + ) + self._set_prompt_hint("Type a number, or `cancel`.") + + async def _select_checkin_to_reset(self, raw_value: str) -> None: + selected: dict[str, Any] | None = self._select_numbered_item(raw_value) + if selected is None: + return + self.pending_flow = {"type": "checkin_reset_confirm", "candidate": selected} + self._write_dailybot( + f"Delete today's response for **{self._checkin_edit_label(selected)}**?\n" + "1. Delete\n2. Cancel" + ) + self._set_prompt_hint("Type 1 to delete, 2 to cancel.") + + async def _confirm_checkin_reset(self, raw_value: str) -> None: + assert self.pending_flow is not None + normalized: str = raw_value.strip().lower() + if normalized not in {"1", "yes", "y", "delete", "2", "no", "n"}: + self._write_error("Type 1 to delete or 2 to cancel.") + return + if normalized in {"2", "no", "n"}: + self.pending_flow = None + self._set_prompt_hint() + self._write_system("Check-in delete cancelled.") + return + candidate: dict[str, Any] = self.pending_flow["candidate"] + checkin: dict[str, Any] = candidate.get("checkin") or {} + followup_uuid: str = str(checkin.get("id") or checkin.get("followup_uuid") or "") + self.pending_flow = None + self._set_prompt_hint() + self._set_loading(True) + try: + await asyncio.to_thread( + self.client.delete_checkin_response, + followup_uuid, + response_date=date.today().isoformat(), + ) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + return + finally: + self._set_loading(False) + self._write_dailybot(f"Done — deleted today's response for **{self._checkin_name(checkin)}**.") + + async def _handle_terminal_action_intent(self, intent: TerminalActionIntent) -> None: + await self._start_terminal_flow_by_name(intent.action, {"args": list(intent.args)}) + + async def _handle_chat_actions(self, actions: list[dict[str, Any]]) -> None: + for action in actions: + action_type: str = str(action.get("type") or "") + flow_name: str = str(action.get("flow") or action.get("name") or "") + if action_type == "terminal_flow" and flow_name: + self.pending_flow = { + "type": "terminal_action_confirm", + "flow": flow_name, + "params": action.get("params") or {}, + } + self._write_dailybot( + f"Start the **{flow_name}** terminal flow?\n1. Yes\n2. No" + ) + self._set_prompt_hint("Type 1 to start, 2 to skip.") + return + action_names: str = ", ".join(str(action.get("name", "action")) for action in actions) + self._write_system(f"Suggested action: {action_names}") + + async def _confirm_terminal_action(self, raw_value: str) -> None: + assert self.pending_flow is not None + normalized: str = raw_value.strip().lower() + if normalized not in {"1", "yes", "y", "2", "no", "n"}: + self._write_error("Type 1 to start or 2 to skip.") + return + flow_name: str = str(self.pending_flow.get("flow") or "") + params: dict[str, Any] = self.pending_flow.get("params") or {} + self.pending_flow = None + self._set_prompt_hint() + if normalized in {"2", "no", "n"}: + self._write_system("Skipped suggested action.") + return + await self._start_terminal_flow_by_name(flow_name, params) + + async def _start_terminal_flow_by_name(self, flow_name: str, params: dict[str, Any]) -> None: + normalized: str = flow_name.strip().lower().replace("-", "_") + if normalized in {"checkins", "checkin", "complete_checkin", "form_checkin"}: + await self._start_checkins_flow() + return + if normalized in {"checkin_edit", "edit_checkin"}: + await self._start_checkin_edit_flow() + return + if normalized in {"checkin_reset", "reset_checkin", "checkin_delete"}: + await self._start_checkin_reset_flow() + return + if normalized in {"kudos", "send_kudos"}: + receiver_query: str = str(params.get("receiver_query") or params.get("receivers") or "") + message: str = str(params.get("message") or "") + if receiver_query: + await self._start_kudos_flow(KudosIntent(receiver_query=receiver_query, message=message)) + else: + await self._start_kudos_menu() + return + if normalized in {"forms", "form"}: + await self._start_forms_flow() + return + if normalized.startswith("form_"): + await self._start_forms_flow(normalized.removeprefix("form_")) + return + if normalized == "users": + await self._show_users() + return + if normalized == "teams": + await self._show_teams() + return + if normalized == "team": + await self._start_team_detail_flow(str(params.get("query") or "")) + return + if normalized == "dashboard": + self._show_dashboard() + return + if normalized == "mood": + await self._start_mood_flow() + return + self._write_dailybot("That suggested action is not supported in the terminal yet.") + async def _submit_report(self, message: str) -> None: if message.startswith("/"): command = parse_command(message) @@ -319,6 +1704,286 @@ async def _submit_report(self, message: str) -> None: count = int(result.get("followups_count", 0)) self._write_dailybot(f"Progress update submitted to {count} check-in(s).") + def _select_numbered_item(self, raw_value: str) -> dict[str, Any] | None: + assert self.pending_flow is not None + items: list[dict[str, Any]] = self.pending_flow.get("items") or [] + try: + selected_index: int = int(raw_value.strip()) + except ValueError: + self._write_error(f"Enter a number between 1 and {len(items)}, or `cancel`.") + return None + if selected_index < 1 or selected_index > len(items): + self._write_error(f"Enter a number between 1 and {len(items)}, or `cancel`.") + return None + return items[selected_index - 1] + + def _select_numbered_items(self, raw_value: str) -> list[dict[str, Any]] | None: + assert self.pending_flow is not None + items: list[dict[str, Any]] = self.pending_flow.get("items") or [] + raw_parts: list[str] = [ + part.strip() for part in raw_value.replace(" ", ",").split(",") if part.strip() + ] + if not raw_parts: + self._write_error(f"Enter a number between 1 and {len(items)}, or `cancel`.") + return None + selected: list[dict[str, Any]] = [] + for raw_part in raw_parts: + try: + selected_index: int = int(raw_part) + except ValueError: + self._write_error(f"Enter numbers between 1 and {len(items)}, separated by commas.") + return None + if selected_index < 1 or selected_index > len(items): + self._write_error(f"Enter numbers between 1 and {len(items)}, separated by commas.") + return None + item: dict[str, Any] = items[selected_index - 1] + if item not in selected: + selected.append(item) + return selected + + def _ask_current_checkin_question(self) -> None: + assert self.pending_flow is not None + questions: list[dict[str, Any]] = self.pending_flow["questions"] + index: int = int(self.pending_flow["index"]) + question: dict[str, Any] = questions[index] + prompt: str = self._question_label(question, index) + choices: list[str] = self._question_choices(question) + lines: list[str] = [f"Question {index + 1}/{len(questions)}: {prompt}"] + if self._question_type(question) == "boolean": + lines.append("1. Yes") + lines.append("2. No") + self._set_prompt_hint("Type 1 for yes, 2 for no, or `cancel`.") + elif choices: + for choice_index, choice in enumerate(choices, start=1): + lines.append(f"{choice_index}. {choice}") + self._set_prompt_hint("Type a number, or `cancel`.") + else: + self._set_prompt_hint("Type your answer, or `cancel`.") + self._write_dailybot("\n".join(lines)) + + def _ask_current_checkin_edit_question(self) -> None: + assert self.pending_flow is not None + questions: list[dict[str, Any]] = self.pending_flow["questions"] + index: int = int(self.pending_flow["index"]) + question: dict[str, Any] = questions[index] + prompt: str = self._question_label(question, index) + choices: list[str] = self._question_choices(question) + existing_value: Any = self._existing_response_value(index) + existing_label: str = self._format_answer(existing_value) + lines: list[str] = [ + f"Question {index + 1}/{len(questions)}: {prompt}", + f"Current answer: {existing_label}", + ] + if self._question_type(question) == "boolean": + lines.append("1. Yes") + lines.append("2. No") + self._set_prompt_hint("Type 1/2, Enter to keep, or `cancel`.") + elif choices: + for choice_index, choice in enumerate(choices, start=1): + lines.append(f"{choice_index}. {choice}") + self._set_prompt_hint("Type a number, Enter to keep, or `cancel`.") + else: + self._set_prompt_hint("Type a replacement, Enter to keep, or `cancel`.") + self._write_dailybot("\n".join(lines)) + + def _existing_response_value(self, index: int) -> Any: + assert self.pending_flow is not None + existing_responses: list[dict[str, Any]] = self.pending_flow.get("existing_responses") or [] + if index < len(existing_responses): + return existing_responses[index].get("response") + return "" + + @staticmethod + def _format_answer(value: Any) -> str: + if value is None or value == "" or value == {} or value == []: + return "_empty_" + if isinstance(value, bool): + return "Yes" if value else "No" + if isinstance(value, dict): + for key in ( + "response", + "value", + "text", + "answer", + "label", + "name", + "title", + "content", + ): + nested_value: Any = value.get(key) + if nested_value not in (None, "", {}, []): + return DailybotChatApp._format_answer(nested_value) + parts: list[str] = [] + for key, nested_value in value.items(): + formatted_value: str = DailybotChatApp._format_answer(nested_value) + if formatted_value != "_empty_": + parts.append(f"{key}: {formatted_value}") + return ", ".join(parts) if parts else "_empty_" + if isinstance(value, list): + parts = [ + DailybotChatApp._format_answer(item) + for item in value + if item not in (None, "", {}, []) + ] + return ", ".join(part for part in parts if part != "_empty_") or "_empty_" + return str(value) + + def _parse_question_answer(self, raw_value: str, question: dict[str, Any]) -> Any: + value: str = raw_value.strip() + question_type: str = self._question_type(question) + choices: list[str] = self._question_choices(question) + + if question_type == "boolean": + normalized: str = value.lower() + if normalized in {"1", "yes", "y", "true"}: + return True + if normalized in {"2", "no", "n", "false"}: + return False + self._write_error("Type 1 for yes or 2 for no.") + return None + + if choices: + try: + selected_index: int = int(value) + except ValueError: + self._write_error(f"Enter a number between 1 and {len(choices)}.") + return None + if selected_index < 1 or selected_index > len(choices): + self._write_error(f"Enter a number between 1 and {len(choices)}.") + return None + return choices[selected_index - 1] + + if question_type == "numeric": + try: + return float(value) if "." in value else int(value) + except ValueError: + self._write_error("Enter a valid number.") + return None + + return value + + @staticmethod + def _build_checkin_responses( + questions: list[dict[str, Any]], + answers: list[Any], + ) -> list[dict[str, Any]]: + responses: list[dict[str, Any]] = [] + for index, (question, answer) in enumerate(zip(questions, answers, strict=True)): + question_uuid: str = str(question.get("uuid") or question.get("id") or "") + responses.append( + { + "uuid": question_uuid, + "index": index, + "response": answer, + } + ) + return responses + + @staticmethod + def _format_numbered_menu( + heading: str, + items: list[dict[str, Any]], + label_fn: Any, + ) -> str: + lines: list[str] = [heading] + for index, item in enumerate(items, start=1): + lines.append(f"{index}. {label_fn(item)}") + return "\n".join(lines) + + @classmethod + def _checkin_label(cls, checkin: dict[str, Any]) -> str: + question_count: int = len(checkin.get("template_questions", [])) + return ( + f"{cls._checkin_name(checkin)} " + f"({question_count} question{'s' if question_count != 1 else ''})" + ) + + @classmethod + def _checkin_edit_label(cls, candidate: dict[str, Any]) -> str: + response: dict[str, Any] = candidate.get("response") or {} + updated_at: str = str(response.get("updated_at") or response.get("created_at") or "") + suffix: str = f" — {updated_at[:10]}" if updated_at else "" + return f"{cls._checkin_name(candidate.get('checkin') or {})}{suffix}" + + @staticmethod + def _checkin_name(checkin: dict[str, Any]) -> str: + return str(checkin.get("followup_name") or checkin.get("name") or "Check-in") + + @staticmethod + def _user_label(user: dict[str, Any]) -> str: + return str( + user.get("full_name") + or user.get("name") + or user.get("display_name") + or user.get("email") + or user.get("uuid") + or "Unknown user" + ) + + @staticmethod + def _team_label(team: dict[str, Any]) -> str: + return str(team.get("name") or team.get("uuid") or team.get("id") or "Unknown team") + + @staticmethod + def _join_labels(items: list[dict[str, Any]], label_fn: Any) -> str: + return ", ".join(label_fn(item) for item in items) + + def _kudos_confirmation_text(self, flow: dict[str, Any]) -> str: + users: list[dict[str, Any]] = flow.get("users") or [] + teams: list[dict[str, Any]] = flow.get("teams") or [] + user_labels: str = self._join_labels(users, self._user_label) if users else "None" + team_labels: str = self._join_labels(teams, self._team_label) if teams else "None" + company_value: str = str(flow.get("company_value") or "None") + return ( + "**Send kudos?**\n\n" + f"- Users: {user_labels}\n" + f"- Teams: {team_labels}\n" + f"- Message: {flow.get('message')}\n" + f"- Company value: {company_value}\n\n" + "1. Send\n2. Cancel" + ) + + @staticmethod + def _question_label(question: dict[str, Any], index: int) -> str: + for key in ("question", "text", "label", "name", "title"): + value: Any = question.get(key) + if value: + return str(value) + return f"Question {index + 1}" + + @staticmethod + def _question_choices(question: dict[str, Any]) -> list[str]: + raw_choices: Any = question.get("choices") or question.get("options") or [] + if not isinstance(raw_choices, list): + return [] + choices: list[str] = [] + for choice in raw_choices: + if isinstance(choice, str): + choices.append(choice) + elif isinstance(choice, dict): + label: Any = choice.get("label") or choice.get("text") or choice.get("value") + if label: + choices.append(str(label)) + return choices + + @staticmethod + def _question_type(question: dict[str, Any]) -> str: + raw_type: str = str(question.get("question_type") or question.get("type") or "") + normalized: str = raw_type.strip().lower().replace("-", "_").replace(" ", "_") + if normalized in {"boolean", "bool", "yes_no", "yes/no", "toggle"}: + return "boolean" + if normalized in {"number", "numeric", "integer", "int", "float", "decimal"}: + return "numeric" + return "text" + + def _get_current_user_uuid(self) -> str | None: + data: dict[str, Any] = self.client.auth_status() + user_raw: Any = data.get("user", {}) + if isinstance(user_raw, dict): + value: Any = user_raw.get("uuid") or user_raw.get("id") + return str(value) if value else None + return None + def _write_user(self, content: str) -> None: log = self.query_one("#transcript", RichLog) log.write(Text(self.user_label, style="bold #7aa2f7")) @@ -350,6 +2015,10 @@ def _set_loading(self, loading: bool) -> None: if not loading: prompt.focus() + def _set_prompt_hint(self, hint: str | None = None) -> None: + prompt = self.query_one("#prompt", Input) + prompt.placeholder = hint or "> Ask Dailybot anything..." + def _remember_input(self, value: str) -> None: if not self.input_history or self.input_history[-1] != value: self.input_history.append(value) @@ -387,6 +2056,160 @@ def _replace_prompt_value(self, value: str) -> None: prompt.value = value prompt.cursor_position = len(value) + async def _complete_prompt(self, *, reverse: bool = False) -> None: + prompt = self.query_one("#prompt", Input) + value: str = prompt.value + cursor_position: int = prompt.cursor_position + + if self._can_cycle_completion(value, cursor_position): + self._cycle_completion(prompt, reverse=reverse) + return + + command_completion: tuple[str, int] | None = self._command_completion( + value, + cursor_position, + reverse=reverse, + ) + if command_completion is not None: + self._apply_completion(prompt, command_completion[0], command_completion[1]) + return + + mention_items: list[str] = await self._load_mention_completion_items() + mention_completion: tuple[str, int] | None = self._mention_completion( + value, + cursor_position, + mention_items, + reverse=reverse, + ) + if mention_completion is not None: + self._apply_completion(prompt, mention_completion[0], mention_completion[1]) + return + self.completion_state = None + + def _command_completion( + self, + value: str, + cursor_position: int, + *, + reverse: bool = False, + ) -> tuple[str, int] | None: + before_cursor: str = value[:cursor_position] + after_cursor: str = value[cursor_position:] + if not before_cursor.startswith("/") or "\n" in before_cursor: + return None + prefix: str = before_cursor.lower() + matches: list[str] = [ + command for command in COMMAND_COMPLETIONS if command.lower().startswith(prefix) + ] + if not matches: + return None + index: int = len(matches) - 1 if reverse else 0 + completed_value: str = matches[index] + after_cursor + self.completion_state = { + "matches": matches, + "index": index, + "last_value": completed_value, + "last_cursor": len(matches[index]), + } + return completed_value, len(matches[index]) + + def _mention_completion( + self, + value: str, + cursor_position: int, + mention_items: list[str], + *, + reverse: bool = False, + ) -> tuple[str, int] | None: + before_cursor: str = value[:cursor_position] + after_cursor: str = value[cursor_position:] + mention_start: int = before_cursor.rfind("@") + if mention_start < 0: + return None + if mention_start > 0 and not before_cursor[mention_start - 1].isspace(): + return None + raw_prefix: str = before_cursor[mention_start + 1 :] + if any(char.isspace() for char in raw_prefix): + return None + prefix: str = raw_prefix.lower() + matches: list[str] = [ + item for item in mention_items if item.removeprefix("@").lower().startswith(prefix) + ] + if not matches: + return None + index: int = len(matches) - 1 if reverse else 0 + completed_value: str = before_cursor[:mention_start] + matches[index] + after_cursor + cursor: int = mention_start + len(matches[index]) + self.completion_state = { + "matches": matches, + "index": index, + "last_value": completed_value, + "last_cursor": cursor, + } + return completed_value, cursor + + async def _load_mention_completion_items(self) -> list[str]: + if self.mention_completion_items is not None: + return self.mention_completion_items + try: + users, teams = await asyncio.gather( + asyncio.to_thread(self.client.list_users), + asyncio.to_thread(self.client.list_teams), + ) + except APIError as exc: + self._write_error(self._format_api_error(exc)) + self.mention_completion_items = [] + return [] + user_items: list[str] = [f"@{self._user_label(user)}" for user in users] + team_items: list[str] = [f"@{self._team_label(team)} team" for team in teams] + self.mention_completion_items = [*user_items, *team_items] + return self.mention_completion_items + + def _can_cycle_completion(self, value: str, cursor_position: int) -> bool: + if self.completion_state is None: + return False + return ( + value == self.completion_state.get("last_value") + and cursor_position == self.completion_state.get("last_cursor") + ) + + def _cycle_completion(self, prompt: Input, *, reverse: bool = False) -> None: + assert self.completion_state is not None + matches: list[str] = self.completion_state.get("matches") or [] + if not matches: + self.completion_state = None + return + step: int = -1 if reverse else 1 + index: int = (int(self.completion_state.get("index") or 0) + step) % len(matches) + value: str = prompt.value + cursor_position: int = prompt.cursor_position + current_match: str = str(matches[int(self.completion_state.get("index") or 0)]) + replacement: str = matches[index] + start: int = max(0, cursor_position - len(current_match)) + completed_value: str = value[:start] + replacement + value[cursor_position:] + self.completion_state["index"] = index + self.completion_state["last_value"] = completed_value + self.completion_state["last_cursor"] = start + len(replacement) + self._set_prompt_completion(prompt, completed_value, start + len(replacement)) + + def _apply_completion(self, prompt: Input, value: str, cursor_position: int) -> None: + if self.completion_state is None: + self.completion_state = { + "matches": [value], + "index": 0, + "last_value": value, + "last_cursor": cursor_position, + } + else: + self.completion_state["last_value"] = value + self.completion_state["last_cursor"] = cursor_position + self._set_prompt_completion(prompt, value, cursor_position) + + @staticmethod + def _set_prompt_completion(prompt: Input, value: str, cursor_position: int) -> None: + prompt.value = value + prompt.cursor_position = cursor_position + def _refresh_status(self) -> None: self.query_one("#status", Static).update( f"Session {self.session.session_id[:8]} | " diff --git a/dailybot_cli/tui/commands.py b/dailybot_cli/tui/commands.py index a39a4c5..231f384 100644 --- a/dailybot_cli/tui/commands.py +++ b/dailybot_cli/tui/commands.py @@ -2,34 +2,92 @@ COMMAND_CHECKINS: Final[str] = "checkins" COMMAND_CLEAR: Final[str] = "clear" +COMMAND_DASHBOARD: Final[str] = "dashboard" COMMAND_EXIT: Final[str] = "exit" +COMMAND_FORM: Final[str] = "form" +COMMAND_FORMS: Final[str] = "forms" COMMAND_HELP: Final[str] = "help" +COMMAND_KUDOS: Final[str] = "kudos" +COMMAND_MOOD: Final[str] = "mood" COMMAND_QUIT: Final[str] = "quit" COMMAND_REPORT: Final[str] = "report" COMMAND_STATUS: Final[str] = "status" +COMMAND_TEAM: Final[str] = "team" +COMMAND_TEAMS: Final[str] = "teams" +COMMAND_TIMEOFF: Final[str] = "timeoff" +COMMAND_USERS: Final[str] = "users" EXIT_COMMANDS: Final[set[str]] = {COMMAND_EXIT, COMMAND_QUIT} KNOWN_COMMANDS: Final[set[str]] = { COMMAND_CHECKINS, COMMAND_CLEAR, + COMMAND_DASHBOARD, COMMAND_EXIT, + COMMAND_FORM, + COMMAND_FORMS, COMMAND_HELP, + COMMAND_KUDOS, + COMMAND_MOOD, COMMAND_QUIT, COMMAND_REPORT, COMMAND_STATUS, + COMMAND_TEAM, + COMMAND_TEAMS, + COMMAND_TIMEOFF, + COMMAND_USERS, } HELP_TEXT: Final[str] = """Commands /help - Show this help. /clear - Clear the local terminal transcript. -/status - Show the current login/session status. -/checkins - Show pending check-ins. +/status - Show login status and pending check-ins. +/dashboard - Show the Dailybot dashboard link. +/checkins - Complete pending check-ins. +/checkin edit - Edit today's submitted check-in. +/checkin reset - Delete today's submitted check-in after confirmation. +/kudos - Send kudos to users or teams. +/forms - List forms and choose an action. +/form submit - Submit a form response. +/form responses - Browse your form responses. +/form update - Update one of your form responses. +/form transition - Move a workflow form response. +/form delete - Delete a form response after confirmation. +/users - Browse organization members. +/teams - Browse teams. +/team - Pick a team and show its members. +/mood - Track today's mood. /report - Submit a free-text progress update. +/timeoff - Time-off native flow is not available in this terminal yet. /exit - Leave the session. Natural language goes to Dailybot. """ +TERMINAL_COMMANDS: Final[list[dict[str, str]]] = [ + {"name": "/help", "description": "Show the command catalog."}, + {"name": "/clear", "description": "Clear the local terminal transcript."}, + {"name": "/status", "description": "Show login status and pending check-ins."}, + {"name": "/dashboard", "description": "Show the Dailybot dashboard URL."}, + {"name": "/checkins", "description": "Complete pending check-ins with numbered prompts."}, + {"name": "/checkin edit", "description": "Edit today's submitted check-in answers."}, + {"name": "/checkin reset", "description": "Delete today's submitted check-in response after confirmation."}, + {"name": "/kudos", "description": "Send kudos to users or teams from the terminal."}, + {"name": "/forms", "description": "List forms and submit or manage responses."}, + {"name": "/form submit", "description": "Submit a form response question by question."}, + {"name": "/form responses", "description": "Browse form responses visible to the user."}, + {"name": "/form update", "description": "Update a form response question by question."}, + {"name": "/form transition", "description": "Move a workflow form response to an allowed state."}, + {"name": "/form delete", "description": "Delete a form response after confirmation."}, + {"name": "/users", "description": "Browse organization members."}, + {"name": "/teams", "description": "Browse teams."}, + {"name": "/team", "description": "Show details and members for a team."}, + {"name": "/mood", "description": "Track today's mood score."}, + {"name": "/report", "description": "Submit a free-text progress update."}, + {"name": "/timeoff", "description": "Explain time-off terminal flow availability."}, + {"name": "/exit", "description": "Leave the session."}, +] +COMMAND_COMPLETIONS: Final[list[str]] = [command["name"] for command in TERMINAL_COMMANDS] + def parse_command(raw_value: str) -> str | None: """Return a normalized slash command, or None for normal chat text.""" @@ -38,3 +96,12 @@ def parse_command(raw_value: str) -> str | None: return None command: str = value[1:].split(maxsplit=1)[0].lower() return command or COMMAND_HELP + + +def parse_command_args(raw_value: str) -> list[str]: + """Return slash-command arguments without the command itself.""" + value: str = raw_value.strip() + if not value.startswith("/"): + return [] + parts: list[str] = value[1:].split() + return parts[1:] diff --git a/dailybot_cli/tui/intents.py b/dailybot_cli/tui/intents.py new file mode 100644 index 0000000..6170a06 --- /dev/null +++ b/dailybot_cli/tui/intents.py @@ -0,0 +1,205 @@ +"""Small deterministic intents for terminal-native chat flows.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from typing import Any + +CHECKINS_INTENT_PHRASES: frozenset[str] = frozenset( + { + "checkin", + "check-in", + "checkins", + "check-ins", + "show checkins", + "show check-ins", + "show my checkins", + "show my check-ins", + "list checkins", + "list check-ins", + "my checkins", + "my check-ins", + } +) + +_CHECKINS_QUESTION_RE: re.Pattern[str] = re.compile( + r"\b(what|which|show|list|do i have|missing|pending)\b.*\b(checkins?|check-ins?|standups?|stand-ups?)\b", + re.IGNORECASE, +) + +_KUDOS_PATTERNS: tuple[re.Pattern[str], ...] = ( + re.compile( + r"^(?:please\s+)?(?:give|send)\s+kudos\s+to\s+(?P.+?)(?:\s+(?:for|because|about)\s+(?P.+))?$", + re.IGNORECASE, + ), + re.compile( + r"^(?:please\s+)?(?:kudos|shout\s*out|shoutout|props|congratulate|thank|thanks)\s+(?:to\s+)?(?P.+?)(?:\s+(?:for|because|about)\s+(?P.+))?$", + re.IGNORECASE, + ), +) + + +@dataclass(frozen=True) +class TerminalCheckinIntent: + """Parsed check-in request that should stay inside the terminal UI.""" + + action: str + + +@dataclass(frozen=True) +class KudosIntent: + """Parsed kudos request from terminal chat text.""" + + receiver_query: str + message: str + receiver_kind: str = "auto" + + +@dataclass(frozen=True) +class TerminalActionIntent: + """Parsed terminal-native action that should bypass general chat.""" + + action: str + args: tuple[str, ...] = () + + +def is_checkins_intent(raw_value: str) -> bool: + """Return True when the user is asking for the native check-ins flow.""" + value: str = _normalize(raw_value) + return value in CHECKINS_INTENT_PHRASES or bool(_CHECKINS_QUESTION_RE.search(value)) + + +def parse_terminal_checkin_intent(raw_value: str) -> TerminalCheckinIntent | None: + """Parse check-in commands that chatbot platforms expose without a slash.""" + value: str = _normalize(raw_value).lstrip("/") + if not value.startswith(("checkin", "check-in", "checkins", "check-ins")): + return None + + parts: list[str] = value.split() + if len(parts) == 1: + return TerminalCheckinIntent(action="complete") + + action: str = parts[1] + if action in {"list", "show", "complete", "start", "submit"}: + return TerminalCheckinIntent(action="complete") + if action in {"edit", "reset", "delete"}: + return TerminalCheckinIntent(action=action) + return None + + +def parse_kudos_intent(raw_value: str) -> KudosIntent | None: + """Parse simple kudos requests that should use the native CLI flow.""" + value: str = raw_value.strip() + if not value: + return None + for pattern in _KUDOS_PATTERNS: + match = pattern.match(value) + if match is None: + continue + receiver_query: str = _clean_receiver(match.group("receiver") or "") + if not receiver_query: + return None + message: str = (match.group("message") or "").strip(" .") + receiver_kind: str = "team" if _looks_like_team_receiver(receiver_query) else "auto" + receiver_query = _clean_team_receiver(receiver_query) + return KudosIntent( + receiver_query=receiver_query, + message=message, + receiver_kind=receiver_kind, + ) + return None + + +def parse_terminal_action_intent(raw_value: str) -> TerminalActionIntent | None: + """Parse simple natural language requests for terminal-native flows.""" + value: str = _normalize(raw_value).lstrip("/") + if value in {"forms", "form list", "list forms", "show forms", "my forms"}: + return TerminalActionIntent(action="forms") + if value.startswith(("submit form", "fill form", "fill out form")): + return TerminalActionIntent(action="form_submit") + if value.startswith(("form responses", "show form responses", "list form responses")): + return TerminalActionIntent(action="form_responses") + if value.startswith(("update form", "edit form response")): + return TerminalActionIntent(action="form_update") + if value.startswith(("transition form", "move form response")): + return TerminalActionIntent(action="form_transition") + if value.startswith(("delete form", "delete form response")): + return TerminalActionIntent(action="form_delete") + if value in {"users", "list users", "show users", "team members", "list members"}: + return TerminalActionIntent(action="users") + if value in {"teams", "list teams", "show teams"}: + return TerminalActionIntent(action="teams") + if value.startswith("team "): + return TerminalActionIntent(action="team", args=(value.removeprefix("team ").strip(),)) + if value in {"dashboard", "open dashboard", "dailybot dashboard"}: + return TerminalActionIntent(action="dashboard") + if value in {"mood", "track mood", "record mood", "how am i feeling"}: + return TerminalActionIntent(action="mood") + if value.startswith(("timeoff", "time off", "pto")): + return TerminalActionIntent(action="timeoff") + return None + + +def matching_users(users: list[dict[str, Any]], query: str) -> list[dict[str, Any]]: + """Return active users whose name/email/handle matches the query.""" + normalized_query: str = _normalize(query).lstrip("@") + if not normalized_query: + return [] + + query_parts: list[str] = [ + part for part in re.split(r"\s+", normalized_query) if part + ] + matches: list[dict[str, Any]] = [] + for user in users: + if not user.get("is_active", True): + continue + haystack: str = _normalize( + " ".join( + str(user.get(key) or "") + for key in ( + "full_name", + "name", + "display_name", + "email", + "username", + "handle", + ) + ) + ).lstrip("@") + if normalized_query in haystack or all(part in haystack for part in query_parts): + matches.append(user) + return matches + + +def matching_teams(teams: list[dict[str, Any]], query: str) -> list[dict[str, Any]]: + """Return teams whose name/UUID matches the query.""" + normalized_query: str = _normalize(query).lstrip("@") + if not normalized_query: + return [] + matches: list[dict[str, Any]] = [] + for team in teams: + haystack: str = _normalize( + " ".join(str(team.get(key) or "") for key in ("name", "uuid", "id")) + ) + if normalized_query in haystack: + matches.append(team) + return matches + + +def _normalize(value: str) -> str: + return " ".join(value.strip().lower().split()) + + +def _clean_receiver(value: str) -> str: + receiver: str = value.strip(" .") + receiver = re.sub(r"^(?:@dailybot\s+)?", "", receiver, flags=re.IGNORECASE).strip() + return receiver + + +def _looks_like_team_receiver(value: str) -> bool: + return bool(re.search(r"\b(team|squad|group|department)\b", value, re.IGNORECASE)) + + +def _clean_team_receiver(value: str) -> str: + return re.sub(r"\b(team|squad|group|department)\b", "", value, flags=re.IGNORECASE).strip() diff --git a/tests/api_client_test.py b/tests/api_client_test.py index 40712f4..19d7668 100644 --- a/tests/api_client_test.py +++ b/tests/api_client_test.py @@ -163,6 +163,133 @@ def test_complete_checkin(self, client: DailyBotClient) -> None: assert "Bearer test-token" in call_kwargs["headers"]["Authorization"] assert result["uuid"] == "response-uuid" + def test_list_checkins(self, client: DailyBotClient) -> None: + mock_response: MagicMock = MagicMock(spec=httpx.Response) + mock_response.status_code = 200 + mock_response.json.return_value = { + "results": [{"id": "followup-uuid", "name": "Daily Standup"}], + "next": None, + } + + with patch("httpx.get", return_value=mock_response) as mock_get: + result: list[dict[str, Any]] = client.list_checkins() + + assert result[0]["id"] == "followup-uuid" + assert mock_get.call_args[0][0].endswith("/v1/checkins/") + + def test_get_template_with_followup_context(self, client: DailyBotClient) -> None: + mock_response: MagicMock = MagicMock(spec=httpx.Response) + mock_response.status_code = 200 + mock_response.json.return_value = {"id": "template-uuid", "questions": {"fields": []}} + + with patch("httpx.get", return_value=mock_response) as mock_get: + result: dict[str, Any] = client.get_template( + "template-uuid", + followup_uuid="followup-uuid", + ) + + assert result["id"] == "template-uuid" + assert mock_get.call_args[1]["params"] == { + "render_special_vars": "true", + "followup_id": "followup-uuid", + } + + def test_get_checkin(self, client: DailyBotClient) -> None: + mock_response: MagicMock = MagicMock(spec=httpx.Response) + mock_response.status_code = 200 + mock_response.json.return_value = {"id": "followup-uuid", "template": "template-uuid"} + + with patch("httpx.get", return_value=mock_response) as mock_get: + result: dict[str, Any] = client.get_checkin("followup-uuid") + + assert result["template"] == "template-uuid" + assert mock_get.call_args[0][0].endswith("/v1/checkins/followup-uuid/") + + def test_list_checkin_responses(self, client: DailyBotClient) -> None: + mock_response: MagicMock = MagicMock(spec=httpx.Response) + mock_response.status_code = 200 + mock_response.json.return_value = { + "results": [{"uuid": "response-uuid"}], + "next": None, + } + + with patch("httpx.get", return_value=mock_response) as mock_get: + result: list[dict[str, Any]] = client.list_checkin_responses( + "followup-uuid", + date_start="2026-07-01", + date_end="2026-07-01", + ) + + assert result[0]["uuid"] == "response-uuid" + assert mock_get.call_args[0][0].endswith("/v1/checkins/followup-uuid/responses/") + assert mock_get.call_args[1]["params"] == { + "date_start": "2026-07-01", + "date_end": "2026-07-01", + } + + def test_update_checkin_response(self, client: DailyBotClient) -> None: + mock_response: MagicMock = MagicMock(spec=httpx.Response) + mock_response.status_code = 200 + mock_response.json.return_value = {"uuid": "response-uuid"} + responses: list[dict[str, Any]] = [{"uuid": "q-0", "index": 0, "response": "Edited"}] + + with patch("httpx.put", return_value=mock_response) as mock_put: + result: dict[str, Any] = client.update_checkin_response( + "followup-uuid", + responses, + last_question_index=0, + ) + + assert result["uuid"] == "response-uuid" + assert mock_put.call_args[0][0].endswith("/v1/checkins/followup-uuid/responses/") + assert mock_put.call_args[1]["json"] == { + "responses": responses, + "last_question_index": 0, + } + + def test_delete_checkin_response(self, client: DailyBotClient) -> None: + mock_response: MagicMock = MagicMock(spec=httpx.Response) + mock_response.status_code = 200 + mock_response.json.return_value = {"deleted": True} + + with patch("httpx.request", return_value=mock_response) as mock_request: + result: dict[str, Any] = client.delete_checkin_response( + "followup-uuid", + response_date="2026-07-01", + ) + + assert result["deleted"] is True + assert mock_request.call_args[0][0] == "DELETE" + assert mock_request.call_args[0][1].endswith("/v1/checkins/followup-uuid/responses/") + assert mock_request.call_args[1]["params"] == { + "date_start": "2026-07-01", + "date_end": "2026-07-01", + } + + def test_track_mood(self, client: DailyBotClient) -> None: + mock_response: MagicMock = MagicMock(spec=httpx.Response) + mock_response.status_code = 200 + mock_response.json.return_value = {"details": "The mood response has been tracked"} + + with patch("httpx.post", return_value=mock_response) as mock_post: + result: dict[str, Any] = client.track_mood(5) + + assert "tracked" in result["details"] + assert mock_post.call_args[0][0].endswith("/v1/mood/track/") + assert mock_post.call_args[1]["json"] == {"score": 5} + + def test_get_mood(self, client: DailyBotClient) -> None: + mock_response: MagicMock = MagicMock(spec=httpx.Response) + mock_response.status_code = 200 + mock_response.json.return_value = {"motivation": {"score": 4}} + + with patch("httpx.get", return_value=mock_response) as mock_get: + result: dict[str, Any] = client.get_mood("2026-07-01") + + assert result["motivation"]["score"] == 4 + assert mock_get.call_args[0][0].endswith("/v1/mood/track/") + assert mock_get.call_args[1]["params"] == {"date": "2026-07-01"} + def test_list_forms(self, client: DailyBotClient) -> None: mock_response: MagicMock = MagicMock(spec=httpx.Response) mock_response.status_code = 200 @@ -309,11 +436,11 @@ def test_give_kudos(self, client: DailyBotClient) -> None: ) call_kwargs: dict[str, Any] = mock_post.call_args[1] - assert call_kwargs["json"]["user_uuid_receivers"] == ["user-uuid"] - assert "team_uuid_receivers" not in call_kwargs["json"] + assert call_kwargs["json"]["receivers"] == ["user-uuid"] + assert call_kwargs["json"]["users_receivers"] == ["user-uuid"] + assert "teams_receivers" not in call_kwargs["json"] assert call_kwargs["json"]["company_value"] == "value-uuid" assert "by_dailybot" not in call_kwargs["json"] - assert "receivers" not in call_kwargs["json"] assert result["uuid"] == "kudos-uuid" def test_give_kudos_team(self, client: DailyBotClient) -> None: @@ -329,8 +456,9 @@ def test_give_kudos_team(self, client: DailyBotClient) -> None: ) call_kwargs: dict[str, Any] = mock_post.call_args[1] - assert call_kwargs["json"]["user_uuid_receivers"] == ["user-uuid"] - assert call_kwargs["json"]["team_uuid_receivers"] == ["team-uuid"] + assert call_kwargs["json"]["receivers"] == ["user-uuid", "team-uuid"] + assert call_kwargs["json"]["users_receivers"] == ["user-uuid"] + assert call_kwargs["json"]["teams_receivers"] == ["team-uuid"] def test_list_teams(self, client: DailyBotClient) -> None: mock_response: MagicMock = MagicMock(spec=httpx.Response) diff --git a/tests/tui_app_flows_test.py b/tests/tui_app_flows_test.py new file mode 100644 index 0000000..6c2b0fd --- /dev/null +++ b/tests/tui_app_flows_test.py @@ -0,0 +1,359 @@ +import asyncio +from datetime import date +from unittest.mock import AsyncMock, MagicMock + +from dailybot_cli.tui.app import DailybotChatApp +from dailybot_cli.tui.intents import KudosIntent, TerminalCheckinIntent + + +def test_kudos_flow_uses_numbered_user_selection() -> None: + client: MagicMock = MagicMock() + client.list_users.return_value = [ + {"uuid": "self-uuid", "full_name": "Me"}, + {"uuid": "peer-uuid", "full_name": "Andres Prieto"}, + ] + client.list_teams.return_value = [] + client.auth_status.return_value = {"user": {"uuid": "self-uuid"}} + app = DailybotChatApp(client) + _stub_output(app) + + asyncio.run( + app._start_kudos_flow( + KudosIntent(receiver_query="Andres", message="this amazing new feature") + ) + ) + asyncio.run(app._select_kudos_user("1")) + asyncio.run(app._set_kudos_value("")) + asyncio.run(app._confirm_kudos("1")) + + client.give_kudos.assert_called_once_with( + content="this amazing new feature", + user_uuid_receivers=["peer-uuid"], + team_uuid_receivers=None, + company_value=None, + ) + + +def test_checkins_flow_submits_answers_after_numbered_selection() -> None: + client: MagicMock = MagicMock() + client.get_status.return_value = { + "pending_checkins": [ + { + "followup_uuid": "followup-uuid", + "followup_name": "Daily Standup", + "template_questions": [ + { + "uuid": "q1", + "question": ( + "Previous plan ({previous_response_date}): " + "{previous_response_#2}" + ), + }, + {"uuid": "q2", "question": "Blocked?", "question_type": "boolean"}, + ], + } + ] + } + client.get_checkin.return_value = { + "id": "followup-uuid", + "name": "Daily Standup", + "template": "template-uuid", + } + client.get_template.return_value = { + "questions": { + "fields": [ + { + "uuid": "q1", + "question": "Previous plan (Jun 30): Continue with the CLI", + }, + {"uuid": "q2", "question": "Blocked?", "question_type": "boolean"}, + ] + } + } + app = DailybotChatApp(client) + _stub_output(app) + + asyncio.run(app._start_checkins_flow()) + asyncio.run(app._select_checkin("1")) + asyncio.run(app._answer_checkin_question("Built the CLI flow")) + asyncio.run(app._answer_checkin_question("2")) + + client.complete_checkin.assert_called_once_with( + "followup-uuid", + [ + {"uuid": "q1", "index": 0, "response": "Built the CLI flow"}, + {"uuid": "q2", "index": 1, "response": False}, + ], + 1, + ) + question_prompt: str = app._write_dailybot.call_args_list[1].args[0] + assert "Continue with the CLI" in question_prompt + assert "{previous_response" not in question_prompt + + +def test_checkin_edit_flow_keeps_and_replaces_answers() -> None: + client: MagicMock = MagicMock() + client.list_checkins.return_value = [{"id": "followup-uuid", "name": "Daily Standup"}] + client.list_checkin_responses.return_value = [ + { + "uuid": "daily-uuid", + "responses": [ + {"question": "What did you do?", "response": {"value": "Old work"}}, + {"question": "Blocked?", "response": True}, + ], + } + ] + client.get_checkin.return_value = { + "id": "followup-uuid", + "name": "Daily Standup", + "template": "template-uuid", + } + client.get_template.return_value = { + "questions": { + "fields": [ + {"uuid": "q1", "question": "What did you do?"}, + {"uuid": "q2", "question": "Blocked?", "question_type": "boolean"}, + ] + } + } + app = DailybotChatApp(client) + _stub_output(app) + + asyncio.run(app._handle_terminal_checkin_intent(TerminalCheckinIntent(action="edit"))) + asyncio.run(app._select_checkin_to_edit("1")) + asyncio.run(app._answer_checkin_edit_question("")) + asyncio.run(app._answer_checkin_edit_question("2")) + + question_prompt: str = app._write_dailybot.call_args_list[1].args[0] + assert "Current answer: Old work" in question_prompt + assert "Current answer: {}" not in question_prompt + client.update_checkin_response.assert_called_once_with( + "followup-uuid", + [ + {"uuid": "q1", "index": 0, "response": {"value": "Old work"}}, + {"uuid": "q2", "index": 1, "response": False}, + ], + 1, + ) + + +def test_form_submit_flow_collects_answers() -> None: + client: MagicMock = MagicMock() + client.list_forms.return_value = [ + { + "id": "form-uuid", + "name": "Code Release", + "questions": [{"uuid": "q1", "question": "What shipped?"}], + } + ] + app = DailybotChatApp(client) + _stub_output(app) + + asyncio.run(app._start_forms_flow("submit")) + asyncio.run(app._select_form_action("1")) + asyncio.run(app._answer_form_submit_question("Interactive CLI")) + + client.submit_form_response.assert_called_once_with( + "form-uuid", + {"q1": "Interactive CLI"}, + ) + + +def test_form_update_flow_keeps_existing_answers() -> None: + client: MagicMock = MagicMock() + form: dict[str, object] = { + "id": "form-uuid", + "name": "Code Release", + "questions": [{"uuid": "q1", "question": "What shipped?"}], + } + response: dict[str, object] = { + "id": "response-uuid", + "content": {"q1": {"value": "Old summary"}}, + } + app = DailybotChatApp(client) + _stub_output(app) + + asyncio.run(app._start_form_update_flow(form, response)) + asyncio.run(app._answer_form_update_question("")) + + client.update_form_response.assert_called_once_with( + "form-uuid", + "response-uuid", + {"q1": {"value": "Old summary"}}, + ) + + +def test_form_transition_flow_uses_allowed_transition() -> None: + client: MagicMock = MagicMock() + form: dict[str, object] = {"id": "form-uuid", "name": "Code Release"} + response: dict[str, object] = { + "id": "response-uuid", + "allowed_transitions": [{"to_state": "qa", "label": "QA"}], + } + app = DailybotChatApp(client) + _stub_output(app) + + app._start_form_transition_flow(form, response) + asyncio.run(app._select_form_transition("1")) + asyncio.run(app._set_form_transition_note("Ready for QA")) + + client.transition_form_response.assert_called_once_with( + "form-uuid", + "response-uuid", + "qa", + "Ready for QA", + ) + + +def test_form_delete_flow_confirms_before_delete() -> None: + client: MagicMock = MagicMock() + form: dict[str, object] = {"id": "form-uuid", "name": "Code Release"} + response: dict[str, object] = {"id": "response-uuid"} + app = DailybotChatApp(client) + _stub_output(app) + + app._start_form_delete_flow(form, response) + asyncio.run(app._confirm_form_delete("1")) + + client.delete_form_response.assert_called_once_with("form-uuid", "response-uuid") + + +def test_checkin_reset_flow_confirms_before_delete() -> None: + client: MagicMock = MagicMock() + app = DailybotChatApp(client) + _stub_output(app) + app.pending_flow = { + "type": "checkin_reset_confirm", + "candidate": {"checkin": {"id": "followup-uuid", "name": "Daily Standup"}}, + } + + asyncio.run(app._confirm_checkin_reset("1")) + + client.delete_checkin_response.assert_called_once_with( + "followup-uuid", + response_date=date.today().isoformat(), + ) + + +def test_mood_flow_tracks_score() -> None: + client: MagicMock = MagicMock() + app = DailybotChatApp(client) + _stub_output(app) + + asyncio.run(app._start_mood_flow()) + asyncio.run(app._select_mood("5")) + + client.track_mood.assert_called_once_with(5) + + +def test_structured_terminal_action_launches_native_flow() -> None: + client: MagicMock = MagicMock() + app = DailybotChatApp(client) + _stub_output(app) + app._start_kudos_menu = AsyncMock() # type: ignore[method-assign] + + asyncio.run(app._handle_chat_actions([{"type": "terminal_flow", "flow": "kudos"}])) + asyncio.run(app._confirm_terminal_action("1")) + + app._start_kudos_menu.assert_called_once() + + +def test_slash_tab_completion_cycles_matching_commands() -> None: + client: MagicMock = MagicMock() + app = DailybotChatApp(client) + + completed = app._command_completion("/fo", 3) + + assert completed is not None + assert completed[0] == "/forms" + prompt = _FakePrompt(value=completed[0], cursor_position=completed[1]) + app._cycle_completion(prompt) # type: ignore[arg-type] + assert prompt.value == "/form submit" + + +def test_slash_shift_tab_completion_cycles_backwards() -> None: + client: MagicMock = MagicMock() + app = DailybotChatApp(client) + + completed = app._command_completion("/fo", 3) + + assert completed is not None + prompt = _FakePrompt(value=completed[0], cursor_position=completed[1]) + app._cycle_completion(prompt, reverse=True) # type: ignore[arg-type] + assert prompt.value == "/form delete" + + +def test_slash_shift_tab_completion_starts_from_last_match() -> None: + client: MagicMock = MagicMock() + app = DailybotChatApp(client) + + completed = app._command_completion("/fo", 3, reverse=True) + + assert completed is not None + assert completed[0] == "/form delete" + + +def test_mention_tab_completion_supports_users_and_teams() -> None: + client: MagicMock = MagicMock() + app = DailybotChatApp(client) + + user_completed = app._mention_completion( + "kudos @an", + len("kudos @an"), + ["@Andres Prieto", "@Engineering team"], + ) + team_completed = app._mention_completion( + "kudos @eng", + len("kudos @eng"), + ["@Andres Prieto", "@Engineering team"], + ) + + assert user_completed is not None + assert user_completed[0] == "kudos @Andres Prieto" + assert team_completed is not None + assert team_completed[0] == "kudos @Engineering team" + + +def test_mention_tab_completion_cycles_matches() -> None: + client: MagicMock = MagicMock() + app = DailybotChatApp(client) + + completed = app._mention_completion( + "kudos @and", + len("kudos @and"), + ["@Andres Prieto", "@Andrea Dailybot"], + ) + + assert completed is not None + prompt = _FakePrompt(value=completed[0], cursor_position=completed[1]) + app._cycle_completion(prompt) # type: ignore[arg-type] + assert prompt.value == "kudos @Andrea Dailybot" + + +def test_mention_shift_tab_completion_cycles_backwards() -> None: + client: MagicMock = MagicMock() + app = DailybotChatApp(client) + + completed = app._mention_completion( + "kudos @and", + len("kudos @and"), + ["@Andres Prieto", "@Andrea Dailybot"], + ) + + assert completed is not None + prompt = _FakePrompt(value=completed[0], cursor_position=completed[1]) + app._cycle_completion(prompt, reverse=True) # type: ignore[arg-type] + assert prompt.value == "kudos @Andrea Dailybot" + + +def _stub_output(app: DailybotChatApp) -> None: + app._write_dailybot = MagicMock() # type: ignore[method-assign] + app._write_error = MagicMock() # type: ignore[method-assign] + app._set_loading = MagicMock() # type: ignore[method-assign] + app._set_prompt_hint = MagicMock() # type: ignore[method-assign] + + +class _FakePrompt: + def __init__(self, *, value: str, cursor_position: int) -> None: + self.value = value + self.cursor_position = cursor_position diff --git a/tests/tui_intents_test.py b/tests/tui_intents_test.py new file mode 100644 index 0000000..e38ef13 --- /dev/null +++ b/tests/tui_intents_test.py @@ -0,0 +1,75 @@ +from dailybot_cli.tui.intents import ( + is_checkins_intent, + matching_teams, + matching_users, + parse_kudos_intent, + parse_terminal_action_intent, + parse_terminal_checkin_intent, +) + + +def test_checkins_plain_text_routes_to_native_flow() -> None: + assert is_checkins_intent("checkins") + assert is_checkins_intent("show my check-ins") + assert is_checkins_intent("what checkins do I have pending?") + + +def test_parse_kudos_intent_with_message() -> None: + intent = parse_kudos_intent("give kudos to Andres for this amazing new feature") + + assert intent is not None + assert intent.receiver_query == "Andres" + assert intent.message == "this amazing new feature" + assert intent.receiver_kind == "auto" + + +def test_parse_kudos_intent_without_message() -> None: + intent = parse_kudos_intent("send kudos to Jane Doe") + + assert intent is not None + assert intent.receiver_query == "Jane Doe" + assert intent.message == "" + + +def test_parse_kudos_intent_detects_team_receiver() -> None: + intent = parse_kudos_intent("give kudos to Engineering team for shipping") + + assert intent is not None + assert intent.receiver_query == "Engineering" + assert intent.receiver_kind == "team" + + +def test_matching_users_filters_by_name_and_email() -> None: + users = [ + {"uuid": "1", "full_name": "Andres Prieto", "email": "andres@example.com"}, + {"uuid": "2", "full_name": "Jane Doe", "email": "jane@example.com"}, + ] + + assert matching_users(users, "andres") == [users[0]] + assert matching_users(users, "jane@example.com") == [users[1]] + + +def test_matching_teams_filters_by_name() -> None: + teams = [ + {"uuid": "1", "name": "Engineering"}, + {"uuid": "2", "name": "Customer Success"}, + ] + + assert matching_teams(teams, "engineer") == [teams[0]] + + +def test_parse_terminal_checkin_intent_accepts_slash_aliases() -> None: + assert parse_terminal_checkin_intent("/checkin").action == "complete" + assert parse_terminal_checkin_intent("/checkin list").action == "complete" + assert parse_terminal_checkin_intent("/checkin complete").action == "complete" + assert parse_terminal_checkin_intent("/checkin edit").action == "edit" + assert parse_terminal_checkin_intent("/checkin reset").action == "reset" + + +def test_parse_terminal_action_intent_routes_common_actions() -> None: + assert parse_terminal_action_intent("forms").action == "forms" + assert parse_terminal_action_intent("submit form").action == "form_submit" + assert parse_terminal_action_intent("list users").action == "users" + assert parse_terminal_action_intent("show teams").action == "teams" + assert parse_terminal_action_intent("open dashboard").action == "dashboard" + assert parse_terminal_action_intent("track mood").action == "mood" From 776bd9d9fce88441b7550cb328be2bded1960f23 Mon Sep 17 00:00:00 2001 From: AndresMpa Date: Thu, 2 Jul 2026 13:28:57 +0000 Subject: [PATCH 2/3] fix(interactive): satisfy ruff format for terminal flows Co-authored-by: Cursor --- dailybot_cli/tui/app.py | 71 ++++++++++++++++++++++++------------ dailybot_cli/tui/commands.py | 10 ++++- dailybot_cli/tui/intents.py | 4 +- tests/tui_app_flows_test.py | 3 +- 4 files changed, 57 insertions(+), 31 deletions(-) diff --git a/dailybot_cli/tui/app.py b/dailybot_cli/tui/app.py index f73b3d4..a52d478 100644 --- a/dailybot_cli/tui/app.py +++ b/dailybot_cli/tui/app.py @@ -550,7 +550,9 @@ def _load_checkin_edit_context(self, selected: dict[str, Any]) -> dict[str, Any] return {"checkin": checkin_detail, "questions": [], "existing_responses": []} template = self.client.get_template(template_uuid, followup_uuid=followup_uuid) questions: list[dict[str, Any]] = (template.get("questions") or {}).get("fields") or [] - existing_responses: list[dict[str, Any]] = selected.get("response", {}).get("responses") or [] + existing_responses: list[dict[str, Any]] = ( + selected.get("response", {}).get("responses") or [] + ) return { "checkin": checkin_detail, "questions": questions, @@ -592,7 +594,9 @@ def _load_checkin_questions(self, checkin: dict[str, Any]) -> list[dict[str, Any if not template_uuid: return checkin.get("template_questions") or [] template = self.client.get_template(template_uuid, followup_uuid=followup_uuid) - rendered_questions: list[dict[str, Any]] = (template.get("questions") or {}).get("fields") or [] + rendered_questions: list[dict[str, Any]] = (template.get("questions") or {}).get( + "fields" + ) or [] return rendered_questions or checkin.get("template_questions") or [] async def _answer_checkin_question(self, raw_value: str) -> None: @@ -693,7 +697,9 @@ async def _start_kudos_menu(self) -> None: "teams": teams, "current_uuid": current_uuid, } - self._write_dailybot("Who should receive kudos?\n1. One or more users\n2. One or more teams") + self._write_dailybot( + "Who should receive kudos?\n1. One or more users\n2. One or more teams" + ) self._set_prompt_hint("Type 1 or 2, or `cancel`.") async def _start_kudos_flow(self, intent: KudosIntent) -> None: @@ -872,7 +878,9 @@ def _ask_kudos_value( "message": message, "current_uuid": current_uuid, } - self._write_dailybot("Optional: type a company value for this kudos, or press Enter to skip.") + self._write_dailybot( + "Optional: type a company value for this kudos, or press Enter to skip." + ) self._set_prompt_hint("Type a value, press Enter to skip, or `cancel`.") async def _set_kudos_value(self, raw_value: str) -> None: @@ -898,7 +906,11 @@ async def _confirm_kudos(self, raw_value: str) -> None: message: str = str(self.pending_flow.get("message") or "") company_value: str | None = self.pending_flow.get("company_value") user_uuids: list[str] = [str(user.get("uuid") or "") for user in users if user.get("uuid")] - team_uuids: list[str] = [str(team.get("uuid") or team.get("id") or "") for team in teams if team.get("uuid") or team.get("id")] + team_uuids: list[str] = [ + str(team.get("uuid") or team.get("id") or "") + for team in teams + if team.get("uuid") or team.get("id") + ] self.pending_flow = None self._set_prompt_hint() self._set_loading(True) @@ -1164,7 +1176,9 @@ async def _answer_form_update_question(self, raw_value: str) -> None: index: int = int(self.pending_flow["index"]) question: dict[str, Any] = questions[index] existing_value: Any = self._existing_form_response_value(question) - answer: Any = existing_value if raw_value == "" else self._parse_question_answer(raw_value, question) + answer: Any = ( + existing_value if raw_value == "" else self._parse_question_answer(raw_value, question) + ) if answer is None: return self.pending_flow["answers"].append(answer) @@ -1312,7 +1326,9 @@ def _ask_current_form_question(self) -> None: questions: list[dict[str, Any]] = self.pending_flow["questions"] index: int = int(self.pending_flow["index"]) question: dict[str, Any] = questions[index] - lines: list[str] = [f"Question {index + 1}/{len(questions)}: {self._question_label(question, index)}"] + lines: list[str] = [ + f"Question {index + 1}/{len(questions)}: {self._question_label(question, index)}" + ] choices: list[str] = self._question_choices(question) if self._question_type(question) == "boolean": lines.extend(["1. Yes", "2. No"]) @@ -1350,7 +1366,9 @@ def _ask_current_form_update_question(self) -> None: def _existing_form_response_value(self, question: dict[str, Any]) -> Any: assert self.pending_flow is not None response: dict[str, Any] = self.pending_flow.get("response") or {} - content: dict[str, Any] = response.get("content") if isinstance(response.get("content"), dict) else {} + content: dict[str, Any] = ( + response.get("content") if isinstance(response.get("content"), dict) else {} + ) question_uuid: str = str(question.get("uuid") or question.get("id") or "") return content.get(question_uuid, "") @@ -1378,7 +1396,16 @@ def _build_form_content( def _form_label(cls, form: dict[str, Any]) -> str: questions_count: int = len(cls._form_questions(form)) suffix: str = f" ({questions_count} question{'s' if questions_count != 1 else ''})" - return str(form.get("name") or form.get("title") or form.get("uuid") or form.get("id") or "Form") + suffix + return ( + str( + form.get("name") + or form.get("title") + or form.get("uuid") + or form.get("id") + or "Form" + ) + + suffix + ) @classmethod def _form_response_label(cls, response: dict[str, Any]) -> str: @@ -1513,12 +1540,7 @@ def _show_dashboard(self) -> None: async def _start_mood_flow(self) -> None: self.pending_flow = {"type": "mood_select"} self._write_dailybot( - "How is your mood today?\n" - "1. Very low\n" - "2. Low\n" - "3. Okay\n" - "4. Good\n" - "5. Great" + "How is your mood today?\n1. Very low\n2. Low\n3. Okay\n4. Good\n5. Great" ) self._set_prompt_hint("Type a number from 1 to 5, or `cancel`.") @@ -1604,7 +1626,9 @@ async def _confirm_checkin_reset(self, raw_value: str) -> None: return finally: self._set_loading(False) - self._write_dailybot(f"Done — deleted today's response for **{self._checkin_name(checkin)}**.") + self._write_dailybot( + f"Done — deleted today's response for **{self._checkin_name(checkin)}**." + ) async def _handle_terminal_action_intent(self, intent: TerminalActionIntent) -> None: await self._start_terminal_flow_by_name(intent.action, {"args": list(intent.args)}) @@ -1619,9 +1643,7 @@ async def _handle_chat_actions(self, actions: list[dict[str, Any]]) -> None: "flow": flow_name, "params": action.get("params") or {}, } - self._write_dailybot( - f"Start the **{flow_name}** terminal flow?\n1. Yes\n2. No" - ) + self._write_dailybot(f"Start the **{flow_name}** terminal flow?\n1. Yes\n2. No") self._set_prompt_hint("Type 1 to start, 2 to skip.") return action_names: str = ", ".join(str(action.get("name", "action")) for action in actions) @@ -1657,7 +1679,9 @@ async def _start_terminal_flow_by_name(self, flow_name: str, params: dict[str, A receiver_query: str = str(params.get("receiver_query") or params.get("receivers") or "") message: str = str(params.get("message") or "") if receiver_query: - await self._start_kudos_flow(KudosIntent(receiver_query=receiver_query, message=message)) + await self._start_kudos_flow( + KudosIntent(receiver_query=receiver_query, message=message) + ) else: await self._start_kudos_menu() return @@ -2168,10 +2192,9 @@ async def _load_mention_completion_items(self) -> list[str]: def _can_cycle_completion(self, value: str, cursor_position: int) -> bool: if self.completion_state is None: return False - return ( - value == self.completion_state.get("last_value") - and cursor_position == self.completion_state.get("last_cursor") - ) + return value == self.completion_state.get( + "last_value" + ) and cursor_position == self.completion_state.get("last_cursor") def _cycle_completion(self, prompt: Input, *, reverse: bool = False) -> None: assert self.completion_state is not None diff --git a/dailybot_cli/tui/commands.py b/dailybot_cli/tui/commands.py index 231f384..4da038d 100644 --- a/dailybot_cli/tui/commands.py +++ b/dailybot_cli/tui/commands.py @@ -70,13 +70,19 @@ {"name": "/dashboard", "description": "Show the Dailybot dashboard URL."}, {"name": "/checkins", "description": "Complete pending check-ins with numbered prompts."}, {"name": "/checkin edit", "description": "Edit today's submitted check-in answers."}, - {"name": "/checkin reset", "description": "Delete today's submitted check-in response after confirmation."}, + { + "name": "/checkin reset", + "description": "Delete today's submitted check-in response after confirmation.", + }, {"name": "/kudos", "description": "Send kudos to users or teams from the terminal."}, {"name": "/forms", "description": "List forms and submit or manage responses."}, {"name": "/form submit", "description": "Submit a form response question by question."}, {"name": "/form responses", "description": "Browse form responses visible to the user."}, {"name": "/form update", "description": "Update a form response question by question."}, - {"name": "/form transition", "description": "Move a workflow form response to an allowed state."}, + { + "name": "/form transition", + "description": "Move a workflow form response to an allowed state.", + }, {"name": "/form delete", "description": "Delete a form response after confirmation."}, {"name": "/users", "description": "Browse organization members."}, {"name": "/teams", "description": "Browse teams."}, diff --git a/dailybot_cli/tui/intents.py b/dailybot_cli/tui/intents.py index 6170a06..ff7751b 100644 --- a/dailybot_cli/tui/intents.py +++ b/dailybot_cli/tui/intents.py @@ -147,9 +147,7 @@ def matching_users(users: list[dict[str, Any]], query: str) -> list[dict[str, An if not normalized_query: return [] - query_parts: list[str] = [ - part for part in re.split(r"\s+", normalized_query) if part - ] + query_parts: list[str] = [part for part in re.split(r"\s+", normalized_query) if part] matches: list[dict[str, Any]] = [] for user in users: if not user.get("is_active", True): diff --git a/tests/tui_app_flows_test.py b/tests/tui_app_flows_test.py index 6c2b0fd..3a0f171 100644 --- a/tests/tui_app_flows_test.py +++ b/tests/tui_app_flows_test.py @@ -45,8 +45,7 @@ def test_checkins_flow_submits_answers_after_numbered_selection() -> None: { "uuid": "q1", "question": ( - "Previous plan ({previous_response_date}): " - "{previous_response_#2}" + "Previous plan ({previous_response_date}): {previous_response_#2}" ), }, {"uuid": "q2", "question": "Blocked?", "question_type": "boolean"}, From e27cfc7a1256ffe935eec38dfe13a64186ce2fe8 Mon Sep 17 00:00:00 2001 From: AndresMpa Date: Thu, 2 Jul 2026 13:42:11 +0000 Subject: [PATCH 3/3] fix(interactive): satisfy mypy for form responses Co-authored-by: Cursor --- dailybot_cli/tui/app.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/dailybot_cli/tui/app.py b/dailybot_cli/tui/app.py index a52d478..521d2c6 100644 --- a/dailybot_cli/tui/app.py +++ b/dailybot_cli/tui/app.py @@ -1366,9 +1366,8 @@ def _ask_current_form_update_question(self) -> None: def _existing_form_response_value(self, question: dict[str, Any]) -> Any: assert self.pending_flow is not None response: dict[str, Any] = self.pending_flow.get("response") or {} - content: dict[str, Any] = ( - response.get("content") if isinstance(response.get("content"), dict) else {} - ) + raw_content: Any = response.get("content") + content: dict[str, Any] = raw_content if isinstance(raw_content, dict) else {} question_uuid: str = str(question.get("uuid") or question.get("id") or "") return content.get(question_uuid, "")