setup the initial app and deploy

.dockerignore

@@ -0,0 +1,10 @@
+.git
+.github
+.pytest_cache
+__pycache__
+*.pyc
+.env
+.env.example
+extension
+tests
+flowpilot-vertex-key.json

.env

@@ -0,0 +1,7 @@
+AI_PROVIDER=vertex_ai
+ALLOW_ORIGINS=*
+GMAIL_POLL_SECONDS=30
+VERTEX_PROJECT_ID=flow-pilot-493104
+VERTEX_LOCATION=us-central1
+VERTEX_MODEL=gemini-2.5-pro
+GOOGLE_APPLICATION_CREDENTIALS=

.env.example

@@ -0,0 +1,7 @@
+AI_PROVIDER=vertex_ai
+ALLOW_ORIGINS=*
+GMAIL_POLL_SECONDS=30
+VERTEX_PROJECT_ID=your-gcp-project-id
+VERTEX_LOCATION=us-central1
+VERTEX_MODEL=gemini-2.5-pro
+GOOGLE_APPLICATION_CREDENTIALS=

.github/workflows/deploy.yml

@@ -0,0 +1,61 @@
+name: Deploy Backend
+
+on:
+  push:
+    branches:
+      - main
+  workflow_dispatch:
+
+env:
+  PROJECT_ID: flow-pilot-493104
+  PROJECT_NUMBER: "707525727191"
+  REGION: us-central1
+  SERVICE: flowpilot-backend
+  REPOSITORY: flowpilot
+  IMAGE: backend
+  WIF_PROVIDER: projects/707525727191/locations/global/workloadIdentityPools/github-pool/providers/github-provider
+  SERVICE_ACCOUNT: github-deployer@flow-pilot-493104.iam.gserviceaccount.com
+
+jobs:
+  deploy:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      id-token: write
+
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Authenticate to Google Cloud
+        uses: google-github-actions/auth@v3
+        with:
+          workload_identity_provider: ${{ env.WIF_PROVIDER }}
+          service_account: ${{ env.SERVICE_ACCOUNT }}
+
+      - name: Setup gcloud
+        uses: google-github-actions/setup-gcloud@v2
+
+      - name: Configure Docker auth
+        run: gcloud auth configure-docker ${{ env.REGION }}-docker.pkg.dev --quiet
+
+      - name: Build image
+        run: |
+          docker build \
+            -t ${{ env.REGION }}-docker.pkg.dev/${{ env.PROJECT_ID }}/${{ env.REPOSITORY }}/${{ env.IMAGE }}:${{ github.sha }} \
+            .
+
+      - name: Push image
+        run: |
+          docker push \
+            ${{ env.REGION }}-docker.pkg.dev/${{ env.PROJECT_ID }}/${{ env.REPOSITORY }}/${{ env.IMAGE }}:${{ github.sha }}
+
+      - name: Deploy to Cloud Run
+        uses: google-github-actions/deploy-cloudrun@v3
+        with:
+          service: ${{ env.SERVICE }}
+          region: ${{ env.REGION }}
+          image: ${{ env.REGION }}-docker.pkg.dev/${{ env.PROJECT_ID }}/${{ env.REPOSITORY }}/${{ env.IMAGE }}:${{ github.sha }}
+          flags: >-
+            --allow-unauthenticated
+            --set-env-vars=AI_PROVIDER=vertex_ai,VERTEX_PROJECT_ID=${{ env.PROJECT_ID }},VERTEX_LOCATION=${{ env.REGION }},VERTEX_MODEL=gemini-2.5-pro

Dockerfile

@@ -0,0 +1,14 @@
+FROM python:3.12-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+ENV PORT=8080
+
+WORKDIR /app
+
+COPY backend/requirements.txt /app/backend/requirements.txt
+RUN pip install --no-cache-dir -r /app/backend/requirements.txt
+
+COPY backend /app/backend
+
+CMD ["sh", "-c", "uvicorn backend.main:app --host 0.0.0.0 --port ${PORT:-8080}"]

README.md

@@ -1 +1,78 @@
-# flow-pilot
+# FlowPilot
+
+FlowPilot is a Gmail-first automation layer for small business owners. The current scaffold includes:
+
+- A FastAPI backend with analysis, workflow suggestion, workflow build, deploy, upload, status, and escalation endpoints.
+- A primitive-based workflow engine that compiles and executes JSON workflows.
+- A Chrome extension scaffold that injects a sidebar into Gmail and walks through the onboarding-to-live journey.
+- Lightweight in-memory storage and tests for the core owner flow.
+
+## Project Structure
+
+```text
+backend/
+extension/
+tests/
+```
+
+## Backend
+
+Run locally:
+
+```bash
+pip install -r backend/requirements.txt
+uvicorn backend.main:app --reload
+```
+
+Create a `.env` in the repo root first. FlowPilot is now wired to read Vertex AI settings from that file.
+
+```bash
+cp .env.example .env
+```
+
+## Extension
+
+Load `extension/` as an unpacked Chrome extension. The sidebar injects into Gmail and points to `http://localhost:8000/api`.
+
+## Deploying
+
+The repo includes a Cloud Run GitHub Actions workflow at [`.github/workflows/deploy.yml`](/Users/sivasankernp/Desktop/flow-pilot/.github/workflows/deploy.yml:1).
+
+Before the action will work, finish these one-time Google Cloud steps:
+
+```bash
+gcloud iam workload-identity-pools create github-pool \
+  --project=flow-pilot-493104 \
+  --location=global \
+  --display-name="GitHub Pool"
+```
+
+```bash
+gcloud iam workload-identity-pools providers create-oidc github-provider \
+  --project=flow-pilot-493104 \
+  --location=global \
+  --workload-identity-pool=github-pool \
+  --display-name="GitHub Provider" \
+  --issuer-uri="https://token.actions.githubusercontent.com" \
+  --attribute-mapping="google.subject=assertion.sub,attribute.repository=assertion.repository,attribute.actor=assertion.actor"
+```
+
+```bash
+gcloud iam service-accounts add-iam-policy-binding \
+  github-deployer@flow-pilot-493104.iam.gserviceaccount.com \
+  --project=flow-pilot-493104 \
+  --role="roles/iam.workloadIdentityUser" \
+  --member="principalSet://iam.googleapis.com/projects/707525727191/locations/global/workloadIdentityPools/github-pool/attribute.repository/DevelopedBy-Siva/flow-pilot"
+```
+
+The workflow builds the Docker image, pushes it to Artifact Registry, and deploys the FastAPI backend to Cloud Run.
+
+## Notes
+
+- Backend storage is currently in-memory for demo speed; the folder layout is ready for SQLite migration work.
+- AI services are still mocked for workflow behavior, but config is now set up for Vertex AI credentials and model selection.
+- Fill in your real Vertex values in `.env`, especially `VERTEX_PROJECT_ID` and `VERTEX_LOCATION`.
+- If you authenticate with `gcloud auth application-default login`, leave `GOOGLE_APPLICATION_CREDENTIALS` blank.
+- Only set `GOOGLE_APPLICATION_CREDENTIALS` when you have a real service account JSON path available.
+- The backend now attempts live Vertex AI calls through the `google-genai` SDK when credentials are configured.
+- If the SDK is missing or `VERTEX_PROJECT_ID` is still a placeholder, FlowPilot falls back to local deterministic mock logic so tests and scaffolding still work.

backend/__init__.py

@@ -0,0 +1 @@
+"""FlowPilot backend package."""

backend/ai/__init__.py

@@ -0,0 +1 @@
+"""AI helpers."""

backend/ai/analyzer.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+
+from backend.ai.client import render_prompt, vertex_client
+from backend.ai.prompts import ANALYZE_PROMPT, CUSTOM_TASK_PROMPT
+from backend.models.schemas import TaskCategory
+
+
+def analyze_business_description(description: str) -> dict:
+    if vertex_client.is_ready():
+        return vertex_client.generate_json(render_prompt(ANALYZE_PROMPT, description=description))
+
+    lower = description.lower()
+    fully_automatable = []
+    ai_assisted = []
+    manual = []
+    if "order" in lower:
+        fully_automatable.append(
+            {
+                "id": "task_order_processing",
+                "name": "Order processing",
+                "description": "Read order emails, check stock, log rows, and reply.",
+                "time_estimate": "10-15 hours/week",
+                "pain_level": "high",
+            }
+        )
+    if "friday" in lower or "weekly" in lower:
+        fully_automatable.append(
+            {
+                "id": "task_weekly_summary",
+                "name": "Weekly order summary report",
+                "description": "Compile the week's orders and send a summary.",
+                "time_estimate": "1-2 hours/week",
+                "pain_level": "medium",
+            }
+        )
+    ai_assisted.append(
+        {
+            "id": "task_availability_inquiries",
+            "name": "Availability inquiries",
+            "description": "Draft replies using current stock and pricing.",
+            "time_estimate": "2-3 hours/week",
+            "pain_level": "medium",
+        }
+    )
+    ai_assisted.append(
+        {
+            "id": "task_low_inventory_alerts",
+            "name": "Low inventory alerts",
+            "description": "Watch for thin inventory and propose next steps.",
+            "time_estimate": "30 minutes/week",
+            "pain_level": "medium",
+        }
+    )
+    manual.append(
+        {
+            "id": "task_special_requests",
+            "name": "Custom and relationship-sensitive requests",
+            "description": "Gift boxes, delivery arrangements, and VIP relationship notes.",
+            "reason_manual": "These need human judgment, pricing discretion, and a personal touch.",
+        }
+    )
+    return {
+        "business_type": "Email-driven small business",
+        "summary": "FlowPilot found repetitive inbox work that maps well to Gmail, Sheets, and scheduled automations.",
+        "tasks": {
+            "fully_automatable": fully_automatable,
+            "ai_assisted": ai_assisted,
+            "manual": manual,
+        },
+    }
+
+
+def analyze_custom_task(business_description: str, existing_workflows: list[dict], custom_task: str) -> dict:
+    if vertex_client.is_ready():
+        return vertex_client.generate_json(
+            render_prompt(
+                CUSTOM_TASK_PROMPT,
+                business_description=business_description,
+                existing_workflows=existing_workflows,
+                primitives_list=[
+                    "email_received",
+                    "schedule",
+                    "file_uploaded",
+                    "manual_trigger",
+                    "ai_extract",
+                    "ai_classify",
+                    "ai_compose",
+                    "sheets_read",
+                    "sheets_write",
+                    "sheets_update",
+                    "send_email",
+                    "notify_owner",
+                    "condition",
+                    "loop",
+                    "wait_for_input",
+                ],
+                custom_task=custom_task,
+            )
+        )
+
+    lower = custom_task.lower()
+    feasible = any(word in lower for word in ("flag", "priority", "sheet", "email"))
+    category = TaskCategory.FULLY_AUTOMATABLE if feasible else TaskCategory.MANUAL
+    return {
+        "feasible": feasible,
+        "category": category,
+        "task_name": "Restaurant priority handling",
+        "task_description": custom_task,
+        "missing_capabilities": [] if feasible else ["A matching primitive for this request is missing."],
+        "reason_if_manual": None if feasible else "This task needs a capability outside the current primitives.",
+        "time_estimate": "1 hour/week",
+        "pain_level": "medium",
+    }

backend/ai/classifier.py

@@ -0,0 +1,27 @@
+from backend.ai.client import render_prompt, vertex_client
+from backend.ai.prompts import CLASSIFY_PROMPT
+
+
+def classify_email(from_email: str, subject: str, body: str, business_context: str) -> dict:
+    if vertex_client.is_ready():
+        return vertex_client.generate_json(
+            render_prompt(
+                CLASSIFY_PROMPT,
+                categories="order, availability_inquiry, complaint, greeting, other",
+                from_email=from_email,
+                subject=subject,
+                body=body,
+                business_context=business_context,
+            )
+        )
+
+    text = f"{subject} {body}".lower()
+    category = "order" if "order" in text else "availability_inquiry" if "available" in text else "other"
+    priority = "high" if "restaurant" in from_email.lower() else "medium"
+    return {
+        "category": category,
+        "priority": priority,
+        "requires_response": True,
+        "sentiment": "neutral",
+        "reasoning": "Derived from keywords and sender profile.",
+    }

backend/ai/client.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+import json
+import os
+from typing import Any
+
+from backend.config import get_settings
+
+try:
+    from google import genai
+except ImportError:  # pragma: no cover - depends on optional package
+    genai = None
+
+
+class VertexAIClient:
+    def __init__(self) -> None:
+        self.settings = get_settings()
+        self._client = None
+
+    def generate_json(self, prompt: str) -> dict[str, Any]:
+        text = self.generate_text(prompt)
+        return _extract_json_object(text)
+
+    def generate_text(self, prompt: str) -> str:
+        client = self._get_client()
+        response = client.models.generate_content(
+            model=self.settings.vertex_model,
+            contents=prompt,
+        )
+        text = getattr(response, "text", None)
+        if not text:
+            raise RuntimeError("Vertex AI returned an empty response")
+        return text.strip()
+
+    def is_ready(self) -> bool:
+        placeholders = {"your-gcp-project-id"}
+        if self.settings.ai_provider != "vertex_ai":
+            return False
+        if genai is None:
+            return False
+        if self.settings.vertex_project_id in placeholders:
+            return False
+        credentials_path = self.settings.google_application_credentials.strip()
+        if credentials_path and credentials_path != "/absolute/path/to/service-account.json":
+            return os.path.exists(credentials_path)
+        return True
+
+    def _get_client(self):
+        if genai is None:
+            raise RuntimeError("google-genai is not installed")
+        if self._client is None:
+            self._client = genai.Client(
+                vertexai=True,
+                project=self.settings.vertex_project_id,
+                location=self.settings.vertex_location,
+            )
+        return self._client
+
+
+def _extract_json_object(text: str) -> dict[str, Any]:
+    cleaned = text.strip()
+    if cleaned.startswith("```"):
+        parts = cleaned.split("```")
+        cleaned = next((part for part in parts if "{" in part and "}" in part), cleaned)
+        cleaned = cleaned.replace("json", "", 1).strip()
+    start = cleaned.find("{")
+    end = cleaned.rfind("}")
+    if start == -1 or end == -1 or end < start:
+        raise ValueError("No JSON object found in model response")
+    return json.loads(cleaned[start : end + 1])
+
+
+vertex_client = VertexAIClient()
+
+
+def render_prompt(template: str, **kwargs: Any) -> str:
+    rendered = template
+    for key, value in kwargs.items():
+        rendered = rendered.replace(f"{{{key}}}", str(value))
+    return rendered

backend/ai/composer.py

@@ -0,0 +1,28 @@
+import json
+
+from backend.ai.client import render_prompt, vertex_client
+from backend.ai.prompts import COMPOSE_PROMPT
+
+
+def compose_reply(context: dict, tone: str = "friendly") -> str:
+    if vertex_client.is_ready():
+        return vertex_client.generate_text(
+            render_prompt(
+                COMPOSE_PROMPT,
+                business_name=context.get("business_name", "FlowPilot customer"),
+                business_type=context.get("business_type", "small business"),
+                owner_name=context.get("owner_name", "Raj"),
+                tone=tone,
+                context=json.dumps(context, indent=2),
+                data=json.dumps(context, indent=2),
+            )
+        )
+
+    customer = context.get("customer_name", "there")
+    items = context.get("items", [])
+    item_summary = ", ".join(f"{item['quantity']} x {item['name']}" for item in items) or "your request"
+    return (
+        f"Hi {customer},\n\n"
+        f"Thanks for your order. We have {item_summary} ready and will follow up with pickup details shortly.\n\n"
+        "Raj"
+    )

backend/ai/extractor.py

@@ -0,0 +1,29 @@
+from backend.ai.client import render_prompt, vertex_client
+from backend.ai.prompts import EXTRACT_PROMPT
+
+
+def extract_email_data(from_email: str, subject: str, body: str) -> dict:
+    if vertex_client.is_ready():
+        return vertex_client.generate_json(
+            render_prompt(
+                EXTRACT_PROMPT,
+                from_email=from_email,
+                subject=subject,
+                body=body,
+            )
+        )
+
+    is_order = "order" in subject.lower() or "order" in body.lower()
+    return {
+        "is_order": is_order,
+        "message_type": "order" if is_order else "question",
+        "confidence": 0.91 if is_order else 0.72,
+        "customer_name": from_email.split("@")[0].replace(".", " ").title(),
+        "customer_email": from_email,
+        "items": [{"name": "Honeycrisp apples", "quantity": 2}] if is_order else [],
+        "pickup_date": "next available pickup",
+        "special_requests": "",
+        "needs_clarification": not is_order,
+        "clarification_question": "Can you confirm the items and quantities?" if not is_order else "",
+        "suggested_response": "Thanks for the note. I can help with that.",
+    }

backend/ai/prompts.py

@@ -0,0 +1,313 @@
+ANALYZE_PROMPT = """
+You are a business process analyst for small businesses.
+
+The owner described their business and daily operations below.
+Analyze it and categorize every task/operation into exactly
+three categories:
+
+1. FULLY_AUTOMATABLE - tasks that can run without human
+   involvement using these capabilities:
+   - Reading and understanding emails
+   - Extracting structured data from text
+   - Reading/writing Google Sheets
+   - Sending email replies
+   - Running on a schedule
+   - Processing uploaded files (CSV/Excel/JSON)
+
+2. AI_ASSISTED - tasks where AI can do the heavy lifting
+   but a human should review or approve before action is taken.
+   This includes anything with ambiguity, financial decisions,
+   or relationship sensitivity.
+
+3. MANUAL - tasks that require human creativity, judgment,
+   physical presence, or personal touch. Explain WHY each
+   manual task can't be automated.
+
+For each task also provide:
+- time_estimate: how much time this task takes per week
+- pain_level: low/medium/high
+- description: one line explaining the task
+
+Respond in this exact JSON format:
+{
+  "business_type": "...",
+  "summary": "...",
+  "tasks": {
+    "fully_automatable": [
+      {
+        "id": "task_1",
+        "name": "...",
+        "description": "...",
+        "time_estimate": "...",
+        "pain_level": "high"
+      }
+    ],
+    "ai_assisted": [...],
+    "manual": [
+      {
+        "id": "task_5",
+        "name": "...",
+        "description": "...",
+        "reason_manual": "..."
+      }
+    ]
+  }
+}
+
+No markdown. No explanation. JSON only.
+
+Owner's description:
+{description}
+""".strip()
+
+SUGGEST_WORKFLOWS_PROMPT = """
+You are a workflow architect for small business automation.
+
+The owner selected this task to automate:
+Task: {task_name}
+Description: {task_description}
+Category: {category}
+
+Available primitives for building workflows:
+TRIGGERS: email_received, schedule, file_uploaded, manual_trigger
+AI ACTIONS: ai_extract, ai_classify, ai_compose
+DATA ACTIONS: sheets_read, sheets_write, sheets_update
+COMM ACTIONS: send_email, notify_owner
+LOGIC: condition, loop, wait_for_input
+
+The owner's tools:
+- Gmail (for email)
+- Google Sheets (spreadsheet: {spreadsheet_info})
+- Uploaded data files: {uploaded_files}
+
+Generate exactly 3 workflow options with DIFFERENT levels
+of automation and human involvement:
+
+Option A - Maximum automation (owner does nothing)
+Option B - Balanced (AI handles most, owner approves key steps)
+Option C - Minimal (AI organizes/prepares, owner acts)
+
+Respond in this exact JSON format:
+{
+  "options": [
+    {
+      "id": "A",
+      "name": "...",
+      "approach": "full_auto | balanced | minimal",
+      "steps_summary": [
+        "Step 1: ...",
+        "Step 2: ...",
+        "Step 3: ..."
+      ],
+      "owner_involvement": "what the owner does (or 'nothing')",
+      "time_saved": "estimated time saved per week",
+      "risk_note": "what could go wrong"
+    }
+  ]
+}
+
+No markdown. JSON only.
+""".strip()
+
+BUILD_WORKFLOW_PROMPT = """
+You are a workflow compiler. Convert the selected workflow
+option into an executable workflow JSON.
+
+Selected option:
+{selected_option}
+
+Task: {task_name}
+Owner config:
+  - email: {owner_email}
+  - spreadsheet_id: {spreadsheet_id}
+  - inventory_sheet: {inventory_sheet}
+  - orders_sheet: {orders_sheet}
+  - item_column: {item_column}
+  - stock_column: {stock_column}
+  - uploaded data: {uploaded_data_summary}
+
+AVAILABLE PRIMITIVES (you can ONLY use these):
+
+TRIGGERS:
+  email_received - fires when a new email arrives
+  schedule - fires on a cron schedule
+  file_uploaded - fires when owner uploads a file
+  manual_trigger - fires when owner clicks "run now"
+
+AI ACTIONS:
+  ai_extract - send text to AI, get structured JSON back
+    params: input, extract_fields, output_var
+  ai_classify - send text to AI, get category back
+    params: input, categories, output_var
+  ai_compose - send context to AI, get composed text back
+    params: context, tone, output_var
+
+DATA ACTIONS:
+  sheets_read - read rows from a Google Sheet
+    params: spreadsheet_id, sheet_name, filter (optional), output_var
+  sheets_write - append a new row to a sheet
+    params: spreadsheet_id, sheet_name, data
+  sheets_update - update existing cells
+    params: spreadsheet_id, sheet_name, lookup_column,
+            lookup_value, update_column, update_value
+
+COMM ACTIONS:
+  send_email - send an email via Gmail
+    params: to, subject, body
+  notify_owner - send alert to owner's extension sidebar
+    params: message, severity (info/warning/critical),
+            options (optional - clickable choices)
+
+LOGIC:
+  condition - if/else branching
+    params: check, if_true (step_id), if_false (step_id)
+  loop - iterate over a list
+    params: items, step_id (executed for each item)
+  wait_for_input - pause workflow until owner responds
+    params: prompt, options, output_var
+
+VARIABLE REFERENCES:
+  Use {{step_id.output_var}} to reference output from
+  a previous step.
+  Use {{trigger.email.from}}, {{trigger.email.subject}},
+  {{trigger.email.body}} for email trigger data.
+  Use {{config.xxx}} for owner config values.
+
+Generate a valid workflow JSON following this exact schema:
+{
+  "name": "...",
+  "description": "...",
+  "trigger": {
+    "type": "email_received | schedule | file_uploaded | manual_trigger",
+    "config": {}
+  },
+  "steps": [
+    {
+      "id": "step_1",
+      "action": "primitive_name",
+      "params": {},
+      "on_error": "notify_owner | skip | abort"
+    }
+  ]
+}
+
+Rules:
+- Every step must have a unique id
+- Every variable reference must point to a step that
+  runs BEFORE the current step
+- If a task needs a primitive that doesn't exist in the
+  list above, DO NOT include it. Add a notify_owner step
+  explaining what needs manual handling.
+- Include on_error for every step
+
+No markdown. No explanation. Valid JSON only.
+""".strip()
+
+EXTRACT_PROMPT = """
+Extract order/request details from this email.
+
+Return JSON:
+{
+  "is_order": true,
+  "message_type": "order | question | complaint | greeting | other",
+  "confidence": 0.0,
+  "customer_name": "...",
+  "customer_email": "...",
+  "items": [{"name": "...", "quantity": 1}],
+  "pickup_date": "...",
+  "special_requests": "...",
+  "needs_clarification": false,
+  "clarification_question": "...",
+  "suggested_response": "..."
+}
+
+If confidence < 0.7, set needs_clarification: true.
+If not an order, set is_order: false and provide
+message_type and suggested_response.
+
+Email:
+From: {from_email}
+Subject: {subject}
+Body: {body}
+""".strip()
+
+CLASSIFY_PROMPT = """
+Classify this email into one of these categories:
+{categories}
+
+Also determine:
+- priority: low/medium/high/critical
+- requires_response: true/false
+- sentiment: positive/neutral/negative
+
+Email:
+From: {from_email}
+Subject: {subject}
+Body: {body}
+
+Context about the business:
+{business_context}
+
+Respond in JSON only:
+{
+  "category": "...",
+  "priority": "...",
+  "requires_response": true,
+  "sentiment": "...",
+  "reasoning": "..."
+}
+""".strip()
+
+COMPOSE_PROMPT = """
+Write an email reply for this business context.
+
+Business: {business_name} ({business_type})
+Owner: {owner_name}
+Tone: {tone}
+
+Context:
+{context}
+
+Data available:
+{data}
+
+Rules:
+- Keep it short and warm
+- Include all relevant details (items, quantities,
+  prices, dates)
+- Match the tone specified
+- Do not include a subject line
+- Do not include "Dear" or overly formal greetings
+- Sign off with the owner's name
+
+Write the email body only.
+""".strip()
+
+CUSTOM_TASK_PROMPT = """
+The business owner described a custom task they want handled.
+
+Their business: {business_description}
+Their existing workflows: {existing_workflows}
+Available primitives: {primitives_list}
+
+Custom task description:
+{custom_task}
+
+First determine:
+1. Can this task be accomplished using ONLY the
+   available primitives?
+2. If yes, categorize it: fully_automatable or ai_assisted
+3. If no, explain what's missing and categorize as manual
+
+Respond in JSON:
+{
+  "feasible": true,
+  "category": "fully_automatable | ai_assisted | manual",
+  "task_name": "short name",
+  "task_description": "one line",
+  "missing_capabilities": [],
+  "reason_if_manual": null,
+  "time_estimate": "...",
+  "pain_level": "low/medium/high"
+}
+""".strip()

backend/ai/workflow_builder.py

@@ -0,0 +1,142 @@
+import json
+
+from backend.ai.client import render_prompt, vertex_client
+from backend.ai.prompts import BUILD_WORKFLOW_PROMPT
+from backend.models.schemas import BuildWorkflowRequest, WorkflowDefinition, WorkflowStep
+
+
+def build_workflow_definition(request: BuildWorkflowRequest, owner: dict) -> dict:
+    if vertex_client.is_ready():
+        prompt = render_prompt(
+            BUILD_WORKFLOW_PROMPT,
+            selected_option=json.dumps(request.selected_option, indent=2),
+            task_name=request.task_name,
+            owner_email=owner.get("email", "owner@example.com"),
+            spreadsheet_id=owner.get("spreadsheet_id", ""),
+            inventory_sheet=owner.get("spreadsheet_config", {}).get("inventory_sheet", "Inventory"),
+            orders_sheet=owner.get("spreadsheet_config", {}).get("orders_sheet", "Orders"),
+            item_column="item",
+            stock_column="stock",
+            uploaded_data_summary=json.dumps(owner.get("uploaded_data_summary", []), indent=2),
+        )
+        return vertex_client.generate_json(prompt)
+
+    trigger_type = "schedule" if "summary" in request.task_name.lower() else "email_received"
+    steps: list[WorkflowStep] = []
+    if trigger_type == "email_received":
+        steps = [
+            WorkflowStep(
+                id="step_extract",
+                action="ai_extract",
+                params={
+                    "input": "{{trigger.email.body}}",
+                    "extract_fields": ["customer_name", "items", "pickup_date", "needs_clarification"],
+                    "output_var": "order_data",
+                },
+                on_error="notify_owner",
+            ),
+            WorkflowStep(
+                id="step_read_inventory",
+                action="sheets_read",
+                params={
+                    "spreadsheet_id": "{{config.spreadsheet_id}}",
+                    "sheet_name": "Inventory",
+                    "output_var": "inventory_rows",
+                },
+                on_error="abort",
+            ),
+            WorkflowStep(
+                id="step_check_stock",
+                action="condition",
+                params={
+                    "check": "not {{step_extract.order_data.needs_clarification}}",
+                    "if_true": "step_write_order",
+                    "if_false": "step_notify_ambiguity",
+                },
+                on_error="abort",
+            ),
+            WorkflowStep(
+                id="step_write_order",
+                action="sheets_write",
+                params={
+                    "spreadsheet_id": "{{config.spreadsheet_id}}",
+                    "sheet_name": "Orders",
+                    "data": {
+                        "customer": "{{step_extract.order_data.customer_name}}",
+                        "items": "{{step_extract.order_data.items}}",
+                        "priority": "high" if "restaurant" in request.task_name.lower() else "normal",
+                    },
+                },
+                on_error="notify_owner",
+            ),
+            WorkflowStep(
+                id="step_compose_reply",
+                action="ai_compose",
+                params={
+                    "context": "{{step_extract.order_data}}",
+                    "tone": owner.get("preferred_tone", "friendly"),
+                    "output_var": "reply_body",
+                },
+                on_error="notify_owner",
+            ),
+            WorkflowStep(
+                id="step_send_email",
+                action="send_email",
+                params={
+                    "to": "{{trigger.email.from}}",
+                    "subject": "Order confirmation",
+                    "body": "{{step_compose_reply.reply_body}}",
+                },
+                on_error="notify_owner",
+            ),
+            WorkflowStep(
+                id="step_notify_ambiguity",
+                action="notify_owner",
+                params={
+                    "message": "This order needs clarification before FlowPilot can continue.",
+                    "severity": "warning",
+                    "options": ["Reply myself", "Ask customer for details"],
+                },
+                on_error="skip",
+            ),
+        ]
+    else:
+        steps = [
+            WorkflowStep(
+                id="step_collect_orders",
+                action="sheets_read",
+                params={
+                    "spreadsheet_id": "{{config.spreadsheet_id}}",
+                    "sheet_name": "Orders",
+                    "output_var": "order_rows",
+                },
+                on_error="abort",
+            ),
+            WorkflowStep(
+                id="step_compose_summary",
+                action="ai_compose",
+                params={
+                    "context": "{{step_collect_orders.order_rows}}",
+                    "tone": "helpful",
+                    "output_var": "summary_body",
+                },
+                on_error="notify_owner",
+            ),
+            WorkflowStep(
+                id="step_send_summary",
+                action="send_email",
+                params={
+                    "to": owner.get("email", "owner@example.com"),
+                    "subject": "Weekly order summary",
+                    "body": "{{step_compose_summary.summary_body}}",
+                },
+                on_error="notify_owner",
+            ),
+        ]
+    workflow = WorkflowDefinition(
+        name=request.task_name,
+        description=request.selected_option.get("name", request.task_description),
+        trigger={"type": trigger_type, "config": {"cron": "0 8 * * FRI"} if trigger_type == "schedule" else {}},
+        steps=steps,
+    )
+    return workflow.model_dump()

backend/ai/workflow_suggester.py

@@ -0,0 +1,72 @@
+from backend.ai.client import render_prompt, vertex_client
+from backend.ai.prompts import SUGGEST_WORKFLOWS_PROMPT
+
+
+def suggest_workflow_options(
+    task_name: str,
+    task_description: str,
+    category: str,
+    spreadsheet_info: dict,
+    uploaded_files: list[dict],
+) -> dict:
+    if vertex_client.is_ready():
+        return vertex_client.generate_json(
+            render_prompt(
+                SUGGEST_WORKFLOWS_PROMPT,
+                task_name=task_name,
+                task_description=task_description,
+                category=category,
+                spreadsheet_info=spreadsheet_info,
+                uploaded_files=uploaded_files,
+            )
+        )
+
+    return {
+        "options": [
+            {
+                "id": "A",
+                "name": f"{task_name} Full Auto",
+                "approach": "full_auto",
+                "steps_summary": [
+                    "Trigger from a new input automatically.",
+                    "Use AI to extract or classify the content.",
+                    "Update Sheets and send the output without waiting.",
+                ],
+                "owner_involvement": "nothing unless FlowPilot raises an alert",
+                "time_saved": "5-10 hours/week",
+                "risk_note": "Incorrect extraction could act before review.",
+            },
+            {
+                "id": "B",
+                "name": f"{task_name} Review First",
+                "approach": "balanced",
+                "steps_summary": [
+                    "Trigger from inbox or schedule.",
+                    "AI prepares the work and checks current data.",
+                    "Owner approves before write-back or reply.",
+                ],
+                "owner_involvement": "one approval on important actions",
+                "time_saved": "3-6 hours/week",
+                "risk_note": "Still depends on owner response time.",
+            },
+            {
+                "id": "C",
+                "name": f"{task_name} Log Only",
+                "approach": "minimal",
+                "steps_summary": [
+                    "Trigger from the chosen source.",
+                    "Extract or summarize the information.",
+                    "Record it for the owner to finish manually.",
+                ],
+                "owner_involvement": "send final reply or finish task manually",
+                "time_saved": "1-3 hours/week",
+                "risk_note": "Most value still depends on manual follow-through.",
+            },
+        ],
+        "context": {
+            "category": category,
+            "spreadsheet_connected": spreadsheet_info.get("connected", False),
+            "uploaded_files": [file["filename"] for file in uploaded_files],
+            "task_description": task_description,
+        },
+    }

backend/api/__init__.py

@@ -0,0 +1 @@
+"""API package."""

backend/api/middleware.py

@@ -0,0 +1,6 @@
+from fastapi import HTTPException
+
+
+def ensure_owner(owner_id: str | None) -> None:
+    if not owner_id:
+        raise HTTPException(status_code=400, detail="owner_id is required")

backend/api/routes.py

@@ -0,0 +1,121 @@
+from fastapi import APIRouter, HTTPException
+
+from backend.ai.analyzer import analyze_business_description, analyze_custom_task
+from backend.ai.workflow_builder import build_workflow_definition
+from backend.ai.workflow_suggester import suggest_workflow_options
+from backend.engine.compiler import compile_workflow
+from backend.engine.executor import WorkflowExecutor
+from backend.integrations.file_parser import parse_uploaded_payload
+from backend.models.schemas import (
+    AnalyzeRequest,
+    BuildWorkflowRequest,
+    BuildWorkflowResponse,
+    CustomTaskRequest,
+    DeployRequest,
+    DeploymentResponse,
+    EscalationReplyRequest,
+    EscalationResponse,
+    FileUploadRequest,
+    FileUploadResponse,
+    OwnerStatusResponse,
+    WorkflowSuggestionRequest,
+)
+from backend.storage.database import db
+
+router = APIRouter()
+executor = WorkflowExecutor()
+
+
+@router.post("/analyze")
+def analyze(request: AnalyzeRequest) -> dict:
+    analysis = analyze_business_description(request.description)
+    owner = db.ensure_owner(request.owner_id, request.owner_email)
+    owner["business_description"] = request.description
+    owner["business_analysis"] = analysis
+    db.save_owner(owner)
+    return analysis
+
+
+@router.post("/custom-task")
+def custom_task(request: CustomTaskRequest) -> dict:
+    owner = db.get_owner(request.owner_id)
+    return analyze_custom_task(
+        business_description=owner.get("business_description", ""),
+        existing_workflows=db.list_workflows(request.owner_id),
+        custom_task=request.custom_task,
+    )
+
+
+@router.post("/suggest-workflows")
+def suggest_workflows(request: WorkflowSuggestionRequest) -> dict:
+    owner = db.get_owner(request.owner_id)
+    return suggest_workflow_options(
+        task_name=request.task_name,
+        task_description=request.task_description,
+        category=request.category,
+        spreadsheet_info=owner.get("spreadsheet_config", {"connected": False}),
+        uploaded_files=db.list_data_files(request.owner_id),
+    )
+
+
+@router.post("/build-workflow", response_model=BuildWorkflowResponse)
+def build_workflow(request: BuildWorkflowRequest) -> BuildWorkflowResponse:
+    owner = db.get_owner(request.owner_id)
+    workflow_json = build_workflow_definition(request=request, owner=owner)
+    compiled = compile_workflow(workflow_json)
+    return BuildWorkflowResponse(workflow=compiled)
+
+
+@router.post("/deploy", response_model=DeploymentResponse)
+def deploy_workflow(request: DeployRequest) -> DeploymentResponse:
+    owner = db.get_owner(request.owner_id)
+    if not owner:
+        raise HTTPException(status_code=404, detail="owner not found")
+    deployments = []
+    for workflow in request.workflows:
+        compiled = compile_workflow(workflow.model_dump())
+        saved = db.save_workflow(request.owner_id, compiled)
+        deployments.append(saved)
+    owner["state"] = "live"
+    db.save_owner(owner)
+    return DeploymentResponse(
+        status="deployed",
+        workflows=deployments,
+        message="All workflows are live.",
+    )
+
+
+@router.post("/upload-data", response_model=FileUploadResponse)
+def upload_data(request: FileUploadRequest) -> FileUploadResponse:
+    parsed = parse_uploaded_payload(request.filename, request.content, request.purpose)
+    record = db.save_data_file(request.owner_id, request.filename, request.file_type, request.purpose, parsed)
+    return FileUploadResponse(file=record)
+
+
+@router.get("/status", response_model=OwnerStatusResponse)
+def status(owner_id: str) -> OwnerStatusResponse:
+    owner = db.get_owner(owner_id)
+    return OwnerStatusResponse(
+        owner=owner,
+        workflows=db.list_workflows(owner_id),
+        recent_executions=db.list_execution_logs(owner_id),
+    )
+
+
+@router.get("/escalations")
+def escalations(owner_id: str) -> dict:
+    return {"items": db.list_escalations(owner_id)}
+
+
+@router.post("/escalation-reply", response_model=EscalationResponse)
+def escalation_reply(request: EscalationReplyRequest) -> EscalationResponse:
+    escalation = db.resolve_escalation(request.escalation_id, request.response)
+    return EscalationResponse(escalation=escalation)
+
+
+@router.post("/simulate-run")
+def simulate_run(owner_id: str, workflow_id: str, trigger: dict) -> dict:
+    workflow = db.get_workflow(owner_id, workflow_id)
+    result = executor.execute(workflow, trigger, db=db, owner_id=owner_id)
+    db.save_execution_log(owner_id, workflow_id, trigger, result["steps"], result["outcome"], result.get("error"))
+    return result

backend/config.py

@@ -0,0 +1,32 @@
+import os
+from functools import lru_cache
+
+from pydantic import BaseModel, Field
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+class Settings(BaseModel):
+    app_name: str = "FlowPilot"
+    api_prefix: str = "/api"
+    database_url: str = "sqlite:///./flowpilot.db"
+    allow_origins: list[str] = Field(default_factory=lambda: _split_csv(os.getenv("ALLOW_ORIGINS", "*")))
+    ai_provider: str = os.getenv("AI_PROVIDER", "vertex_ai")
+    gmail_poll_seconds: int = int(os.getenv("GMAIL_POLL_SECONDS", "30"))
+    vertex_project_id: str = os.getenv("VERTEX_PROJECT_ID", "your-gcp-project-id")
+    vertex_location: str = os.getenv("VERTEX_LOCATION", "us-central1")
+    vertex_model: str = os.getenv("VERTEX_MODEL", "gemini-2.5-pro")
+    google_application_credentials: str = os.getenv(
+        "GOOGLE_APPLICATION_CREDENTIALS",
+        "",
+    )
+
+
+def _split_csv(value: str) -> list[str]:
+    return [item.strip() for item in value.split(",") if item.strip()]
+
+
+@lru_cache
+def get_settings() -> Settings:
+    return Settings()

backend/engine/__init__.py

@@ -0,0 +1,5 @@
+"""Workflow engine exports."""
+
+from backend.engine.compiler import compile_workflow
+
+__all__ = ["compile_workflow"]