From 222bac0e97161669eb55dc65fc897814699bd432 Mon Sep 17 00:00:00 2001 From: Kazet111 Date: Sun, 19 Apr 2026 04:00:37 +0200 Subject: [PATCH 01/58] feat(automations): implement scheduled automation cycles and run details UI --- app/core/config/settings.py | 2 + .../20260415_160000_add_automations_tables.py | 189 ++ ...18_190000_add_automation_runs_cycle_key.py | 87 + ...00000_add_automation_run_cycle_metadata.py | 56 + app/db/models.py | 116 + app/dependencies.py | 25 + app/main.py | 6 + app/modules/automations/__init__.py | 2 + app/modules/automations/api.py | 353 +++ app/modules/automations/repository.py | 983 +++++++++ app/modules/automations/scheduler.py | 82 + app/modules/automations/schemas.py | 175 ++ app/modules/automations/service.py | 1894 +++++++++++++++++ app/modules/dashboard/api.py | 23 +- frontend/src/App.tsx | 2 + .../__integration__/automations-flow.test.tsx | 97 + frontend/src/components/layout/app-header.tsx | 1 + .../components/account-multi-select.tsx | 14 +- frontend/src/features/api-keys/schemas.ts | 10 +- .../automations/account-display.test.ts | 57 + .../features/automations/account-display.ts | 113 + frontend/src/features/automations/api.ts | 140 ++ .../components/automation-job-dialog.tsx | 757 +++++++ .../components/automation-list-filters.tsx | 159 ++ .../components/automations-page.tsx | 930 ++++++++ .../components/run-details-dialog.tsx | 369 ++++ .../components/run-status-utils.ts | 42 + .../hooks/use-automation-listing.ts | 291 +++ .../automations/hooks/use-automations.test.ts | 75 + .../automations/hooks/use-automations.ts | 107 + .../src/features/automations/schemas.test.ts | 87 + frontend/src/features/automations/schemas.ts | 146 ++ .../features/automations/time-utils.test.ts | 36 + .../src/features/automations/time-utils.ts | 65 + .../components/recent-requests-table.tsx | 11 +- .../src/test/mocks/handler-coverage.test.ts | 11 + frontend/src/test/mocks/handlers.ts | 528 +++++ .../context.md | 146 ++ .../add-automations-scheduled-pings/design.md | 134 ++ .../proposal.md | 64 + .../specs/automations/spec.md | 148 ++ .../specs/frontend-architecture/spec.md | 63 + .../add-automations-scheduled-pings/tasks.md | 30 + tests/integration/test_automations_api.py | 1068 ++++++++++ tests/unit/test_automations_service.py | 131 ++ 45 files changed, 9817 insertions(+), 8 deletions(-) create mode 100644 app/db/alembic/versions/20260415_160000_add_automations_tables.py create mode 100644 app/db/alembic/versions/20260418_190000_add_automation_runs_cycle_key.py create mode 100644 app/db/alembic/versions/20260419_000000_add_automation_run_cycle_metadata.py create mode 100644 app/modules/automations/__init__.py create mode 100644 app/modules/automations/api.py create mode 100644 app/modules/automations/repository.py create mode 100644 app/modules/automations/scheduler.py create mode 100644 app/modules/automations/schemas.py create mode 100644 app/modules/automations/service.py create mode 100644 frontend/src/__integration__/automations-flow.test.tsx create mode 100644 frontend/src/features/automations/account-display.test.ts create mode 100644 frontend/src/features/automations/account-display.ts create mode 100644 frontend/src/features/automations/api.ts create mode 100644 frontend/src/features/automations/components/automation-job-dialog.tsx create mode 100644 frontend/src/features/automations/components/automation-list-filters.tsx create mode 100644 frontend/src/features/automations/components/automations-page.tsx create mode 100644 frontend/src/features/automations/components/run-details-dialog.tsx create mode 100644 frontend/src/features/automations/components/run-status-utils.ts create mode 100644 frontend/src/features/automations/hooks/use-automation-listing.ts create mode 100644 frontend/src/features/automations/hooks/use-automations.test.ts create mode 100644 frontend/src/features/automations/hooks/use-automations.ts create mode 100644 frontend/src/features/automations/schemas.test.ts create mode 100644 frontend/src/features/automations/schemas.ts create mode 100644 frontend/src/features/automations/time-utils.test.ts create mode 100644 frontend/src/features/automations/time-utils.ts create mode 100644 openspec/changes/add-automations-scheduled-pings/context.md create mode 100644 openspec/changes/add-automations-scheduled-pings/design.md create mode 100644 openspec/changes/add-automations-scheduled-pings/proposal.md create mode 100644 openspec/changes/add-automations-scheduled-pings/specs/automations/spec.md create mode 100644 openspec/changes/add-automations-scheduled-pings/specs/frontend-architecture/spec.md create mode 100644 openspec/changes/add-automations-scheduled-pings/tasks.md create mode 100644 tests/integration/test_automations_api.py create mode 100644 tests/unit/test_automations_service.py diff --git a/app/core/config/settings.py b/app/core/config/settings.py index 3023626487..86c0534acd 100644 --- a/app/core/config/settings.py +++ b/app/core/config/settings.py @@ -211,6 +211,8 @@ class Settings(BaseSettings): sticky_session_cleanup_interval_seconds: int = Field(default=300, gt=0) quota_planner_scheduler_enabled: bool = True quota_planner_tick_seconds: int = Field(default=300, gt=0) + automations_scheduler_enabled: bool = True + automations_scheduler_interval_seconds: int = Field(default=30, gt=0) encryption_key_file: Path = DEFAULT_ENCRYPTION_KEY_FILE database_migrations_fail_fast: bool = True log_proxy_request_shape: bool = False diff --git a/app/db/alembic/versions/20260415_160000_add_automations_tables.py b/app/db/alembic/versions/20260415_160000_add_automations_tables.py new file mode 100644 index 0000000000..585c8fee69 --- /dev/null +++ b/app/db/alembic/versions/20260415_160000_add_automations_tables.py @@ -0,0 +1,189 @@ +"""add automations tables + +Revision ID: 20260415_160000_add_automations_tables +Revises: 20260415_160000_add_request_logs_response_lookup_index +Create Date: 2026-04-15 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.engine import Connection + +revision = "20260415_160000_add_automations_tables" +down_revision = "20260415_160000_add_request_logs_response_lookup_index" +branch_labels = None +depends_on = None + + +def _table_exists(connection: Connection, table_name: str) -> bool: + inspector = sa.inspect(connection) + return inspector.has_table(table_name) + + +def _index_names(connection: Connection, table_name: str) -> set[str]: + inspector = sa.inspect(connection) + if not inspector.has_table(table_name): + return set() + return {str(index["name"]) for index in inspector.get_indexes(table_name) if index.get("name")} + + +def _column_names(connection: Connection, table_name: str) -> set[str]: + inspector = sa.inspect(connection) + if not inspector.has_table(table_name): + return set() + return {str(column["name"]) for column in inspector.get_columns(table_name) if column.get("name")} + + +def upgrade() -> None: + bind = op.get_bind() + + if not _table_exists(bind, "automation_jobs"): + op.create_table( + "automation_jobs", + sa.Column("id", sa.String(), nullable=False), + sa.Column("name", sa.String(length=200), nullable=False), + sa.Column("enabled", sa.Boolean(), nullable=False, server_default=sa.true()), + sa.Column("schedule_type", sa.String(length=32), nullable=False, server_default=sa.text("'daily'")), + sa.Column("schedule_time", sa.String(length=5), nullable=False), + sa.Column("schedule_timezone", sa.String(length=64), nullable=False), + sa.Column( + "schedule_days", + sa.String(length=64), + nullable=False, + server_default=sa.text("'mon,tue,wed,thu,fri,sat,sun'"), + ), + sa.Column("schedule_threshold_minutes", sa.Integer(), nullable=False, server_default=sa.text("0")), + sa.Column("include_paused_accounts", sa.Boolean(), nullable=False, server_default=sa.false()), + sa.Column("model", sa.String(), nullable=False), + sa.Column("reasoning_effort", sa.String(length=16), nullable=True), + sa.Column("prompt", sa.Text(), nullable=False, server_default=sa.text("'ping'")), + sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.func.now()), + sa.Column("updated_at", sa.DateTime(), nullable=False, server_default=sa.func.now()), + sa.PrimaryKeyConstraint("id"), + ) + else: + job_columns = _column_names(bind, "automation_jobs") + if "schedule_days" not in job_columns: + op.add_column( + "automation_jobs", + sa.Column( + "schedule_days", + sa.String(length=64), + nullable=False, + server_default=sa.text("'mon,tue,wed,thu,fri,sat,sun'"), + ), + ) + if "reasoning_effort" not in job_columns: + op.add_column( + "automation_jobs", + sa.Column("reasoning_effort", sa.String(length=16), nullable=True), + ) + if "schedule_threshold_minutes" not in job_columns: + op.add_column( + "automation_jobs", + sa.Column("schedule_threshold_minutes", sa.Integer(), nullable=False, server_default=sa.text("0")), + ) + if "include_paused_accounts" not in job_columns: + op.add_column( + "automation_jobs", + sa.Column("include_paused_accounts", sa.Boolean(), nullable=False, server_default=sa.false()), + ) + + if not _table_exists(bind, "automation_job_accounts"): + op.create_table( + "automation_job_accounts", + sa.Column("job_id", sa.String(), nullable=False), + sa.Column("account_id", sa.String(), nullable=False), + sa.Column("position", sa.Integer(), nullable=False), + sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.func.now()), + sa.ForeignKeyConstraint(["account_id"], ["accounts.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["job_id"], ["automation_jobs.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("job_id", "account_id"), + sa.UniqueConstraint("job_id", "position", name="uq_automation_job_accounts_position"), + ) + + if not _table_exists(bind, "automation_runs"): + op.create_table( + "automation_runs", + sa.Column("id", sa.String(), nullable=False), + sa.Column("job_id", sa.String(), nullable=False), + sa.Column("trigger", sa.String(length=16), nullable=False), + sa.Column("slot_key", sa.String(length=128), nullable=False), + sa.Column("scheduled_for", sa.DateTime(), nullable=False), + sa.Column("started_at", sa.DateTime(), nullable=False), + sa.Column("finished_at", sa.DateTime(), nullable=True), + sa.Column("status", sa.String(length=16), nullable=False, server_default=sa.text("'running'")), + sa.Column("account_id", sa.String(), nullable=True), + sa.Column("error_code", sa.String(length=100), nullable=True), + sa.Column("error_message", sa.Text(), nullable=True), + sa.Column("attempt_count", sa.Integer(), nullable=False, server_default=sa.text("0")), + sa.Column("created_at", sa.DateTime(), nullable=False, server_default=sa.func.now()), + sa.ForeignKeyConstraint(["account_id"], ["accounts.id"], ondelete="SET NULL"), + sa.ForeignKeyConstraint(["job_id"], ["automation_jobs.id"], ondelete="CASCADE"), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("slot_key", name="uq_automation_runs_slot_key"), + ) + + jobs_indexes = _index_names(bind, "automation_jobs") + if "idx_automation_jobs_enabled" not in jobs_indexes: + op.create_index("idx_automation_jobs_enabled", "automation_jobs", ["enabled"], unique=False) + + job_accounts_indexes = _index_names(bind, "automation_job_accounts") + if "idx_automation_job_accounts_account_id" not in job_accounts_indexes: + op.create_index( + "idx_automation_job_accounts_account_id", + "automation_job_accounts", + ["account_id"], + unique=False, + ) + + runs_indexes = _index_names(bind, "automation_runs") + if "idx_automation_runs_job_id_started_at" not in runs_indexes: + op.create_index( + "idx_automation_runs_job_id_started_at", + "automation_runs", + ["job_id", "started_at"], + unique=False, + ) + if "idx_automation_runs_status_started_at" not in runs_indexes: + op.create_index( + "idx_automation_runs_status_started_at", + "automation_runs", + ["status", "started_at"], + unique=False, + ) + if "idx_automation_runs_scheduled_for" not in runs_indexes: + op.create_index( + "idx_automation_runs_scheduled_for", + "automation_runs", + ["scheduled_for"], + unique=False, + ) + + +def downgrade() -> None: + bind = op.get_bind() + + if _table_exists(bind, "automation_runs"): + runs_indexes = _index_names(bind, "automation_runs") + if "idx_automation_runs_scheduled_for" in runs_indexes: + op.drop_index("idx_automation_runs_scheduled_for", table_name="automation_runs") + if "idx_automation_runs_status_started_at" in runs_indexes: + op.drop_index("idx_automation_runs_status_started_at", table_name="automation_runs") + if "idx_automation_runs_job_id_started_at" in runs_indexes: + op.drop_index("idx_automation_runs_job_id_started_at", table_name="automation_runs") + op.drop_table("automation_runs") + + if _table_exists(bind, "automation_job_accounts"): + job_accounts_indexes = _index_names(bind, "automation_job_accounts") + if "idx_automation_job_accounts_account_id" in job_accounts_indexes: + op.drop_index("idx_automation_job_accounts_account_id", table_name="automation_job_accounts") + op.drop_table("automation_job_accounts") + + if _table_exists(bind, "automation_jobs"): + jobs_indexes = _index_names(bind, "automation_jobs") + if "idx_automation_jobs_enabled" in jobs_indexes: + op.drop_index("idx_automation_jobs_enabled", table_name="automation_jobs") + op.drop_table("automation_jobs") diff --git a/app/db/alembic/versions/20260418_190000_add_automation_runs_cycle_key.py b/app/db/alembic/versions/20260418_190000_add_automation_runs_cycle_key.py new file mode 100644 index 0000000000..065b0dcfb6 --- /dev/null +++ b/app/db/alembic/versions/20260418_190000_add_automation_runs_cycle_key.py @@ -0,0 +1,87 @@ +"""add cycle key to automation runs + +Revision ID: 20260418_190000_add_automation_runs_cycle_key +Revises: 20260415_160000_add_automations_tables +Create Date: 2026-04-18 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.engine import Connection + +revision = "20260418_190000_add_automation_runs_cycle_key" +down_revision = "20260415_160000_add_automations_tables" +branch_labels = None +depends_on = None + + +def _table_exists(connection: Connection, table_name: str) -> bool: + inspector = sa.inspect(connection) + return inspector.has_table(table_name) + + +def _index_names(connection: Connection, table_name: str) -> set[str]: + inspector = sa.inspect(connection) + if not inspector.has_table(table_name): + return set() + return {str(index["name"]) for index in inspector.get_indexes(table_name) if index.get("name")} + + +def _column_names(connection: Connection, table_name: str) -> set[str]: + inspector = sa.inspect(connection) + if not inspector.has_table(table_name): + return set() + return {str(column["name"]) for column in inspector.get_columns(table_name) if column.get("name")} + + +def upgrade() -> None: + bind = op.get_bind() + if not _table_exists(bind, "automation_runs"): + return + + run_columns = _column_names(bind, "automation_runs") + if "cycle_key" not in run_columns: + op.add_column( + "automation_runs", + sa.Column("cycle_key", sa.String(length=160), nullable=True), + ) + op.execute("UPDATE automation_runs SET cycle_key = slot_key WHERE cycle_key IS NULL OR cycle_key = ''") + if bind.dialect.name == "sqlite": + with op.batch_alter_table("automation_runs", recreate="always") as batch_op: + batch_op.alter_column( + "cycle_key", + existing_type=sa.String(length=160), + nullable=False, + ) + else: + op.alter_column( + "automation_runs", + "cycle_key", + existing_type=sa.String(length=160), + nullable=False, + ) + + run_indexes = _index_names(bind, "automation_runs") + if "idx_automation_runs_cycle_key_started_at" not in run_indexes: + op.create_index( + "idx_automation_runs_cycle_key_started_at", + "automation_runs", + ["cycle_key", "started_at"], + unique=False, + ) + + +def downgrade() -> None: + bind = op.get_bind() + if not _table_exists(bind, "automation_runs"): + return + + run_indexes = _index_names(bind, "automation_runs") + if "idx_automation_runs_cycle_key_started_at" in run_indexes: + op.drop_index("idx_automation_runs_cycle_key_started_at", table_name="automation_runs") + + run_columns = _column_names(bind, "automation_runs") + if "cycle_key" in run_columns: + op.drop_column("automation_runs", "cycle_key") diff --git a/app/db/alembic/versions/20260419_000000_add_automation_run_cycle_metadata.py b/app/db/alembic/versions/20260419_000000_add_automation_run_cycle_metadata.py new file mode 100644 index 0000000000..dae02e244d --- /dev/null +++ b/app/db/alembic/versions/20260419_000000_add_automation_run_cycle_metadata.py @@ -0,0 +1,56 @@ +"""add automation run cycle metadata columns + +Revision ID: 20260419_000000_add_automation_run_cycle_metadata +Revises: 20260418_190000_add_automation_runs_cycle_key +Create Date: 2026-04-19 +""" + +from __future__ import annotations + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.engine import Connection + +revision = "20260419_000000_add_automation_run_cycle_metadata" +down_revision = "20260418_190000_add_automation_runs_cycle_key" +branch_labels = None +depends_on = None + + +def _table_exists(connection: Connection, table_name: str) -> bool: + inspector = sa.inspect(connection) + return inspector.has_table(table_name) + + +def _column_names(connection: Connection, table_name: str) -> set[str]: + inspector = sa.inspect(connection) + if not inspector.has_table(table_name): + return set() + return {str(column["name"]) for column in inspector.get_columns(table_name) if column.get("name")} + + +def upgrade() -> None: + bind = op.get_bind() + if not _table_exists(bind, "automation_runs"): + return + + run_columns = _column_names(bind, "automation_runs") + if "cycle_expected_accounts" not in run_columns: + op.add_column("automation_runs", sa.Column("cycle_expected_accounts", sa.Integer(), nullable=True)) + if "cycle_window_end" not in run_columns: + op.add_column("automation_runs", sa.Column("cycle_window_end", sa.DateTime(), nullable=True)) + + op.execute("UPDATE automation_runs SET cycle_expected_accounts = 0 WHERE cycle_expected_accounts IS NULL") + op.execute("UPDATE automation_runs SET cycle_window_end = scheduled_for WHERE cycle_window_end IS NULL") + + +def downgrade() -> None: + bind = op.get_bind() + if not _table_exists(bind, "automation_runs"): + return + + run_columns = _column_names(bind, "automation_runs") + if "cycle_window_end" in run_columns: + op.drop_column("automation_runs", "cycle_window_end") + if "cycle_expected_accounts" in run_columns: + op.drop_column("automation_runs", "cycle_expected_accounts") diff --git a/app/db/models.py b/app/db/models.py index 0eef688c84..fccb7702c0 100644 --- a/app/db/models.py +++ b/app/db/models.py @@ -779,6 +779,116 @@ class ApiKeyUsageReservationItem(Base): limit: Mapped[ApiKeyLimit] = relationship("ApiKeyLimit") +class AutomationJob(Base): + __tablename__ = "automation_jobs" + + id: Mapped[str] = mapped_column(String, primary_key=True) + name: Mapped[str] = mapped_column(String(200), nullable=False) + enabled: Mapped[bool] = mapped_column(Boolean, default=True, server_default=true(), nullable=False) + schedule_type: Mapped[str] = mapped_column( + String(32), + nullable=False, + default="daily", + server_default=text("'daily'"), + ) + schedule_time: Mapped[str] = mapped_column(String(5), nullable=False) + schedule_timezone: Mapped[str] = mapped_column(String(64), nullable=False) + schedule_days: Mapped[str] = mapped_column( + String(64), + nullable=False, + default="mon,tue,wed,thu,fri,sat,sun", + server_default=text("'mon,tue,wed,thu,fri,sat,sun'"), + ) + schedule_threshold_minutes: Mapped[int] = mapped_column( + Integer, + nullable=False, + default=0, + server_default=text("0"), + ) + include_paused_accounts: Mapped[bool] = mapped_column( + Boolean, + nullable=False, + default=False, + server_default=false(), + ) + model: Mapped[str] = mapped_column(String, nullable=False) + reasoning_effort: Mapped[str | None] = mapped_column(String(16), nullable=True) + prompt: Mapped[str] = mapped_column(Text, nullable=False, default="ping", server_default=text("'ping'")) + created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), nullable=False) + updated_at: Mapped[datetime] = mapped_column( + DateTime, + server_default=func.now(), + onupdate=func.now(), + nullable=False, + ) + + account_links: Mapped[list["AutomationJobAccount"]] = relationship( + "AutomationJobAccount", + back_populates="job", + cascade="all, delete-orphan", + ) + runs: Mapped[list["AutomationRun"]] = relationship( + "AutomationRun", + back_populates="job", + cascade="all, delete-orphan", + ) + + +class AutomationJobAccount(Base): + __tablename__ = "automation_job_accounts" + __table_args__ = ( + UniqueConstraint("job_id", "position", name="uq_automation_job_accounts_position"), + ) + + job_id: Mapped[str] = mapped_column( + String, + ForeignKey("automation_jobs.id", ondelete="CASCADE"), + primary_key=True, + ) + account_id: Mapped[str] = mapped_column( + String, + ForeignKey("accounts.id", ondelete="CASCADE"), + primary_key=True, + ) + position: Mapped[int] = mapped_column(Integer, nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), nullable=False) + + job: Mapped[AutomationJob] = relationship("AutomationJob", back_populates="account_links") + account: Mapped[Account] = relationship("Account") + + +class AutomationRun(Base): + __tablename__ = "automation_runs" + + id: Mapped[str] = mapped_column(String, primary_key=True) + job_id: Mapped[str] = mapped_column( + String, + ForeignKey("automation_jobs.id", ondelete="CASCADE"), + nullable=False, + ) + trigger: Mapped[str] = mapped_column(String(16), nullable=False) + slot_key: Mapped[str] = mapped_column(String(128), nullable=False, unique=True) + cycle_key: Mapped[str] = mapped_column(String(160), nullable=False) + cycle_expected_accounts: Mapped[int | None] = mapped_column(Integer, nullable=True) + cycle_window_end: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + scheduled_for: Mapped[datetime] = mapped_column(DateTime, nullable=False) + started_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + finished_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + status: Mapped[str] = mapped_column(String(16), nullable=False, default="running", server_default=text("'running'")) + account_id: Mapped[str | None] = mapped_column( + String, + ForeignKey("accounts.id", ondelete="SET NULL"), + nullable=True, + ) + error_code: Mapped[str | None] = mapped_column(String(100), nullable=True) + error_message: Mapped[str | None] = mapped_column(Text, nullable=True) + attempt_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0, server_default=text("0")) + created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now(), nullable=False) + + job: Mapped[AutomationJob] = relationship("AutomationJob", back_populates="runs") + account: Mapped[Account | None] = relationship("Account") + + class RateLimitAttempt(Base): __tablename__ = "rate_limit_attempts" @@ -1114,6 +1224,12 @@ class HttpBridgeSessionAlias(Base): QuotaWindowObservation.account_id, QuotaWindowObservation.observed_at.desc(), ) +Index("idx_automation_jobs_enabled", AutomationJob.enabled) +Index("idx_automation_job_accounts_account_id", AutomationJobAccount.account_id) +Index("idx_automation_runs_job_id_started_at", AutomationRun.job_id, AutomationRun.started_at) +Index("idx_automation_runs_status_started_at", AutomationRun.status, AutomationRun.started_at) +Index("idx_automation_runs_scheduled_for", AutomationRun.scheduled_for) +Index("idx_automation_runs_cycle_key_started_at", AutomationRun.cycle_key, AutomationRun.started_at) Index("idx_http_bridge_sessions_owner_state", HttpBridgeSessionRecord.owner_instance_id, HttpBridgeSessionRecord.state) Index("idx_http_bridge_sessions_lease", HttpBridgeSessionRecord.lease_expires_at) Index("idx_http_bridge_sessions_last_seen", HttpBridgeSessionRecord.last_seen_at.desc()) diff --git a/app/dependencies.py b/app/dependencies.py index 18942507ae..05d531d86d 100644 --- a/app/dependencies.py +++ b/app/dependencies.py @@ -16,6 +16,8 @@ from app.modules.api_keys.service import ApiKeysService from app.modules.audit.repository import AuditRepository from app.modules.audit.service import AuditLogsService +from app.modules.automations.repository import AutomationsRepository +from app.modules.automations.service import AutomationsService from app.modules.dashboard.repository import DashboardRepository from app.modules.dashboard.service import DashboardService from app.modules.dashboard_auth.repository import DashboardAuthRepository @@ -137,6 +139,14 @@ class ReportsContext: service: ReportsService +@dataclass(slots=True) +class AutomationsContext: + session: AsyncSession + repository: AutomationsRepository + accounts_repository: AccountsRepository + service: AutomationsService + + def get_accounts_context( session: AsyncSession = Depends(get_session), ) -> AccountsContext: @@ -306,3 +316,18 @@ def get_reports_context( repository = ReportsRepository(session) service = ReportsService(repository) return ReportsContext(session=session, repository=repository, service=service) + + +def get_automations_context( + session: AsyncSession = Depends(get_session), +) -> AutomationsContext: + repository = AutomationsRepository(session) + accounts_repository = AccountsRepository(session) + request_logs_repository = RequestLogsRepository(session) + service = AutomationsService(repository, accounts_repository, request_logs_repository) + return AutomationsContext( + session=session, + repository=repository, + accounts_repository=accounts_repository, + service=service, + ) diff --git a/app/main.py b/app/main.py index 393110e54b..901c2bacad 100644 --- a/app/main.py +++ b/app/main.py @@ -46,6 +46,8 @@ from app.modules.api_keys import api as api_keys_api from app.modules.api_keys.reset_scheduler import build_api_key_limit_reset_scheduler from app.modules.audit import api as audit_api +from app.modules.automations import api as automations_api +from app.modules.automations.scheduler import build_automations_scheduler from app.modules.conversation_archive import api as conversation_archive_api from app.modules.dashboard import api as dashboard_api from app.modules.dashboard_auth import api as dashboard_auth_api @@ -151,12 +153,14 @@ async def lifespan(app: FastAPI): sticky_session_cleanup_scheduler = build_sticky_session_cleanup_scheduler() quota_planner_scheduler = build_quota_planner_scheduler() auth_guardian_scheduler = build_auth_guardian_scheduler() + automations_scheduler = build_automations_scheduler() await usage_scheduler.start() await api_key_limit_reset_scheduler.start() await model_scheduler.start() await sticky_session_cleanup_scheduler.start() await quota_planner_scheduler.start() await auth_guardian_scheduler.start() + await automations_scheduler.start() if settings.metrics_enabled and PROMETHEUS_AVAILABLE: import uvicorn @@ -313,6 +317,7 @@ async def _activate_bridge_membership(svc: RingMembershipService, iid: str) -> N await cache_poller.stop() await quota_planner_scheduler.stop() await auth_guardian_scheduler.stop() + await automations_scheduler.stop() await sticky_session_cleanup_scheduler.stop() await model_scheduler.stop() await api_key_limit_reset_scheduler.stop() @@ -399,6 +404,7 @@ def create_app() -> FastAPI: app.include_router(settings_api.router) app.include_router(firewall_api.router) app.include_router(sticky_sessions_api.router) + app.include_router(automations_api.router) app.include_router(api_keys_api.router) app.include_router(health_api.router) diff --git a/app/modules/automations/__init__.py b/app/modules/automations/__init__.py new file mode 100644 index 0000000000..3ce55a69ae --- /dev/null +++ b/app/modules/automations/__init__.py @@ -0,0 +1,2 @@ +from __future__ import annotations + diff --git a/app/modules/automations/api.py b/app/modules/automations/api.py new file mode 100644 index 0000000000..bf9b7b6040 --- /dev/null +++ b/app/modules/automations/api.py @@ -0,0 +1,353 @@ +from __future__ import annotations + +from typing import Literal, cast + +from fastapi import APIRouter, Body, Depends, Query + +from app.core.auth.dependencies import set_dashboard_error_format, validate_dashboard_session +from app.core.exceptions import DashboardBadRequestError, DashboardNotFoundError +from app.dependencies import AutomationsContext, get_automations_context +from app.modules.automations.schemas import ( + AutomationDeleteResponse, + AutomationJobCreateRequest, + AutomationJobFilterOptionsResponse, + AutomationJobResponse, + AutomationJobsListResponse, + AutomationJobUpdateRequest, + AutomationRunAccountStateResponse, + AutomationRunDetailsResponse, + AutomationRunFilterOptionsResponse, + AutomationRunResponse, + AutomationRunsListResponse, + AutomationScheduleResponse, +) +from app.modules.automations.service import ( + AutomationJobCreateInput, + AutomationJobData, + AutomationJobUpdateInput, + AutomationNotFoundError, + AutomationRunAccountStateData, + AutomationRunData, + AutomationRunDetailsData, + AutomationValidationError, +) + +router = APIRouter( + prefix="/api/automations", + tags=["dashboard"], + dependencies=[Depends(validate_dashboard_session), Depends(set_dashboard_error_format)], +) + + +@router.get("", response_model=AutomationJobsListResponse) +async def list_automations( + limit: int = Query(default=25, ge=1, le=1000), + offset: int = Query(default=0, ge=0), + search: str | None = Query(default=None), + account_id: list[str] | None = Query(default=None, alias="accountId"), + model: list[str] | None = Query(default=None), + status: list[str] | None = Query(default=None), + schedule_type: list[str] | None = Query(default=None, alias="scheduleType"), + context: AutomationsContext = Depends(get_automations_context), +) -> AutomationJobsListResponse: + page = await context.service.list_jobs_page( + limit=limit, + offset=offset, + search=search, + account_ids=account_id, + models=model, + statuses=status, + schedule_types=schedule_type, + ) + return AutomationJobsListResponse( + items=[_to_job_response(job) for job in page.items], + total=page.total, + has_more=page.has_more, + ) + + +@router.get("/options", response_model=AutomationJobFilterOptionsResponse) +async def list_automation_filter_options( + search: str | None = Query(default=None), + account_id: list[str] | None = Query(default=None, alias="accountId"), + model: list[str] | None = Query(default=None), + schedule_type: list[str] | None = Query(default=None, alias="scheduleType"), + status: list[str] | None = Query(default=None), + context: AutomationsContext = Depends(get_automations_context), +) -> AutomationJobFilterOptionsResponse: + options = await context.service.list_job_filter_options( + search=search, + account_ids=account_id, + models=model, + statuses=status, + schedule_types=schedule_type, + ) + return AutomationJobFilterOptionsResponse( + account_ids=options.account_ids, + models=options.models, + statuses=options.statuses, + schedule_types=options.schedule_types, + ) + + +@router.get("/runs", response_model=AutomationRunsListResponse) +async def list_automations_runs( + limit: int = Query(default=25, ge=1, le=1000), + offset: int = Query(default=0, ge=0), + search: str | None = Query(default=None), + account_id: list[str] | None = Query(default=None, alias="accountId"), + model: list[str] | None = Query(default=None), + status: list[str] | None = Query(default=None), + trigger: list[str] | None = Query(default=None), + automation_id: list[str] | None = Query(default=None, alias="automationId"), + context: AutomationsContext = Depends(get_automations_context), +) -> AutomationRunsListResponse: + page = await context.service.list_runs_page( + limit=limit, + offset=offset, + search=search, + account_ids=account_id, + models=model, + statuses=status, + triggers=trigger, + job_ids=automation_id, + ) + return AutomationRunsListResponse( + items=[_to_run_response(run) for run in page.items], + total=page.total, + has_more=page.has_more, + ) + + +@router.get("/runs/options", response_model=AutomationRunFilterOptionsResponse) +async def list_automation_run_filter_options( + search: str | None = Query(default=None), + account_id: list[str] | None = Query(default=None, alias="accountId"), + model: list[str] | None = Query(default=None), + status: list[str] | None = Query(default=None), + trigger: list[str] | None = Query(default=None), + automation_id: list[str] | None = Query(default=None, alias="automationId"), + context: AutomationsContext = Depends(get_automations_context), +) -> AutomationRunFilterOptionsResponse: + options = await context.service.list_run_filter_options( + search=search, + account_ids=account_id, + models=model, + statuses=status, + triggers=trigger, + job_ids=automation_id, + ) + return AutomationRunFilterOptionsResponse( + account_ids=options.account_ids, + models=options.models, + statuses=options.statuses, + triggers=options.triggers, + ) + + +@router.get("/runs/{run_id}/details", response_model=AutomationRunDetailsResponse) +async def get_automation_run_details( + run_id: str, + context: AutomationsContext = Depends(get_automations_context), +) -> AutomationRunDetailsResponse: + try: + details = await context.service.get_run_details(run_id) + except AutomationNotFoundError as exc: + raise DashboardNotFoundError("Automation run not found", code="automation_run_not_found") from exc + return _to_run_details_response(details) + + +@router.post("", response_model=AutomationJobResponse) +async def create_automation( + payload: AutomationJobCreateRequest = Body(...), + context: AutomationsContext = Depends(get_automations_context), +) -> AutomationJobResponse: + try: + job = await context.service.create_job( + AutomationJobCreateInput( + name=payload.name, + enabled=payload.enabled, + include_paused_accounts=payload.include_paused_accounts, + schedule_type=payload.schedule.type, + schedule_time=payload.schedule.time, + schedule_timezone=payload.schedule.timezone, + schedule_threshold_minutes=payload.schedule.threshold_minutes, + schedule_days=list(payload.schedule.days), + model=payload.model, + reasoning_effort=payload.reasoning_effort, + prompt=payload.prompt, + account_ids=payload.account_ids, + ) + ) + except AutomationValidationError as exc: + raise DashboardBadRequestError(str(exc), code=exc.code) from exc + return _to_job_response(job) + + +@router.patch("/{automation_id}", response_model=AutomationJobResponse) +async def update_automation( + automation_id: str, + payload: AutomationJobUpdateRequest, + context: AutomationsContext = Depends(get_automations_context), +) -> AutomationJobResponse: + try: + updated = await context.service.update_job( + automation_id, + AutomationJobUpdateInput( + name=payload.name, + enabled=payload.enabled, + include_paused_accounts=payload.include_paused_accounts, + schedule_type=payload.schedule.type if payload.schedule else None, + schedule_time=payload.schedule.time if payload.schedule else None, + schedule_timezone=payload.schedule.timezone if payload.schedule else None, + schedule_threshold_minutes=payload.schedule.threshold_minutes if payload.schedule else None, + schedule_days=list(payload.schedule.days) if payload.schedule else None, + model=payload.model, + reasoning_effort=payload.reasoning_effort, + reasoning_effort_set="reasoning_effort" in payload.model_fields_set, + prompt=payload.prompt, + account_ids=payload.account_ids, + ), + ) + except AutomationNotFoundError as exc: + raise DashboardNotFoundError("Automation not found", code="automation_not_found") from exc + except AutomationValidationError as exc: + raise DashboardBadRequestError(str(exc), code=exc.code) from exc + return _to_job_response(updated) + + +@router.delete("/{automation_id}", response_model=AutomationDeleteResponse) +async def delete_automation( + automation_id: str, + context: AutomationsContext = Depends(get_automations_context), +) -> AutomationDeleteResponse: + deleted = await context.service.delete_job(automation_id) + if not deleted: + raise DashboardNotFoundError("Automation not found", code="automation_not_found") + return AutomationDeleteResponse(status="deleted") + + +@router.post("/{automation_id}/run-now", response_model=AutomationRunResponse, status_code=202) +async def run_automation_now( + automation_id: str, + context: AutomationsContext = Depends(get_automations_context), +) -> AutomationRunResponse: + try: + run = await context.service.run_now(automation_id) + except AutomationNotFoundError as exc: + raise DashboardNotFoundError("Automation not found", code="automation_not_found") from exc + return _to_run_response(run) + + +@router.get("/{automation_id}/runs", response_model=AutomationRunsListResponse) +async def list_automation_runs( + automation_id: str, + limit: int = Query(default=20, ge=1, le=200), + context: AutomationsContext = Depends(get_automations_context), +) -> AutomationRunsListResponse: + try: + runs = await context.service.list_runs(automation_id, limit=limit) + except AutomationNotFoundError as exc: + raise DashboardNotFoundError("Automation not found", code="automation_not_found") from exc + return AutomationRunsListResponse(items=[_to_run_response(run) for run in runs], total=len(runs), has_more=False) + + +def _to_job_response(job: AutomationJobData) -> AutomationJobResponse: + return AutomationJobResponse( + id=job.id, + name=job.name, + enabled=job.enabled, + include_paused_accounts=job.include_paused_accounts, + schedule=AutomationScheduleResponse( + type=_schedule_type_literal(job.schedule.type), + time=job.schedule.time, + timezone=job.schedule.timezone, + threshold_minutes=job.schedule.threshold_minutes, + days=_schedule_days_literals(job.schedule.days), + ), + model=job.model, + reasoning_effort=job.reasoning_effort, + prompt=job.prompt, + account_ids=job.account_ids, + next_run_at=job.next_run_at, + last_run=_to_run_response(job.last_run) if job.last_run else None, + ) + + +def _to_run_response(run: AutomationRunData) -> AutomationRunResponse: + return AutomationRunResponse( + id=run.id, + job_id=run.job_id, + job_name=run.job_name, + model=run.model, + reasoning_effort=run.reasoning_effort, + trigger=_run_trigger_literal(run.trigger), + status=_run_status_literal(run.status), + scheduled_for=run.scheduled_for, + started_at=run.started_at, + finished_at=run.finished_at, + account_id=run.account_id, + error_code=run.error_code, + error_message=run.error_message, + attempt_count=run.attempt_count, + effective_status=_run_status_literal(run.effective_status) if run.effective_status else None, + total_accounts=run.total_accounts, + completed_accounts=run.completed_accounts, + pending_accounts=run.pending_accounts, + cycle_key=run.cycle_key, + ) + + +def _to_run_details_response(details: AutomationRunDetailsData) -> AutomationRunDetailsResponse: + return AutomationRunDetailsResponse( + run=_to_run_response(details.run), + accounts=[_to_run_account_state_response(entry) for entry in details.accounts], + total_accounts=details.total_accounts, + completed_accounts=details.completed_accounts, + pending_accounts=details.pending_accounts, + ) + + +def _to_run_account_state_response(entry: AutomationRunAccountStateData) -> AutomationRunAccountStateResponse: + return AutomationRunAccountStateResponse( + account_id=entry.account_id, + status=_run_account_state_literal(entry.status), + run_id=entry.run_id, + scheduled_for=entry.scheduled_for, + started_at=entry.started_at, + finished_at=entry.finished_at, + error_code=entry.error_code, + error_message=entry.error_message, + ) + + +def _schedule_type_literal(value: str) -> Literal["daily"]: + if value == "daily": + return "daily" + raise RuntimeError(f"Unexpected automation schedule type: {value}") + + +def _run_trigger_literal(value: str) -> Literal["scheduled", "manual"]: + if value in {"scheduled", "manual"}: + return cast(Literal["scheduled", "manual"], value) + raise RuntimeError(f"Unexpected automation run trigger: {value}") + + +def _run_status_literal(value: str) -> Literal["running", "success", "failed", "partial"]: + if value in {"running", "success", "failed", "partial"}: + return cast(Literal["running", "success", "failed", "partial"], value) + raise RuntimeError(f"Unexpected automation run status: {value}") + + +def _run_account_state_literal(value: str) -> Literal["pending", "running", "success", "failed", "partial"]: + if value in {"pending", "running", "success", "failed", "partial"}: + return cast(Literal["pending", "running", "success", "failed", "partial"], value) + raise RuntimeError(f"Unexpected automation run account state: {value}") + + +def _schedule_days_literals(value: list[str]) -> list[Literal["mon", "tue", "wed", "thu", "fri", "sat", "sun"]]: + allowed = {"mon", "tue", "wed", "thu", "fri", "sat", "sun"} + invalid = [day for day in value if day not in allowed] + if invalid: + raise RuntimeError(f"Unexpected automation schedule day values: {invalid!r}") + return [cast(Literal["mon", "tue", "wed", "thu", "fri", "sat", "sun"], day) for day in value] diff --git a/app/modules/automations/repository.py b/app/modules/automations/repository.py new file mode 100644 index 0000000000..fb8a0ea532 --- /dev/null +++ b/app/modules/automations/repository.py @@ -0,0 +1,983 @@ +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass +from datetime import datetime +from uuid import uuid4 + +from sqlalchemy import and_, case, delete, exists, func, or_, select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload + +from app.db.models import Account, AutomationJob, AutomationJobAccount, AutomationRun + +DEFAULT_AUTOMATION_SCHEDULE_DAYS = ["mon", "tue", "wed", "thu", "fri", "sat", "sun"] + + +@dataclass(frozen=True, slots=True) +class AutomationRunRecord: + id: str + job_id: str + job_name: str | None + model: str | None + reasoning_effort: str | None + trigger: str + status: str + slot_key: str + cycle_key: str + cycle_expected_accounts: int | None + cycle_window_end: datetime | None + scheduled_for: datetime + started_at: datetime + finished_at: datetime | None + account_id: str | None + error_code: str | None + error_message: str | None + attempt_count: int + + +@dataclass(frozen=True, slots=True) +class AutomationJobRecord: + id: str + name: str + enabled: bool + schedule_type: str + schedule_time: str + schedule_timezone: str + schedule_days: list[str] + schedule_threshold_minutes: int + include_paused_accounts: bool + model: str + reasoning_effort: str | None + prompt: str + account_ids: list[str] + created_at: datetime + updated_at: datetime + + +@dataclass(frozen=True, slots=True) +class AutomationJobsFilterOptionsRecord: + account_ids: list[str] + models: list[str] + statuses: list[str] + schedule_types: list[str] + + +@dataclass(frozen=True, slots=True) +class AutomationRunsFilterOptionsRecord: + account_ids: list[str] + models: list[str] + statuses: list[str] + triggers: list[str] + + +class AutomationsRepository: + def __init__(self, session: AsyncSession) -> None: + self._session = session + + async def list_jobs(self) -> list[AutomationJobRecord]: + result = await self._session.execute( + select(AutomationJob) + .options(selectinload(AutomationJob.account_links)) + .order_by(AutomationJob.created_at.desc(), AutomationJob.id.asc()) + ) + jobs = list(result.scalars().all()) + return [self._job_from_model(job) for job in jobs] + + async def list_jobs_page( + self, + *, + limit: int, + offset: int, + search: str | None = None, + account_ids: Sequence[str] | None = None, + models: Sequence[str] | None = None, + statuses: Sequence[str] | None = None, + schedule_types: Sequence[str] | None = None, + ) -> tuple[list[AutomationJobRecord], int]: + conditions = self._build_job_conditions( + search=search, + account_ids=account_ids, + models=models, + statuses=statuses, + schedule_types=schedule_types, + ) + stmt = ( + select(AutomationJob) + .options(selectinload(AutomationJob.account_links)) + .order_by(AutomationJob.created_at.desc(), AutomationJob.id.asc()) + .offset(offset) + .limit(limit) + ) + if conditions: + stmt = stmt.where(and_(*conditions)) + result = await self._session.execute(stmt) + jobs = [self._job_from_model(job) for job in result.scalars().all()] + + count_stmt = select(func.count(AutomationJob.id)) + if conditions: + count_stmt = count_stmt.where(and_(*conditions)) + total = int((await self._session.execute(count_stmt)).scalar_one() or 0) + return jobs, total + + async def list_job_filter_options( + self, + *, + search: str | None = None, + account_ids: Sequence[str] | None = None, + models: Sequence[str] | None = None, + statuses: Sequence[str] | None = None, + schedule_types: Sequence[str] | None = None, + ) -> AutomationJobsFilterOptionsRecord: + conditions = self._build_job_conditions( + search=search, + account_ids=account_ids, + models=models, + statuses=statuses, + schedule_types=schedule_types, + ) + + account_stmt = ( + select(AutomationJobAccount.account_id) + .distinct() + .join(AutomationJob, AutomationJob.id == AutomationJobAccount.job_id) + .order_by(AutomationJobAccount.account_id.asc()) + ) + model_stmt = select(AutomationJob.model).distinct().order_by(AutomationJob.model.asc()) + type_stmt = ( + select(AutomationJob.schedule_type) + .distinct() + .order_by(AutomationJob.schedule_type.asc()) + ) + status_stmt = select(AutomationJob.enabled).distinct() + if conditions: + clause = and_(*conditions) + account_stmt = account_stmt.where(clause) + model_stmt = model_stmt.where(clause) + type_stmt = type_stmt.where(clause) + status_stmt = status_stmt.where(clause) + + account_ids_rows = await self._session.execute(account_stmt) + model_rows = await self._session.execute(model_stmt) + type_rows = await self._session.execute(type_stmt) + status_rows = await self._session.execute(status_stmt) + + statuses = sorted( + { + "enabled" if bool(value) else "disabled" + for (value,) in status_rows.all() + } + ) + return AutomationJobsFilterOptionsRecord( + account_ids=[value for (value,) in account_ids_rows.all() if value], + models=[value for (value,) in model_rows.all() if value], + statuses=statuses, + schedule_types=[value for (value,) in type_rows.all() if value], + ) + + async def list_enabled_jobs(self) -> list[AutomationJobRecord]: + result = await self._session.execute( + select(AutomationJob) + .where(AutomationJob.enabled.is_(True)) + .options(selectinload(AutomationJob.account_links)) + .order_by(AutomationJob.created_at.asc(), AutomationJob.id.asc()) + ) + jobs = list(result.scalars().all()) + return [self._job_from_model(job) for job in jobs] + + async def get_job(self, job_id: str) -> AutomationJobRecord | None: + result = await self._session.execute( + select(AutomationJob).where(AutomationJob.id == job_id).options(selectinload(AutomationJob.account_links)) + ) + job = result.scalar_one_or_none() + if job is None: + return None + return self._job_from_model(job) + + async def get_jobs_by_ids(self, job_ids: Sequence[str]) -> dict[str, AutomationJobRecord]: + normalized_job_ids = [job_id for job_id in job_ids if job_id] + if not normalized_job_ids: + return {} + result = await self._session.execute( + select(AutomationJob) + .where(AutomationJob.id.in_(list(dict.fromkeys(normalized_job_ids)))) + .options(selectinload(AutomationJob.account_links)) + ) + jobs = [self._job_from_model(job) for job in result.scalars().all()] + return {job.id: job for job in jobs} + + async def create_job( + self, + *, + name: str, + enabled: bool, + schedule_type: str, + schedule_time: str, + schedule_timezone: str, + schedule_days: Sequence[str], + schedule_threshold_minutes: int, + include_paused_accounts: bool, + model: str, + reasoning_effort: str | None, + prompt: str, + account_ids: Sequence[str], + ) -> AutomationJobRecord: + job = AutomationJob( + id=f"job_{uuid4().hex}", + name=name, + enabled=enabled, + schedule_type=schedule_type, + schedule_time=schedule_time, + schedule_timezone=schedule_timezone, + schedule_days=_serialize_schedule_days(schedule_days), + schedule_threshold_minutes=schedule_threshold_minutes, + include_paused_accounts=include_paused_accounts, + model=model, + reasoning_effort=reasoning_effort, + prompt=prompt, + ) + job.account_links = [ + AutomationJobAccount(job_id=job.id, account_id=account_id, position=index) + for index, account_id in enumerate(account_ids) + ] + self._session.add(job) + await self._session.commit() + await self._session.refresh(job) + await self._session.refresh(job, attribute_names=["account_links"]) + return self._job_from_model(job) + + async def update_job( + self, + job_id: str, + *, + name: str | None = None, + enabled: bool | None = None, + schedule_type: str | None = None, + schedule_time: str | None = None, + schedule_timezone: str | None = None, + schedule_days: Sequence[str] | None = None, + schedule_threshold_minutes: int | None = None, + include_paused_accounts: bool | None = None, + model: str | None = None, + reasoning_effort: str | None = None, + reasoning_effort_set: bool = False, + prompt: str | None = None, + account_ids: Sequence[str] | None = None, + ) -> AutomationJobRecord | None: + result = await self._session.execute( + select(AutomationJob).where(AutomationJob.id == job_id).options(selectinload(AutomationJob.account_links)) + ) + job = result.scalar_one_or_none() + if job is None: + return None + + if name is not None: + job.name = name + if enabled is not None: + job.enabled = enabled + if schedule_type is not None: + job.schedule_type = schedule_type + if schedule_time is not None: + job.schedule_time = schedule_time + if schedule_timezone is not None: + job.schedule_timezone = schedule_timezone + if schedule_days is not None: + job.schedule_days = _serialize_schedule_days(schedule_days) + if schedule_threshold_minutes is not None: + job.schedule_threshold_minutes = schedule_threshold_minutes + if include_paused_accounts is not None: + job.include_paused_accounts = include_paused_accounts + if model is not None: + job.model = model + if reasoning_effort_set: + job.reasoning_effort = reasoning_effort + if prompt is not None: + job.prompt = prompt + if account_ids is not None: + await self._session.execute( + delete(AutomationJobAccount).where(AutomationJobAccount.job_id == job.id) + ) + await self._session.flush() + self._session.add_all( + [ + AutomationJobAccount(job_id=job.id, account_id=account_id, position=index) + for index, account_id in enumerate(account_ids) + ] + ) + + await self._session.commit() + await self._session.refresh(job) + await self._session.refresh(job, attribute_names=["account_links"]) + return self._job_from_model(job) + + async def delete_job(self, job_id: str) -> bool: + result = await self._session.execute( + delete(AutomationJob).where(AutomationJob.id == job_id).returning(AutomationJob.id) + ) + await self._session.commit() + return result.scalar_one_or_none() is not None + + async def list_existing_account_ids(self, account_ids: Sequence[str]) -> set[str]: + if not account_ids: + return set() + result = await self._session.execute(select(Account.id).where(Account.id.in_(list(account_ids)))) + return set(result.scalars().all()) + + async def claim_run( + self, + *, + job_id: str, + trigger: str, + slot_key: str, + cycle_key: str, + cycle_expected_accounts: int | None, + cycle_window_end: datetime | None, + scheduled_for: datetime, + started_at: datetime, + account_id: str | None = None, + ) -> AutomationRunRecord | None: + run = AutomationRun( + id=f"run_{uuid4().hex}", + job_id=job_id, + trigger=trigger, + slot_key=slot_key, + cycle_key=cycle_key, + cycle_expected_accounts=cycle_expected_accounts, + cycle_window_end=cycle_window_end, + scheduled_for=scheduled_for, + started_at=started_at, + status="running", + account_id=account_id, + attempt_count=0, + ) + self._session.add(run) + try: + await self._session.commit() + except IntegrityError: + await self._session.rollback() + return None + await self._session.refresh(run) + return self._run_from_model(run) + + async def complete_run( + self, + run_id: str, + *, + status: str, + finished_at: datetime, + account_id: str | None, + error_code: str | None, + error_message: str | None, + attempt_count: int, + ) -> AutomationRunRecord: + run = await self._session.get(AutomationRun, run_id) + if run is None: + raise LookupError(f"Automation run not found: {run_id}") + run.status = status + run.finished_at = finished_at + run.account_id = account_id + run.error_code = error_code + run.error_message = error_message + run.attempt_count = attempt_count + await self._session.commit() + await self._session.refresh(run) + return self._run_from_model(run) + + async def list_runs(self, job_id: str, *, limit: int) -> list[AutomationRunRecord]: + result = await self._session.execute( + select(AutomationRun, AutomationJob.name, AutomationJob.model, AutomationJob.reasoning_effort) + .join(AutomationJob, AutomationJob.id == AutomationRun.job_id) + .where(AutomationRun.job_id == job_id) + .order_by(AutomationRun.started_at.desc(), AutomationRun.id.desc()) + .limit(limit) + ) + return [ + self._run_from_model(run, job_name=job_name, model=model, reasoning_effort=reasoning_effort) + for run, job_name, model, reasoning_effort in result.all() + ] + + async def get_run(self, run_id: str) -> AutomationRunRecord | None: + result = await self._session.execute( + select(AutomationRun, AutomationJob.name, AutomationJob.model, AutomationJob.reasoning_effort) + .join(AutomationJob, AutomationJob.id == AutomationRun.job_id) + .where(AutomationRun.id == run_id) + .limit(1) + ) + row = result.one_or_none() + if row is None: + return None + run, job_name, model, reasoning_effort = row + return self._run_from_model(run, job_name=job_name, model=model, reasoning_effort=reasoning_effort) + + async def list_runs_for_cycle_window( + self, + *, + job_id: str, + trigger: str, + scheduled_from: datetime, + scheduled_to: datetime, + ) -> list[AutomationRunRecord]: + result = await self._session.execute( + select(AutomationRun, AutomationJob.name, AutomationJob.model, AutomationJob.reasoning_effort) + .join(AutomationJob, AutomationJob.id == AutomationRun.job_id) + .where(AutomationRun.job_id == job_id) + .where(AutomationRun.trigger == trigger) + .where(AutomationRun.scheduled_for >= scheduled_from) + .where(AutomationRun.scheduled_for <= scheduled_to) + .order_by(AutomationRun.started_at.desc(), AutomationRun.id.desc()) + ) + return [ + self._run_from_model(run, job_name=job_name, model=model, reasoning_effort=reasoning_effort) + for run, job_name, model, reasoning_effort in result.all() + ] + + async def list_runs_for_cycle_key(self, *, cycle_key: str) -> list[AutomationRunRecord]: + result = await self._session.execute( + select(AutomationRun, AutomationJob.name, AutomationJob.model, AutomationJob.reasoning_effort) + .join(AutomationJob, AutomationJob.id == AutomationRun.job_id) + .where(AutomationRun.cycle_key == cycle_key) + .order_by(AutomationRun.started_at.desc(), AutomationRun.id.desc()) + ) + return [ + self._run_from_model(run, job_name=job_name, model=model, reasoning_effort=reasoning_effort) + for run, job_name, model, reasoning_effort in result.all() + ] + + async def list_runs_for_manual_cycle( + self, + *, + job_id: str, + slot_key_prefix: str, + ) -> list[AutomationRunRecord]: + result = await self._session.execute( + select(AutomationRun, AutomationJob.name, AutomationJob.model, AutomationJob.reasoning_effort) + .join(AutomationJob, AutomationJob.id == AutomationRun.job_id) + .where(AutomationRun.job_id == job_id) + .where(AutomationRun.trigger == "manual") + .where(AutomationRun.slot_key.like(f"{slot_key_prefix}%")) + .order_by(AutomationRun.started_at.desc(), AutomationRun.id.desc()) + ) + return [ + self._run_from_model(run, job_name=job_name, model=model, reasoning_effort=reasoning_effort) + for run, job_name, model, reasoning_effort in result.all() + ] + + async def list_due_manual_runs(self, *, now_utc: datetime, limit: int = 500) -> list[AutomationRunRecord]: + result = await self._session.execute( + select(AutomationRun, AutomationJob.name, AutomationJob.model, AutomationJob.reasoning_effort) + .join(AutomationJob, AutomationJob.id == AutomationRun.job_id) + .where(AutomationRun.trigger == "manual") + .where(AutomationRun.status == "running") + .where(AutomationRun.finished_at.is_(None)) + .where(AutomationRun.account_id.is_not(None)) + .where(AutomationRun.scheduled_for <= now_utc) + .order_by(AutomationRun.scheduled_for.asc(), AutomationRun.started_at.asc(), AutomationRun.id.asc()) + .limit(limit) + ) + return [ + self._run_from_model(run, job_name=job_name, model=model, reasoning_effort=reasoning_effort) + for run, job_name, model, reasoning_effort in result.all() + ] + + async def list_runs_page( + self, + *, + limit: int, + offset: int, + search: str | None = None, + account_ids: Sequence[str] | None = None, + models: Sequence[str] | None = None, + statuses: Sequence[str] | None = None, + triggers: Sequence[str] | None = None, + job_ids: Sequence[str] | None = None, + ) -> tuple[list[AutomationRunRecord], int]: + conditions = self._build_run_conditions( + search=search, + account_ids=account_ids, + models=models, + statuses=statuses, + triggers=triggers, + job_ids=job_ids, + ) + stmt = ( + select(AutomationRun, AutomationJob.name, AutomationJob.model, AutomationJob.reasoning_effort) + .join(AutomationJob, AutomationJob.id == AutomationRun.job_id) + .order_by(AutomationRun.started_at.desc(), AutomationRun.id.desc()) + .offset(offset) + .limit(limit) + ) + if conditions: + stmt = stmt.where(and_(*conditions)) + result = await self._session.execute(stmt) + runs = [ + self._run_from_model(run, job_name=job_name, model=model, reasoning_effort=reasoning_effort) + for run, job_name, model, reasoning_effort in result.all() + ] + + count_stmt = ( + select(func.count(AutomationRun.id)) + .select_from(AutomationRun) + .join(AutomationJob, AutomationJob.id == AutomationRun.job_id) + ) + if conditions: + count_stmt = count_stmt.where(and_(*conditions)) + total = int((await self._session.execute(count_stmt)).scalar_one() or 0) + return runs, total + + async def list_runs_filtered( + self, + *, + search: str | None = None, + account_ids: Sequence[str] | None = None, + models: Sequence[str] | None = None, + statuses: Sequence[str] | None = None, + triggers: Sequence[str] | None = None, + job_ids: Sequence[str] | None = None, + ) -> list[AutomationRunRecord]: + conditions = self._build_run_conditions( + search=search, + account_ids=account_ids, + models=models, + statuses=statuses, + triggers=triggers, + job_ids=job_ids, + ) + stmt = ( + select(AutomationRun, AutomationJob.name, AutomationJob.model, AutomationJob.reasoning_effort) + .join(AutomationJob, AutomationJob.id == AutomationRun.job_id) + .order_by(AutomationRun.started_at.desc(), AutomationRun.id.desc()) + ) + if conditions: + stmt = stmt.where(and_(*conditions)) + result = await self._session.execute(stmt) + return [ + self._run_from_model(run, job_name=job_name, model=model, reasoning_effort=reasoning_effort) + for run, job_name, model, reasoning_effort in result.all() + ] + + async def list_run_cycles_page( + self, + *, + limit: int, + offset: int, + now_utc: datetime, + search: str | None = None, + account_ids: Sequence[str] | None = None, + models: Sequence[str] | None = None, + statuses: Sequence[str] | None = None, + triggers: Sequence[str] | None = None, + job_ids: Sequence[str] | None = None, + ) -> tuple[list[AutomationRunRecord], int]: + conditions = self._build_run_conditions( + search=search, + account_ids=account_ids, + models=models, + statuses=None, + triggers=triggers, + job_ids=job_ids, + ) + filtered_runs_stmt = ( + select( + AutomationRun.id.label("run_id"), + AutomationRun.cycle_key.label("cycle_key"), + AutomationRun.status.label("status"), + AutomationRun.account_id.label("account_id"), + AutomationRun.started_at.label("started_at"), + AutomationRun.scheduled_for.label("scheduled_for"), + AutomationRun.cycle_window_end.label("cycle_window_end"), + AutomationRun.cycle_expected_accounts.label("cycle_expected_accounts"), + ) + .join(AutomationJob, AutomationJob.id == AutomationRun.job_id) + ) + if conditions: + filtered_runs_stmt = filtered_runs_stmt.where(and_(*conditions)) + filtered_runs = filtered_runs_stmt.subquery() + candidate_cycles = select(filtered_runs.c.cycle_key).distinct().subquery() + + cycle_runs_stmt = ( + select( + AutomationRun.id.label("run_id"), + AutomationRun.cycle_key.label("cycle_key"), + AutomationRun.status.label("status"), + AutomationRun.account_id.label("account_id"), + AutomationRun.started_at.label("started_at"), + AutomationRun.scheduled_for.label("scheduled_for"), + AutomationRun.cycle_window_end.label("cycle_window_end"), + AutomationRun.cycle_expected_accounts.label("cycle_expected_accounts"), + ) + .join(AutomationJob, AutomationJob.id == AutomationRun.job_id) + .join(candidate_cycles, candidate_cycles.c.cycle_key == AutomationRun.cycle_key) + ) + cycle_runs = cycle_runs_stmt.subquery() + + ranked_stmt = ( + select( + filtered_runs.c.run_id, + filtered_runs.c.cycle_key, + filtered_runs.c.status.label("fallback_status"), + func.row_number() + .over( + partition_by=filtered_runs.c.cycle_key, + order_by=(filtered_runs.c.started_at.desc(), filtered_runs.c.run_id.desc()), + ) + .label("cycle_rank"), + ) + ) + ranked = ranked_stmt.subquery() + + cycle_agg_stmt = select( + cycle_runs.c.cycle_key.label("cycle_key"), + func.max(cycle_runs.c.started_at).label("cycle_started_at"), + func.count( + func.distinct( + case( + (cycle_runs.c.status != "running", cycle_runs.c.account_id), + else_=None, + ) + ) + ).label("completed_accounts"), + func.sum(case((cycle_runs.c.status == "success", 1), else_=0)).label("success_count"), + func.sum(case((cycle_runs.c.status == "failed", 1), else_=0)).label("failed_count"), + func.sum(case((cycle_runs.c.status == "partial", 1), else_=0)).label("partial_count"), + func.sum(case((cycle_runs.c.status == "running", 1), else_=0)).label("running_count"), + func.max(func.coalesce(cycle_runs.c.cycle_expected_accounts, 0)).label("expected_accounts"), + func.max(func.coalesce(cycle_runs.c.cycle_window_end, cycle_runs.c.scheduled_for)).label("window_end"), + ).group_by(cycle_runs.c.cycle_key) + cycle_agg = cycle_agg_stmt.subquery() + + cycle_rows_stmt = ( + select( + ranked.c.run_id, + ranked.c.cycle_key, + cycle_agg.c.cycle_started_at, + ranked.c.fallback_status, + cycle_agg.c.completed_accounts, + cycle_agg.c.success_count, + cycle_agg.c.failed_count, + cycle_agg.c.partial_count, + cycle_agg.c.running_count, + cycle_agg.c.expected_accounts, + cycle_agg.c.window_end, + ) + .join(cycle_agg, cycle_agg.c.cycle_key == ranked.c.cycle_key) + .where(ranked.c.cycle_rank == 1) + ) + cycle_rows = cycle_rows_stmt.subquery() + + effective_total_expr = case( + (cycle_rows.c.expected_accounts > cycle_rows.c.completed_accounts, cycle_rows.c.expected_accounts), + else_=cycle_rows.c.completed_accounts, + ) + pending_expr = effective_total_expr - cycle_rows.c.completed_accounts + effective_status_expr = case( + (cycle_rows.c.running_count > 0, "running"), + (and_(pending_expr > 0, now_utc <= cycle_rows.c.window_end), "running"), + (pending_expr > 0, case((cycle_rows.c.completed_accounts > 0, "partial"), else_="failed")), + ( + and_( + cycle_rows.c.success_count > 0, + cycle_rows.c.failed_count == 0, + cycle_rows.c.partial_count == 0, + ), + "success", + ), + ( + and_( + cycle_rows.c.success_count > 0, + or_(cycle_rows.c.failed_count > 0, cycle_rows.c.partial_count > 0), + ), + "partial", + ), + ( + and_( + cycle_rows.c.failed_count > 0, + cycle_rows.c.success_count == 0, + cycle_rows.c.partial_count == 0, + ), + "failed", + ), + (cycle_rows.c.partial_count > 0, "partial"), + else_=cycle_rows.c.fallback_status, + ).label("effective_status") + + cycles_with_status_stmt = select( + cycle_rows.c.run_id, + cycle_rows.c.cycle_key, + cycle_rows.c.cycle_started_at, + effective_status_expr, + ) + if statuses: + cycles_with_status_stmt = cycles_with_status_stmt.where(effective_status_expr.in_(list(statuses))) + cycles_with_status = cycles_with_status_stmt.subquery() + + page_ids_stmt = ( + select(cycles_with_status.c.run_id) + .order_by(cycles_with_status.c.cycle_started_at.desc(), cycles_with_status.c.run_id.desc()) + .offset(offset) + .limit(limit) + ) + run_ids_rows = await self._session.execute(page_ids_stmt) + run_ids = [value for (value,) in run_ids_rows.all() if value] + if not run_ids: + count_stmt = select(func.count()).select_from(cycles_with_status) + total = int((await self._session.execute(count_stmt)).scalar_one() or 0) + return [], total + + runs_stmt = ( + select(AutomationRun, AutomationJob.name, AutomationJob.model, AutomationJob.reasoning_effort) + .join(AutomationJob, AutomationJob.id == AutomationRun.job_id) + .where(AutomationRun.id.in_(run_ids)) + ) + runs_rows = await self._session.execute(runs_stmt) + runs_by_id = { + run.id: self._run_from_model(run, job_name=job_name, model=model, reasoning_effort=reasoning_effort) + for run, job_name, model, reasoning_effort in runs_rows.all() + } + ordered_runs = [runs_by_id[run_id] for run_id in run_ids if run_id in runs_by_id] + + count_stmt = select(func.count()).select_from(cycles_with_status) + total = int((await self._session.execute(count_stmt)).scalar_one() or 0) + return ordered_runs, total + + async def list_run_filter_options( + self, + *, + search: str | None = None, + account_ids: Sequence[str] | None = None, + models: Sequence[str] | None = None, + statuses: Sequence[str] | None = None, + triggers: Sequence[str] | None = None, + job_ids: Sequence[str] | None = None, + ) -> AutomationRunsFilterOptionsRecord: + conditions = self._build_run_conditions( + search=search, + account_ids=account_ids, + models=models, + statuses=statuses, + triggers=triggers, + job_ids=job_ids, + ) + account_stmt = ( + select(AutomationRun.account_id) + .distinct() + .join(AutomationJob, AutomationJob.id == AutomationRun.job_id) + .where(AutomationRun.account_id.is_not(None)) + .order_by(AutomationRun.account_id.asc()) + ) + model_stmt = ( + select(AutomationJob.model) + .distinct() + .join(AutomationRun, AutomationRun.job_id == AutomationJob.id) + .order_by(AutomationJob.model.asc()) + ) + status_stmt = ( + select(AutomationRun.status) + .distinct() + .join(AutomationJob, AutomationJob.id == AutomationRun.job_id) + .order_by(AutomationRun.status.asc()) + ) + trigger_stmt = ( + select(AutomationRun.trigger) + .distinct() + .join(AutomationJob, AutomationJob.id == AutomationRun.job_id) + .order_by(AutomationRun.trigger.asc()) + ) + if conditions: + clause = and_(*conditions) + account_stmt = account_stmt.where(clause) + model_stmt = model_stmt.where(clause) + status_stmt = status_stmt.where(clause) + trigger_stmt = trigger_stmt.where(clause) + + account_rows = await self._session.execute(account_stmt) + model_rows = await self._session.execute(model_stmt) + status_rows = await self._session.execute(status_stmt) + trigger_rows = await self._session.execute(trigger_stmt) + return AutomationRunsFilterOptionsRecord( + account_ids=[value for (value,) in account_rows.all() if value], + models=[value for (value,) in model_rows.all() if value], + statuses=[value for (value,) in status_rows.all() if value], + triggers=[value for (value,) in trigger_rows.all() if value], + ) + + async def get_latest_runs_by_job_ids(self, job_ids: Sequence[str]) -> dict[str, AutomationRunRecord]: + if not job_ids: + return {} + result = await self._session.execute( + select(AutomationRun) + .where(AutomationRun.job_id.in_(list(job_ids))) + .order_by(AutomationRun.job_id.asc(), AutomationRun.started_at.desc(), AutomationRun.id.desc()) + ) + latest: dict[str, AutomationRunRecord] = {} + for run in result.scalars().all(): + if run.job_id in latest: + continue + latest[run.job_id] = self._run_from_model(run) + return latest + + @staticmethod + def _job_from_model(job: AutomationJob) -> AutomationJobRecord: + sorted_accounts = sorted(job.account_links, key=lambda link: link.position) + return AutomationJobRecord( + id=job.id, + name=job.name, + enabled=job.enabled, + schedule_type=job.schedule_type, + schedule_time=job.schedule_time, + schedule_timezone=job.schedule_timezone, + schedule_days=_parse_schedule_days(job.schedule_days), + schedule_threshold_minutes=job.schedule_threshold_minutes, + include_paused_accounts=job.include_paused_accounts, + model=job.model, + reasoning_effort=job.reasoning_effort, + prompt=job.prompt, + account_ids=[link.account_id for link in sorted_accounts], + created_at=job.created_at, + updated_at=job.updated_at, + ) + + @staticmethod + def _run_from_model( + run: AutomationRun, + *, + job_name: str | None = None, + model: str | None = None, + reasoning_effort: str | None = None, + ) -> AutomationRunRecord: + return AutomationRunRecord( + id=run.id, + job_id=run.job_id, + job_name=job_name, + model=model, + reasoning_effort=reasoning_effort, + trigger=run.trigger, + status=run.status, + slot_key=run.slot_key, + cycle_key=run.cycle_key, + cycle_expected_accounts=run.cycle_expected_accounts, + cycle_window_end=run.cycle_window_end, + scheduled_for=run.scheduled_for, + started_at=run.started_at, + finished_at=run.finished_at, + account_id=run.account_id, + error_code=run.error_code, + error_message=run.error_message, + attempt_count=run.attempt_count, + ) + + @staticmethod + def _build_job_conditions( + *, + search: str | None, + account_ids: Sequence[str] | None, + models: Sequence[str] | None, + statuses: Sequence[str] | None, + schedule_types: Sequence[str] | None, + ) -> list: + conditions = [] + normalized_search = (search or "").strip() + if normalized_search: + like = f"%{normalized_search}%" + conditions.append( + or_( + AutomationJob.id.ilike(like), + AutomationJob.name.ilike(like), + AutomationJob.prompt.ilike(like), + AutomationJob.model.ilike(like), + AutomationJob.reasoning_effort.ilike(like), + ) + ) + normalized_accounts = [value.strip() for value in (account_ids or []) if value and value.strip()] + if normalized_accounts: + matching_account_links = select(AutomationJobAccount.job_id).where( + AutomationJobAccount.account_id.in_(normalized_accounts) + ) + job_has_no_account_links = ~exists( + select(1).where(AutomationJobAccount.job_id == AutomationJob.id) + ) + conditions.append( + or_( + AutomationJob.id.in_(matching_account_links), + job_has_no_account_links, + ) + ) + normalized_models = [value.strip() for value in (models or []) if value and value.strip()] + if normalized_models: + conditions.append(AutomationJob.model.in_(normalized_models)) + normalized_types = [value.strip() for value in (schedule_types or []) if value and value.strip()] + if normalized_types: + conditions.append(AutomationJob.schedule_type.in_(normalized_types)) + normalized_statuses = {value.strip().lower() for value in (statuses or []) if value and value.strip()} + if normalized_statuses and "all" not in normalized_statuses: + enabled_values: list[bool] = [] + if "enabled" in normalized_statuses: + enabled_values.append(True) + if "disabled" in normalized_statuses: + enabled_values.append(False) + if enabled_values: + conditions.append(AutomationJob.enabled.in_(enabled_values)) + else: + conditions.append(AutomationJob.id == "__none__") + return conditions + + @staticmethod + def _build_run_conditions( + *, + search: str | None, + account_ids: Sequence[str] | None, + models: Sequence[str] | None, + statuses: Sequence[str] | None, + triggers: Sequence[str] | None, + job_ids: Sequence[str] | None, + ) -> list: + conditions = [] + normalized_search = (search or "").strip() + if normalized_search: + like = f"%{normalized_search}%" + conditions.append( + or_( + AutomationRun.id.ilike(like), + AutomationRun.job_id.ilike(like), + AutomationRun.account_id.ilike(like), + AutomationRun.error_code.ilike(like), + AutomationRun.error_message.ilike(like), + AutomationJob.name.ilike(like), + AutomationJob.model.ilike(like), + AutomationJob.reasoning_effort.ilike(like), + ) + ) + normalized_accounts = [value.strip() for value in (account_ids or []) if value and value.strip()] + if normalized_accounts: + conditions.append(AutomationRun.account_id.in_(normalized_accounts)) + normalized_models = [value.strip() for value in (models or []) if value and value.strip()] + if normalized_models: + conditions.append(AutomationJob.model.in_(normalized_models)) + normalized_statuses = [value.strip().lower() for value in (statuses or []) if value and value.strip()] + if normalized_statuses: + conditions.append(AutomationRun.status.in_(normalized_statuses)) + normalized_triggers = [value.strip().lower() for value in (triggers or []) if value and value.strip()] + if normalized_triggers: + conditions.append(AutomationRun.trigger.in_(normalized_triggers)) + normalized_job_ids = [value.strip() for value in (job_ids or []) if value and value.strip()] + if normalized_job_ids: + conditions.append(AutomationRun.job_id.in_(normalized_job_ids)) + return conditions + + +def _parse_schedule_days(value: str | None) -> list[str]: + if not value: + return list(DEFAULT_AUTOMATION_SCHEDULE_DAYS) + parsed = [part.strip().lower() for part in value.split(",") if part.strip()] + if not parsed: + return list(DEFAULT_AUTOMATION_SCHEDULE_DAYS) + return parsed + + +def _serialize_schedule_days(days: Sequence[str]) -> str: + if not days: + return ",".join(DEFAULT_AUTOMATION_SCHEDULE_DAYS) + normalized = [day.strip().lower() for day in days if day.strip()] + if not normalized: + return ",".join(DEFAULT_AUTOMATION_SCHEDULE_DAYS) + return ",".join(normalized) diff --git a/app/modules/automations/scheduler.py b/app/modules/automations/scheduler.py new file mode 100644 index 0000000000..7829420e2b --- /dev/null +++ b/app/modules/automations/scheduler.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import asyncio +import contextlib +import importlib +import logging +from dataclasses import dataclass, field +from typing import Protocol, cast + +from app.core.config.settings import get_settings +from app.db.session import get_background_session +from app.modules.accounts.repository import AccountsRepository +from app.modules.automations.repository import AutomationsRepository +from app.modules.automations.service import AutomationsService +from app.modules.request_logs.repository import RequestLogsRepository + +logger = logging.getLogger(__name__) + + +class _LeaderElectionLike(Protocol): + async def try_acquire(self) -> bool: ... + + +def _get_leader_election() -> _LeaderElectionLike: + module = importlib.import_module("app.core.scheduling.leader_election") + return cast(_LeaderElectionLike, module.get_leader_election()) + + +@dataclass(slots=True) +class AutomationsScheduler: + interval_seconds: int + enabled: bool + _task: asyncio.Task[None] | None = None + _stop: asyncio.Event = field(default_factory=asyncio.Event) + _lock: asyncio.Lock = field(default_factory=asyncio.Lock) + + async def start(self) -> None: + if not self.enabled: + return + if self._task and not self._task.done(): + return + self._stop.clear() + self._task = asyncio.create_task(self._run_loop()) + + async def stop(self) -> None: + if not self._task: + return + self._stop.set() + self._task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await self._task + self._task = None + + async def _run_loop(self) -> None: + while not self._stop.is_set(): + await self._run_due_once() + try: + await asyncio.wait_for(self._stop.wait(), timeout=self.interval_seconds) + except asyncio.TimeoutError: + continue + + async def _run_due_once(self) -> None: + if not await _get_leader_election().try_acquire(): + return + async with self._lock: + try: + async with get_background_session() as session: + repository = AutomationsRepository(session) + accounts_repository = AccountsRepository(session) + request_logs_repository = RequestLogsRepository(session) + service = AutomationsService(repository, accounts_repository, request_logs_repository) + await service.run_due_jobs() + except Exception: + logger.exception("Automations scheduler loop failed") + + +def build_automations_scheduler() -> AutomationsScheduler: + settings = get_settings() + return AutomationsScheduler( + interval_seconds=settings.automations_scheduler_interval_seconds, + enabled=settings.automations_scheduler_enabled, + ) diff --git a/app/modules/automations/schemas.py b/app/modules/automations/schemas.py new file mode 100644 index 0000000000..02922b09a3 --- /dev/null +++ b/app/modules/automations/schemas.py @@ -0,0 +1,175 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Literal + +from pydantic import Field, field_validator + +from app.modules.shared.schemas import DashboardModel + +AUTOMATION_SCHEDULE_TYPES = ("daily",) +AUTOMATION_WEEKDAY_CODES = ("mon", "tue", "wed", "thu", "fri", "sat", "sun") +AUTOMATION_RUN_STATUSES = ("running", "success", "failed", "partial") +AUTOMATION_RUN_TRIGGERS = ("scheduled", "manual") +AUTOMATION_REASONING_EFFORTS = ("minimal", "low", "medium", "high", "xhigh") + +AutomationWeekday = Literal["mon", "tue", "wed", "thu", "fri", "sat", "sun"] + + +class AutomationScheduleRequest(DashboardModel): + type: Literal["daily"] = "daily" + time: str = Field(pattern=r"^\d{2}:\d{2}$") + timezone: str = Field(min_length=1, max_length=64) + threshold_minutes: int = Field(default=0, ge=0, le=240) + days: list[AutomationWeekday] = Field( + default_factory=lambda: list(AUTOMATION_WEEKDAY_CODES), + min_length=1, + max_length=7, + ) + + @field_validator("days") + @classmethod + def _validate_days(cls, value: list[AutomationWeekday]) -> list[AutomationWeekday]: + deduped: list[AutomationWeekday] = [] + seen: set[str] = set() + for day in value: + if day in seen: + continue + seen.add(day) + deduped.append(day) + if not deduped: + raise ValueError("At least one schedule day is required") + return deduped + + +class AutomationScheduleResponse(DashboardModel): + type: Literal["daily"] + time: str + timezone: str + threshold_minutes: int + days: list[AutomationWeekday] + + +class AutomationJobCreateRequest(DashboardModel): + name: str = Field(min_length=1, max_length=200) + enabled: bool = True + include_paused_accounts: bool = Field(default=False, alias="includePausedAccounts") + schedule: AutomationScheduleRequest + model: str = Field(min_length=1) + reasoning_effort: str | None = Field(default=None, pattern=r"(?i)^(minimal|low|medium|high|xhigh)$") + prompt: str | None = Field(default=None, max_length=1000) + account_ids: list[str] = Field(default_factory=list, max_length=128, alias="accountIds") + + @field_validator("account_ids") + @classmethod + def _validate_unique_account_ids(cls, value: list[str]) -> list[str]: + normalized = [account_id.strip() for account_id in value if account_id.strip()] + if len(set(normalized)) != len(normalized): + raise ValueError("Duplicate account IDs are not allowed") + return normalized + + +class AutomationJobUpdateRequest(DashboardModel): + name: str | None = Field(default=None, min_length=1, max_length=200) + enabled: bool | None = None + include_paused_accounts: bool | None = Field(default=None, alias="includePausedAccounts") + schedule: AutomationScheduleRequest | None = None + model: str | None = Field(default=None, min_length=1) + reasoning_effort: str | None = Field(default=None, pattern=r"(?i)^(minimal|low|medium|high|xhigh)$") + prompt: str | None = Field(default=None, max_length=1000) + account_ids: list[str] | None = Field(default=None, max_length=128, alias="accountIds") + + @field_validator("account_ids") + @classmethod + def _validate_unique_account_ids(cls, value: list[str] | None) -> list[str] | None: + if value is None: + return None + normalized = [account_id.strip() for account_id in value if account_id.strip()] + if len(set(normalized)) != len(normalized): + raise ValueError("Duplicate account IDs are not allowed") + return normalized + + +class AutomationRunResponse(DashboardModel): + id: str + job_id: str + job_name: str | None = None + model: str | None = None + reasoning_effort: str | None = None + trigger: Literal["scheduled", "manual"] + status: Literal["running", "success", "failed", "partial"] + scheduled_for: datetime + started_at: datetime + finished_at: datetime | None = None + account_id: str | None = None + error_code: str | None = None + error_message: str | None = None + attempt_count: int = Field(ge=0) + effective_status: Literal["running", "success", "failed", "partial"] | None = None + total_accounts: int | None = Field(default=None, ge=0) + completed_accounts: int | None = Field(default=None, ge=0) + pending_accounts: int | None = Field(default=None, ge=0) + cycle_key: str | None = None + + +class AutomationJobResponse(DashboardModel): + id: str + name: str + enabled: bool + include_paused_accounts: bool = False + schedule: AutomationScheduleResponse + model: str + reasoning_effort: str | None = None + prompt: str + account_ids: list[str] + next_run_at: datetime | None = None + last_run: AutomationRunResponse | None = None + + +class AutomationJobsListResponse(DashboardModel): + items: list[AutomationJobResponse] = Field(default_factory=list) + total: int = 0 + has_more: bool = False + + +class AutomationRunsListResponse(DashboardModel): + items: list[AutomationRunResponse] = Field(default_factory=list) + total: int = 0 + has_more: bool = False + + +class AutomationJobFilterOptionsResponse(DashboardModel): + account_ids: list[str] = Field(default_factory=list) + models: list[str] = Field(default_factory=list) + statuses: list[str] = Field(default_factory=list) + schedule_types: list[str] = Field(default_factory=list) + + +class AutomationRunFilterOptionsResponse(DashboardModel): + account_ids: list[str] = Field(default_factory=list) + models: list[str] = Field(default_factory=list) + statuses: list[str] = Field(default_factory=list) + triggers: list[str] = Field(default_factory=list) + + +class AutomationRunAccountStateResponse(DashboardModel): + account_id: str + status: Literal["pending", "running", "success", "failed", "partial"] + run_id: str | None = None + scheduled_for: datetime | None = None + started_at: datetime | None = None + finished_at: datetime | None = None + error_code: str | None = None + error_message: str | None = None + + +class AutomationRunDetailsResponse(DashboardModel): + run: AutomationRunResponse + accounts: list[AutomationRunAccountStateResponse] = Field(default_factory=list) + total_accounts: int = Field(ge=0) + completed_accounts: int = Field(ge=0) + pending_accounts: int = Field(ge=0) + + +class AutomationDeleteResponse(DashboardModel): + status: Literal["deleted"] diff --git a/app/modules/automations/service.py b/app/modules/automations/service.py new file mode 100644 index 0000000000..6236baf491 --- /dev/null +++ b/app/modules/automations/service.py @@ -0,0 +1,1894 @@ +from __future__ import annotations + +import logging +import os +import random +import time +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from hashlib import sha1 +from uuid import uuid4 +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +from app.core.auth.refresh import RefreshError +from app.core.balancer import PERMANENT_FAILURE_CODES +from app.core.clients.proxy import ProxyResponseError +from app.core.clients.proxy import compact_responses as core_compact_responses +from app.core.crypto import TokenEncryptor +from app.core.openai.model_registry import get_model_registry +from app.core.openai.requests import ResponsesCompactRequest, ResponsesReasoning +from app.core.utils.time import naive_utc_to_epoch, utcnow +from app.db.models import Account, AccountStatus +from app.modules.accounts.auth_manager import AuthManager +from app.modules.accounts.repository import AccountsRepository +from app.modules.automations.repository import AutomationJobRecord, AutomationRunRecord, AutomationsRepository +from app.modules.request_logs.repository import RequestLogsRepository + +AUTOMATION_SCHEDULE_DAILY = "daily" +AUTOMATION_RUN_TRIGGER_SCHEDULED = "scheduled" +AUTOMATION_RUN_TRIGGER_MANUAL = "manual" +AUTOMATION_RUN_STATUS_RUNNING = "running" +AUTOMATION_RUN_STATUS_SUCCESS = "success" +AUTOMATION_RUN_STATUS_FAILED = "failed" +AUTOMATION_RUN_STATUS_PARTIAL = "partial" +DEFAULT_AUTOMATION_PROMPT = "ping" +AUTOMATION_SERVER_DEFAULT_TIMEZONE = "server_default" +AUTOMATION_SERVER_DEFAULT_TIMEZONE_ALIASES = frozenset({"server_default", "server default", "default"}) +AUTOMATION_WEEKDAY_CODES = ("mon", "tue", "wed", "thu", "fri", "sat", "sun") +AUTOMATION_DEFAULT_WEEKDAYS = list(AUTOMATION_WEEKDAY_CODES) +AUTOMATION_DEFAULT_THRESHOLD_MINUTES = 0 +AUTOMATION_MAX_THRESHOLD_MINUTES = 240 +_WEEKDAY_INDEX_TO_CODE = {index: code for index, code in enumerate(AUTOMATION_WEEKDAY_CODES)} + +_RETRYABLE_ACCOUNT_FAILURE_CODES = frozenset( + { + "rate_limit_exceeded", + "usage_limit_reached", + "insufficient_quota", + "usage_not_included", + "quota_exceeded", + "account_deactivated", + "invalid_api_key", + "authentication_error", + "upstream_unavailable", + "server_error", + "upstream_error", + *PERMANENT_FAILURE_CODES.keys(), + } +) +_AUTOMATION_ALWAYS_SKIPPED_ACCOUNT_STATUSES = frozenset( + { + AccountStatus.DEACTIVATED, + AccountStatus.RATE_LIMITED, + AccountStatus.QUOTA_EXCEEDED, + } +) + +logger = logging.getLogger(__name__) + + +class AutomationValidationError(ValueError): + def __init__(self, message: str, *, code: str) -> None: + super().__init__(message) + self.code = code + + +class AutomationNotFoundError(LookupError): + pass + + +@dataclass(frozen=True, slots=True) +class AutomationScheduleData: + type: str + time: str + timezone: str + days: list[str] + threshold_minutes: int + + +@dataclass(frozen=True, slots=True) +class AutomationRunData: + id: str + job_id: str + job_name: str | None + model: str | None + reasoning_effort: str | None + trigger: str + status: str + scheduled_for: datetime + started_at: datetime + finished_at: datetime | None + account_id: str | None + error_code: str | None + error_message: str | None + attempt_count: int + effective_status: str | None = None + total_accounts: int | None = None + completed_accounts: int | None = None + pending_accounts: int | None = None + cycle_key: str | None = None + + +@dataclass(frozen=True, slots=True) +class AutomationJobData: + id: str + name: str + enabled: bool + include_paused_accounts: bool + schedule: AutomationScheduleData + model: str + reasoning_effort: str | None + prompt: str + account_ids: list[str] + next_run_at: datetime | None + last_run: AutomationRunData | None + + +@dataclass(frozen=True, slots=True) +class AutomationJobsPageData: + items: list[AutomationJobData] + total: int + has_more: bool + + +@dataclass(frozen=True, slots=True) +class AutomationRunsPageData: + items: list[AutomationRunData] + total: int + has_more: bool + + +@dataclass(frozen=True, slots=True) +class AutomationJobFilterOptionsData: + account_ids: list[str] + models: list[str] + statuses: list[str] + schedule_types: list[str] + + +@dataclass(frozen=True, slots=True) +class AutomationRunFilterOptionsData: + account_ids: list[str] + models: list[str] + statuses: list[str] + triggers: list[str] + + +@dataclass(frozen=True, slots=True) +class AutomationRunAccountStateData: + account_id: str + status: str + run_id: str | None + scheduled_for: datetime | None + started_at: datetime | None + finished_at: datetime | None + error_code: str | None + error_message: str | None + + +@dataclass(frozen=True, slots=True) +class AutomationRunDetailsData: + run: AutomationRunData + accounts: list[AutomationRunAccountStateData] + total_accounts: int + completed_accounts: int + pending_accounts: int + + +@dataclass(frozen=True, slots=True) +class _AutomationRunCycleSummary: + cycle_key: str + effective_status: str + total_accounts: int + completed_accounts: int + pending_accounts: int + accounts: list[AutomationRunAccountStateData] + + +@dataclass(frozen=True, slots=True) +class AutomationJobCreateInput: + name: str + enabled: bool + include_paused_accounts: bool + schedule_type: str + schedule_time: str + schedule_timezone: str + schedule_days: list[str] | None + schedule_threshold_minutes: int | None + model: str + reasoning_effort: str | None + prompt: str | None + account_ids: list[str] + + +@dataclass(frozen=True, slots=True) +class AutomationJobUpdateInput: + name: str | None = None + enabled: bool | None = None + include_paused_accounts: bool | None = None + schedule_type: str | None = None + schedule_time: str | None = None + schedule_timezone: str | None = None + schedule_days: list[str] | None = None + schedule_threshold_minutes: int | None = None + model: str | None = None + reasoning_effort: str | None = None + reasoning_effort_set: bool = False + prompt: str | None = None + account_ids: list[str] | None = None + + +def parse_schedule_time_hhmm(value: str) -> tuple[int, int]: + parts = value.split(":", maxsplit=1) + if len(parts) != 2: + raise AutomationValidationError("Schedule time must use HH:MM format", code="invalid_schedule_time") + hour_raw, minute_raw = parts + if len(hour_raw) != 2 or len(minute_raw) != 2: + raise AutomationValidationError("Schedule time must use HH:MM format", code="invalid_schedule_time") + if not (hour_raw.isdigit() and minute_raw.isdigit()): + raise AutomationValidationError("Schedule time must use HH:MM format", code="invalid_schedule_time") + hour = int(hour_raw) + minute = int(minute_raw) + if hour < 0 or hour > 23 or minute < 0 or minute > 59: + raise AutomationValidationError("Schedule time is out of range", code="invalid_schedule_time") + return hour, minute + + +def normalize_schedule_time(value: str) -> str: + hour, minute = parse_schedule_time_hhmm(value.strip()) + return f"{hour:02d}:{minute:02d}" + + +def validate_timezone(value: str) -> str: + normalized = value.strip() + if not normalized: + raise AutomationValidationError("Schedule timezone is required", code="invalid_schedule_timezone") + if normalized.lower() in AUTOMATION_SERVER_DEFAULT_TIMEZONE_ALIASES: + return AUTOMATION_SERVER_DEFAULT_TIMEZONE + try: + ZoneInfo(normalized) + except ZoneInfoNotFoundError as exc: + raise AutomationValidationError( + f"Unknown timezone: {normalized}", + code="invalid_schedule_timezone", + ) from exc + return normalized + + +def normalize_schedule_days(value: list[str] | None) -> list[str]: + if value is None: + return list(AUTOMATION_DEFAULT_WEEKDAYS) + + normalized: list[str] = [] + for day in value: + day_code = day.strip().lower() + if not day_code: + continue + if day_code not in AUTOMATION_WEEKDAY_CODES: + raise AutomationValidationError( + f"Unsupported schedule day: {day}", + code="invalid_schedule_days", + ) + if day_code not in normalized: + normalized.append(day_code) + + if not normalized: + raise AutomationValidationError( + "At least one schedule day is required", + code="invalid_schedule_days", + ) + return normalized + + +def normalize_schedule_threshold_minutes(value: int | None) -> int: + if value is None: + return AUTOMATION_DEFAULT_THRESHOLD_MINUTES + if value < 0: + raise AutomationValidationError( + "Schedule threshold must be greater than or equal to 0", + code="invalid_schedule_threshold", + ) + if value > AUTOMATION_MAX_THRESHOLD_MINUTES: + raise AutomationValidationError( + f"Schedule threshold cannot exceed {AUTOMATION_MAX_THRESHOLD_MINUTES} minutes", + code="invalid_schedule_threshold", + ) + return value + + +def compute_latest_due_slot_utc( + now_utc: datetime, + *, + schedule_time: str, + timezone_name: str, + schedule_days: list[str], +) -> datetime: + normalized_now = _normalize_now_utc(now_utc) + hour, minute = parse_schedule_time_hhmm(schedule_time) + timezone = ZoneInfo(resolve_schedule_timezone_name(timezone_name)) + allowed_days = set(normalize_schedule_days(schedule_days)) + + local_now = normalized_now.replace(tzinfo=UTC).astimezone(timezone) + local_candidate = datetime( + local_now.year, + local_now.month, + local_now.day, + hour, + minute, + tzinfo=timezone, + ) + if local_candidate > local_now: + local_candidate -= timedelta(days=1) + + for _ in range(8): + if _local_weekday_code(local_candidate) in allowed_days: + return local_candidate.astimezone(UTC).replace(tzinfo=None) + local_candidate -= timedelta(days=1) + + raise RuntimeError("Failed to resolve latest due slot for automation schedule") + + +def compute_due_slot_for_current_local_day_utc( + now_utc: datetime, + *, + schedule_time: str, + timezone_name: str, + schedule_days: list[str], +) -> datetime | None: + normalized_now = _normalize_now_utc(now_utc) + hour, minute = parse_schedule_time_hhmm(schedule_time) + timezone = ZoneInfo(resolve_schedule_timezone_name(timezone_name)) + allowed_days = set(normalize_schedule_days(schedule_days)) + + local_now = normalized_now.replace(tzinfo=UTC).astimezone(timezone) + local_due = datetime( + local_now.year, + local_now.month, + local_now.day, + hour, + minute, + tzinfo=timezone, + ) + if _local_weekday_code(local_due) not in allowed_days: + return None + if local_due > local_now: + return None + return local_due.astimezone(UTC).replace(tzinfo=None) + + +def compute_next_run_utc( + now_utc: datetime, + *, + schedule_time: str, + timezone_name: str, + schedule_days: list[str], +) -> datetime: + normalized_now = _normalize_now_utc(now_utc) + hour, minute = parse_schedule_time_hhmm(schedule_time) + timezone = ZoneInfo(resolve_schedule_timezone_name(timezone_name)) + allowed_days = set(normalize_schedule_days(schedule_days)) + + local_now = normalized_now.replace(tzinfo=UTC).astimezone(timezone) + local_candidate = datetime( + local_now.year, + local_now.month, + local_now.day, + hour, + minute, + tzinfo=timezone, + ) + if local_candidate <= local_now: + local_candidate += timedelta(days=1) + + for _ in range(8): + if _local_weekday_code(local_candidate) in allowed_days: + return local_candidate.astimezone(UTC).replace(tzinfo=None) + local_candidate += timedelta(days=1) + + raise RuntimeError("Failed to resolve next run for automation schedule") + + +def _normalize_now_utc(value: datetime) -> datetime: + if value.tzinfo is None: + return value + return value.astimezone(UTC).replace(tzinfo=None) + + +def _local_weekday_code(value: datetime) -> str: + return _WEEKDAY_INDEX_TO_CODE[value.weekday()] + + +def resolve_schedule_timezone_name(value: str) -> str: + normalized = value.strip() + if normalized.lower() in AUTOMATION_SERVER_DEFAULT_TIMEZONE_ALIASES: + return _resolve_server_timezone_name() + return normalized + + +def _resolve_server_timezone_name() -> str: + candidates: list[str] = [] + + env_timezone = os.getenv("TZ", "").strip() + if env_timezone: + candidates.append(env_timezone) + + local_timezone = datetime.now().astimezone().tzinfo + local_key = getattr(local_timezone, "key", None) + if isinstance(local_key, str) and local_key.strip(): + candidates.append(local_key.strip()) + + if local_timezone is not None: + local_label = str(local_timezone).strip() + if local_label: + candidates.append(local_label) + + candidates.append("UTC") + + for candidate in candidates: + try: + ZoneInfo(candidate) + return candidate + except ZoneInfoNotFoundError: + continue + + return "UTC" + + +class AutomationsService: + def __init__( + self, + repository: AutomationsRepository, + accounts_repository: AccountsRepository, + request_logs_repository: RequestLogsRepository | None = None, + ) -> None: + self._repository = repository + self._accounts_repository = accounts_repository + self._request_logs_repository = request_logs_repository + self._auth_manager = AuthManager(accounts_repository) + self._encryptor = TokenEncryptor() + + async def list_jobs(self, *, now_utc: datetime | None = None) -> list[AutomationJobData]: + now = now_utc or utcnow() + jobs = await self._repository.list_jobs() + latest_runs = await self._repository.get_latest_runs_by_job_ids([job.id for job in jobs]) + latest_run_data = await self._enrich_runs_with_progress(list(latest_runs.values())) + latest_run_data_by_job_id = {run.job_id: run for run in latest_run_data} + return [self._to_job_data(job, latest_run_data_by_job_id.get(job.id), now_utc=now) for job in jobs] + + async def list_jobs_page( + self, + *, + limit: int, + offset: int, + search: str | None = None, + account_ids: list[str] | None = None, + models: list[str] | None = None, + statuses: list[str] | None = None, + schedule_types: list[str] | None = None, + now_utc: datetime | None = None, + ) -> AutomationJobsPageData: + now = now_utc or utcnow() + jobs, total = await self._repository.list_jobs_page( + limit=limit, + offset=offset, + search=search, + account_ids=account_ids, + models=models, + statuses=_normalize_job_status_filters(statuses), + schedule_types=schedule_types, + ) + latest_runs = await self._repository.get_latest_runs_by_job_ids([job.id for job in jobs]) + latest_run_data = await self._enrich_runs_with_progress(list(latest_runs.values())) + latest_run_data_by_job_id = {run.job_id: run for run in latest_run_data} + items = [self._to_job_data(job, latest_run_data_by_job_id.get(job.id), now_utc=now) for job in jobs] + return AutomationJobsPageData(items=items, total=total, has_more=offset + limit < total) + + async def list_job_filter_options( + self, + *, + search: str | None = None, + account_ids: list[str] | None = None, + models: list[str] | None = None, + statuses: list[str] | None = None, + schedule_types: list[str] | None = None, + ) -> AutomationJobFilterOptionsData: + options = await self._repository.list_job_filter_options( + search=search, + account_ids=account_ids, + models=models, + statuses=_normalize_job_status_filters(statuses), + schedule_types=schedule_types, + ) + accounts = await self._accounts_repository.list_accounts() + available_account_ids = sorted( + { + *options.account_ids, + *(account.id for account in accounts), + } + ) + return AutomationJobFilterOptionsData( + account_ids=available_account_ids, + models=options.models, + statuses=options.statuses, + schedule_types=options.schedule_types, + ) + + async def create_job( + self, payload: AutomationJobCreateInput, *, now_utc: datetime | None = None + ) -> AutomationJobData: + normalized = await self._normalize_create_input(payload) + prompt = normalized.prompt if normalized.prompt is not None else DEFAULT_AUTOMATION_PROMPT + record = await self._repository.create_job( + name=normalized.name, + enabled=normalized.enabled, + include_paused_accounts=normalized.include_paused_accounts, + schedule_type=normalized.schedule_type, + schedule_time=normalized.schedule_time, + schedule_timezone=normalized.schedule_timezone, + schedule_days=normalized.schedule_days or list(AUTOMATION_DEFAULT_WEEKDAYS), + schedule_threshold_minutes=normalized.schedule_threshold_minutes + if normalized.schedule_threshold_minutes is not None + else AUTOMATION_DEFAULT_THRESHOLD_MINUTES, + model=normalized.model, + reasoning_effort=normalized.reasoning_effort, + prompt=prompt, + account_ids=normalized.account_ids, + ) + now = now_utc or utcnow() + return self._to_job_data(record, None, now_utc=now) + + async def update_job( + self, + job_id: str, + payload: AutomationJobUpdateInput, + *, + now_utc: datetime | None = None, + ) -> AutomationJobData: + existing = await self._repository.get_job(job_id) + if existing is None: + raise AutomationNotFoundError(job_id) + normalized = await self._normalize_update_input(payload, existing=existing) + updated = await self._repository.update_job( + job_id, + name=normalized.name, + enabled=normalized.enabled, + include_paused_accounts=normalized.include_paused_accounts, + schedule_type=normalized.schedule_type, + schedule_time=normalized.schedule_time, + schedule_timezone=normalized.schedule_timezone, + schedule_days=normalized.schedule_days, + schedule_threshold_minutes=normalized.schedule_threshold_minutes, + model=normalized.model, + reasoning_effort=normalized.reasoning_effort, + reasoning_effort_set=normalized.reasoning_effort_set, + prompt=normalized.prompt, + account_ids=normalized.account_ids, + ) + if updated is None: + raise AutomationNotFoundError(job_id) + latest_runs = await self._repository.get_latest_runs_by_job_ids([job_id]) + latest_run_data = await self._enrich_runs_with_progress(list(latest_runs.values())) + latest_run_data_by_job_id = {run.job_id: run for run in latest_run_data} + now = now_utc or utcnow() + return self._to_job_data(updated, latest_run_data_by_job_id.get(job_id), now_utc=now) + + async def delete_job(self, job_id: str) -> bool: + return await self._repository.delete_job(job_id) + + async def list_runs(self, job_id: str, *, limit: int = 20) -> list[AutomationRunData]: + job = await self._repository.get_job(job_id) + if job is None: + raise AutomationNotFoundError(job_id) + runs = await self._repository.list_runs(job_id, limit=limit) + return await self._enrich_runs_with_progress(runs) + + async def list_runs_page( + self, + *, + limit: int, + offset: int, + search: str | None = None, + account_ids: list[str] | None = None, + models: list[str] | None = None, + statuses: list[str] | None = None, + triggers: list[str] | None = None, + job_ids: list[str] | None = None, + ) -> AutomationRunsPageData: + runs, total = await self._repository.list_run_cycles_page( + limit=limit, + offset=offset, + now_utc=utcnow(), + search=search, + account_ids=account_ids, + models=models, + statuses=_normalize_run_status_filters(statuses), + triggers=_normalize_run_trigger_filters(triggers), + job_ids=job_ids, + ) + items = await self._enrich_runs_with_progress(runs) + return AutomationRunsPageData(items=items, total=total, has_more=offset + limit < total) + + async def list_run_filter_options( + self, + *, + search: str | None = None, + account_ids: list[str] | None = None, + models: list[str] | None = None, + statuses: list[str] | None = None, + triggers: list[str] | None = None, + job_ids: list[str] | None = None, + ) -> AutomationRunFilterOptionsData: + options = await self._repository.list_run_filter_options( + search=search, + account_ids=account_ids, + models=models, + statuses=None, + triggers=_normalize_run_trigger_filters(triggers), + job_ids=job_ids, + ) + accounts = await self._accounts_repository.list_accounts() + available_account_ids = sorted( + { + *options.account_ids, + *(account.id for account in accounts), + } + ) + return AutomationRunFilterOptionsData( + account_ids=available_account_ids, + models=options.models, + statuses=_all_run_statuses(), + triggers=_all_run_triggers(), + ) + + async def get_run_details(self, run_id: str) -> AutomationRunDetailsData: + run = await self._repository.get_run(run_id) + if run is None: + raise AutomationNotFoundError(run_id) + jobs_by_id = await self._repository.get_jobs_by_ids([run.job_id]) + job = jobs_by_id.get(run.job_id) + if job is None: + raise AutomationNotFoundError(run.job_id) + summary = await self._build_cycle_summary_for_run( + run=run, + job=job, + now_utc=utcnow(), + ) + return AutomationRunDetailsData( + run=self._to_run_data(run, summary=summary), + accounts=summary.accounts, + total_accounts=summary.total_accounts, + completed_accounts=summary.completed_accounts, + pending_accounts=summary.pending_accounts, + ) + + async def run_now(self, job_id: str, *, now_utc: datetime | None = None) -> AutomationRunData: + job = await self._repository.get_job(job_id) + if job is None: + raise AutomationNotFoundError(job_id) + now = now_utc or utcnow() + cycle_id = uuid4().hex + cycle_key = _manual_cycle_key(job.id, cycle_id) + account_ids = await self._resolve_job_account_ids_for_dispatch(job) + if not account_ids: + slot_key = f"manual:{job.id}:{cycle_id}:none" + claim = await self._repository.claim_run( + job_id=job.id, + trigger=AUTOMATION_RUN_TRIGGER_MANUAL, + slot_key=slot_key, + cycle_key=cycle_key, + cycle_expected_accounts=0, + cycle_window_end=now, + scheduled_for=now, + started_at=now, + ) + if claim is None: + raise RuntimeError("Failed to claim manual automation run") + return await self._execute_claimed_run(job, claim) + + threshold = max(0, job.schedule_threshold_minutes) + dispatch_plan = _build_dispatch_plan( + job_id=job.id, + due_slot=now, + account_ids=account_ids, + threshold_minutes=threshold, + ) + expected_accounts_count = len(dispatch_plan) + cycle_window_end = now + timedelta(minutes=threshold) + + representative_claim: AutomationRunRecord | None = None + for account_id, scheduled_for in dispatch_plan: + slot_key = _manual_slot_key(job.id, cycle_id, account_id) + claim = await self._repository.claim_run( + job_id=job.id, + trigger=AUTOMATION_RUN_TRIGGER_MANUAL, + slot_key=slot_key, + cycle_key=cycle_key, + cycle_expected_accounts=expected_accounts_count, + cycle_window_end=cycle_window_end, + scheduled_for=scheduled_for, + started_at=now, + account_id=account_id, + ) + if claim is None: + continue + representative_claim = claim + if representative_claim is not None: + representative_run = await self._repository.get_run(representative_claim.id) + if representative_run is None: + representative_run = representative_claim + summary = await self._build_cycle_summary_for_run( + run=representative_run, + job=job, + now_utc=now, + ) + return self._to_run_data(representative_run, summary=summary) + raise RuntimeError("Failed to claim manual automation run") + + async def run_due_jobs(self, *, now_utc: datetime | None = None) -> int: + now = now_utc or utcnow() + executed = await self._run_due_manual_runs(now_utc=now) + jobs = await self._repository.list_enabled_jobs() + for job in jobs: + due_slot = compute_due_slot_for_current_local_day_utc( + now, + schedule_time=job.schedule_time, + timezone_name=job.schedule_timezone, + schedule_days=job.schedule_days, + ) + if due_slot is None: + continue + dispatch_plan = await self._build_scheduled_dispatch_plan(job, due_slot) + cycle_expected_accounts = len(dispatch_plan) + cycle_window_end = due_slot + timedelta(minutes=max(0, job.schedule_threshold_minutes)) + for account_id, scheduled_for in dispatch_plan: + if scheduled_for > now: + continue + slot_key = _scheduled_slot_key( + job.id, + scheduled_for, + account_id=account_id, + due_slot=due_slot, + ) + claim = await self._repository.claim_run( + job_id=job.id, + trigger=AUTOMATION_RUN_TRIGGER_SCHEDULED, + slot_key=slot_key, + cycle_key=_scheduled_cycle_key(job.id, due_slot), + cycle_expected_accounts=cycle_expected_accounts, + cycle_window_end=cycle_window_end, + scheduled_for=scheduled_for, + started_at=now, + account_id=account_id, + ) + if claim is None: + continue + await self._execute_claimed_run(job, claim, forced_account_id=account_id) + executed += 1 + return executed + + async def _run_due_manual_runs(self, *, now_utc: datetime) -> int: + due_runs = await self._repository.list_due_manual_runs(now_utc=now_utc) + if not due_runs: + return 0 + jobs_by_id = await self._repository.get_jobs_by_ids([run.job_id for run in due_runs]) + executed = 0 + for run in due_runs: + job = jobs_by_id.get(run.job_id) + if job is None or run.account_id is None: + continue + await self._execute_claimed_run(job, run, forced_account_id=run.account_id) + executed += 1 + return executed + + async def _execute_claimed_run( + self, + job: AutomationJobRecord, + run: AutomationRunRecord, + *, + forced_account_id: str | None = None, + ) -> AutomationRunData: + had_prior_failures = False + attempt_count = 0 + last_error_code: str | None = "no_available_accounts" + last_error_message: str | None = "No available accounts configured for automation job" + last_attempted_account_id: str | None = forced_account_id + cached_accounts_by_id: dict[str, Account] | None = None + if forced_account_id is not None: + account_ids_to_try = [forced_account_id] + else: + account_ids_to_try = list(job.account_ids) + if not account_ids_to_try: + accounts = await self._accounts_repository.list_accounts() + account_ids_to_try = [ + account.id + for account in accounts + if self._is_account_eligible_for_automation( + account, + include_paused_accounts=job.include_paused_accounts, + ) + ] + cached_accounts_by_id = {account.id: account for account in accounts} + + for account_id in account_ids_to_try: + request_started_at: float | None = None + if cached_accounts_by_id is None: + account = await self._accounts_repository.get_by_id(account_id) + else: + account = cached_accounts_by_id.get(account_id) + if account is None: + had_prior_failures = True + last_error_code = "account_not_found" + last_error_message = f"Account '{account_id}' not found" + continue + if not self._is_account_eligible_for_automation( + account, + include_paused_accounts=job.include_paused_accounts, + ): + continue + + try: + attempt_count += 1 + last_attempted_account_id = account_id + account = await self._auth_manager.ensure_fresh(account) + access_token = self._encryptor.decrypt(account.access_token_encrypted) + ping_request = ResponsesCompactRequest( + model=job.model, + input=job.prompt, + instructions="Automation ping", + reasoning=ResponsesReasoning(effort=job.reasoning_effort) if job.reasoning_effort else None, + ) + request_started_at = time.monotonic() + compact_response = await core_compact_responses( + ping_request, + headers={}, + access_token=access_token, + account_id=account.chatgpt_account_id, + ) + latency_ms = _elapsed_ms(request_started_at) + request_id = _automation_request_id(getattr(compact_response, "id", None), run.id, attempt_count) + ( + input_tokens, + output_tokens, + cached_input_tokens, + reasoning_tokens, + service_tier, + ) = _extract_compact_usage_fields(compact_response) + await self._write_request_log( + account_id=account.id, + request_id=request_id, + model=job.model, + reasoning_effort=job.reasoning_effort, + latency_ms=latency_ms, + status="success", + input_tokens=input_tokens, + output_tokens=output_tokens, + cached_input_tokens=cached_input_tokens, + reasoning_tokens=reasoning_tokens, + service_tier=service_tier, + ) + status = AUTOMATION_RUN_STATUS_PARTIAL if had_prior_failures else AUTOMATION_RUN_STATUS_SUCCESS + completed = await self._repository.complete_run( + run.id, + status=status, + finished_at=utcnow(), + account_id=account.id, + error_code=None, + error_message=None, + attempt_count=attempt_count, + ) + return self._to_run_data(completed) + except RefreshError as exc: + had_prior_failures = True + last_error_code = exc.code or "authentication_error" + last_error_message = exc.message + if forced_account_id is None and self._is_retryable_account_failure(last_error_code): + continue + break + except ProxyResponseError as exc: + had_prior_failures = True + error_code, error_message = _extract_proxy_error(exc) + last_error_code = error_code + last_error_message = error_message + await self._write_request_log( + account_id=account.id, + request_id=_automation_request_id(None, run.id, attempt_count), + model=job.model, + reasoning_effort=job.reasoning_effort, + latency_ms=_elapsed_ms(request_started_at), + status="error", + error_code=error_code, + error_message=error_message, + ) + if forced_account_id is None and self._is_retryable_account_failure(error_code): + continue + break + except Exception as exc: + had_prior_failures = True + last_error_code = "automation_ping_failed" + last_error_message = str(exc) or "Automation ping failed" + if request_started_at is not None: + await self._write_request_log( + account_id=account.id, + request_id=_automation_request_id(None, run.id, attempt_count), + model=job.model, + reasoning_effort=job.reasoning_effort, + latency_ms=_elapsed_ms(request_started_at), + status="error", + error_code=last_error_code, + error_message=last_error_message, + ) + continue + + completed = await self._repository.complete_run( + run.id, + status=AUTOMATION_RUN_STATUS_FAILED, + finished_at=utcnow(), + account_id=last_attempted_account_id, + error_code=last_error_code, + error_message=last_error_message, + attempt_count=attempt_count, + ) + return self._to_run_data(completed) + + async def _enrich_runs_with_progress(self, runs: list[AutomationRunRecord]) -> list[AutomationRunData]: + if not runs: + return [] + jobs_by_id = await self._repository.get_jobs_by_ids([run.job_id for run in runs]) + cycle_cache: dict[str, _AutomationRunCycleSummary] = {} + items: list[AutomationRunData] = [] + now = utcnow() + for run in runs: + job = jobs_by_id.get(run.job_id) + summary = None + if job is not None: + summary = await self._build_cycle_summary_for_run( + run=run, + job=job, + now_utc=now, + cycle_cache=cycle_cache, + ) + items.append(self._to_run_data(run, summary=summary)) + return items + + async def _build_cycle_summary_for_run( + self, + *, + run: AutomationRunRecord, + job: AutomationJobRecord, + now_utc: datetime, + cycle_cache: dict[str, _AutomationRunCycleSummary] | None = None, + ) -> _AutomationRunCycleSummary: + if run.trigger == AUTOMATION_RUN_TRIGGER_SCHEDULED: + return await self._build_scheduled_cycle_summary( + run=run, + job=job, + now_utc=now_utc, + cycle_cache=cycle_cache, + ) + return await self._build_manual_cycle_summary( + run=run, + job=job, + now_utc=now_utc, + cycle_cache=cycle_cache, + ) + + async def _build_manual_cycle_summary( + self, + *, + run: AutomationRunRecord, + job: AutomationJobRecord, + now_utc: datetime, + cycle_cache: dict[str, _AutomationRunCycleSummary] | None = None, + ) -> _AutomationRunCycleSummary: + parsed = _parse_manual_cycle_key(run.slot_key) + cycle_key = run.cycle_key.strip() if run.cycle_key and run.cycle_key.strip() else None + if cycle_key is None and parsed is not None: + cycle_id, _slot_key_prefix = parsed + cycle_key = _manual_cycle_key(job.id, cycle_id) + if cycle_key is None: + return _AutomationRunCycleSummary( + cycle_key=f"manual:{run.id}", + effective_status=run.status, + total_accounts=1 if run.account_id else 0, + completed_accounts=1 if run.account_id else 0, + pending_accounts=0, + accounts=[ + AutomationRunAccountStateData( + account_id=run.account_id, + status=run.status, + run_id=run.id, + scheduled_for=run.scheduled_for, + started_at=run.started_at, + finished_at=run.finished_at, + error_code=run.error_code, + error_message=run.error_message, + ) + ] + if run.account_id + else [], + ) + + if cycle_cache is not None and cycle_key in cycle_cache: + return cycle_cache[cycle_key] + + cycle_runs = await self._repository.list_runs_for_cycle_key(cycle_key=cycle_key) + latest_run_by_account_id: dict[str, AutomationRunRecord] = {} + for cycle_run in cycle_runs: + if not cycle_run.account_id: + continue + if cycle_run.account_id in latest_run_by_account_id: + continue + latest_run_by_account_id[cycle_run.account_id] = cycle_run + + expected_account_ids: list[str] = [] + seen_account_ids: set[str] = set() + for cycle_run in sorted( + (entry for entry in cycle_runs if entry.account_id), + key=lambda entry: (entry.scheduled_for, entry.account_id or "", entry.id), + ): + if cycle_run.account_id in seen_account_ids: + continue + seen_account_ids.add(cycle_run.account_id) + expected_account_ids.append(cycle_run.account_id) + + if not expected_account_ids and run.account_id: + expected_account_ids = [run.account_id] + scheduled_for_by_account_id = { + account_id: entry.scheduled_for + for account_id, entry in latest_run_by_account_id.items() + } + expected_set = set(expected_account_ids) + observed_only = [ + account_id + for account_id in latest_run_by_account_id + if account_id not in expected_set + ] + all_account_ids = [*expected_account_ids, *observed_only] + + account_states: list[AutomationRunAccountStateData] = [] + for account_id in all_account_ids: + observed_run = latest_run_by_account_id.get(account_id) + if observed_run is None: + account_states.append( + AutomationRunAccountStateData( + account_id=account_id, + status="pending", + run_id=None, + scheduled_for=scheduled_for_by_account_id.get(account_id), + started_at=None, + finished_at=None, + error_code=None, + error_message=None, + ) + ) + continue + account_status = observed_run.status + if observed_run.status == AUTOMATION_RUN_STATUS_RUNNING and observed_run.scheduled_for > now_utc: + account_status = "pending" + account_states.append( + AutomationRunAccountStateData( + account_id=account_id, + status=account_status, + run_id=observed_run.id, + scheduled_for=observed_run.scheduled_for, + started_at=observed_run.started_at, + finished_at=observed_run.finished_at, + error_code=observed_run.error_code, + error_message=observed_run.error_message, + ) + ) + + expected_accounts_hint = max( + [entry.cycle_expected_accounts or 0 for entry in cycle_runs], + default=run.cycle_expected_accounts or 0, + ) + total_accounts = max(len(all_account_ids), expected_accounts_hint) + completed_accounts = sum( + 1 + for account_id in all_account_ids + if (entry := latest_run_by_account_id.get(account_id)) is not None + and entry.status != AUTOMATION_RUN_STATUS_RUNNING + ) + pending_accounts = max(0, total_accounts - completed_accounts) + status_counts = { + "success": 0, + "failed": 0, + "partial": 0, + "running": 0, + } + for entry in latest_run_by_account_id.values(): + if entry.status in status_counts: + status_counts[entry.status] += 1 + effective_status = _resolve_effective_status( + pending_accounts=pending_accounts, + completed_accounts=completed_accounts, + success_count=status_counts["success"], + failed_count=status_counts["failed"], + partial_count=status_counts["partial"], + running_count=status_counts["running"], + fallback_status=run.status, + now_utc=now_utc, + window_end_utc=run.cycle_window_end or run.scheduled_for, + ) + summary = _AutomationRunCycleSummary( + cycle_key=cycle_key, + effective_status=effective_status, + total_accounts=total_accounts, + completed_accounts=completed_accounts, + pending_accounts=pending_accounts, + accounts=account_states, + ) + if cycle_cache is not None: + cycle_cache[cycle_key] = summary + return summary + + async def _build_scheduled_cycle_summary( + self, + *, + run: AutomationRunRecord, + job: AutomationJobRecord, + now_utc: datetime, + cycle_cache: dict[str, _AutomationRunCycleSummary] | None = None, + ) -> _AutomationRunCycleSummary: + due_slot = compute_latest_due_slot_utc( + run.scheduled_for, + schedule_time=job.schedule_time, + timezone_name=job.schedule_timezone, + schedule_days=job.schedule_days, + ) + cycle_key = ( + run.cycle_key.strip() + if run.cycle_key and run.cycle_key.strip() + else _scheduled_cycle_key(job.id, due_slot) + ) + if cycle_cache is not None and cycle_key in cycle_cache: + return cycle_cache[cycle_key] + + threshold = max(0, job.schedule_threshold_minutes) + window_end = due_slot + timedelta(minutes=threshold) + cycle_runs = await self._repository.list_runs_for_cycle_window( + job_id=job.id, + trigger=AUTOMATION_RUN_TRIGGER_SCHEDULED, + scheduled_from=due_slot, + scheduled_to=window_end, + ) + + latest_run_by_account_id: dict[str, AutomationRunRecord] = {} + for cycle_run in cycle_runs: + if not cycle_run.account_id: + continue + if cycle_run.account_id in latest_run_by_account_id: + continue + latest_run_by_account_id[cycle_run.account_id] = cycle_run + + if now_utc <= window_end: + dispatch_plan = await self._build_scheduled_dispatch_plan(job, due_slot) + scheduled_for_by_account_id = { + account_id: scheduled_for for account_id, scheduled_for in dispatch_plan + } + expected_account_ids = list(scheduled_for_by_account_id.keys()) + else: + expected_account_ids = [] + scheduled_for_by_account_id = {} + for cycle_run in sorted( + (entry for entry in cycle_runs if entry.account_id), + key=lambda entry: (entry.scheduled_for, entry.account_id or "", entry.id), + ): + account_id = cycle_run.account_id + if account_id is None or account_id in scheduled_for_by_account_id: + continue + scheduled_for_by_account_id[account_id] = cycle_run.scheduled_for + expected_account_ids.append(account_id) + if not expected_account_ids and run.account_id: + expected_account_ids = [run.account_id] + scheduled_for_by_account_id[run.account_id] = run.scheduled_for + observed_account_ids = [ + account_id + for account_id in latest_run_by_account_id + if account_id not in expected_account_ids + ] + all_account_ids = [*expected_account_ids, *observed_account_ids] + + account_states: list[AutomationRunAccountStateData] = [] + for account_id in all_account_ids: + observed_run = latest_run_by_account_id.get(account_id) + if observed_run is None: + account_states.append( + AutomationRunAccountStateData( + account_id=account_id, + status="pending", + run_id=None, + scheduled_for=scheduled_for_by_account_id.get(account_id), + started_at=None, + finished_at=None, + error_code=None, + error_message=None, + ) + ) + continue + account_states.append( + AutomationRunAccountStateData( + account_id=account_id, + status=observed_run.status, + run_id=observed_run.id, + scheduled_for=observed_run.scheduled_for, + started_at=observed_run.started_at, + finished_at=observed_run.finished_at, + error_code=observed_run.error_code, + error_message=observed_run.error_message, + ) + ) + + expected_accounts_hint = max( + [entry.cycle_expected_accounts or 0 for entry in cycle_runs], + default=run.cycle_expected_accounts or 0, + ) + total_accounts = max(len(all_account_ids), expected_accounts_hint) + completed_accounts = len(latest_run_by_account_id) + pending_accounts = max(0, total_accounts - completed_accounts) + status_counts = { + "success": 0, + "failed": 0, + "partial": 0, + "running": 0, + } + for entry in latest_run_by_account_id.values(): + if entry.status in status_counts: + status_counts[entry.status] += 1 + effective_status = _resolve_effective_status( + pending_accounts=pending_accounts, + completed_accounts=completed_accounts, + success_count=status_counts["success"], + failed_count=status_counts["failed"], + partial_count=status_counts["partial"], + running_count=status_counts["running"], + fallback_status=run.status, + now_utc=now_utc, + window_end_utc=window_end, + ) + summary = _AutomationRunCycleSummary( + cycle_key=cycle_key, + effective_status=effective_status, + total_accounts=total_accounts, + completed_accounts=completed_accounts, + pending_accounts=pending_accounts, + accounts=account_states, + ) + if cycle_cache is not None: + cycle_cache[cycle_key] = summary + return summary + + async def _normalize_create_input(self, payload: AutomationJobCreateInput) -> AutomationJobCreateInput: + name = _normalize_non_empty(payload.name, field_label="name", code="invalid_name") + model = _normalize_non_empty(payload.model, field_label="model", code="invalid_model") + schedule_type = _normalize_schedule_type(payload.schedule_type) + schedule_time = normalize_schedule_time(payload.schedule_time) + schedule_timezone = validate_timezone(payload.schedule_timezone) + schedule_days = normalize_schedule_days(payload.schedule_days) + schedule_threshold_minutes = normalize_schedule_threshold_minutes(payload.schedule_threshold_minutes) + reasoning_effort = _normalize_reasoning_effort(payload.reasoning_effort, model_slug=model) + prompt = _normalize_prompt(payload.prompt) + account_ids = await self._normalize_account_ids( + payload.account_ids, + include_paused_accounts=payload.include_paused_accounts, + ) + return AutomationJobCreateInput( + name=name, + enabled=payload.enabled, + include_paused_accounts=payload.include_paused_accounts, + schedule_type=schedule_type, + schedule_time=schedule_time, + schedule_timezone=schedule_timezone, + schedule_days=schedule_days, + schedule_threshold_minutes=schedule_threshold_minutes, + model=model, + reasoning_effort=reasoning_effort, + prompt=prompt, + account_ids=account_ids, + ) + + async def _normalize_update_input( + self, + payload: AutomationJobUpdateInput, + *, + existing: AutomationJobRecord, + ) -> AutomationJobUpdateInput: + if payload.name is None: + name = None + else: + name = _normalize_non_empty(payload.name, field_label="name", code="invalid_name") + if payload.model is None: + model = None + else: + model = _normalize_non_empty(payload.model, field_label="model", code="invalid_model") + model_for_reasoning = model if model is not None else existing.model + if payload.reasoning_effort_set: + reasoning_effort = _normalize_reasoning_effort(payload.reasoning_effort, model_slug=model_for_reasoning) + else: + reasoning_effort = None + if payload.schedule_type is None: + schedule_type = None + else: + schedule_type = _normalize_schedule_type(payload.schedule_type) + if payload.schedule_time is None: + schedule_time = None + else: + schedule_time = normalize_schedule_time(payload.schedule_time) + if payload.schedule_timezone is None: + schedule_timezone = None + else: + schedule_timezone = validate_timezone(payload.schedule_timezone) + if payload.schedule_days is None: + schedule_days = None + else: + schedule_days = normalize_schedule_days(payload.schedule_days) + if payload.schedule_threshold_minutes is None: + schedule_threshold_minutes = None + else: + schedule_threshold_minutes = normalize_schedule_threshold_minutes(payload.schedule_threshold_minutes) + if payload.prompt is None: + prompt = None + else: + prompt = _normalize_prompt(payload.prompt) + if payload.include_paused_accounts is None: + include_paused_accounts = None + else: + include_paused_accounts = payload.include_paused_accounts + if payload.account_ids is None: + account_ids = None + else: + account_ids = await self._normalize_account_ids( + payload.account_ids, + include_paused_accounts=( + payload.include_paused_accounts + if payload.include_paused_accounts is not None + else existing.include_paused_accounts + ), + ) + + if ( + name is None + and payload.enabled is None + and include_paused_accounts is None + and schedule_type is None + and schedule_time is None + and schedule_timezone is None + and schedule_days is None + and schedule_threshold_minutes is None + and model is None + and not payload.reasoning_effort_set + and prompt is None + and account_ids is None + ): + return AutomationJobUpdateInput( + name=existing.name, + enabled=existing.enabled, + include_paused_accounts=existing.include_paused_accounts, + schedule_type=existing.schedule_type, + schedule_time=existing.schedule_time, + schedule_timezone=existing.schedule_timezone, + schedule_days=existing.schedule_days, + schedule_threshold_minutes=existing.schedule_threshold_minutes, + model=existing.model, + reasoning_effort=existing.reasoning_effort, + reasoning_effort_set=True, + prompt=existing.prompt, + account_ids=existing.account_ids, + ) + + return AutomationJobUpdateInput( + name=name, + enabled=payload.enabled, + include_paused_accounts=include_paused_accounts, + schedule_type=schedule_type, + schedule_time=schedule_time, + schedule_timezone=schedule_timezone, + schedule_days=schedule_days, + schedule_threshold_minutes=schedule_threshold_minutes, + model=model, + reasoning_effort=reasoning_effort, + reasoning_effort_set=payload.reasoning_effort_set, + prompt=prompt, + account_ids=account_ids, + ) + + async def _normalize_account_ids( + self, + account_ids: list[str], + *, + include_paused_accounts: bool = False, + ) -> list[str]: + normalized = [account_id.strip() for account_id in account_ids if account_id.strip()] + if not normalized: + accounts = await self._accounts_repository.list_accounts() + await self._reactivate_accounts_if_reset_elapsed(accounts, now_utc=utcnow()) + has_eligible_accounts = any( + self._is_account_eligible_for_automation( + account, + include_paused_accounts=include_paused_accounts, + ) + for account in accounts + ) + if not has_eligible_accounts: + raise AutomationValidationError( + "No eligible accounts available for this automation", + code="invalid_account_ids", + ) + return [] + deduped = list(dict.fromkeys(normalized)) + existing = await self._repository.list_existing_account_ids(deduped) + missing = [account_id for account_id in deduped if account_id not in existing] + if missing: + raise AutomationValidationError( + f"Unknown account IDs: {', '.join(missing)}", + code="invalid_account_ids", + ) + return deduped + + def _to_job_data( + self, + job: AutomationJobRecord, + last_run: AutomationRunData | None, + *, + now_utc: datetime, + ) -> AutomationJobData: + next_run_at = None + if job.enabled: + next_run_at = compute_next_run_utc( + now_utc, + schedule_time=job.schedule_time, + timezone_name=job.schedule_timezone, + schedule_days=job.schedule_days, + ) + return AutomationJobData( + id=job.id, + name=job.name, + enabled=job.enabled, + include_paused_accounts=job.include_paused_accounts, + schedule=AutomationScheduleData( + type=job.schedule_type, + time=job.schedule_time, + timezone=job.schedule_timezone, + days=job.schedule_days, + threshold_minutes=job.schedule_threshold_minutes, + ), + model=job.model, + reasoning_effort=job.reasoning_effort, + prompt=job.prompt, + account_ids=job.account_ids, + next_run_at=next_run_at, + last_run=last_run, + ) + + async def _build_scheduled_dispatch_plan( + self, + job: AutomationJobRecord, + due_slot: datetime, + ) -> list[tuple[str, datetime]]: + account_ids = await self._resolve_job_account_ids_for_dispatch(job) + if not account_ids: + return [] + return _build_dispatch_plan( + job_id=job.id, + due_slot=due_slot, + account_ids=account_ids, + threshold_minutes=job.schedule_threshold_minutes, + ) + + async def _resolve_job_account_ids_for_dispatch(self, job: AutomationJobRecord) -> list[str]: + now_utc = utcnow() + account_ids = list(job.account_ids) + accounts = await self._accounts_repository.list_accounts() + await self._reactivate_accounts_if_reset_elapsed(accounts, now_utc=now_utc) + accounts_by_id = {account.id: account for account in accounts} + candidate_accounts: list[Account] + if account_ids: + candidate_accounts = [ + accounts_by_id[account_id] + for account_id in account_ids + if account_id in accounts_by_id + ] + else: + candidate_accounts = accounts + return [ + account.id + for account in candidate_accounts + if self._is_account_eligible_for_automation( + account, + include_paused_accounts=job.include_paused_accounts, + ) + ] + + async def _reactivate_accounts_if_reset_elapsed( + self, + accounts: list[Account], + *, + now_utc: datetime, + ) -> None: + now_epoch = naive_utc_to_epoch(now_utc) + for account in accounts: + if account.status not in {AccountStatus.RATE_LIMITED, AccountStatus.QUOTA_EXCEEDED}: + continue + if account.reset_at is None or account.reset_at > now_epoch: + continue + updated = await self._accounts_repository.update_status_if_current( + account.id, + AccountStatus.ACTIVE, + None, + None, + blocked_at=None, + expected_status=account.status, + expected_deactivation_reason=account.deactivation_reason, + expected_reset_at=account.reset_at, + expected_blocked_at=account.blocked_at, + ) + if updated: + account.status = AccountStatus.ACTIVE + account.deactivation_reason = None + account.reset_at = None + account.blocked_at = None + + @staticmethod + def _to_run_data( + run: AutomationRunRecord, + *, + summary: _AutomationRunCycleSummary | None = None, + ) -> AutomationRunData: + scheduled_for = ( + run.started_at + if summary is not None and run.trigger == AUTOMATION_RUN_TRIGGER_MANUAL + else run.scheduled_for + ) + return AutomationRunData( + id=run.id, + job_id=run.job_id, + job_name=run.job_name, + model=run.model, + reasoning_effort=run.reasoning_effort, + trigger=run.trigger, + status=run.status, + scheduled_for=scheduled_for, + started_at=run.started_at, + finished_at=run.finished_at, + account_id=run.account_id, + error_code=run.error_code, + error_message=run.error_message, + attempt_count=run.attempt_count, + effective_status=summary.effective_status if summary is not None else None, + total_accounts=summary.total_accounts if summary is not None else None, + completed_accounts=summary.completed_accounts if summary is not None else None, + pending_accounts=summary.pending_accounts if summary is not None else None, + cycle_key=summary.cycle_key if summary is not None else None, + ) + + @staticmethod + def _is_retryable_account_failure(error_code: str | None) -> bool: + if not error_code: + return False + return error_code.lower() in _RETRYABLE_ACCOUNT_FAILURE_CODES + + @staticmethod + def _is_account_eligible_for_automation( + account: Account, + *, + include_paused_accounts: bool, + ) -> bool: + if account.status in _AUTOMATION_ALWAYS_SKIPPED_ACCOUNT_STATUSES: + return False + if account.status == AccountStatus.PAUSED and not include_paused_accounts: + return False + return True + + async def _write_request_log( + self, + *, + account_id: str | None, + request_id: str, + model: str, + reasoning_effort: str | None = None, + latency_ms: int | None, + status: str, + error_code: str | None = None, + error_message: str | None = None, + input_tokens: int | None = None, + output_tokens: int | None = None, + cached_input_tokens: int | None = None, + reasoning_tokens: int | None = None, + service_tier: str | None = None, + ) -> None: + if self._request_logs_repository is None: + return + try: + await self._request_logs_repository.add_log( + account_id=account_id, + request_id=request_id, + model=model, + input_tokens=input_tokens, + output_tokens=output_tokens, + cached_input_tokens=cached_input_tokens, + reasoning_tokens=reasoning_tokens, + reasoning_effort=reasoning_effort, + latency_ms=latency_ms, + status=status, + error_code=error_code, + error_message=error_message, + service_tier=service_tier, + transport="automation", + ) + except Exception: + logger.warning( + "Failed to persist automation request log account_id=%s request_id=%s", + account_id, + request_id, + exc_info=True, + ) + + +def _normalize_schedule_type(value: str) -> str: + normalized = value.strip().lower() + if normalized != AUTOMATION_SCHEDULE_DAILY: + raise AutomationValidationError( + f"Unsupported schedule type: {value}", + code="invalid_schedule_type", + ) + return normalized + + +def _normalize_non_empty(value: str, *, field_label: str, code: str) -> str: + normalized = value.strip() + if not normalized: + raise AutomationValidationError(f"{field_label} is required", code=code) + return normalized + + +def _normalize_prompt(value: str | None) -> str: + if value is None: + return DEFAULT_AUTOMATION_PROMPT + normalized = value.strip() + if not normalized: + raise AutomationValidationError("prompt cannot be empty", code="invalid_prompt") + return normalized + + +def _normalize_reasoning_effort(value: str | None, *, model_slug: str) -> str | None: + if value is None: + return None + normalized = value.strip().lower() + if not normalized: + return None + allowed = {"minimal", "low", "medium", "high", "xhigh"} + if normalized not in allowed: + raise AutomationValidationError( + f"Unsupported reasoning effort: {value}", + code="invalid_reasoning_effort", + ) + + registry = get_model_registry() + model = registry.get_models_with_fallback().get(model_slug) + if model is None: + return normalized + supported = {level.effort.strip().lower() for level in model.supported_reasoning_levels if level.effort.strip()} + if supported and normalized not in supported: + raise AutomationValidationError( + f"Reasoning effort '{normalized}' is not supported by model '{model_slug}'", + code="invalid_reasoning_effort", + ) + return normalized + + +def _build_dispatch_plan( + *, + job_id: str, + due_slot: datetime, + account_ids: list[str], + threshold_minutes: int, +) -> list[tuple[str, datetime]]: + deduped_account_ids = list(dict.fromkeys(account_ids)) + if not deduped_account_ids: + return [] + offsets = _pick_dispatch_offsets_seconds( + job_id=job_id, + due_slot=due_slot, + account_count=len(deduped_account_ids), + threshold_minutes=threshold_minutes, + ) + return [ + (account_id, due_slot + timedelta(seconds=offset_seconds)) + for account_id, offset_seconds in zip(deduped_account_ids, offsets, strict=False) + ] + + +def _pick_dispatch_offsets_seconds( + *, + job_id: str, + due_slot: datetime, + account_count: int, + threshold_minutes: int, +) -> list[int]: + if account_count <= 0: + return [] + if threshold_minutes <= 0: + return [0] * account_count + + max_offset_seconds = threshold_minutes * 60 + available_offsets = list(range(1, max_offset_seconds + 1)) + seed = f"{job_id}:{due_slot.isoformat()}" + rng = random.Random(seed) + + if account_count <= len(available_offsets): + return rng.sample(available_offsets, account_count) + + offsets = available_offsets.copy() + rng.shuffle(offsets) + for _ in range(account_count - len(available_offsets)): + offsets.append(rng.choice(available_offsets)) + rng.shuffle(offsets) + return offsets + + +def _scheduled_slot_key( + job_id: str, + scheduled_for: datetime, + *, + account_id: str | None = None, + due_slot: datetime | None = None, +) -> str: + if account_id is None: + return f"scheduled:{job_id}:{scheduled_for.isoformat()}Z" + seed = f"{job_id}:{(due_slot or scheduled_for).isoformat()}:{account_id}:{scheduled_for.isoformat()}" + digest = sha1(seed.encode("utf-8")).hexdigest()[:20] + return f"scheduled:{job_id}:{digest}" + + +def _scheduled_cycle_key(job_id: str, due_slot: datetime) -> str: + return f"scheduled:{job_id}:{due_slot.isoformat()}" + + +def _manual_cycle_key(job_id: str, cycle_id: str) -> str: + return f"manual:{job_id}:{cycle_id}" + + +def _manual_slot_key(job_id: str, cycle_id: str, account_id: str) -> str: + seed = f"{job_id}:{cycle_id}:{account_id}" + digest = sha1(seed.encode("utf-8")).hexdigest()[:20] + return f"{_manual_cycle_key(job_id, cycle_id)}:{digest}" + + +def _parse_manual_cycle_key(slot_key: str) -> tuple[str, str] | None: + parts = slot_key.split(":") + if len(parts) != 4: + return None + trigger, job_id, cycle_id, _digest = parts + if trigger != "manual" or not job_id or not cycle_id: + return None + return cycle_id, f"manual:{job_id}:{cycle_id}:" + + +def _resolve_effective_status( + *, + pending_accounts: int, + completed_accounts: int, + success_count: int, + failed_count: int, + partial_count: int, + running_count: int, + fallback_status: str, + now_utc: datetime, + window_end_utc: datetime, +) -> str: + if running_count > 0: + return AUTOMATION_RUN_STATUS_RUNNING + if pending_accounts > 0 and now_utc <= window_end_utc: + return AUTOMATION_RUN_STATUS_RUNNING + if pending_accounts > 0: + return AUTOMATION_RUN_STATUS_PARTIAL if completed_accounts > 0 else AUTOMATION_RUN_STATUS_FAILED + if success_count > 0 and failed_count == 0 and partial_count == 0: + return AUTOMATION_RUN_STATUS_SUCCESS + if success_count > 0 and (failed_count > 0 or partial_count > 0): + return AUTOMATION_RUN_STATUS_PARTIAL + if failed_count > 0 and success_count == 0 and partial_count == 0: + return AUTOMATION_RUN_STATUS_FAILED + if partial_count > 0: + return AUTOMATION_RUN_STATUS_PARTIAL + return fallback_status + + +def _extract_proxy_error(exc: ProxyResponseError) -> tuple[str, str]: + payload = exc.payload + if isinstance(payload, dict): + error = payload.get("error") + if isinstance(error, dict): + code_value = error.get("code") + message_value = error.get("message") + code = ( + code_value.strip().lower() + if isinstance(code_value, str) and code_value.strip() + else "upstream_error" + ) + if isinstance(message_value, str) and message_value.strip(): + return code, message_value.strip() + return code, "Upstream request failed" + return "upstream_error", "Upstream request failed" + + +def _normalize_job_status_filters(values: list[str] | None) -> list[str] | None: + if not values: + return None + normalized = sorted({value.strip().lower() for value in values if value and value.strip()}) + if not normalized or "all" in normalized: + return None + return normalized + + +def _normalize_run_status_filters(values: list[str] | None) -> list[str] | None: + if not values: + return None + normalized = sorted({value.strip().lower() for value in values if value and value.strip()}) + if not normalized or "all" in normalized: + return None + return [value for value in normalized if value in _all_run_statuses()] or None + + +def _normalize_run_trigger_filters(values: list[str] | None) -> list[str] | None: + if not values: + return None + normalized = sorted({value.strip().lower() for value in values if value and value.strip()}) + if not normalized or "all" in normalized: + return None + return [value for value in normalized if value in _all_run_triggers()] or None + + +def _all_run_statuses() -> list[str]: + return [ + AUTOMATION_RUN_STATUS_RUNNING, + AUTOMATION_RUN_STATUS_SUCCESS, + AUTOMATION_RUN_STATUS_FAILED, + AUTOMATION_RUN_STATUS_PARTIAL, + ] + + +def _all_run_triggers() -> list[str]: + return [AUTOMATION_RUN_TRIGGER_SCHEDULED, AUTOMATION_RUN_TRIGGER_MANUAL] + + +def _automation_request_id(response_id: str | None, run_id: str, attempt_count: int) -> str: + if isinstance(response_id, str): + normalized = response_id.strip() + if normalized: + return normalized + return f"automation-{run_id}-attempt-{attempt_count}" + + +def _elapsed_ms(started_at: float | None) -> int | None: + if started_at is None: + return None + return max(0, int((time.monotonic() - started_at) * 1000)) + + +def _extract_compact_usage_fields( + compact_response: object, +) -> tuple[int | None, int | None, int | None, int | None, str | None]: + usage = getattr(compact_response, "usage", None) + input_tokens = _coerce_int(getattr(usage, "input_tokens", None)) + output_tokens = _coerce_int(getattr(usage, "output_tokens", None)) + input_details = getattr(usage, "input_tokens_details", None) + output_details = getattr(usage, "output_tokens_details", None) + cached_input_tokens = _coerce_int(getattr(input_details, "cached_tokens", None)) + reasoning_tokens = _coerce_int(getattr(output_details, "reasoning_tokens", None)) + + service_tier: str | None = None + model_extra = getattr(compact_response, "model_extra", None) + if isinstance(model_extra, dict): + service_tier_raw = model_extra.get("service_tier") + if isinstance(service_tier_raw, str): + normalized = service_tier_raw.strip() + if normalized: + service_tier = normalized + + return input_tokens, output_tokens, cached_input_tokens, reasoning_tokens, service_tier + + +def _coerce_int(value: object) -> int | None: + if isinstance(value, int): + return value + return None diff --git a/app/modules/dashboard/api.py b/app/modules/dashboard/api.py index e3d07dba18..03da1f42db 100644 --- a/app/modules/dashboard/api.py +++ b/app/modules/dashboard/api.py @@ -39,8 +39,29 @@ async def list_models() -> dict: models_by_slug = registry.get_models_with_fallback() if not models_by_slug: return {"models": []} + allowed_efforts = {"minimal", "low", "medium", "high", "xhigh"} + + def _normalize_effort(value: str | None) -> str | None: + if not isinstance(value, str): + return None + normalized = value.strip().lower() + if normalized in allowed_efforts: + return normalized + return None + models = [ - {"id": slug, "name": model.display_name or slug} + { + "id": slug, + "name": model.display_name or slug, + "supportedReasoningEfforts": list( + dict.fromkeys( + effort + for effort in (_normalize_effort(level.effort) for level in model.supported_reasoning_levels) + if effort is not None + ) + ), + "defaultReasoningEffort": _normalize_effort(model.default_reasoning_level), + } for slug, model in models_by_slug.items() if is_public_model(model, None) ] diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index aee89a2b5b..ea46ac93ee 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -7,6 +7,7 @@ import { TooltipProvider } from "@/components/ui/tooltip"; import { AuthGate } from "@/features/auth/components/auth-gate"; import { useAuthStore } from "@/features/auth/hooks/use-auth"; import { AccountsPage } from "@/features/accounts/components/accounts-page"; +import { AutomationsPage } from "@/features/automations/components/automations-page"; import { ApisPage } from "@/features/apis/components/apis-page"; import { DashboardPage } from "@/features/dashboard/components/dashboard-page"; import { ReportsPage } from "@/features/reports/components/reports-page"; @@ -51,6 +52,7 @@ export default function App() { } /> } /> } /> + } /> } /> } /> } /> diff --git a/frontend/src/__integration__/automations-flow.test.tsx b/frontend/src/__integration__/automations-flow.test.tsx new file mode 100644 index 0000000000..cc45763ae9 --- /dev/null +++ b/frontend/src/__integration__/automations-flow.test.tsx @@ -0,0 +1,97 @@ +import { screen, waitFor, within } from "@testing-library/react"; +import userEvent from "@testing-library/user-event"; +import { beforeEach, describe, expect, it } from "vitest"; + +import App from "@/App"; +import { renderWithProviders } from "@/test/utils"; + +function getJobRow(jobName: string): HTMLElement { + const cell = screen.getByText(jobName); + const row = cell.closest("tr"); + if (!row) { + throw new Error(`Row for job '${jobName}' not found`); + } + return row; +} + +describe("automations page integration", () => { + beforeEach(() => { + window.history.pushState({}, "", "/automations"); + }); + + it("navigates to automations from the header navigation", async () => { + const user = userEvent.setup(); + window.history.pushState({}, "", "/dashboard"); + renderWithProviders(); + + await user.click(await screen.findByRole("link", { name: "Automations" })); + + expect(await screen.findByRole("heading", { name: "Automations" })).toBeInTheDocument(); + expect(window.location.pathname).toBe("/automations"); + }); + + it("validates form input, creates a job, updates it, and renders run history", async () => { + const user = userEvent.setup(); + renderWithProviders(); + + expect(await screen.findByRole("heading", { name: "Automations" })).toBeInTheDocument(); + expect(await screen.findByText("No automations")).toBeInTheDocument(); + await user.click(screen.getByRole("button", { name: "Add automation" })); + + expect(await screen.findByRole("heading", { name: "Add automation" })).toBeInTheDocument(); + expect(screen.getByRole("heading", { name: "Basics" })).toBeInTheDocument(); + expect(screen.getByRole("heading", { name: "Schedule" })).toBeInTheDocument(); + expect(screen.getByRole("heading", { name: "Content / Execution" })).toBeInTheDocument(); + + await user.type(screen.getByPlaceholderText("Automation name"), "Daily smoke ping"); + await user.click(screen.getByRole("button", { name: "Accounts" })); + await user.click(await screen.findByRole("menuitemcheckbox", { name: "primary@example.com" })); + await user.keyboard("{Escape}"); + + await user.click(screen.getByRole("button", { name: "Create automation" })); + + expect(await screen.findByText("Daily smoke ping")).toBeInTheDocument(); + expect(screen.getByText("Runs will appear here after automation jobs execute.")).toBeInTheDocument(); + + await user.click(within(getJobRow("Daily smoke ping")).getByRole("switch")); + await waitFor(() => { + expect(within(getJobRow("Daily smoke ping")).getByText("Disabled")).toBeInTheDocument(); + }); + + await user.click(within(getJobRow("Daily smoke ping")).getByRole("button", { name: "Edit Daily smoke ping" })); + expect(await screen.findByRole("heading", { name: "Edit automation" })).toBeInTheDocument(); + const nameInput = screen.getByLabelText("Name"); + await user.clear(nameInput); + await user.type(nameInput, "Daily smoke ping edited"); + await user.click(screen.getByRole("button", { name: "Save changes" })); + expect(await screen.findByText("Daily smoke ping edited")).toBeInTheDocument(); + + await user.click(within(getJobRow("Daily smoke ping edited")).getByRole("button", { name: "Run now Daily smoke ping edited" })); + expect(await screen.findByRole("alertdialog", { name: "Run automation now" })).toBeInTheDocument(); + await user.click(screen.getByRole("button", { name: "Run now" })); + + const recentRunsSection = screen.getByRole("heading", { name: "Recent runs" }).closest("section"); + if (!recentRunsSection) { + throw new Error("Recent runs section not found"); + } + expect(await within(recentRunsSection).findByText("manual")).toBeInTheDocument(); + expect(await within(recentRunsSection).findByText("success")).toBeInTheDocument(); + await waitFor(() => { + expect(screen.queryByText("Runs will appear here after automation jobs execute.")).not.toBeInTheDocument(); + }); + }); + + it("creates automation with default all-accounts selection", async () => { + const user = userEvent.setup(); + renderWithProviders(); + + expect(await screen.findByRole("heading", { name: "Automations" })).toBeInTheDocument(); + await user.click(screen.getByRole("button", { name: "Add automation" })); + await user.type(screen.getByPlaceholderText("Automation name"), "All accounts job"); + await user.click(screen.getByRole("button", { name: "Create automation" })); + + const row = getJobRow("All accounts job"); + expect(within(row).getByText("All accounts")).toBeInTheDocument(); + expect(screen.queryByText("No accounts available. Add at least one account.")).not.toBeInTheDocument(); + }); +}); diff --git a/frontend/src/components/layout/app-header.tsx b/frontend/src/components/layout/app-header.tsx index b83a6cd483..9bd7fe2043 100644 --- a/frontend/src/components/layout/app-header.tsx +++ b/frontend/src/components/layout/app-header.tsx @@ -12,6 +12,7 @@ const NAV_ITEMS = [ { to: "/dashboard", label: "Dashboard" }, { to: "/reports", label: "Reports" }, { to: "/accounts", label: "Accounts" }, + { to: "/automations", label: "Automations" }, { to: "/apis", label: "APIs" }, { to: "/settings", label: "Settings" }, ] as const; diff --git a/frontend/src/features/api-keys/components/account-multi-select.tsx b/frontend/src/features/api-keys/components/account-multi-select.tsx index fe96b144a9..916d0f8055 100644 --- a/frontend/src/features/api-keys/components/account-multi-select.tsx +++ b/frontend/src/features/api-keys/components/account-multi-select.tsx @@ -22,8 +22,11 @@ export type AccountMultiSelectProps = { value: string[]; onChange: (value: string[]) => void; placeholder?: string; + triggerId?: string; + ariaInvalid?: boolean; + ariaDescribedBy?: string; + triggerClassName?: string; }; - type LimitChip = { key: string; label: string; @@ -122,6 +125,10 @@ export function AccountMultiSelect({ value, onChange, placeholder = "All accounts", + triggerId, + ariaInvalid = false, + ariaDescribedBy, + triggerClassName, }: AccountMultiSelectProps) { const { accountsQuery } = useAccounts(); const accounts = useMemo(() => accountsQuery.data ?? [], [accountsQuery.data]); @@ -183,7 +190,10 @@ export function AccountMultiSelect({ + ); + })} + +

Choose weekdays when this automation should run.

+ + + + +
+
+

Content / Execution

+
+
+
+ +