Notifications

Discord notification system for task updates and agent events.

Notification formatting for Discord messages about task lifecycle events.

String formatters (format_*) produce human-readable markdown strings that are easy to unit test without a live Discord connection and are used for logging and plain-text fallback.

Embed formatters (format_*_embed) produce discord.Embed objects for rich Discord presentation — color-coded by severity, with structured fields for task metadata. The orchestrator passes both versions through the notification callback so the bot can choose the appropriate format.

Interactive views (TaskFailedView, TaskApprovalView, AgentQuestionView) attach action buttons to notification embeds so users can retry, skip, approve tasks, or reply to agent questions directly from Discord without memorizing slash commands.

classify_error pattern-matches raw error messages against known failure modes and returns an actionable fix suggestion -- this turns opaque stack traces into guidance the user can act on immediately from Discord.

Classes

TaskStartedView

TaskStartedView(task_id: str, handler=None)

Bases: View

Action button attached to task-started notifications.

Provides a one-click "Stop Task" button so the user can cancel a running task directly from the notification message without needing to find the task ID or type a slash command — especially useful on mobile.

Source code in src/discord/notifications.py

def __init__(self, task_id: str, handler=None) -> None:
    super().__init__(timeout=86400)  # 24 hours
    self.task_id = task_id
    self._handler = handler

TaskFailedView

TaskFailedView(task_id: str, handler=None)

Bases: View

Action buttons attached to failed task notifications.

Provides one-click Retry and Skip buttons so the user doesn't have to remember /restart-task or /skip-task slash commands. The handler is passed at creation time and called when buttons are pressed.

Source code in src/discord/notifications.py

def __init__(self, task_id: str, handler=None) -> None:
    super().__init__(timeout=3600)  # 1 hour
    self.task_id = task_id
    self._handler = handler

TaskApprovalView

TaskApprovalView(task_id: str, handler=None)

Bases: View

Action buttons attached to PR-created / awaiting-approval notifications.

Provides one-click Approve and Restart buttons for tasks in AWAITING_APPROVAL status.

Source code in src/discord/notifications.py

def __init__(self, task_id: str, handler=None) -> None:
    super().__init__(timeout=86400)  # 24 hours
    self.task_id = task_id
    self._handler = handler

AgentReplyModal

AgentReplyModal(task_id: str, handler=None)

Bases: Modal

Modal dialog for replying to an agent's question.

Opens a text input where the user can type their response. On submit the reply is forwarded via CommandHandler.execute("provide_input", …) so the agent's next execution cycle receives the user's answer.

Source code in src/discord/notifications.py

def __init__(self, task_id: str, handler=None) -> None:
    super().__init__()
    self.task_id = task_id
    self._handler = handler

TaskBlockedView

TaskBlockedView(task_id: str, handler=None)

Bases: View

Action buttons for blocked task notifications.

Provides Restart and Skip buttons for tasks that have exhausted retries.

Source code in src/discord/notifications.py

def __init__(self, task_id: str, handler=None) -> None:
    super().__init__(timeout=86400)  # 24 hours
    self.task_id = task_id
    self._handler = handler

AgentQuestionModal

AgentQuestionModal(task_id: str, handler=None)

Bases: Modal

Modal dialog for typing a reply to an agent question.

Opened when the user clicks the "Reply" button on an agent question notification. On submit, calls CommandHandler.execute("provide_input", ...) to transition the task from WAITING_INPUT → READY with the user's reply appended to the task description.

Source code in src/discord/notifications.py

def __init__(self, task_id: str, handler=None) -> None:
    super().__init__()
    self.task_id = task_id
    self._handler = handler

AgentQuestionView

AgentQuestionView(task_id: str, handler=None)

Bases: View

Action buttons attached to agent question notifications.

Provides a one-click "Reply" button that opens a modal text input so the user can answer the agent's question without memorizing slash commands. A secondary "Skip Task" button allows skipping the task entirely.

Source code in src/discord/notifications.py

def __init__(self, task_id: str, handler=None) -> None:
    super().__init__(timeout=86400)  # 24 hours
    self.task_id = task_id
    self._handler = handler

Functions

format_server_started

format_server_started() -> str

Plain-text message indicating the server is back online.

Source code in src/discord/notifications.py

def format_server_started() -> str:
    """Plain-text message indicating the server is back online."""
    return "✅ **AgentQueue is back online** — the server has started and is ready to process tasks."

format_server_started_embed

format_server_started_embed() -> discord.Embed

Rich embed announcing the server is back online.

Uses a green success embed to clearly signal that the system is operational and ready to accept work.

Source code in src/discord/notifications.py

def format_server_started_embed() -> discord.Embed:
    """Rich embed announcing the server is back online.

    Uses a green success embed to clearly signal that the system is
    operational and ready to accept work.
    """
    return success_embed(
        "Server Online",
        description=(
            "AgentQueue has started and is ready to process tasks.\n\n"
            "All systems are operational — commands, notifications, and "
            "task orchestration are now available."
        ),
    )

classify_error

classify_error(error_message: str | None) -> tuple[str, str]

Return (error_type_label, fix_suggestion) for a given error message.

Falls back to a generic label when no pattern matches.

Source code in src/discord/notifications.py

def classify_error(error_message: str | None) -> tuple[str, str]:
    """Return (error_type_label, fix_suggestion) for a given error message.

    Falls back to a generic label when no pattern matches.
    """
    if not error_message:
        return "Unknown error", "Check daemon logs for details."
    lowered = error_message.lower()
    for keyword, label, suggestion in _ERROR_PATTERNS:
        if keyword.lower() in lowered:
            return label, suggestion
    return "Unexpected error", "Check daemon logs (`~/.agent-queue/daemon.log`) for full details."

format_chain_stuck

format_chain_stuck(blocked_task: Task, stuck_tasks: list[Task]) -> str

Format a notification about downstream tasks stuck because of a blocked task.

Source code in src/discord/notifications.py

def format_chain_stuck(
    blocked_task: Task,
    stuck_tasks: list[Task],
) -> str:
    """Format a notification about downstream tasks stuck because of a blocked task."""
    task_list = ", ".join(f"`{t.id}`" for t in stuck_tasks[:5])
    if len(stuck_tasks) > 5:
        task_list += f" +{len(stuck_tasks) - 5} more"
    return (
        f"⛓️ **Chain Stuck:** `{blocked_task.id}` BLOCKED → "
        f"{len(stuck_tasks)} stuck: {task_list}\n"
        f"`/skip-task {blocked_task.id}` or `/restart-task {blocked_task.id}`"
    )

format_stuck_defined_task

format_stuck_defined_task(task: Task, blocking_deps: list[tuple[str, str, str]], stuck_hours: float) -> str

Format a notification for a DEFINED task stuck waiting on dependencies.

Source code in src/discord/notifications.py

def format_stuck_defined_task(
    task: Task,
    blocking_deps: list[tuple[str, str, str]],
    stuck_hours: float,
) -> str:
    """Format a notification for a DEFINED task stuck waiting on dependencies."""
    if blocking_deps:
        blockers = ", ".join(
            f"`{dep_id}` ({dep_status})" for dep_id, _, dep_status in blocking_deps[:3]
        )
        if len(blocking_deps) > 3:
            blockers += f" +{len(blocking_deps) - 3} more"
        return (
            f"⏳ **Stuck:** `{task.id}` — {task.title} "
            f"(DEFINED {stuck_hours:.1f}h, blocked by {blockers})\n"
            f"`/skip-task` or `/restart-task` the blocker to unblock"
        )
    return (
        f"⏳ **Stuck:** `{task.id}` — {task.title} "
        f"(DEFINED {stuck_hours:.1f}h, no unmet deps found — possible bug)"
    )

format_plan_generated

format_plan_generated(parent_task: Task, generated_tasks: list[Task], *, workspace_path: str | None = None, chained: bool = True) -> str

Format a plain-text notification for auto-generated plan subtasks.

Returns a human-readable markdown string listing all tasks created from a plan file, suitable for logging and fallback display.

Source code in src/discord/notifications.py

def format_plan_generated(
    parent_task: Task,
    generated_tasks: list[Task],
    *,
    workspace_path: str | None = None,
    chained: bool = True,
) -> str:
    """Format a plain-text notification for auto-generated plan subtasks.

    Returns a human-readable markdown string listing all tasks created
    from a plan file, suitable for logging and fallback display.
    """
    count = len(generated_tasks)
    lines = [
        f"📋 **Plan Generated — {count} Task{'s' if count != 1 else ''} Created**",
        f"Parent: `{parent_task.id}` — {parent_task.title}",
        f"Project: `{parent_task.project_id}`",
    ]
    if workspace_path:
        lines.append(f"Workspace: `{workspace_path}`")
    if chained and count > 1:
        chain_str = " → ".join(f"`{t.id}`" for t in generated_tasks)
        lines.append(f"Chain: {chain_str}")
    lines.append("")
    for idx, t in enumerate(generated_tasks, 1):
        type_emoji = ""
        if t.task_type:
            type_emoji = TASK_TYPE_EMOJIS.get(t.task_type.value, "") + " "
        lines.append(
            f"**{idx}.** {type_emoji}`{t.id}` — {t.title} (priority: {t.priority})"
        )
    return "\n".join(lines)

format_task_started_embed

format_task_started_embed(task: Task, agent: Agent, workspace: Workspace | None = None) -> discord.Embed

Rich embed version of :func:format_task_started.

Uses the IN_PROGRESS status color (amber) to visually indicate that the task is now actively being worked on by an agent.

Source code in src/discord/notifications.py

def format_task_started_embed(task: Task, agent: Agent, workspace: Workspace | None = None) -> discord.Embed:
    """Rich embed version of :func:`format_task_started`.

    Uses the IN_PROGRESS status color (amber) to visually indicate that
    the task is now actively being worked on by an agent.
    """
    fields: list[tuple[str, str, bool]] = [
        ("Task ID", f"`{task.id}`", True),
        ("Project", f"`{task.project_id}`", True),
        ("Agent", agent.name, True),
        ("Status", "\U0001F7E1 IN_PROGRESS", True),
    ]
    if workspace:
        label = workspace.name or workspace.workspace_path
        fields.append(("Workspace", f"`{label}`", True))
    if task.branch_name:
        fields.append(("Branch", f"`{task.branch_name}`", True))
    return status_embed(
        TaskStatus.IN_PROGRESS.value,
        f"Task Started — {task.title}",
        fields=fields,
    )

format_task_completed_embed

format_task_completed_embed(task: Task, agent: Agent, output: AgentOutput) -> discord.Embed

Rich embed version of :func:format_task_completed.

Source code in src/discord/notifications.py

def format_task_completed_embed(
    task: Task, agent: Agent, output: AgentOutput,
) -> discord.Embed:
    """Rich embed version of :func:`format_task_completed`."""
    fields: list[tuple[str, str, bool]] = [
        ("Task ID", f"`{task.id}`", True),
        ("Project", f"`{task.project_id}`", True),
        ("Agent", agent.name, True),
        ("Tokens Used", f"{output.tokens_used:,}", True),
    ]
    if output.summary:
        fields.append((
            "Summary",
            truncate(output.summary, LIMIT_FIELD_VALUE),
            False,
        ))
    if output.files_changed:
        files_text = ", ".join(f"`{f}`" for f in output.files_changed)
        fields.append((
            "Files Changed",
            truncate(files_text, LIMIT_FIELD_VALUE),
            False,
        ))
    return success_embed(f"Task Completed — {task.title}", fields=fields)

format_task_failed_embed

format_task_failed_embed(task: Task, agent: Agent, output: AgentOutput) -> discord.Embed

Rich embed version of :func:format_task_failed.

Source code in src/discord/notifications.py

def format_task_failed_embed(
    task: Task, agent: Agent, output: AgentOutput,
) -> discord.Embed:
    """Rich embed version of :func:`format_task_failed`."""
    error_type, suggestion = classify_error(output.error_message)
    fields: list[tuple[str, str, bool]] = [
        ("Task ID", f"`{task.id}`", True),
        ("Project", f"`{task.project_id}`", True),
        ("Agent", agent.name, True),
        ("Retries", f"{task.retry_count}/{task.max_retries}", True),
        ("Error Type", f"**{error_type}**", True),
    ]
    if output.error_message:
        # Code block markers consume ~8 chars; cap snippet to leave room.
        snippet = output.error_message[:300]
        if len(output.error_message) > 300:
            snippet += "\u2026"
        fields.append(("Error Detail", f"```\n{snippet}\n```", False))
    fields.append(("Suggestion", f"\U0001F4A1 {suggestion}", False))
    fields.append((
        "Next Step",
        f"Use `/agent-error {task.id}` for the full error log.",
        False,
    ))
    return error_embed(f"Task Failed — {task.title}", fields=fields)

format_task_blocked_embed

format_task_blocked_embed(task: Task, last_error: str | None = None) -> discord.Embed

Rich embed version of :func:format_task_blocked.

Source code in src/discord/notifications.py

def format_task_blocked_embed(
    task: Task, last_error: str | None = None,
) -> discord.Embed:
    """Rich embed version of :func:`format_task_blocked`."""
    fields: list[tuple[str, str, bool]] = [
        ("Task ID", f"`{task.id}`", True),
        ("Project", f"`{task.project_id}`", True),
        ("Status", f"Max retries ({task.max_retries}) exhausted", False),
    ]
    if last_error:
        error_type, suggestion = classify_error(last_error)
        fields.append(("Last Error Type", f"**{error_type}**", True))
        fields.append(("Suggestion", f"\U0001F4A1 {suggestion}", False))
    fields.append((
        "Action Required",
        f"Use `/agent-error {task.id}` to inspect the last error.",
        False,
    ))
    return critical_embed(f"Task Blocked — {task.title}", fields=fields)

format_pr_created_embed

format_pr_created_embed(task: Task, pr_url: str) -> discord.Embed

Rich embed version of :func:format_pr_created.

Source code in src/discord/notifications.py

def format_pr_created_embed(task: Task, pr_url: str) -> discord.Embed:
    """Rich embed version of :func:`format_pr_created`."""
    fields: list[tuple[str, str, bool]] = [
        ("Task ID", f"`{task.id}`", True),
        ("Project", f"`{task.project_id}`", True),
        ("Status", "AWAITING_APPROVAL", True),
        ("Pull Request", f"[Review and merge to complete]({pr_url})", False),
    ]
    return info_embed(f"PR Created — {task.title}", fields=fields, url=pr_url)

format_agent_question_embed

format_agent_question_embed(task: Task, agent: Agent, question: str) -> discord.Embed

Rich embed version of :func:format_agent_question.

Source code in src/discord/notifications.py

def format_agent_question_embed(
    task: Task, agent: Agent, question: str,
) -> discord.Embed:
    """Rich embed version of :func:`format_agent_question`."""
    fields: list[tuple[str, str, bool]] = [
        ("Task ID", f"`{task.id}`", True),
        ("Project", f"`{task.project_id}`", True),
        ("Agent", agent.name, True),
        ("Question", f"> {truncate(question, LIMIT_FIELD_VALUE - 2)}", False),
    ]
    return warning_embed(f"Agent Question — {task.title}", fields=fields)

format_chain_stuck_embed

format_chain_stuck_embed(blocked_task: Task, stuck_tasks: list[Task]) -> discord.Embed

Rich embed version of :func:format_chain_stuck.

Source code in src/discord/notifications.py

def format_chain_stuck_embed(
    blocked_task: Task,
    stuck_tasks: list[Task],
) -> discord.Embed:
    """Rich embed version of :func:`format_chain_stuck`."""
    task_list = "\n".join(
        f"\u2022 `{t.id}` \u2014 {t.title}" for t in stuck_tasks[:10]
    )
    if len(stuck_tasks) > 10:
        task_list += f"\n+{len(stuck_tasks) - 10} more"
    fields: list[tuple[str, str, bool]] = [
        ("Blocked Task", f"`{blocked_task.id}` \u2014 {blocked_task.title}", False),
        ("Project", f"`{blocked_task.project_id}`", True),
        ("Affected Tasks", str(len(stuck_tasks)), True),
        ("Downstream Tasks", truncate(task_list, LIMIT_FIELD_VALUE), False),
        (
            "Actions",
            f"`/skip-task {blocked_task.id}` or `/restart-task {blocked_task.id}`",
            False,
        ),
    ]
    return critical_embed(
        "Chain Stuck",
        description=(
            f"Task `{blocked_task.id}` is BLOCKED, preventing "
            f"{len(stuck_tasks)} downstream task(s) from running."
        ),
        fields=fields,
    )

format_stuck_defined_task_embed

format_stuck_defined_task_embed(task: Task, blocking_deps: list[tuple[str, str, str]], stuck_hours: float) -> discord.Embed

Rich embed version of :func:format_stuck_defined_task.

Source code in src/discord/notifications.py

def format_stuck_defined_task_embed(
    task: Task,
    blocking_deps: list[tuple[str, str, str]],
    stuck_hours: float,
) -> discord.Embed:
    """Rich embed version of :func:`format_stuck_defined_task`."""
    fields: list[tuple[str, str, bool]] = [
        ("Task ID", f"`{task.id}`", True),
        ("Project", f"`{task.project_id}`", True),
        ("Stuck Duration", f"{stuck_hours:.1f} hours", True),
    ]
    if blocking_deps:
        blockers = "\n".join(
            f"\u2022 `{dep_id}` ({dep_status})"
            for dep_id, _, dep_status in blocking_deps[:5]
        )
        if len(blocking_deps) > 5:
            blockers += f"\n+{len(blocking_deps) - 5} more"
        fields.append((
            "Blocking Dependencies",
            truncate(blockers, LIMIT_FIELD_VALUE),
            False,
        ))
        fields.append((
            "Actions",
            "`/skip-task` or `/restart-task` the blocker to unblock",
            False,
        ))
    else:
        fields.append((
            "Note",
            "No unmet dependencies found \u2014 possible bug",
            False,
        ))
    return warning_embed(
        f"Task Stuck — {task.title}",
        description=f"DEFINED for {stuck_hours:.1f}h, waiting on dependencies.",
        fields=fields,
    )

format_budget_warning_embed

format_budget_warning_embed(project_name: str, usage: int, limit: int) -> discord.Embed

Rich embed version of :func:format_budget_warning.

The embed color shifts from amber to orange to red as the budget utilization increases, providing an at-a-glance severity indicator.

Source code in src/discord/notifications.py

def format_budget_warning_embed(
    project_name: str, usage: int, limit: int,
) -> discord.Embed:
    """Rich embed version of :func:`format_budget_warning`.

    The embed color shifts from amber to orange to red as the budget
    utilization increases, providing an at-a-glance severity indicator.
    """
    pct = (usage / limit * 100) if limit > 0 else 0
    remaining = max(0, limit - usage)

    # Dynamic color: amber → orange → red as budget depletes
    if pct >= 95:
        color = 0xE74C3C   # Red
    elif pct >= 80:
        color = 0xE67E22   # Orange
    else:
        color = 0xF39C12   # Amber

    fields: list[tuple[str, str, bool]] = [
        ("Project", f"**{project_name}**", True),
        ("Used", f"{usage:,} tokens", True),
        ("Limit", f"{limit:,} tokens", True),
        ("Remaining", f"{remaining:,} tokens ({100 - pct:.0f}%)", True),
    ]
    return warning_embed(
        f"Budget Warning — {pct:.0f}% Used",
        fields=fields,
        color_override=color,
    )

format_plan_generated_embed

format_plan_generated_embed(parent_task: Task, generated_tasks: list[Task], *, workspace_path: str | None = None, chained: bool = True) -> discord.Embed

Rich embed for auto-generated plan subtasks.

Builds a visually structured embed that displays all relevant metadata for each auto-generated task: title, priority, project, workspace, task type, and dependency chain — making it easy to scan at a glance.

Parameters

parent_task: The task whose completion produced the plan file. generated_tasks: The list of newly created subtasks from the plan. workspace_path: The workspace directory used by the parent task (shown when available). chained: Whether tasks are chained in a sequential dependency order.

Source code in src/discord/notifications.py

def format_plan_generated_embed(
    parent_task: Task,
    generated_tasks: list[Task],
    *,
    workspace_path: str | None = None,
    chained: bool = True,
) -> discord.Embed:
    """Rich embed for auto-generated plan subtasks.

    Builds a visually structured embed that displays all relevant metadata
    for each auto-generated task: title, priority, project, workspace,
    task type, and dependency chain — making it easy to scan at a glance.

    Parameters
    ----------
    parent_task:
        The task whose completion produced the plan file.
    generated_tasks:
        The list of newly created subtasks from the plan.
    workspace_path:
        The workspace directory used by the parent task (shown when available).
    chained:
        Whether tasks are chained in a sequential dependency order.
    """
    count = len(generated_tasks)
    plural = "s" if count != 1 else ""

    # --- Description block ---------------------------------------------------
    desc_lines: list[str] = [
        f"Task `{parent_task.id}` completed with an implementation plan.",
        f"**{count}** subtask{plural} {'have' if count != 1 else 'has'} been "
        f"created{' and chained for sequential execution' if chained and count > 1 else ''}.",
    ]

    # Show dependency chain as a compact arrow diagram
    if chained and count > 1:
        chain_ids = " → ".join(f"`{t.id}`" for t in generated_tasks)
        desc_lines.append(f"\n**Execution order:** {chain_ids}")

    description = "\n".join(desc_lines)

    # --- Header fields -------------------------------------------------------
    fields: list[tuple[str, str, bool]] = [
        ("Parent Task", f"`{parent_task.id}`\n{truncate(parent_task.title, 80)}", True),
        ("Project", f"`{parent_task.project_id}`", True),
    ]

    if workspace_path:
        # Show only the last 2 path components to keep it readable
        short_path = workspace_path
        parts = workspace_path.replace("\\", "/").rstrip("/").split("/")
        if len(parts) > 2:
            short_path = "…/" + "/".join(parts[-2:])
        fields.append(("Workspace", f"`{short_path}`", True))

    # --- Separator -----------------------------------------------------------
    fields.append(("─── Subtasks ───", "\u200b", False))  # zero-width space value

    # --- Per-task fields -----------------------------------------------------
    for idx, t in enumerate(generated_tasks, 1):
        # Build the field name with step number and optional type emoji
        type_emoji = ""
        if t.task_type:
            type_emoji = TASK_TYPE_EMOJIS.get(t.task_type.value, "")
            if type_emoji:
                type_emoji += " "

        field_name = f"{type_emoji}Step {idx}/{count}: {truncate(t.title, 80)}"

        # Build the field value with key metadata
        detail_parts: list[str] = [f"**ID:** `{t.id}`"]

        detail_parts.append(f"**Priority:** {t.priority}")

        if t.task_type:
            detail_parts.append(
                f"**Type:** {TASK_TYPE_EMOJIS.get(t.task_type.value, '')} "
                f"`{t.task_type.value}`"
            )

        if t.requires_approval:
            detail_parts.append("🔒 **Requires approval**")

        # Show dependency info for chained tasks (except the first)
        if chained and idx > 1:
            prev = generated_tasks[idx - 2]
            detail_parts.append(f"⏳ Depends on `{prev.id}`")

        # Show a snippet of the description if available
        if t.description:
            # Extract first meaningful line (skip headers/blanks)
            snippet = _extract_description_snippet(t.description, max_len=120)
            if snippet:
                detail_parts.append(f"*{snippet}*")

        field_value = "\n".join(detail_parts)
        fields.append((field_name, truncate(field_value, 1024), False))

    # --- Build the embed with plan-generation color (teal/cyan) --------------
    # Use a distinctive teal color (0x1ABC9C) to stand out from standard
    # status notifications (success=green, error=red, etc.)
    _PLAN_GENERATED_COLOR = 0x1ABC9C

    embed = make_embed(
        EmbedStyle.INFO,
        f"Plan Generated — {count} New Task{plural}",
        description=truncate(description, LIMIT_DESCRIPTION),
        fields=fields,
        color_override=_PLAN_GENERATED_COLOR,
    )

    return embed

format_merge_conflict

format_merge_conflict(task: Task, branch_name: str, default_branch: str) -> str

Plain-text notification for a merge conflict during sync-and-merge.

Source code in src/discord/notifications.py

def format_merge_conflict(task: Task, branch_name: str, default_branch: str) -> str:
    """Plain-text notification for a merge conflict during sync-and-merge."""
    return (
        f"**Merge Conflict:** Task `{task.id}` — {task.title}\n"
        f"Project: `{task.project_id}`\n"
        f"Branch `{branch_name}` has conflicts with `{default_branch}`.\n"
        f"Manual resolution needed — check out the branch locally and resolve conflicts."
    )

format_merge_conflict_embed

format_merge_conflict_embed(task: Task, branch_name: str, default_branch: str) -> discord.Embed

Rich embed for a merge conflict during sync-and-merge.

Source code in src/discord/notifications.py

def format_merge_conflict_embed(
    task: Task, branch_name: str, default_branch: str,
) -> discord.Embed:
    """Rich embed for a merge conflict during sync-and-merge."""
    fields: list[tuple[str, str, bool]] = [
        ("Task ID", f"`{task.id}`", True),
        ("Project", f"`{task.project_id}`", True),
        ("Branch", f"`{branch_name}`", True),
        ("Target", f"`{default_branch}`", True),
        (
            "Action Required",
            f"Branch `{branch_name}` has conflicts with `{default_branch}`.\n"
            f"Check out the branch locally and resolve conflicts manually.",
            False,
        ),
    ]
    return error_embed(f"Merge Conflict — {task.title}", fields=fields)

format_push_failed

format_push_failed(task: Task, default_branch: str, error_detail: str) -> str

Plain-text notification for a push failure after retries.

Source code in src/discord/notifications.py

def format_push_failed(
    task: Task, default_branch: str, error_detail: str,
) -> str:
    """Plain-text notification for a push failure after retries."""
    return (
        f"**Push Failed:** Task `{task.id}` — {task.title}\n"
        f"Project: `{task.project_id}`\n"
        f"Could not push `{default_branch}` after retries. "
        f"The workspace may be diverged and require manual intervention.\n"
        f"```\n{error_detail[:300]}\n```"
    )

format_push_failed_embed

format_push_failed_embed(task: Task, default_branch: str, error_detail: str) -> discord.Embed

Rich embed for a push failure after retries.

Source code in src/discord/notifications.py

def format_push_failed_embed(
    task: Task, default_branch: str, error_detail: str,
) -> discord.Embed:
    """Rich embed for a push failure after retries."""
    snippet = error_detail[:300]
    if len(error_detail) > 300:
        snippet += "…"
    fields: list[tuple[str, str, bool]] = [
        ("Task ID", f"`{task.id}`", True),
        ("Project", f"`{task.project_id}`", True),
        ("Branch", f"`{default_branch}`", True),
        ("Error Detail", f"```\n{snippet}\n```", False),
        (
            "Action Required",
            "Push failed after retries. The workspace may be diverged.\n"
            "Inspect the workspace and push manually, or restart the task.",
            False,
        ),
    ]
    return warning_embed(f"Push Failed — {task.title}", fields=fields)