356 lines
13 KiB
Python
356 lines
13 KiB
Python
from datetime import datetime
|
|
from enum import StrEnum
|
|
from typing import Annotated, Literal, Self, final
|
|
|
|
import sqlalchemy as sa
|
|
from pydantic import BaseModel, Field, model_validator
|
|
from sqlalchemy.orm import Mapped, mapped_column, relationship
|
|
|
|
from core.workflow.human_input_adapter import DeliveryMethodType, InstantMessageProvider, InstantMessageRecipientType
|
|
from core.workflow.nodes.human_input.enums import HumanInputFormKind, HumanInputFormStatus
|
|
from libs.helper import generate_string
|
|
|
|
from .base import Base, DefaultFieldsMixin
|
|
from .types import EnumText, StringUUID
|
|
|
|
_token_length = 22
|
|
# A 32-character string can store a base64-encoded value with 192 bits of entropy
|
|
# or a base62-encoded value with over 180 bits of entropy, providing sufficient
|
|
# uniqueness for most use cases.
|
|
_token_field_length = 32
|
|
_email_field_length = 330
|
|
|
|
|
|
def _generate_token() -> str:
|
|
return generate_string(_token_length)
|
|
|
|
|
|
class HumanInputForm(DefaultFieldsMixin, Base):
|
|
__tablename__ = "human_input_forms"
|
|
__table_args__ = (
|
|
sa.Index(
|
|
"human_input_forms_workflow_run_id_node_id_idx",
|
|
"workflow_run_id",
|
|
"node_id",
|
|
),
|
|
sa.Index("human_input_forms_status_expiration_time_idx", "status", "expiration_time"),
|
|
sa.Index("human_input_forms_status_created_at_idx", "status", "created_at"),
|
|
)
|
|
|
|
tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
|
|
app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
|
|
workflow_run_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
|
|
# ENG-635: a RUNTIME form is tagged with its owning workflow run and/or its
|
|
# conversation. Workflow / Human-Input / agent-node forms always set
|
|
# workflow_run_id, and ALSO set conversation_id when the run has a conversation
|
|
# (chatflow / advanced-chat). Agent v2 chat ask_human forms set only
|
|
# conversation_id (the new Agent App has no workflow_run_id). At least one is set;
|
|
# resume routing prefers workflow_run_id when both are present.
|
|
conversation_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
|
|
form_kind: Mapped[HumanInputFormKind] = mapped_column(
|
|
EnumText(HumanInputFormKind),
|
|
nullable=False,
|
|
default=HumanInputFormKind.RUNTIME,
|
|
)
|
|
|
|
# The human input node the current form corresponds to.
|
|
node_id: Mapped[str] = mapped_column(sa.String(60), nullable=False)
|
|
form_definition: Mapped[str] = mapped_column(sa.Text, nullable=False)
|
|
rendered_content: Mapped[str] = mapped_column(sa.Text, nullable=False)
|
|
status: Mapped[HumanInputFormStatus] = mapped_column(
|
|
EnumText(HumanInputFormStatus),
|
|
nullable=False,
|
|
default=HumanInputFormStatus.WAITING,
|
|
)
|
|
|
|
expiration_time: Mapped[datetime] = mapped_column(
|
|
sa.DateTime,
|
|
nullable=False,
|
|
)
|
|
|
|
# Submission-related fields (nullable until a submission happens).
|
|
selected_action_id: Mapped[str | None] = mapped_column(sa.String(200), nullable=True)
|
|
submitted_data: Mapped[str | None] = mapped_column(sa.Text, nullable=True)
|
|
submitted_at: Mapped[datetime | None] = mapped_column(sa.DateTime, nullable=True)
|
|
submission_user_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
|
|
submission_end_user_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
|
|
|
|
completed_by_recipient_id: Mapped[str | None] = mapped_column(
|
|
StringUUID,
|
|
nullable=True,
|
|
)
|
|
|
|
deliveries: Mapped[list["HumanInputDelivery"]] = relationship(
|
|
"HumanInputDelivery",
|
|
primaryjoin="HumanInputForm.id == foreign(HumanInputDelivery.form_id)",
|
|
uselist=True,
|
|
back_populates="form",
|
|
lazy="raise",
|
|
)
|
|
completed_by_recipient: Mapped["HumanInputFormRecipient | None"] = relationship(
|
|
"HumanInputFormRecipient",
|
|
primaryjoin="HumanInputForm.completed_by_recipient_id == foreign(HumanInputFormRecipient.id)",
|
|
lazy="raise",
|
|
viewonly=True,
|
|
)
|
|
|
|
|
|
class HumanInputDelivery(DefaultFieldsMixin, Base):
|
|
__tablename__ = "human_input_form_deliveries"
|
|
__table_args__ = (
|
|
sa.Index(
|
|
None,
|
|
"form_id",
|
|
),
|
|
)
|
|
|
|
form_id: Mapped[str] = mapped_column(
|
|
StringUUID,
|
|
nullable=False,
|
|
)
|
|
delivery_method_type: Mapped[DeliveryMethodType] = mapped_column(
|
|
EnumText(DeliveryMethodType),
|
|
nullable=False,
|
|
)
|
|
delivery_config_id: Mapped[str | None] = mapped_column(StringUUID, nullable=True)
|
|
channel_payload: Mapped[str] = mapped_column(sa.Text, nullable=False)
|
|
|
|
form: Mapped[HumanInputForm] = relationship(
|
|
"HumanInputForm",
|
|
uselist=False,
|
|
foreign_keys=[form_id],
|
|
primaryjoin="HumanInputDelivery.form_id == HumanInputForm.id",
|
|
back_populates="deliveries",
|
|
lazy="raise",
|
|
)
|
|
|
|
recipients: Mapped[list["HumanInputFormRecipient"]] = relationship(
|
|
"HumanInputFormRecipient",
|
|
primaryjoin="HumanInputDelivery.id == foreign(HumanInputFormRecipient.delivery_id)",
|
|
uselist=True,
|
|
back_populates="delivery",
|
|
# Require explicit preloading
|
|
lazy="raise",
|
|
)
|
|
|
|
|
|
class ApprovalChannel(StrEnum):
|
|
"""Where a paused human input form can be approved, surfaced to API callers."""
|
|
|
|
EMAIL = "email"
|
|
IM = "im"
|
|
WEB_APP = "web_app"
|
|
CONSOLE = "console"
|
|
|
|
|
|
class RecipientType(StrEnum):
|
|
# Second value = the approval channel this recipient maps to (surfaced in `approval_channels`).
|
|
EMAIL_MEMBER = "email_member", ApprovalChannel.EMAIL
|
|
EMAIL_EXTERNAL = "email_external", ApprovalChannel.EMAIL
|
|
INSTANT_MESSAGE = "instant_message", ApprovalChannel.IM
|
|
# STANDALONE_WEB_APP is used by the standalone web app.
|
|
#
|
|
# It's not used while running workflows / chatflows containing HumanInput
|
|
# node inside console.
|
|
STANDALONE_WEB_APP = "standalone_web_app", ApprovalChannel.WEB_APP
|
|
# CONSOLE is used while running workflows / chatflows containing HumanInput
|
|
# node inside console. (E.G. running installed apps or debugging workflows / chatflows)
|
|
CONSOLE = "console", ApprovalChannel.CONSOLE
|
|
# BACKSTAGE is used for backstage input inside console.
|
|
BACKSTAGE = "backstage", ApprovalChannel.CONSOLE
|
|
|
|
_approval_channel: ApprovalChannel
|
|
|
|
def __new__(cls, value: str, approval_channel: ApprovalChannel) -> "RecipientType":
|
|
member = str.__new__(cls, value)
|
|
member._value_ = value
|
|
member._approval_channel = approval_channel
|
|
return member
|
|
|
|
@property
|
|
def approval_channel(self) -> ApprovalChannel:
|
|
return self._approval_channel
|
|
|
|
|
|
@final
|
|
class EmailMemberRecipientPayload(BaseModel):
|
|
TYPE: Literal[RecipientType.EMAIL_MEMBER] = RecipientType.EMAIL_MEMBER
|
|
user_id: str
|
|
|
|
# The `email` field here is only used for mail sending.
|
|
email: str
|
|
|
|
|
|
@final
|
|
class EmailExternalRecipientPayload(BaseModel):
|
|
TYPE: Literal[RecipientType.EMAIL_EXTERNAL] = RecipientType.EMAIL_EXTERNAL
|
|
email: str
|
|
|
|
|
|
@final
|
|
class InstantMessageRecipientPayload(BaseModel):
|
|
TYPE: Literal[RecipientType.INSTANT_MESSAGE] = RecipientType.INSTANT_MESSAGE
|
|
provider: InstantMessageProvider
|
|
recipient_kind: InstantMessageRecipientType
|
|
channel_id: str | None = None
|
|
user_id: str | None = None
|
|
|
|
@model_validator(mode="after")
|
|
def validate_target(self) -> Self:
|
|
match self.recipient_kind:
|
|
case InstantMessageRecipientType.CHANNEL:
|
|
if self.channel_id is None or self.user_id is not None:
|
|
raise ValueError("channel recipients must include channel_id and must not include user_id")
|
|
case InstantMessageRecipientType.USER:
|
|
if self.user_id is None or self.channel_id is not None:
|
|
raise ValueError("user recipients must include user_id and must not include channel_id")
|
|
return self
|
|
|
|
|
|
@final
|
|
class StandaloneWebAppRecipientPayload(BaseModel):
|
|
TYPE: Literal[RecipientType.STANDALONE_WEB_APP] = RecipientType.STANDALONE_WEB_APP
|
|
|
|
|
|
@final
|
|
class ConsoleRecipientPayload(BaseModel):
|
|
TYPE: Literal[RecipientType.CONSOLE] = RecipientType.CONSOLE
|
|
account_id: str | None = None
|
|
|
|
|
|
@final
|
|
class BackstageRecipientPayload(BaseModel):
|
|
TYPE: Literal[RecipientType.BACKSTAGE] = RecipientType.BACKSTAGE
|
|
account_id: str | None = None
|
|
|
|
|
|
@final
|
|
class ConsoleDeliveryPayload(BaseModel):
|
|
type: Literal["console"] = "console"
|
|
internal: bool = True
|
|
|
|
|
|
RecipientPayload = Annotated[
|
|
EmailMemberRecipientPayload
|
|
| EmailExternalRecipientPayload
|
|
| InstantMessageRecipientPayload
|
|
| StandaloneWebAppRecipientPayload
|
|
| ConsoleRecipientPayload
|
|
| BackstageRecipientPayload,
|
|
Field(discriminator="TYPE"),
|
|
]
|
|
|
|
|
|
class HumanInputFormRecipient(DefaultFieldsMixin, Base):
|
|
__tablename__ = "human_input_form_recipients"
|
|
__table_args__ = (
|
|
sa.Index(None, "form_id"),
|
|
sa.Index(None, "delivery_id"),
|
|
)
|
|
|
|
form_id: Mapped[str] = mapped_column(
|
|
StringUUID,
|
|
nullable=False,
|
|
)
|
|
delivery_id: Mapped[str] = mapped_column(
|
|
StringUUID,
|
|
nullable=False,
|
|
)
|
|
recipient_type: Mapped["RecipientType"] = mapped_column(EnumText(RecipientType), nullable=False)
|
|
recipient_payload: Mapped[str] = mapped_column(sa.Text, nullable=False)
|
|
|
|
# Token primarily used for authenticated resume links (email, etc.).
|
|
access_token: Mapped[str | None] = mapped_column(
|
|
sa.VARCHAR(_token_field_length),
|
|
nullable=False,
|
|
default=_generate_token,
|
|
unique=True,
|
|
)
|
|
|
|
delivery: Mapped[HumanInputDelivery] = relationship(
|
|
"HumanInputDelivery",
|
|
uselist=False,
|
|
foreign_keys=[delivery_id],
|
|
back_populates="recipients",
|
|
primaryjoin="HumanInputFormRecipient.delivery_id == HumanInputDelivery.id",
|
|
# Require explicit preloading
|
|
lazy="raise",
|
|
)
|
|
|
|
form: Mapped[HumanInputForm] = relationship(
|
|
"HumanInputForm",
|
|
uselist=False,
|
|
foreign_keys=[form_id],
|
|
primaryjoin="HumanInputFormRecipient.form_id == HumanInputForm.id",
|
|
# Require explicit preloading
|
|
lazy="raise",
|
|
)
|
|
|
|
@classmethod
|
|
def new(
|
|
cls,
|
|
form_id: str,
|
|
delivery_id: str,
|
|
payload: RecipientPayload,
|
|
) -> Self:
|
|
recipient_model = cls(
|
|
form_id=form_id,
|
|
delivery_id=delivery_id,
|
|
recipient_type=payload.TYPE,
|
|
recipient_payload=payload.model_dump_json(),
|
|
access_token=_generate_token(),
|
|
)
|
|
return recipient_model
|
|
|
|
|
|
class HumanInputFormUploadToken(DefaultFieldsMixin, Base):
|
|
"""Upload authorization token bound to one human input form recipient.
|
|
|
|
HITL upload tokens are intentionally separate from app/service bearer tokens.
|
|
The token is stored as an opaque random value so upload endpoints can perform
|
|
a direct lookup without entering the normal Web App authentication chain.
|
|
Upload ownership is resolved from the form's workflow run initiator instead
|
|
of being persisted on the token row itself.
|
|
"""
|
|
|
|
__tablename__ = "human_input_form_upload_tokens"
|
|
__table_args__ = (
|
|
sa.UniqueConstraint("token", name="human_input_form_upload_tokens_token_key"),
|
|
sa.Index("human_input_form_upload_tokens_form_id_idx", "form_id"),
|
|
)
|
|
|
|
tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
|
|
app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
|
|
form_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
|
|
recipient_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
|
|
token: Mapped[str] = mapped_column(sa.String(255), nullable=False)
|
|
|
|
form: Mapped[HumanInputForm] = relationship(
|
|
"HumanInputForm",
|
|
uselist=False,
|
|
foreign_keys=[form_id],
|
|
primaryjoin="foreign(HumanInputFormUploadToken.form_id) == HumanInputForm.id",
|
|
lazy="raise",
|
|
)
|
|
|
|
|
|
class HumanInputFormUploadFile(DefaultFieldsMixin, Base):
|
|
"""Association between a human input form and a file uploaded through its token.
|
|
|
|
Ownership remains on ``UploadFile`` itself; this table only records the
|
|
durable form/token/file linkage needed by Human Input flows.
|
|
"""
|
|
|
|
__tablename__ = "human_input_form_upload_files"
|
|
__table_args__ = (
|
|
sa.UniqueConstraint("upload_file_id", name="human_input_form_upload_files_upload_file_id_key"),
|
|
sa.Index("human_input_form_upload_files_form_id_idx", "form_id"),
|
|
sa.Index("human_input_form_upload_files_upload_token_id_idx", "upload_token_id"),
|
|
)
|
|
|
|
tenant_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
|
|
app_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
|
|
form_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
|
|
upload_file_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
|
|
upload_token_id: Mapped[str] = mapped_column(StringUUID, nullable=False)
|