AutoGPT/autogpt_platform/backend/backend/copilot/constants.py

"""Shared constants for the CoPilot module."""

# Special message prefixes for text-based markers (parsed by frontend).
# The hex suffix makes accidental LLM generation of these strings virtually
# impossible, avoiding false-positive marker detection in normal conversation.
COPILOT_ERROR_PREFIX = "[__COPILOT_ERROR_f7a1__]"  # Renders as ErrorCard
COPILOT_RETRYABLE_ERROR_PREFIX = (
    "[__COPILOT_RETRYABLE_ERROR_a9c2__]"  # ErrorCard + retry
)
COPILOT_SYSTEM_PREFIX = "[__COPILOT_SYSTEM_e3b0__]"  # Renders as system info message

# Prefix for all synthetic IDs generated by CoPilot block execution.
# Used to distinguish CoPilot-generated records from real graph execution records
# in PendingHumanReview and other tables.
COPILOT_SYNTHETIC_ID_PREFIX = "copilot-"

# Sub-prefixes for session-scoped and node-scoped synthetic IDs.
COPILOT_SESSION_PREFIX = f"{COPILOT_SYNTHETIC_ID_PREFIX}session-"
COPILOT_NODE_PREFIX = f"{COPILOT_SYNTHETIC_ID_PREFIX}node-"

# Separator used in synthetic node_exec_id to encode node_id.
# Format: "{node_id}:{random_hex}" — extract node_id via rsplit(":", 1)[0]
COPILOT_NODE_EXEC_ID_SEPARATOR = ":"

# Compaction notice messages shown to users.
COMPACTION_DONE_MSG = "Earlier messages were summarized to fit within context limits."
COMPACTION_TOOL_NAME = "context_compaction"


def is_copilot_synthetic_id(id_value: str) -> bool:
    """Check if an ID is a CoPilot synthetic ID (not from a real graph execution)."""
    return id_value.startswith(COPILOT_SYNTHETIC_ID_PREFIX)


def parse_node_id_from_exec_id(node_exec_id: str) -> str:
    """Extract node_id from a synthetic node_exec_id.

    Format: "{node_id}:{random_hex}" → returns "{node_id}".
    """
    return node_exec_id.rsplit(COPILOT_NODE_EXEC_ID_SEPARATOR, 1)[0]


# ---------------------------------------------------------------------------
# Transient Anthropic API error detection
# ---------------------------------------------------------------------------
# Patterns in error text that indicate a transient Anthropic API error
# (ECONNRESET / dropped TCP connection) which is retryable.
_TRANSIENT_ERROR_PATTERNS = (
    "socket connection was closed unexpectedly",
    "ECONNRESET",
    "connection was forcibly closed",
    "network socket disconnected",
)

FRIENDLY_TRANSIENT_MSG = "Anthropic connection interrupted — please retry"


def is_transient_api_error(error_text: str) -> bool:
    """Return True if *error_text* matches a known transient Anthropic API error."""
    lower = error_text.lower()
    return any(pat.lower() in lower for pat in _TRANSIENT_ERROR_PATTERNS)