diff --git a/app/google_docs/services/folder_manager.py b/app/google_docs/services/folder_manager.py index 142ca59..7690666 100644 --- a/app/google_docs/services/folder_manager.py +++ b/app/google_docs/services/folder_manager.py @@ -10,6 +10,7 @@ from app.google_docs.utils.constants import ( AUTOMATED_DOCS_FOLDER_NAME, DEFAULT_MAX_RECURSION_DEPTH, + DRIVE_INACCESSIBLE_STATUSES, ) logger = logging.getLogger(__name__) @@ -116,7 +117,7 @@ def _search_folder_recursive( except HttpError as e: # Log warning for permission/access issues but continue searching - if e.resp.status in [403, 404]: + if e.resp.status in DRIVE_INACCESSIBLE_STATUSES: logger.warning( f"Access denied or folder not found {parent_folder_id}: {e}", ) @@ -156,7 +157,29 @@ async def get_automated_docs_folder( Exception: If folder is not found or API call errors occur. """ - drive_service = self.auth_service.get_drive_service() + drive_service: Any = self.auth_service.get_drive_service() + + # Checked first because the recursive search swallows 403/404: an + # unshared parent would otherwise be reported as a missing subfolder. + try: + drive_service.files().get( + fileId=parent_folder_id, + fields="id", + supportsAllDrives=True, + ).execute() + except HttpError as exc: + # Relabelling a 5xx as a sharing problem sends the user to fix + # the wrong thing, so only report what this probe can detect. + if exc.resp.status not in DRIVE_INACCESSIBLE_STATUSES: + raise + + log_and_raise( + logger, + f"Cannot access Drive folder {parent_folder_id}. Share it with " + f"{self.auth_service.service_account_email} first.", + Exception, + cause=exc, + ) # Try to find the folder recursively folder_id = self._search_folder_recursive( diff --git a/app/google_docs/services/google_auth.py b/app/google_docs/services/google_auth.py index 8d0272b..95120d7 100644 --- a/app/google_docs/services/google_auth.py +++ b/app/google_docs/services/google_auth.py @@ -87,6 +87,15 @@ def _get_credentials(self) -> service_account.Credentials: return self._credentials + @property + def service_account_email(self) -> str: + """The address a Drive folder must be shared with for reports to work.""" + try: + info = json.loads(settings.GOOGLE_SERVICE_ACCOUNT_KEY.get_secret_value()) + except (json.JSONDecodeError, ValueError): + return "the report service account" + return str(info.get("client_email") or "the report service account") + def get_drive_service(self) -> Resource: """Get an authenticated Google Drive API service. diff --git a/app/google_docs/utils/constants.py b/app/google_docs/utils/constants.py index 48e8a5a..27124c8 100644 --- a/app/google_docs/utils/constants.py +++ b/app/google_docs/utils/constants.py @@ -14,6 +14,10 @@ # Folder management DEFAULT_MAX_RECURSION_DEPTH = 10 # Maximum depth for recursive folder search +# Drive hides what the caller can't see, so a 404 here means "not shared" +# rather than "deleted". The search and the parent probe must agree on these. +DRIVE_INACCESSIBLE_STATUSES = (403, 404) + # TODO(namankhare): https://github.com/rtCamp/rt-report-automation/issues/67 # The folder name is subject to change and will be updated once the final # naming decision is made. diff --git a/app/inngest_proxy/controller.py b/app/inngest_proxy/controller.py index 5798f2a..5d32787 100644 --- a/app/inngest_proxy/controller.py +++ b/app/inngest_proxy/controller.py @@ -2,6 +2,7 @@ from fastapi import APIRouter +from app.inngest_proxy.models import RunStatusResponse from app.inngest_proxy.service import InngestProxyService router = APIRouter( @@ -18,32 +19,85 @@ description=( "Check the status of an Inngest run by event ID. " "Proxies the request server-side to the Inngest API to avoid " - "browser CORS restrictions." + "browser CORS restrictions, and normalizes the result into a " + "toast-ready status with structured failure detail." ), + response_model=RunStatusResponse, + response_model_exclude_none=False, responses={ 200: { - "description": "Inngest run status retrieved successfully", + "description": "Run status retrieved successfully", "content": { "application/json": { - "example": { - "data": [ - { + "examples": { + "completed": { + "summary": "Report generated", + "value": { + "event_id": "01KPZ2GTFVR7X2X4V0B1Q9QS6X", "run_id": "01KPZ2GTK23KKQBD1Y3NYYHC8E", - "run_started_at": "2026-04-24T06:23:32.962Z", - "function_id": "9577baf4-7d44-572b-acb4-b04cf05e487d", - "function_version": 0, - "environment_id": "00000000-0000-0000-0000-000000000000", # noqa: E501 + "status": "completed", + "is_terminal": True, + "message": "Report generated successfully.", + "document_url": "https://docs.google.com/document/d/example/edit", + "error": None, + }, + }, + "failed_user_fixable": { + "summary": "Failed for a reason the PM can fix", + "value": { "event_id": "01KPZ2GTFVR7X2X4V0B1Q9QS6X", - "status": "Completed", - "ended_at": "2026-04-24T06:24:18.984143Z", - "output": { - "document_url": "https://docs.google.com/document/d/example/edit", + "run_id": "01KPZ2GTK23KKQBD1Y3NYYHC8E", + "status": "failed", + "is_terminal": True, + "message": ( + "The report bot does not have access to the " + "Slack channel. Invite the bot to the channel, " + "then try again." + ), + "document_url": None, + "error": { + "error_code": "slack_access_denied", + "user_message": ( + "The report bot does not have access to " + "the Slack channel." + ), + "action": ( + "Invite the bot to the channel, then try again." + ), + "is_user_fixable": True, + "technical_detail": "SlackApiError: not_in_channel", + "trace_id": "01KPZ2GTK23KKQBD1Y3NYYHC8E", + "occurred_at": "2026-04-24T06:24:18.984143Z", + }, + }, + }, + "failed_unknown": { + "summary": "Unrecognised failure, escalate with trace ID", + "value": { + "event_id": "01KPZ2GTFVR7X2X4V0B1Q9QS6X", + "run_id": "01KPZ2GTK23KKQBD1Y3NYYHC8E", + "status": "failed", + "is_terminal": True, + "message": ( + "Report generation failed for an unexpected " + "reason. Share the trace ID with engineering " + "so they can check the logs." + ), + "document_url": None, + "error": { + "error_code": "unknown", + "user_message": ( + "Report generation failed for an unexpected " + "reason. Share the trace ID with engineering " + "so they can check the logs." + ), + "action": None, + "is_user_fixable": False, + "technical_detail": "RuntimeError: unexpected", + "trace_id": "01KPZ2GTK23KKQBD1Y3NYYHC8E", + "occurred_at": "2026-04-24T06:24:18.984143Z", }, - } - ], - "metadata": { - "fetched_at": "2026-04-24T12:29:19.445086Z", - "cached_until": "2026-04-24T12:29:34.445086Z", + }, }, }, }, @@ -51,6 +105,6 @@ } }, ) -async def get_run_status(event_id: str): +async def get_run_status(event_id: str) -> RunStatusResponse: """Proxy endpoint to check Inngest run status by event ID.""" return await inngest_proxy_service.get_run_status(event_id) diff --git a/app/inngest_proxy/errors.py b/app/inngest_proxy/errors.py new file mode 100644 index 0000000..c33073f --- /dev/null +++ b/app/inngest_proxy/errors.py @@ -0,0 +1,380 @@ +"""Classification of Inngest run failures into user-facing error details.""" + +from __future__ import annotations + +import json +import re +from enum import StrEnum +from functools import lru_cache +from typing import Any, NamedTuple + +from app.core.config import settings +from app.google_docs.utils.constants import AUTOMATED_DOCS_FOLDER_NAME + + +class RunErrorCode(StrEnum): + """Stable error codes the frontend can branch on.""" + + INVALID_DATE_RANGE = "invalid_date_range" + NO_DATA_FOR_FILTERS = "no_data_for_filters" + SLACK_CHANNEL_NOT_FOUND = "slack_channel_not_found" + SLACK_ACCESS_DENIED = "slack_access_denied" + GITHUB_ACCESS_DENIED = "github_access_denied" + GOOGLE_DRIVE_PERMISSION = "google_drive_permission" + AUTOMATED_DOCS_FOLDER_MISSING = "automated_docs_folder_missing" + DRIVE_FOLDER_NOT_SHARED = "drive_folder_not_shared" + DRIVE_STORAGE_FULL = "drive_storage_full" + GOOGLE_AUTH_FAILED = "google_auth_failed" + INVALID_DRIVE_LINK = "invalid_drive_link" + TEMPLATE_MISMATCH = "template_mismatch" + MISSING_PROJECT_FIELD = "missing_project_field" + INVALID_INPUT = "invalid_input" + LLM_RATE_LIMITED = "llm_rate_limited" + LLM_UNAVAILABLE = "llm_unavailable" + CONTENT_TOO_LARGE = "content_too_large" + UPSTREAM_TIMEOUT = "upstream_timeout" + RUN_CANCELLED = "run_cancelled" + NO_DOCUMENT_PRODUCED = "no_document_produced" + UNKNOWN = "unknown" + + +class ErrorRule(NamedTuple): + """A single classification rule. + + Attributes: + code: The stable error code assigned on a match. + pattern: Case-insensitive regex matched against the raw failure text. + user_message: Plain-language explanation shown in the toast/status. + action: Optional next step for the user, when the cause is user-fixable. + + """ + + code: RunErrorCode + pattern: re.Pattern[str] + user_message: str + action: str | None = None + + +def _rule( + code: RunErrorCode, + pattern: str, + user_message: str, + action: str | None = None, +) -> ErrorRule: + """Build an ErrorRule with a compiled, case-insensitive pattern.""" + return ErrorRule(code, re.compile(pattern, re.IGNORECASE), user_message, action) + + +@lru_cache(maxsize=1) +def get_service_account_email() -> str | None: + """Return the service account address, or None if it can't be read. + + Drive fixes ("share the folder with...") are only actionable if the message + names the address to share with, so it is resolved from the configured + service account key rather than left for the PM to hunt down. + """ + try: + info = json.loads(settings.GOOGLE_SERVICE_ACCOUNT_KEY.get_secret_value()) + except (json.JSONDecodeError, AttributeError, ValueError): + return None + + email = info.get("client_email") if isinstance(info, dict) else None + return str(email) if email else None + + +def _render_action(action: str | None) -> str | None: + """Fill placeholders in a rule's action text with live configuration.""" + if not action: + return action + + if "{service_account}" in action: + email = get_service_account_email() + # Without the address, naming it is worse than describing it generally. + action = action.replace( + "{service_account}", + email or "the report bot's service account", + ) + + return action.replace("{automated_docs_folder}", AUTOMATED_DOCS_FOLDER_NAME) + + +# Drive folder IDs as they appear in our own error text, e.g. +# "... not found in parent folder 1cXu0Turb42LYKqyskt-p_hE-RozsqDCs or its ..." +_DRIVE_FOLDER_ID_PATTERN = re.compile( + r"(?:parent folder|folder id|file not found:)\s+([A-Za-z0-9_-]{15,})", + re.IGNORECASE, +) + + +def extract_drive_folder_id(text: str) -> str | None: + """Pull the Drive folder ID out of a failure message, if it names one.""" + match = _DRIVE_FOLDER_ID_PATTERN.search(text) + return match.group(1) if match else None + + +def drive_folder_url(folder_id: str) -> str: + """Build the Drive URL for a folder ID, so the user can open it directly.""" + return f"https://drive.google.com/drive/folders/{folder_id}" + + +# Ordered most-specific first -- the first matching rule wins, so a narrow +# cause (e.g. "channel_not_found") is not swallowed by a broader one +# (e.g. a generic Slack API error). +ERROR_RULES: tuple[ErrorRule, ...] = ( + _rule( + RunErrorCode.INVALID_DATE_RANGE, + r"start_date must be less than or equal to end_date|invalid date range", + "The selected date range is invalid.", + "Pick a start date on or before the end date, then try again.", + ), + _rule( + RunErrorCode.SLACK_CHANNEL_NOT_FOUND, + r"channel_not_found|no channel (was )?found|channel .* not found", + "The Slack channel for this project could not be found.", + "Check the channel slug on the project in PMS, then try again.", + ), + _rule( + RunErrorCode.SLACK_ACCESS_DENIED, + r"not_in_channel|missing_scope|invalid_auth|account_inactive|token_revoked", + "The report bot does not have access to the Slack channel.", + "Invite the bot to the channel, then try again.", + ), + _rule( + RunErrorCode.NO_DATA_FOR_FILTERS, + r"no standup|no messages|no data (found|available)|empty (result|data)", + "No standup data was found for the selected project and date range.", + "Widen the date range or confirm standups were posted, then try again.", + ), + _rule( + RunErrorCode.GITHUB_ACCESS_DENIED, + r"github.*(401|403|bad credentials|not accessible|installation)" + r"|failed to get installation token|failed to generate jwt", + "GitHub data could not be fetched for this project.", + "Confirm the repository is correct and the GitHub App is installed on it.", + ), + _rule( + RunErrorCode.INVALID_DRIVE_LINK, + r"invalid google drive folder link|drive link cannot be empty" + r"|invalid google docs url", + "The project's Google Drive link is missing or malformed.", + "Fix the drive link on the project in PMS, then try again.", + ), + _rule( + RunErrorCode.AUTOMATED_DOCS_FOLDER_MISSING, + r"folder '?automated docs'? not found|please create the folder manually", + ( + "The project's Drive folder has no '{automated_docs_folder}' subfolder, " + "which is where generated reports are filed." + ), + ( + "Create a folder named '{automated_docs_folder}' inside the project's " + "Drive folder, then try again." + ), + ), + _rule( + RunErrorCode.DRIVE_STORAGE_FULL, + r"storagequotaexceeded|quota.*storage|drive storage", + ( + "The report couldn't be saved to Drive: the service account has no " + "storage of its own, so it can only create files inside a Shared " + "Drive." + ), + ("Move the project's Drive folder into a Shared Drive, then try again."), + ), + _rule( + RunErrorCode.GOOGLE_AUTH_FAILED, + r"failed to refresh service account credentials" + r"|invalid_grant|invalid jwt|service account key", + "The report bot could not authenticate with Google.", + ), + _rule( + RunErrorCode.DRIVE_FOLDER_NOT_SHARED, + # A 404 from Drive on a folder we were handed almost always means "not + # shared with us" rather than "deleted" -- Drive hides what the caller + # cannot see, so this is reported as a sharing problem. + r"cannot access drive folder|file not found: |notfound.*folder" + r"|folder not found|access denied" + r"|does not have permission|the user does not have sufficient permissions", + "The report bot cannot see the project's Google Drive folder.", + ( + "Share the project's Drive folder with {service_account} as an Editor, " + "then try again." + ), + ), + _rule( + RunErrorCode.GOOGLE_DRIVE_PERMISSION, + r"check folder permissions|check document permissions" + r"|insufficient permission|permission denied|forbidden|403", + "The report bot cannot write to the project's Google Drive folder.", + ("Share the folder with {service_account} as an Editor, then try again."), + ), + _rule( + RunErrorCode.TEMPLATE_MISMATCH, + r"check template tags|hours breakdown table shape mismatch" + r"|no document id returned", + "The Google Docs template does not match what the report expects.", + ), + _rule( + RunErrorCode.MISSING_PROJECT_FIELD, + r"missing required field", + "A required project field is missing.", + "Fill in the missing field on the project in PMS, then try again.", + ), + _rule( + RunErrorCode.CONTENT_TOO_LARGE, + r"context (length|window)|too many tokens|maximum context|request too large", + "There was too much data in the selected range to summarize in one report.", + "Narrow the date range, then try again.", + ), + _rule( + RunErrorCode.LLM_RATE_LIMITED, + r"rate limit|429|too many requests|quota exceeded|throttl", + "The AI provider is rate limiting requests right now.", + "Wait a few minutes, then try again.", + ), + _rule( + RunErrorCode.LLM_UNAVAILABLE, + r"overloaded|service unavailable|503|502|api (error|connection error)" + r"|upstream (error|connect)", + "The AI provider is temporarily unavailable.", + "Wait a few minutes, then try again.", + ), + _rule( + RunErrorCode.UPSTREAM_TIMEOUT, + r"timeout|timed out|deadline exceeded", + "A step in report generation timed out.", + "Try again; if it keeps failing, narrow the date range.", + ), + _rule( + RunErrorCode.INVALID_INPUT, + r"validation error|invalid model metadata|cannot be empty" + r"|expected \w+ for|invalid input", + "Some of the submitted report details were invalid.", + "Review the form values, then try again.", + ), +) + +GENERIC_USER_MESSAGE = ( + "Report generation failed for an unexpected reason. " + "Share the trace ID with engineering so they can check the logs." +) + +# Terminal states Inngest reports for a run that will not produce a document. +FAILED_STATUSES = frozenset({"Failed"}) +CANCELLED_STATUSES = frozenset({"Cancelled"}) + +# Cap on how much raw failure text is echoed back, so a multi-thousand-line +# stack trace can't bloat every poll response. +MAX_TECHNICAL_DETAIL_CHARS = 2000 + + +def extract_failure_text(output: Any, *, include_stack: bool = True) -> str: + """Flatten an Inngest run `output` into searchable text. + + Inngest reports a failure as an object with `name`, `message` and `stack` + keys, but a function that raised something unusual can leave a bare string + or an arbitrary dict there, so every shape is handled. + + Args: + output: The `output` value from an Inngest run record. + include_stack: Include the traceback. Wanted when matching rules, + not when building a user-facing detail string. + + Returns: + str: Text suitable for pattern matching, empty if nothing usable. + + """ + if output is None: + return "" + + if isinstance(output, str): + return output + + if isinstance(output, dict): + keys = ("name", "code", "message", "error", "detail") + if include_stack: + keys = (*keys, "stack") + parts = [str(output[key]) for key in keys if output.get(key)] + return "\n".join(dedupe_repeated_text(parts)) if parts else str(output) + + return str(output) + + +def dedupe_repeated_text(parts: list[str]) -> list[str]: + """Drop parts already contained in an earlier part. + + `log_and_raise(..., cause=e)` appends the cause's text to its own message, + so by the time a failure surfaces in Inngest the same sentence can appear + two or three times over ("Failed to generate Google Doc: X: X"). Keeping + every copy makes `technical_detail` hard to read for no added information. + """ + kept: list[str] = [] + for part in parts: + stripped = part.strip() + if not stripped or any(stripped in seen for seen in kept): + continue + # A new part that subsumes an earlier one replaces it. + kept = [seen for seen in kept if seen not in stripped] + kept.append(stripped) + return kept + + +def collapse_repeated_sentence(text: str) -> str: + """Collapse a message that repeats the same trailing sentence. + + `log_and_raise(..., cause=e)` builds its message as "context: {cause}", so + when the cause's own message is already the full sentence the result reads + "Failed to generate Google Doc: X: X". Only the final copy carries new + information; the duplicates are noise in `technical_detail`. + """ + collapsed_lines = [] + + for line in text.splitlines(): + # Repeatedly strip a trailing ": X" whose X already ends the head. + current = line.strip() + while True: + head, found, tail = current.rpartition(": ") + tail = tail.strip() + if not found or not tail or not head.strip().endswith(tail): + break + current = head.strip() + collapsed_lines.append(current) + + return "\n".join(collapsed_lines) + + +def classify_failure(output: Any) -> tuple[RunErrorCode, str, str | None]: + """Map a failed run's output onto a user-facing error code and message. + + Args: + output: The `output` value from a failed Inngest run record. + + Returns: + tuple[RunErrorCode, str, str | None]: The error code, the plain-language + user message, and a suggested user action (None when the cause is + not user-fixable). + + """ + text = extract_failure_text(output) + + for rule in ERROR_RULES: + if rule.pattern.search(text): + return ( + rule.code, + _render_action(rule.user_message) or rule.user_message, + _render_action(rule.action), + ) + + return RunErrorCode.UNKNOWN, GENERIC_USER_MESSAGE, None + + +def truncate_detail(text: str) -> str | None: + """Trim raw failure text to a bounded, single-line-safe detail string.""" + cleaned = text.strip() + if not cleaned: + return None + + if len(cleaned) <= MAX_TECHNICAL_DETAIL_CHARS: + return cleaned + + return f"{cleaned[:MAX_TECHNICAL_DETAIL_CHARS]}… (truncated)" diff --git a/app/inngest_proxy/models/__init__.py b/app/inngest_proxy/models/__init__.py new file mode 100644 index 0000000..83a9394 --- /dev/null +++ b/app/inngest_proxy/models/__init__.py @@ -0,0 +1,5 @@ +"""Models for the Inngest proxy module.""" + +from app.inngest_proxy.models.models import RunErrorDetail, RunStatus, RunStatusResponse + +__all__ = ["RunErrorDetail", "RunStatus", "RunStatusResponse"] diff --git a/app/inngest_proxy/models/models.py b/app/inngest_proxy/models/models.py new file mode 100644 index 0000000..c03d874 --- /dev/null +++ b/app/inngest_proxy/models/models.py @@ -0,0 +1,102 @@ +"""Response models for the Inngest run-status proxy.""" + +from __future__ import annotations + +from enum import StrEnum +from typing import Any + +from pydantic import BaseModel, Field + +# Imported at runtime, not only for typing: Pydantic needs the concrete enum +# to validate and to emit it in the OpenAPI schema. +from app.inngest_proxy.errors import RunErrorCode # noqa: TC001 + + +class RunStatus(StrEnum): + """Coarse run status the frontend renders, normalized from Inngest's own.""" + + PENDING = "pending" + RUNNING = "running" + COMPLETED = "completed" + FAILED = "failed" + CANCELLED = "cancelled" + + +class RunErrorDetail(BaseModel): + """Structured failure detail for a terminal, unsuccessful run.""" + + error_code: RunErrorCode = Field( + description="Stable machine-readable code for this failure class.", + ) + user_message: str = Field( + description="Plain-language explanation safe to show in a toast.", + ) + action: str | None = Field( + default=None, + description="Suggested next step, when the cause is user-fixable.", + ) + is_user_fixable: bool = Field( + description="True when the user can resolve this without engineering.", + ) + technical_detail: str | None = Field( + default=None, + description="Raw failure text from Inngest, truncated. For engineering.", + ) + resource_url: str | None = Field( + default=None, + description=( + "Link to the specific resource the user must fix (e.g. the Google " + "Drive folder that needs an 'Automated Docs' subfolder)." + ), + ) + trace_id: str = Field( + description="Copyable identifier engineering can use to find the logs.", + ) + occurred_at: str | None = Field( + default=None, + description="ISO-8601 timestamp of when the run failed.", + ) + + +class RunStatusResponse(BaseModel): + """Normalized run status for the report-generation frontend.""" + + event_id: str = Field(description="The Inngest event ID that was polled.") + run_id: str | None = Field( + default=None, + description="The Inngest run ID, absent until a run has been created.", + ) + function_id: str | None = Field( + default=None, + description="Which Inngest function produced this run.", + ) + status: RunStatus = Field(description="Coarse run status to render.") + is_terminal: bool = Field( + description="True when polling should stop -- no further status change.", + ) + message: str = Field(description="Human-readable status line for the toast.") + document_url: str | None = Field( + default=None, + description="URL of the generated report, present only on success.", + ) + error: RunErrorDetail | None = Field( + default=None, + description="Failure detail, present only for failed or cancelled runs.", + ) + started_at: str | None = Field( + default=None, + description="ISO-8601 timestamp of when the run started.", + ) + ended_at: str | None = Field( + default=None, + description="ISO-8601 timestamp of when the run ended.", + ) + raw: dict[str, Any] | None = Field( + default=None, + description=( + "The unmodified Inngest run record. Debug aid, omitted unless the " + "deployment runs with DEBUG enabled -- it carries full stack " + "traces and unbounded step output, which `technical_detail` " + "deliberately strips and truncates for user-facing use." + ), + ) diff --git a/app/inngest_proxy/service.py b/app/inngest_proxy/service.py index 41a49ce..f927ae0 100644 --- a/app/inngest_proxy/service.py +++ b/app/inngest_proxy/service.py @@ -1,15 +1,47 @@ """Service layer for proxying requests to the Inngest API.""" +from __future__ import annotations + import logging +from typing import Any import httpx from app.core.config import settings from app.core.utils import log_and_raise from app.inngest_proxy.constants import INNGEST_API_BASE_URL, INNGEST_API_TIMEOUT +from app.inngest_proxy.errors import ( + CANCELLED_STATUSES, + FAILED_STATUSES, + RunErrorCode, + classify_failure, + collapse_repeated_sentence, + drive_folder_url, + extract_drive_folder_id, + extract_failure_text, + truncate_detail, +) +from app.inngest_proxy.models import RunErrorDetail, RunStatus, RunStatusResponse logger = logging.getLogger(__name__) +# Error codes that only engineering can act on -- everything else in the +# rule table is something the PM can fix themselves. +_NOT_USER_FIXABLE = frozenset( + { + RunErrorCode.UNKNOWN, + RunErrorCode.TEMPLATE_MISMATCH, + RunErrorCode.NO_DOCUMENT_PRODUCED, + RunErrorCode.GOOGLE_AUTH_FAILED, + }, +) + +_STATE_MESSAGES = { + RunStatus.PENDING: "Report generation is queued…", + RunStatus.RUNNING: "Report generation is in progress…", + RunStatus.COMPLETED: "Report generated successfully.", +} + class InngestProxyService: """Service for proxying status requests to the Inngest REST API.""" @@ -18,19 +50,53 @@ def __init__(self): """Initialize the InngestProxyService.""" self.signing_key = settings.INNGEST_SIGNING_KEY.get_secret_value() - async def get_run_status(self, event_id: str) -> dict: - """Fetch run status from the Inngest API for a given event ID. + async def get_run_status(self, event_id: str) -> RunStatusResponse: + """Fetch and normalize the run status for a given event ID. Args: event_id (str): The Inngest event ID to look up. Returns: - dict: The JSON response from the Inngest API. + RunStatusResponse: Normalized status with a toast-ready message and, + on failure, structured error detail. Raises: HTTPException: If the Inngest API returns an error or is unreachable. """ + payload = await self._fetch_runs(event_id) + runs = payload.get("data") or [] + run = self._select_run(runs) + + if run is None: + # Inngest has accepted the event but not yet materialized a run. + return RunStatusResponse( + event_id=event_id, + status=RunStatus.PENDING, + is_terminal=False, + message="Waiting for the run to start…", + ) + + return self._build_response(event_id, run) + + @staticmethod + def _select_run(runs: Any) -> dict | None: + """Pick the live run for an event. + + List order isn't guaranteed, and `retries` leaves earlier attempts + alongside the current one, so the newest start wins. + """ + if not isinstance(runs, list): + return None + + candidates = [r for r in runs if isinstance(r, dict)] + if not candidates: + return None + + return max(candidates, key=lambda r: str(r.get("run_started_at") or "")) + + async def _fetch_runs(self, event_id: str) -> dict[str, Any]: + """Fetch the raw runs payload for an event from the Inngest API.""" url = f"{INNGEST_API_BASE_URL}/events/{event_id}/runs" try: @@ -56,3 +122,140 @@ async def get_run_status(self, event_id: str) -> dict: http_status_code=502, cause=exc, ) + + def _build_response(self, event_id: str, run: dict) -> RunStatusResponse: + """Normalize a single Inngest run record into the frontend response.""" + run_id = run.get("run_id") + output = run.get("output") + status = self._derive_state(run.get("status")) + + document_url = None + error = None + + if status is RunStatus.COMPLETED: + document_url = ( + output.get("document_url") if isinstance(output, dict) else None + ) + if not run.get("ended_at"): + # Inngest reports "Completed" before the run has actually + # finished -- `ended_at` and `output` are still null. Treat + # that as in-flight so a mid-run poll doesn't declare either + # success or failure prematurely. + status = RunStatus.RUNNING + document_url = None + elif not document_url: + # A genuinely finished run with no URL is a real failure from + # the user's point of view -- report it as one instead of + # leaving the frontend to guess. + status = RunStatus.FAILED + error = self._build_error( + run, + code=RunErrorCode.NO_DOCUMENT_PRODUCED, + user_message=( + "Report generation finished but no document was produced." + ), + action=None, + ) + elif status is RunStatus.CANCELLED: + error = self._build_error( + run, + code=RunErrorCode.RUN_CANCELLED, + user_message="Report generation was cancelled before it finished.", + action="Trigger the report again.", + ) + elif status is RunStatus.FAILED: + code, user_message, action = classify_failure(output) + error = self._build_error( + run, + code=code, + user_message=user_message, + action=action, + ) + + if error is not None: + # The response only carries the stripped, truncated detail, so the + # full record -- stack trace and all -- has to land in the server + # logs or the trace ID the user copies leads nowhere. + logger.error( + "Inngest run %s (event %s) ended as %s [%s]: %r", + run_id, + event_id, + status.value, + error.error_code.value, + run, + ) + + return RunStatusResponse( + event_id=event_id, + run_id=run_id, + function_id=run.get("function_id"), + status=status, + is_terminal=status + in {RunStatus.COMPLETED, RunStatus.FAILED, RunStatus.CANCELLED}, + message=self._build_message(status, error), + document_url=document_url, + error=error, + started_at=run.get("run_started_at"), + ended_at=run.get("ended_at"), + # Stack traces and unbounded step output must not ride along on + # every poll of a user-facing response -- `technical_detail` is the + # bounded, stack-free version. Kept for local debugging only. + raw=run if settings.DEBUG else None, + ) + + @staticmethod + def _derive_state(status: str | None) -> RunStatus: + """Map Inngest's raw status string onto a coarse run status.""" + if status in FAILED_STATUSES: + return RunStatus.FAILED + if status in CANCELLED_STATUSES: + return RunStatus.CANCELLED + if status == "Completed": + return RunStatus.COMPLETED + if status == "Running": + return RunStatus.RUNNING + return RunStatus.PENDING + + @staticmethod + def _build_error( + run: dict, + code: RunErrorCode, + user_message: str, + action: str | None, + ) -> RunErrorDetail: + """Assemble the structured error detail for a terminal failed run.""" + # The run ID is the trace ID: it is what engineering searches Inngest + # and the application logs by, and it is stable across polls so the + # value the user copies stays valid. + trace_id = run.get("run_id") or run.get("event_id") or "unavailable" + + # Stack traces help classification but are noise in a user-facing + # payload -- the full trace stays available under `raw`. + raw_text = extract_failure_text(run.get("output"), include_stack=False) + match_text = extract_failure_text(run.get("output")) + + # When the failure names a Drive folder, hand the user a direct link to + # it -- the fix ("create a subfolder here") is a click away, and a bare + # folder ID in a log line is not something a PM can act on. + folder_id = extract_drive_folder_id(match_text) + + return RunErrorDetail( + error_code=code, + user_message=user_message, + action=action, + is_user_fixable=code not in _NOT_USER_FIXABLE, + technical_detail=truncate_detail(collapse_repeated_sentence(raw_text)), + resource_url=drive_folder_url(folder_id) if folder_id else None, + trace_id=str(trace_id), + occurred_at=run.get("ended_at"), + ) + + @staticmethod + def _build_message(status: RunStatus, error: RunErrorDetail | None) -> str: + """Build the single status line the toast renders.""" + if error is not None: + if error.action: + return f"{error.user_message} {error.action}" + return error.user_message + + return _STATE_MESSAGES.get(status, "Report generation is in progress…")