import time from sqlalchemy import ( create_engine, Column, Float, Integer, String, Text, DateTime, JSON, ForeignKey, UniqueConstraint, event, ) from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import Session, sessionmaker from datetime import datetime import os from pathlib import Path # DATABASE_URL (Postgres, e.g. Azure Database for PostgreSQL) takes over when set; otherwise this # falls back to the original SQLite-on-local-disk behavior unchanged (Hugging Face Spaces provides # persistent storage at /data; override with HF_DATA_DIR for local development where /data isn't # writable). Two deployments, two databases, same codebase — neither path touches the other. DATABASE_URL = os.getenv("DATABASE_URL", "").strip() if DATABASE_URL: # SQLAlchemy 1.4+/2.x rejects the "postgres://" scheme some providers still hand out (a # Heroku-era convention); normalize to "postgresql://" so it always works regardless of # which form the connection string arrives in. if DATABASE_URL.startswith("postgres://"): DATABASE_URL = "postgresql://" + DATABASE_URL[len("postgres://"):] engine = create_engine(DATABASE_URL, pool_pre_ping=True) else: DATA_DIR = Path(os.getenv("HF_DATA_DIR", "/data")) DATA_DIR.mkdir(parents=True, exist_ok=True) DB_PATH = os.getenv("HF_DATABASE_PATH", str(DATA_DIR / "emailout.db")) engine = create_engine( f"sqlite:///{DB_PATH}", connect_args={"check_same_thread": False, "timeout": 30}, ) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() IS_POSTGRES = engine.dialect.name == "postgresql" if not IS_POSTGRES: @event.listens_for(engine, "connect") def _sqlite_pragmas(dbapi_connection, _connection_record): cursor = dbapi_connection.cursor() cursor.execute("PRAGMA journal_mode=WAL") cursor.execute("PRAGMA synchronous=NORMAL") cursor.execute("PRAGMA busy_timeout=30000") cursor.close() # Dialect-portable SQL fragments for the additive migrations and raw inserts below — Postgres has # no DATETIME type (use TIMESTAMP) and no datetime('now') function (use NOW()). Kept as constants # rather than scattering dialect checks through every migration line. SQL_DATETIME_TYPE = "TIMESTAMP" if IS_POSTGRES else "DATETIME" SQL_NOW = "NOW()" if IS_POSTGRES else "datetime('now')" def db_commit_with_retry(session: Session, *, attempts: int = 5) -> None: """Retry commits when SQLite is busy (common under concurrent webhooks + campaign ticks).""" last_err = None for i in range(max(1, attempts)): try: session.commit() return except Exception as e: last_err = e msg = str(e).lower() if "locked" not in msg and "busy" not in msg: raise session.rollback() time.sleep(min(0.05 * (2**i), 1.0)) if last_err: raise last_err class User(Base): __tablename__ = "users" id = Column(Integer, primary_key=True, index=True) google_sub = Column(String, unique=True, index=True, nullable=True) microsoft_sub = Column(String, unique=True, index=True, nullable=True) # oid:tid — see auth_common.upsert_user_by_provider microsoft_tenant_id = Column(String, nullable=True) email = Column(String, index=True) name = Column(String, nullable=True) picture = Column(String, nullable=True) google_refresh_token = Column(Text, nullable=True) # for Gmail send (invite emails); treat as secret microsoft_refresh_token = Column(Text, nullable=True) # for Graph mail send (invite emails); treat as secret linkedin_profile_display_name = Column(String, nullable=True) default_unipile_account_ref_id = Column(Integer, ForeignKey("unipile_accounts.id"), nullable=True) mailbox_profile_display_name = Column(String, nullable=True) default_mailbox_unipile_account_ref_id = Column(Integer, ForeignKey("unipile_accounts.id"), nullable=True) created_at = Column(DateTime, default=datetime.utcnow) class Tenant(Base): __tablename__ = "tenants" id = Column(Integer, primary_key=True, index=True) name = Column(String, nullable=False) created_at = Column(DateTime, default=datetime.utcnow) class TenantMembership(Base): __tablename__ = "tenant_memberships" id = Column(Integer, primary_key=True, index=True) user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True) role = Column(String, default="member", nullable=False) # admin | member team_id = Column(Integer, ForeignKey("teams.id"), nullable=True, index=True) created_at = Column(DateTime, default=datetime.utcnow) __table_args__ = (UniqueConstraint("user_id", "tenant_id", name="uq_user_tenant"),) class Team(Base): """A sales team headed by one manager (TenantMembership.team_id points here). The manager is themself a member of their own team, so 'my team' queries naturally include their own records alongside the members they head.""" __tablename__ = "teams" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True) name = Column(String, nullable=False) manager_user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class Invitation(Base): __tablename__ = "invitations" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True) email = Column(String, nullable=False, index=True) token_hash = Column(String(64), unique=True, nullable=False, index=True) role = Column(String, default="member") team_id = Column(Integer, ForeignKey("teams.id"), nullable=True) invited_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=True) expires_at = Column(DateTime, nullable=False) created_at = Column(DateTime, default=datetime.utcnow) accepted_at = Column(DateTime, nullable=True) class UploadedFile(Base): __tablename__ = "uploaded_files" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True) file_id = Column(String, unique=True, index=True) filename = Column(String) contact_count = Column(Integer) file_path = Column(String) sequence_plan_json = Column(Text, nullable=True) campaign_name = Column(String, nullable=True) # user-facing campaign name — the backend record of "this campaign exists" status = Column(String, default="draft") # draft | generating | ready | launched # Email send pacing (campaign-setup-time config) — NULL timezone means no business-hours restriction. email_timezone = Column(String, nullable=True) # IANA zone name, e.g. "America/New_York" email_business_hours_start = Column(Integer, default=8) # local hour, inclusive, 0-23 email_business_hours_end = Column(Integer, default=18) # local hour, exclusive, 0-23 email_business_days = Column(String, default="1,2,3,4,5") # CSV ISO weekdays, Mon=1..Sun=7 email_send_interval_seconds = Column(Integer, default=60) # min gap between two email sends for this campaign generation_cancel_requested = Column(Integer, default=0) # 0/1 flag checked mid-stream by the SSE generator created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class Prompt(Base): __tablename__ = "prompts" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True) file_id = Column(String, index=True) product_name = Column(String) prompt_template = Column(Text) prompt_kind = Column(String, default="email") # email | linkedin created_at = Column(DateTime, default=datetime.utcnow) class GeneratedSequence(Base): __tablename__ = "generated_sequences" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True) file_id = Column(String, index=True) sequence_id = Column(Integer) # Contact sequence number email_number = Column(Integer, default=1) # Email number in sequence (1-10) channel = Column(String, default="email") # email | linkedin first_name = Column(String) last_name = Column(String) email = Column(String) company = Column(String) title = Column(String) product = Column(String) subject = Column(String) email_content = Column(Text) step_order = Column(Integer, nullable=True) created_at = Column(DateTime, default=datetime.utcnow) class Contact(Base): __tablename__ = "contacts" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True) file_id = Column(String, index=True) row_index = Column(Integer) first_name = Column(String) last_name = Column(String) email = Column(String, index=True) company = Column(String) title = Column(String) phone = Column(String, default="") source = Column(String, default="apollo_csv") raw_data = Column(JSON) # UniPile LinkedIn outreach (set when a campaign invite/DM is executed) unipile_provider_id = Column(String, nullable=True, index=True) linkedin_invite_sent_at = Column(DateTime, nullable=True) linkedin_invite_pending = Column(Integer, default=0) # 1 = waiting on acceptance webhook linkedin_connection_accepted_at = Column(DateTime, nullable=True) linkedin_last_followup_sent_at = Column(DateTime, nullable=True) linkedin_followup_next_email_number = Column(Integer, nullable=True) outreach_stopped_at = Column(DateTime, nullable=True) outreach_stop_reason = Column(String, nullable=True) owner_user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True) created_at = Column(DateTime, default=datetime.utcnow()) class Company(Base): """Account-level record — either created manually or synced (by name match, one-directionally) from distinct Contact.company values. Not linked to Contact via a foreign key: association is a case-insensitive name match, kept intentionally soft so this table is fully additive and never touches existing Contact/CrmLead/CrmDeal rows or behavior.""" __tablename__ = "companies" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True) name = Column(String, nullable=False, index=True) category = Column(String, default="") # A | B | C | "" type = Column(String, default="") # Existing | New | "" company_name_for_emails = Column(String, default="") industry = Column(String, default="") employees = Column(String, default="") annual_revenue = Column(String, default="") last_raised_at = Column(String, default="") website = Column(String, default="") city = Column(String, default="") state = Column(String, default="") country = Column(String, default="") phone = Column(String, default="") # main company line — company-level, synced like the rest source = Column(String, default="manual") # manual | synced_from_contacts owner_user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class ContactActivity(Base): """A logged Call or Visit against a Contact — who logged it, what was discussed, when. Powers the per-contact and per-company activity timelines. Always tied to a real Contact row (never a bare Lead) so company-level rollups can reuse the same case-insensitive Contact.company name match already used for associated-contacts lists.""" __tablename__ = "contact_activities" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True) contact_id = Column(Integer, ForeignKey("contacts.id"), nullable=False, index=True) user_id = Column(Integer, ForeignKey("users.id"), nullable=True) activity_type = Column(String, nullable=False) # "call" | "visit" notes = Column(Text, default="") created_at = Column(DateTime, default=datetime.utcnow, index=True) class CrmLead(Base): """Lead synced from Smartlead replies (webhook) — CRM pipeline status is local.""" __tablename__ = "crm_leads" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True) smartlead_lead_id = Column(String, index=True) campaign_id = Column(String, index=True) campaign_name = Column(String) email = Column(String, index=True) first_name = Column(String) last_name = Column(String) company_name = Column(String) title = Column(String) phone = Column(String, default="") # the lead's own personal phone — not company-synced last_reply_subject = Column(String) last_reply_body = Column(Text) last_reply_at = Column(DateTime, nullable=True) crm_status = Column(String, default="new_lead") # new_lead|contacted|qualified|unqualified|none contact_id = Column(Integer, nullable=True) # links to contacts.id after "Move to Contacts" owner_user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True) raw_webhook = Column(JSON) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class CrmDeal(Base): """Pipeline deal (often converted from a lead).""" __tablename__ = "crm_deals" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True) name = Column(String, index=True) stage = Column(String, default="new") # new|discovery|proposal|negotiation|won|lost owner_user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True) owner_initials = Column(String, default="") # display cache; derived from User when owner_user_id set deal_value = Column(Integer, nullable=True) # whole USD (nullable) revenue_type = Column(String, default="arr") # arr|qrr|mrr|one_time contact_display = Column(String, default="") # primary person label account_name = Column(String, default="") expected_close_date = Column(DateTime, nullable=True) close_probability = Column(Integer, default=10) # 0–100 country = Column(String, default="") last_interaction_at = Column(DateTime, nullable=True) source_lead_id = Column(Integer, nullable=True) source_campaign_name = Column(String, default="") contact_id = Column(Integer, nullable=True) # When stage is lost: reason + optional detail fields lost_reason = Column(String, default="") # price|no_decision|chose_another|didnt_meet_expectation|unknown lost_price_discount_pct = Column(Float, nullable=True) lost_chose_product_name = Column(String, default="") # When stage is won: invoicing / PO snapshot won_po_number = Column(String, default="") won_customer_legal_name = Column(String, default="") won_customer_address = Column(Text, default="") won_contact_person_name = Column(String, default="") won_channel_partner_name = Column(String, default="") won_note_to_customer = Column(Text, default="") won_note_to_accounts = Column(Text, default="") won_line_items = Column(JSON, nullable=True) won_at = Column(DateTime, nullable=True) # first transition to won (for quarterly attribution) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class SmartleadRun(Base): __tablename__ = "smartlead_runs" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True) file_id = Column(String, index=True) run_id = Column(String, unique=True, index=True) campaign_id = Column(String, index=True) campaign_name = Column(String) mode = Column(String) # 'existing' or 'new' steps_count = Column(Integer) dry_run = Column(Integer, default=0) # 0 = false, 1 = true total_leads = Column(Integer) added_leads = Column(Integer, default=0) skipped_leads = Column(Integer, default=0) failed_leads = Column(Integer, default=0) error_details = Column(Text) # JSON string of errors status = Column(String) # 'pending', 'completed', 'failed' created_at = Column(DateTime, default=datetime.utcnow) completed_at = Column(DateTime, nullable=True) class UnipileAccount(Base): __tablename__ = "unipile_accounts" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True) user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True) label = Column(String, default="") provider = Column(String, default="LINKEDIN") unipile_account_id = Column(String, index=True) status = Column(String, default="pending") auth_mode = Column(String, default="hosted") metadata_json = Column(JSON, nullable=True) # Safe-sending caps (LinkedIn accounts only, harmless/unread on mailbox accounts). Defaults are # conservative research-backed numbers — Unipile's own platform ceiling is ~80-100 connects/day, # but sustained volume well under that is what actually avoids LinkedIn's restriction flags. linkedin_daily_connection_cap = Column(Integer, default=20) linkedin_daily_message_cap = Column(Integer, default=40) linkedin_weekly_connection_cap = Column(Integer, default=100) linkedin_weekly_message_cap = Column(Integer, default=200) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class LinkedinCampaign(Base): __tablename__ = "linkedin_campaigns" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True) user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True) name = Column(String, index=True) status = Column(String, default="draft") # draft | generated | running | completed | failed unipile_account_ref_id = Column(Integer, ForeignKey("unipile_accounts.id"), nullable=False, index=True) file_id = Column(String, index=True, nullable=True) contact_count = Column(Integer, default=0) prompt_template = Column(Text, default="") execution_result = Column(JSON, nullable=True) followup_interval_hours = Column(Integer, default=72) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) executed_at = Column(DateTime, nullable=True) class FileOutreachExecution(Base): """Server-side orchestration for wizard email+LinkedIn sequences (per uploaded CSV file).""" __tablename__ = "file_outreach_executions" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True) user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True) file_id = Column(String, index=True, nullable=False) status = Column(String, default="idle") # idle | running | paused | completed progress_done = Column(Integer, default=0) progress_total = Column(Integer, default=0) stats_json = Column(Text, nullable=True) cursor_json = Column(Text, nullable=True) # delivery_started_at, etc. last_error = Column(Text, nullable=True) linkedin_followup_interval_hours = Column(Integer, default=72) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) __table_args__ = (UniqueConstraint("tenant_id", "file_id", name="uq_outreach_exec_tenant_file"),) class OutreachSendReceipt(Base): """One row per (file, contact, sequence action slot) send attempt.""" __tablename__ = "outreach_send_receipts" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True) file_id = Column(String, index=True, nullable=False) contact_id = Column(Integer, ForeignKey("contacts.id"), nullable=False, index=True) step_slot = Column(Integer, nullable=False) channel = Column(String, nullable=False) action = Column(String, default="") status = Column(String, default="pending") # sent | failed | skipped detail = Column(Text, nullable=True) unipile_account_ref_id = Column(Integer, ForeignKey("unipile_accounts.id"), nullable=True, index=True) unipile_message_id = Column(String, nullable=True) unipile_tracking_id = Column(String, nullable=True, index=True) smtp_message_id = Column(String, nullable=True, index=True) tracking_label = Column(String(220), nullable=True, index=True) opened_at = Column(DateTime, nullable=True) open_count = Column(Integer, default=0) replied_at = Column(DateTime, nullable=True) bounced_at = Column(DateTime, nullable=True) created_at = Column(DateTime, default=datetime.utcnow) __table_args__ = ( UniqueConstraint("tenant_id", "file_id", "contact_id", "step_slot", name="uq_outreach_receipt_slot"), ) class UnipileHostedAuthState(Base): __tablename__ = "unipile_hosted_auth_states" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=True, index=True) user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True) token = Column(String, unique=True, index=True) provider = Column(String, default="LINKEDIN") label = Column(String, default="") expires_at = Column(DateTime, nullable=True) used_at = Column(DateTime, nullable=True) created_at = Column(DateTime, default=datetime.utcnow) class CtaForm(Base): """Embeddable lead-capture form (WordPress / external sites → Leads).""" __tablename__ = "cta_forms" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True) user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True) public_id = Column(String, unique=True, index=True, nullable=False) name = Column(String, nullable=False, default="") fields = Column(JSON, nullable=False, default=list) embed_width = Column(String, default="100%") embed_height = Column(String, default="480") is_active = Column(Integer, default=1) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class LandingPage(Base): """AI-generated public landing pages with CTA form embed.""" __tablename__ = "landing_pages" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True) user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True) public_id = Column(String, unique=True, index=True, nullable=False) name = Column(String, nullable=False, default="") prompt = Column(String, default="") content = Column(JSON, nullable=False, default=dict) cta_form_id = Column(Integer, ForeignKey("cta_forms.id"), nullable=True, index=True) banner_asset = Column(String, nullable=True) logo_asset = Column(String, nullable=True) is_active = Column(Integer, default=1) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class ChatAgentConfig(Base): """One row per tenant: Chat Agent widget config, Drive connection, website source, model choice.""" __tablename__ = "chat_agent_configs" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, unique=True, index=True) is_enabled = Column(Integer, default=1) public_id = Column(String, unique=True, index=True, nullable=False) greeting_message = Column( Text, default="Hi! 👋 I'm happy to help answer any questions. What's your name and email, in case we get disconnected?", ) chat_model_provider = Column(String, default="openai") # openai | gemini website_base_url = Column(String, default="") website_status = Column(String, default="not_configured") # not_configured|idle|running|success|error website_last_indexed_at = Column(DateTime, nullable=True) drive_folder_id = Column(String, default="") drive_folder_url = Column(String, default="") drive_refresh_token = Column(Text, nullable=True) drive_connected_email = Column(String, default="") drive_status = Column(String, default="not_connected") # not_connected|idle|running|success|error drive_last_indexed_at = Column(DateTime, nullable=True) reindex_interval_hours = Column(Integer, default=24) # Opt-in, off by default: lets the agentic loop call a Gemini web-search tool when the # knowledge base doesn't have enough for a detailed answer (e.g. a named competitor, # industry/technology background). Results are always presented as unverified external # content — see SKILLS-SECURITY.md for the residual risk this doesn't eliminate (a visitor's # message can still influence what query gets researched). enable_web_search = Column(Integer, default=0) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class ChatAgentSkill(Base): """ Admin-editable skill, following Anthropic's Agent Skills anatomy (SKILL.md: YAML frontmatter name+description + markdown instructions, plus references/assets/templates/ scripts/evals). `instructions` (the "brain") is always compiled into the chat system prompt; references, assets, and templates are indexed into the same searchable knowledge base as Drive/website content so the agentic search_knowledge_base tool can surface them on demand. `scripts_json` can only be executed via the admin-only, sandboxed "Run script" action (chat_agent_script_runner.py) — the live public chat agent never executes it (see SKILLS-SECURITY.md for why). """ __tablename__ = "chat_agent_skills" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True) text = Column(Text, nullable=False, default="") # legacy; superseded by `instructions` below name = Column(String, default="") description = Column(Text, default="") instructions = Column(Text, default="") references_json = Column(JSON, nullable=True) # [{title, content}] assets_json = Column(JSON, nullable=True) # [{name, content}] templates_json = Column(JSON, nullable=True) # [{name, content}] scripts_json = Column(JSON, nullable=True) # [{name, description, language, content, last_run_*}] evals_json = Column(JSON, nullable=True) # [{prompt, assertion_type, assertion_value, last_run_*}] is_default = Column(Integer, default=0) is_active = Column(Integer, default=1) sort_order = Column(Integer, default=0) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class ChatAgentDocument(Base): """One ingested source item (a Drive file or a crawled website page), converted to markdown.""" __tablename__ = "chat_agent_documents" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True) source_type = Column(String, nullable=False, index=True) # drive | website source_ref = Column(String, nullable=False, index=True) # drive file id | page URL title = Column(String, default="") content_markdown = Column(Text, default="") content_hash = Column(String, default="", index=True) external_modified_at = Column(DateTime, nullable=True) indexed_at = Column(DateTime, nullable=True) status = Column(String, default="ok") # ok | error error_detail = Column(Text, nullable=True) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) __table_args__ = ( UniqueConstraint("tenant_id", "source_type", "source_ref", name="uq_chat_doc_tenant_source"), ) class ChatAgentChunk(Base): """Embedded retrieval unit for a ChatAgentDocument.""" __tablename__ = "chat_agent_chunks" id = Column(Integer, primary_key=True, index=True) document_id = Column(Integer, ForeignKey("chat_agent_documents.id"), nullable=False, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True) chunk_index = Column(Integer, default=0) text = Column(Text, default="") embedding = Column(JSON, nullable=True) created_at = Column(DateTime, default=datetime.utcnow) class ChatAgentConversation(Base): """One website-visitor chat session (public widget).""" __tablename__ = "chat_agent_conversations" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True) session_id = Column(String, unique=True, index=True, nullable=False) visitor_name = Column(String, default="") visitor_email = Column(String, default="", index=True) visitor_company = Column(String, default="") lead_id = Column(Integer, ForeignKey("crm_leads.id"), nullable=True, index=True) page_url = Column(String, default="") started_at = Column(DateTime, default=datetime.utcnow) last_message_at = Column(DateTime, default=datetime.utcnow) class ChatAgentMessage(Base): __tablename__ = "chat_agent_messages" id = Column(Integer, primary_key=True, index=True) conversation_id = Column(Integer, ForeignKey("chat_agent_conversations.id"), nullable=False, index=True) role = Column(String, nullable=False) # user | assistant content = Column(Text, default="") # True when this reply used the search_the_web tool — surfaced to admins in the Test Chat # and Conversations views so they can tell live-web-sourced answers apart from knowledge-base # answers at a glance. used_web_search = Column(Integer, default=0) # Total LLM tokens (prompt + completion, summed across every round of the agentic loop) spent # generating this reply — 0 for user messages and for any reply where the provider didn't # report usage. Surfaced per-message and summed per-conversation in the Conversations tab. total_tokens = Column(Integer, default=0) created_at = Column(DateTime, default=datetime.utcnow) class ChatAgentIndexLog(Base): __tablename__ = "chat_agent_index_logs" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True) source_type = Column(String, nullable=False) # drive | website status = Column(String, default="running") # running | success | error documents_processed = Column(Integer, default=0) chunks_created = Column(Integer, default=0) started_at = Column(DateTime, default=datetime.utcnow) finished_at = Column(DateTime, nullable=True) detail = Column(Text, nullable=True) # --------------------------------------------------------------------------- # Qualifier Agent (RFQ Qualifier) — a deliberately SEPARATE skill/ingestion system from the # Chat Agent tables above, per design decision: each background agent (Qualifier, and later Quote # Estimator) gets its own independent skill model + knowledge base rather than a shared/generalized # one. Structurally the same anatomy as ChatAgentSkill/Document/Chunk/IndexLog (SKILL.md-compatible # skill row, hash-diffed embedded chunks, index-run log) but intentionally not reusing those tables # or any cross-FK to them — this whole block could be lifted into a standalone service/repo without # touching the Chat Agent feature. See qualifier_common.py / qualifier_ingest.py / qualifier_agent.py. # --------------------------------------------------------------------------- class QualifierSkill(Base): """ Admin-editable skill for the RFQ Qualifier agent — same SKILL.md anatomy as ChatAgentSkill (name+description+instructions, references/assets/templates/scripts/evals), compiled/parsed by qualifier_common.py. `instructions` (the qualify policy) is always injected into the agent's system prompt; references/assets/templates are indexed into qualifier_chunks alongside the Wittur pricelist so the agentic loop's search_pricelist-adjacent lookups can surface them. `scripts_json` is admin-run-only via qualifier_script_runner.py — the background agent never executes it. """ __tablename__ = "qualifier_skills" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True) name = Column(String, default="") description = Column(Text, default="") instructions = Column(Text, default="") references_json = Column(JSON, nullable=True) # [{title, content}] assets_json = Column(JSON, nullable=True) # [{name, content}] templates_json = Column(JSON, nullable=True) # [{name, content}] scripts_json = Column(JSON, nullable=True) # [{name, description, language, content, last_run_*}] evals_json = Column(JSON, nullable=True) # [{name, input_ref, expected_decision, expected_project_type, assertion_type, assertion_value, last_run_*}] is_default = Column(Integer, default=0) is_active = Column(Integer, default=1) sort_order = Column(Integer, default=0) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) class QualifierDocument(Base): """One ingested source item for the Qualifier's knowledge base — currently only source_type 'pricelist' (the Wittur pricelist PDF, re-ingested on demand) plus 'skill_reference' / 'skill_asset' / 'skill_template' (a QualifierSkill's own content, indexed the same way).""" __tablename__ = "qualifier_documents" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True) source_type = Column(String, nullable=False, index=True) # pricelist | skill_reference | skill_asset | skill_template source_ref = Column(String, nullable=False, index=True) # e.g. "pricelist:page:3" or "skill:1:reference:0" title = Column(String, default="") content_markdown = Column(Text, default="") content_hash = Column(String, default="", index=True) indexed_at = Column(DateTime, nullable=True) status = Column(String, default="ok") # ok | error error_detail = Column(Text, nullable=True) created_at = Column(DateTime, default=datetime.utcnow) updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) __table_args__ = ( UniqueConstraint("tenant_id", "source_type", "source_ref", name="uq_qualifier_doc_tenant_source"), ) class QualifierChunk(Base): """Embedded retrieval unit for a QualifierDocument — searched by the agent's search_pricelist tool call (qualifier_retrieval.py).""" __tablename__ = "qualifier_chunks" id = Column(Integer, primary_key=True, index=True) document_id = Column(Integer, ForeignKey("qualifier_documents.id"), nullable=False, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True) chunk_index = Column(Integer, default=0) text = Column(Text, default="") embedding = Column(JSON, nullable=True) created_at = Column(DateTime, default=datetime.utcnow) class QualifierIndexLog(Base): __tablename__ = "qualifier_index_logs" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True) source_type = Column(String, nullable=False) # pricelist status = Column(String, default="running") # running | success | error documents_processed = Column(Integer, default=0) chunks_created = Column(Integer, default=0) started_at = Column(DateTime, default=datetime.utcnow) finished_at = Column(DateTime, nullable=True) detail = Column(Text, nullable=True) class QualifierRun(Base): """One qualification run: an RFQ input (uploaded .eml, or PDF/DOCX spec, or a future inbox-polling trigger) processed through the Qualifier agent end to end. This is both the audit trail and the source of the "past runs" list in the dashboard.""" __tablename__ = "qualifier_runs" id = Column(Integer, primary_key=True, index=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False, index=True) skill_id = Column(Integer, ForeignKey("qualifier_skills.id"), nullable=True, index=True) created_by_user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True) trigger_type = Column(String, default="manual_upload") # manual_upload | api | inbox_poll (future) input_type = Column(String, nullable=False, default="") # eml | pdf | docx input_filename = Column(String, default="") input_file_path = Column(String, default="") # storage path/blob ref to the raw upload, see qualifier_storage.py project_name = Column(String, default="") sender_email = Column(String, default="") deadline_text = Column(String, default="") status = Column(String, default="running") # running | complete | error decision = Column(String, nullable=True) # qualify | disqualify | needs_review project_type = Column(String, nullable=True) # modernization | new_construction | unknown confidence = Column(Float, nullable=True) result_json = Column(JSON, nullable=True) # full structured QualifierDecision object extracted_text_excerpt = Column(Text, nullable=True) # what was actually handed to the model, for debugging error_detail = Column(Text, nullable=True) total_tokens = Column(Integer, default=0) started_at = Column(DateTime, default=datetime.utcnow) finished_at = Column(DateTime, nullable=True) def run_migrations(connection_engine): """Add tenant_id to legacy SQLite tables and attach rows to the default tenant.""" from sqlalchemy import inspect, text insp = inspect(connection_engine) tenant_tables = ( "uploaded_files", "prompts", "generated_sequences", "contacts", "crm_leads", "crm_deals", "smartlead_runs", "unipile_accounts", "linkedin_campaigns", "unipile_hosted_auth_states", "cta_forms", "landing_pages", "chat_agent_configs", "chat_agent_skills", "chat_agent_documents", "chat_agent_chunks", "chat_agent_conversations", "chat_agent_index_logs", ) with connection_engine.begin() as conn: if insp.has_table("tenants"): n = conn.execute(text("SELECT COUNT(*) FROM tenants")).scalar() if n == 0: conn.execute( text( f"INSERT INTO tenants (name, created_at) VALUES (:name, {SQL_NOW})" ), {"name": "Default workspace"}, ) default_tid = conn.execute(text("SELECT id FROM tenants ORDER BY id LIMIT 1")).scalar() if default_tid is None: default_tid = 1 insp = inspect(connection_engine) for tbl in tenant_tables: if not insp.has_table(tbl): continue cols = [c["name"] for c in insp.get_columns(tbl)] if "tenant_id" not in cols: conn.execute(text(f"ALTER TABLE {tbl} ADD COLUMN tenant_id INTEGER")) for tbl in tenant_tables: if not insp.has_table(tbl): continue conn.execute( text(f"UPDATE {tbl} SET tenant_id = :tid WHERE tenant_id IS NULL"), {"tid": default_tid}, ) insp = inspect(connection_engine) if insp.has_table("users"): ucols = [c["name"] for c in insp.get_columns("users")] if "google_refresh_token" not in ucols: conn.execute(text("ALTER TABLE users ADD COLUMN google_refresh_token TEXT")) if "linkedin_profile_display_name" not in ucols: conn.execute(text("ALTER TABLE users ADD COLUMN linkedin_profile_display_name TEXT")) if "default_unipile_account_ref_id" not in ucols: conn.execute(text("ALTER TABLE users ADD COLUMN default_unipile_account_ref_id INTEGER")) if "mailbox_profile_display_name" not in ucols: conn.execute(text("ALTER TABLE users ADD COLUMN mailbox_profile_display_name TEXT")) if "default_mailbox_unipile_account_ref_id" not in ucols: conn.execute(text("ALTER TABLE users ADD COLUMN default_mailbox_unipile_account_ref_id INTEGER")) if "microsoft_sub" not in ucols: conn.execute(text("ALTER TABLE users ADD COLUMN microsoft_sub TEXT")) conn.execute(text("CREATE UNIQUE INDEX IF NOT EXISTS ix_users_microsoft_sub ON users (microsoft_sub)")) if "microsoft_tenant_id" not in ucols: conn.execute(text("ALTER TABLE users ADD COLUMN microsoft_tenant_id TEXT")) if "microsoft_refresh_token" not in ucols: conn.execute(text("ALTER TABLE users ADD COLUMN microsoft_refresh_token TEXT")) # google_sub was NOT NULL from day one (every user signed in with Google); a # Microsoft-only user has no Google identity, so relax it. Postgres-only (production) # — instant, no table rewrite. Skipped on SQLite (dev-only; ALTER COLUMN isn't # supported there and no production data is ever on SQLite). if IS_POSTGRES: col_info = next((c for c in insp.get_columns("users") if c["name"] == "google_sub"), None) if col_info is not None and not col_info.get("nullable", True): conn.execute(text("ALTER TABLE users ALTER COLUMN google_sub DROP NOT NULL")) insp = inspect(connection_engine) if insp.has_table("crm_deals"): dcols = [c["name"] for c in insp.get_columns("crm_deals")] if "owner_user_id" not in dcols: conn.execute(text("ALTER TABLE crm_deals ADD COLUMN owner_user_id INTEGER")) if "revenue_type" not in dcols: conn.execute( text("ALTER TABLE crm_deals ADD COLUMN revenue_type TEXT DEFAULT 'arr'") ) conn.execute( text( "UPDATE crm_deals SET revenue_type = 'arr' WHERE revenue_type IS NULL OR revenue_type = ''" ) ) if "lost_reason" not in dcols: conn.execute(text("ALTER TABLE crm_deals ADD COLUMN lost_reason TEXT DEFAULT ''")) if "lost_price_discount_pct" not in dcols: conn.execute(text("ALTER TABLE crm_deals ADD COLUMN lost_price_discount_pct REAL")) if "lost_chose_product_name" not in dcols: conn.execute(text("ALTER TABLE crm_deals ADD COLUMN lost_chose_product_name TEXT DEFAULT ''")) if "won_po_number" not in dcols: conn.execute(text("ALTER TABLE crm_deals ADD COLUMN won_po_number TEXT DEFAULT ''")) if "won_customer_legal_name" not in dcols: conn.execute(text("ALTER TABLE crm_deals ADD COLUMN won_customer_legal_name TEXT DEFAULT ''")) if "won_customer_address" not in dcols: conn.execute(text("ALTER TABLE crm_deals ADD COLUMN won_customer_address TEXT DEFAULT ''")) if "won_contact_person_name" not in dcols: conn.execute(text("ALTER TABLE crm_deals ADD COLUMN won_contact_person_name TEXT DEFAULT ''")) if "won_channel_partner_name" not in dcols: conn.execute(text("ALTER TABLE crm_deals ADD COLUMN won_channel_partner_name TEXT DEFAULT ''")) if "won_note_to_customer" not in dcols: conn.execute(text("ALTER TABLE crm_deals ADD COLUMN won_note_to_customer TEXT DEFAULT ''")) if "won_note_to_accounts" not in dcols: conn.execute(text("ALTER TABLE crm_deals ADD COLUMN won_note_to_accounts TEXT DEFAULT ''")) if "won_line_items" not in dcols: conn.execute(text("ALTER TABLE crm_deals ADD COLUMN won_line_items TEXT")) if "won_at" not in dcols: conn.execute(text(f"ALTER TABLE crm_deals ADD COLUMN won_at {SQL_DATETIME_TYPE}")) conn.execute( text( "UPDATE crm_deals SET won_at = updated_at WHERE stage = 'won' AND won_at IS NULL" ) ) insp = inspect(connection_engine) if insp.has_table("prompts"): pcols = [c["name"] for c in insp.get_columns("prompts")] if "prompt_kind" not in pcols: conn.execute(text("ALTER TABLE prompts ADD COLUMN prompt_kind TEXT DEFAULT 'email'")) conn.execute( text( "UPDATE prompts SET prompt_kind = 'email' WHERE prompt_kind IS NULL OR prompt_kind = ''" ) ) insp = inspect(connection_engine) if insp.has_table("generated_sequences"): gcols = [c["name"] for c in insp.get_columns("generated_sequences")] if "channel" not in gcols: conn.execute(text("ALTER TABLE generated_sequences ADD COLUMN channel TEXT DEFAULT 'email'")) conn.execute( text( "UPDATE generated_sequences SET channel = 'email' WHERE channel IS NULL OR channel = ''" ) ) gcols2 = [c["name"] for c in insp.get_columns("generated_sequences")] if "step_order" not in gcols2: conn.execute(text("ALTER TABLE generated_sequences ADD COLUMN step_order INTEGER")) insp = inspect(connection_engine) if insp.has_table("uploaded_files"): ufcols = [c["name"] for c in insp.get_columns("uploaded_files")] if "sequence_plan_json" not in ufcols: conn.execute(text("ALTER TABLE uploaded_files ADD COLUMN sequence_plan_json TEXT")) if "campaign_name" not in ufcols: conn.execute(text("ALTER TABLE uploaded_files ADD COLUMN campaign_name TEXT")) if "status" not in ufcols: conn.execute(text("ALTER TABLE uploaded_files ADD COLUMN status TEXT DEFAULT 'draft'")) if "updated_at" not in ufcols: conn.execute(text(f"ALTER TABLE uploaded_files ADD COLUMN updated_at {SQL_DATETIME_TYPE}")) if "email_timezone" not in ufcols: conn.execute(text("ALTER TABLE uploaded_files ADD COLUMN email_timezone TEXT")) if "email_business_hours_start" not in ufcols: conn.execute(text("ALTER TABLE uploaded_files ADD COLUMN email_business_hours_start INTEGER DEFAULT 8")) if "email_business_hours_end" not in ufcols: conn.execute(text("ALTER TABLE uploaded_files ADD COLUMN email_business_hours_end INTEGER DEFAULT 18")) if "email_business_days" not in ufcols: conn.execute(text("ALTER TABLE uploaded_files ADD COLUMN email_business_days TEXT DEFAULT '1,2,3,4,5'")) if "email_send_interval_seconds" not in ufcols: conn.execute(text("ALTER TABLE uploaded_files ADD COLUMN email_send_interval_seconds INTEGER DEFAULT 60")) if "generation_cancel_requested" not in ufcols: conn.execute(text("ALTER TABLE uploaded_files ADD COLUMN generation_cancel_requested INTEGER DEFAULT 0")) insp = inspect(connection_engine) if insp.has_table("chat_agent_conversations"): caccols = [c["name"] for c in insp.get_columns("chat_agent_conversations")] if "visitor_company" not in caccols: conn.execute(text("ALTER TABLE chat_agent_conversations ADD COLUMN visitor_company TEXT DEFAULT ''")) insp = inspect(connection_engine) if insp.has_table("unipile_accounts"): uacols = [c["name"] for c in insp.get_columns("unipile_accounts")] if "user_id" not in uacols: conn.execute(text("ALTER TABLE unipile_accounts ADD COLUMN user_id INTEGER")) if "linkedin_daily_connection_cap" not in uacols: conn.execute(text("ALTER TABLE unipile_accounts ADD COLUMN linkedin_daily_connection_cap INTEGER DEFAULT 20")) if "linkedin_daily_message_cap" not in uacols: conn.execute(text("ALTER TABLE unipile_accounts ADD COLUMN linkedin_daily_message_cap INTEGER DEFAULT 40")) if "linkedin_weekly_connection_cap" not in uacols: conn.execute(text("ALTER TABLE unipile_accounts ADD COLUMN linkedin_weekly_connection_cap INTEGER DEFAULT 100")) if "linkedin_weekly_message_cap" not in uacols: conn.execute(text("ALTER TABLE unipile_accounts ADD COLUMN linkedin_weekly_message_cap INTEGER DEFAULT 200")) insp = inspect(connection_engine) if insp.has_table("linkedin_campaigns"): lccols = [c["name"] for c in insp.get_columns("linkedin_campaigns")] if "user_id" not in lccols: conn.execute(text("ALTER TABLE linkedin_campaigns ADD COLUMN user_id INTEGER")) if "followup_interval_hours" not in lccols: conn.execute( text("ALTER TABLE linkedin_campaigns ADD COLUMN followup_interval_hours INTEGER DEFAULT 72") ) conn.execute( text( "UPDATE linkedin_campaigns SET followup_interval_hours = 72 WHERE followup_interval_hours IS NULL" ) ) insp = inspect(connection_engine) if insp.has_table("contacts"): ctcols = [c["name"] for c in insp.get_columns("contacts")] if "unipile_provider_id" not in ctcols: conn.execute(text("ALTER TABLE contacts ADD COLUMN unipile_provider_id TEXT")) if "linkedin_invite_sent_at" not in ctcols: conn.execute(text(f"ALTER TABLE contacts ADD COLUMN linkedin_invite_sent_at {SQL_DATETIME_TYPE}")) if "linkedin_invite_pending" not in ctcols: conn.execute(text("ALTER TABLE contacts ADD COLUMN linkedin_invite_pending INTEGER DEFAULT 0")) if "linkedin_connection_accepted_at" not in ctcols: conn.execute(text(f"ALTER TABLE contacts ADD COLUMN linkedin_connection_accepted_at {SQL_DATETIME_TYPE}")) if "linkedin_last_followup_sent_at" not in ctcols: conn.execute(text(f"ALTER TABLE contacts ADD COLUMN linkedin_last_followup_sent_at {SQL_DATETIME_TYPE}")) if "linkedin_followup_next_email_number" not in ctcols: conn.execute(text("ALTER TABLE contacts ADD COLUMN linkedin_followup_next_email_number INTEGER")) if "outreach_stopped_at" not in ctcols: conn.execute(text(f"ALTER TABLE contacts ADD COLUMN outreach_stopped_at {SQL_DATETIME_TYPE}")) if "outreach_stop_reason" not in ctcols: conn.execute(text("ALTER TABLE contacts ADD COLUMN outreach_stop_reason TEXT")) if "phone" not in ctcols: conn.execute(text("ALTER TABLE contacts ADD COLUMN phone TEXT DEFAULT ''")) if "owner_user_id" not in ctcols: conn.execute(text("ALTER TABLE contacts ADD COLUMN owner_user_id INTEGER")) insp = inspect(connection_engine) if insp.has_table("crm_leads"): leadcols = [c["name"] for c in insp.get_columns("crm_leads")] if "phone" not in leadcols: conn.execute(text("ALTER TABLE crm_leads ADD COLUMN phone TEXT DEFAULT ''")) if "owner_user_id" not in leadcols: conn.execute(text("ALTER TABLE crm_leads ADD COLUMN owner_user_id INTEGER")) insp = inspect(connection_engine) if insp.has_table("tenant_memberships"): tmcols = [c["name"] for c in insp.get_columns("tenant_memberships")] if "team_id" not in tmcols: conn.execute(text("ALTER TABLE tenant_memberships ADD COLUMN team_id INTEGER")) insp = inspect(connection_engine) if insp.has_table("companies"): compcols = [c["name"] for c in insp.get_columns("companies")] if "phone" not in compcols: conn.execute(text("ALTER TABLE companies ADD COLUMN phone TEXT DEFAULT ''")) if "type" not in compcols: conn.execute(text("ALTER TABLE companies ADD COLUMN type TEXT DEFAULT ''")) if "owner_user_id" not in compcols: conn.execute(text("ALTER TABLE companies ADD COLUMN owner_user_id INTEGER")) insp = inspect(connection_engine) if insp.has_table("file_outreach_executions"): ecols = [c["name"] for c in insp.get_columns("file_outreach_executions")] if "linkedin_followup_interval_hours" not in ecols: conn.execute( text( "ALTER TABLE file_outreach_executions ADD COLUMN linkedin_followup_interval_hours INTEGER DEFAULT 72" ) ) insp = inspect(connection_engine) if insp.has_table("outreach_send_receipts"): rcols = [c["name"] for c in insp.get_columns("outreach_send_receipts")] if "tracking_label" not in rcols: conn.execute(text("ALTER TABLE outreach_send_receipts ADD COLUMN tracking_label TEXT")) if "unipile_tracking_id" not in rcols: conn.execute(text("ALTER TABLE outreach_send_receipts ADD COLUMN unipile_tracking_id TEXT")) if "unipile_account_ref_id" not in rcols: conn.execute(text("ALTER TABLE outreach_send_receipts ADD COLUMN unipile_account_ref_id INTEGER")) if "smtp_message_id" not in rcols: conn.execute(text("ALTER TABLE outreach_send_receipts ADD COLUMN smtp_message_id TEXT")) if "opened_at" not in rcols: conn.execute(text(f"ALTER TABLE outreach_send_receipts ADD COLUMN opened_at {SQL_DATETIME_TYPE}")) if "open_count" not in rcols: conn.execute( text("ALTER TABLE outreach_send_receipts ADD COLUMN open_count INTEGER DEFAULT 0") ) if "replied_at" not in rcols: conn.execute(text(f"ALTER TABLE outreach_send_receipts ADD COLUMN replied_at {SQL_DATETIME_TYPE}")) if "bounced_at" not in rcols: conn.execute(text(f"ALTER TABLE outreach_send_receipts ADD COLUMN bounced_at {SQL_DATETIME_TYPE}")) insp = inspect(connection_engine) if insp.has_table("cta_forms"): ccols = [c["name"] for c in insp.get_columns("cta_forms")] if "embed_width" not in ccols: conn.execute(text("ALTER TABLE cta_forms ADD COLUMN embed_width TEXT DEFAULT '100%'")) if "embed_height" not in ccols: conn.execute(text("ALTER TABLE cta_forms ADD COLUMN embed_height TEXT DEFAULT '480'")) insp = inspect(connection_engine) if insp.has_table("chat_agent_skills"): skcols = [c["name"] for c in insp.get_columns("chat_agent_skills")] if "name" not in skcols: conn.execute(text("ALTER TABLE chat_agent_skills ADD COLUMN name TEXT DEFAULT ''")) if "description" not in skcols: conn.execute(text("ALTER TABLE chat_agent_skills ADD COLUMN description TEXT")) if "instructions" not in skcols: conn.execute(text("ALTER TABLE chat_agent_skills ADD COLUMN instructions TEXT")) if "references_json" not in skcols: conn.execute(text("ALTER TABLE chat_agent_skills ADD COLUMN references_json TEXT")) if "assets_json" not in skcols: conn.execute(text("ALTER TABLE chat_agent_skills ADD COLUMN assets_json TEXT")) if "templates_json" not in skcols: conn.execute(text("ALTER TABLE chat_agent_skills ADD COLUMN templates_json TEXT")) if "scripts_json" not in skcols: conn.execute(text("ALTER TABLE chat_agent_skills ADD COLUMN scripts_json TEXT")) if "evals_json" not in skcols: conn.execute(text("ALTER TABLE chat_agent_skills ADD COLUMN evals_json TEXT")) if "sort_order" not in skcols: conn.execute(text("ALTER TABLE chat_agent_skills ADD COLUMN sort_order INTEGER DEFAULT 0")) # Backfill legacy rows (created before `instructions` existed) so every skill has a # real name/description/instructions triple, matching the SKILL.md anatomy. conn.execute( text( "UPDATE chat_agent_skills SET instructions = text " "WHERE (instructions IS NULL OR instructions = '') AND text IS NOT NULL" ) ) conn.execute( text( "UPDATE chat_agent_skills SET name = 'skill-' || id " "WHERE name IS NULL OR name = ''" ) ) conn.execute( text( "UPDATE chat_agent_skills SET description = 'Legacy skill — edit to add a name and description' " "WHERE description IS NULL OR description = ''" ) ) # One-time consolidation: the Skills feature moved from 3 small default guardrail # skills to one comprehensive default skill. A tenant whose skills are EXACTLY the old # 3 defaults (all is_default=1, no custom skills added) gets them cleared here — # ensure_default_skills() (chat_agent_common.py) already reseeds automatically once a # tenant has zero skills, so it picks up the new comprehensive one on next access. Any # tenant with a custom (is_default=0) skill is left untouched, so real admin edits are # never silently destroyed. _old_default_names = sorted(["no-internal-pricing", "no-client-names", "no-sensitive-info"]) _rows = conn.execute(text("SELECT tenant_id, name, is_default FROM chat_agent_skills")).fetchall() _by_tenant: dict = {} for _tid, _name, _is_default in _rows: _by_tenant.setdefault(_tid, []).append((_name, _is_default)) for _tid, _items in _by_tenant.items(): _names = sorted(n for n, _ in _items) _all_default = all(bool(d) for _, d in _items) if _all_default and _names == _old_default_names: conn.execute( text("DELETE FROM chat_agent_skills WHERE tenant_id = :tid"), {"tid": _tid}, ) insp = inspect(connection_engine) if insp.has_table("chat_agent_configs"): cfgcols = [c["name"] for c in insp.get_columns("chat_agent_configs")] if "enable_web_search" not in cfgcols: conn.execute( text("ALTER TABLE chat_agent_configs ADD COLUMN enable_web_search INTEGER DEFAULT 0") ) if insp.has_table("chat_agent_messages"): msgcols = [c["name"] for c in insp.get_columns("chat_agent_messages")] if "used_web_search" not in msgcols: conn.execute( text("ALTER TABLE chat_agent_messages ADD COLUMN used_web_search INTEGER DEFAULT 0") ) if "total_tokens" not in msgcols: conn.execute( text("ALTER TABLE chat_agent_messages ADD COLUMN total_tokens INTEGER DEFAULT 0") ) insp = inspect(connection_engine) if insp.has_table("invitations"): invcols = [c["name"] for c in insp.get_columns("invitations")] if "team_id" not in invcols: conn.execute(text("ALTER TABLE invitations ADD COLUMN team_id INTEGER")) # Create tables then migrate legacy SQLite schemas Base.metadata.create_all(bind=engine) run_migrations(engine) def get_db(): db = SessionLocal() try: yield db finally: db.close()