#!/usr/bin/env python3 """ CLOUDFLARE BUILD SWARM — Phase 2 ================================= Gemini agents that BUILD actual SubAIVA deployment tooling. Unlike Phase 1 (research), Phase 2 produces deployable code. Missions: 1. Stripe webhook handler for tier management 2. Admin dashboard API endpoints 3. Analytics and usage metering Worker 4. AI Gateway configuration for caching/rate-limiting 5. E2E test suite for deployed SubAIVAs 6. wrangler deploy automation script 7. Customer onboarding flow (provision + welcome email via Resend) Usage: python3 scripts/cloudflare_build_swarm.py python3 scripts/cloudflare_build_swarm.py --missions 1,3,5 """ import asyncio import json import sys import os import time from pathlib import Path from datetime import datetime sys.path.insert(0, str(Path(__file__).parent.parent)) from core.gemini_executor import GeminiExecutor # Reference code context for all build agents SUBAIVA_TYPES = """ // src/types.ts — Key types for SubAIVA export interface Env { SUBAIVA: DurableObjectNamespace; MEMORY_VECTORS?: VectorizeIndex; AI?: Ai; ENTITLEMENTS?: KVNamespace; API_TOKEN: string; DEFAULT_MODEL: string; EMBEDDING_MODEL: string; ENVIRONMENT: string; } export const TIER_CONFIGS: Record = { 1: { tier: 1, name: "Starter", maxMemories: 500, maxConversationHistory: 50, hasVoice: false, hasBrowser: false, hasIDE: false, hasMCP: false, model: "@cf/meta/llama-3.1-8b-instruct", memoryDecayDays: 30 }, 2: { tier: 2, name: "Professional", maxMemories: 5000, maxConversationHistory: 200, hasVoice: false, hasBrowser: false, hasIDE: false, hasMCP: false, model: "@cf/meta/llama-3.3-70b-instruct-fp8-fast", memoryDecayDays: 90 }, 3: { tier: 3, name: "Enterprise", maxMemories: 50000, maxConversationHistory: 1000, hasVoice: true, hasBrowser: false, hasIDE: false, hasMCP: false, model: "@cf/meta/llama-3.3-70b-instruct-fp8-fast", memoryDecayDays: 0 }, }; """ MISSIONS = [ { "id": 1, "name": "Stripe_Webhook_Handler", "model": "flash", "prompt": f"""You are a TypeScript expert building a Stripe webhook handler for Cloudflare Workers. CONTEXT: SubAIVA is a multi-tenant AI agent platform where each customer gets a Durable Object. Tiers: - Tier 1 Starter ($497 AUD/mo) — 8B model, 500 memories - Tier 2 Professional ($997 AUD/mo) — 70B model, 5000 memories - Tier 3 Enterprise ($1,497 AUD/mo) — 70B model, 50K memories, voice {SUBAIVA_TYPES} Write a COMPLETE TypeScript file `src/stripe-webhook.ts` that: 1. Handles Stripe webhook events (checkout.session.completed, customer.subscription.updated, customer.subscription.deleted, invoice.payment_succeeded, invoice.payment_failed) 2. Verifies webhook signature using Stripe's crypto verification (Web Crypto API, no Node.js crypto) 3. On subscription created: provisions a new SubAIVA DO (calls /api/provision internally) 4. On subscription updated: updates tier via /api/:customerId/tier 5. On subscription deleted: marks customer inactive in KV 6. On payment failed: downgrades tier or suspends 7. Maps Stripe price IDs to tier numbers via a config object 8. Stores webhook events in a KV namespace for audit trail 9. Returns proper 200 responses (Stripe retries on non-200) Include the Hono route registration code that would go in index.ts. Include the wrangler.jsonc additions needed (STRIPE_WEBHOOK_SECRET var). Output ONLY the TypeScript code with comments. No markdown headers or explanations outside code. """, "output_file": "subaiva-do/src/stripe-webhook.ts", }, { "id": 2, "name": "Admin_Dashboard_API", "model": "flash", "prompt": f"""You are a TypeScript expert building an admin dashboard API for Cloudflare Workers. CONTEXT: SubAIVA is a multi-tenant AI agent platform. We need admin endpoints for the Genesis team to manage all customer SubAIVAs. {SUBAIVA_TYPES} Write a COMPLETE TypeScript file `src/admin.ts` that exports Hono routes for: 1. GET /admin/customers — list all customers (from KV scan with prefix "customer:") 2. GET /admin/customers/:id — get detailed customer status (calls DO getStatus) 3. POST /admin/customers/:id/message — send a message as admin to customer's SubAIVA 4. DELETE /admin/customers/:id — deactivate a customer (mark inactive in KV, don't delete DO data) 5. GET /admin/metrics — aggregate metrics (total customers, total messages, customers per tier) 6. POST /admin/broadcast — send a system message to all customers' DOs 7. GET /admin/health — deep health check (check KV, Vectorize, AI bindings) Requirements: - Separate admin auth (ADMIN_TOKEN env var, not the regular API_TOKEN) - All responses are JSON - Pagination support on /admin/customers (cursor-based, KV list supports this) - Rate limit admin endpoints to 100 req/min - Include error handling Output ONLY the TypeScript code. Include the Hono sub-app export and route registration snippet for index.ts. """, "output_file": "subaiva-do/src/admin.ts", }, { "id": 3, "name": "Usage_Metering_Worker", "model": "flash", "prompt": f"""You are a TypeScript expert building a usage metering system for Cloudflare Workers. CONTEXT: SubAIVA charges per tier ($497/$997/$1,497 AUD/mo). We need to track usage per customer for: - Tier enforcement (cut off at limits) - Stripe usage-based billing potential - Analytics dashboard - Abuse detection {SUBAIVA_TYPES} Write a COMPLETE TypeScript file `src/metering.ts` that: 1. Defines a UsageMeter class that tracks per-customer: - Messages sent today / this month - Memories stored today / this month - Tokens consumed (estimated from message length) - Memory search queries today / this month - WebSocket connection hours this month 2. Storage: Use DO SQLite (add a `usage_events` table and `usage_daily` aggregation table) 3. Provides methods: - recordEvent(customerId, eventType, metadata) — fire-and-forget - getUsage(customerId, period: 'today' | 'month') — returns usage summary - checkLimit(customerId, eventType) — returns object with 'allowed' boolean and 'remaining' number - getDailyAggregates(customerId, days: number) — returns daily rollups 4. Tier limits: - Tier 1: 100 messages/day, 10 memory stores/day, 50 searches/day - Tier 2: 500 messages/day, 50 memory stores/day, 200 searches/day - Tier 3: 2000 messages/day, 200 memory stores/day, unlimited searches 5. Integration: Show how to add metering calls to the existing SubAIVA.chat() and SubAIVA.storeMemory() methods Output ONLY the TypeScript code with clear comments. """, "output_file": "subaiva-do/src/metering.ts", }, { "id": 4, "name": "Deploy_Automation", "model": "flash", "prompt": """You are a DevOps expert writing deployment automation for a Cloudflare Workers + Durable Objects project. CONTEXT: Project is at subaiva-do/ with wrangler.jsonc config. We need a deploy script that: 1. Pre-deploy checks: - Verify wrangler is installed and authenticated - Verify all required KV namespaces exist (create if not) - Verify Vectorize index exists (create if not: subaiva-memory, 768 dimensions, cosine metric) - Run TypeScript type checking - Run any tests 2. Create required resources: - KV namespace: ENTITLEMENTS (production and preview) - Vectorize index: subaiva-memory (768 dimensions, cosine) - Set secrets: API_TOKEN, ADMIN_TOKEN, STRIPE_WEBHOOK_SECRET 3. Deploy: - wrangler deploy with proper flags - Post-deploy smoke test (hit /health, provision a test customer, send a message, verify) 4. Rollback: - If smoke test fails, rollback to previous version - Notify on failure Write this as a bash script `scripts/deploy.sh` with clear sections, error handling, and color output. Also write a `scripts/setup-resources.sh` that creates all Cloudflare resources needed (KV, Vectorize, secrets). Output BOTH scripts. The deploy script should be production-grade. """, "output_file": "subaiva-do/scripts/deploy.sh", }, { "id": 5, "name": "E2E_Test_Suite", "model": "flash", "prompt": f"""You are a TypeScript testing expert writing an E2E test suite for a Cloudflare Workers + Durable Objects application. CONTEXT: SubAIVA is deployed at a URL (local: http://localhost:8787, prod: https://subaiva.workers.dev). We need comprehensive E2E tests. {SUBAIVA_TYPES} Write a COMPLETE TypeScript test file `tests/e2e.test.ts` using Node.js built-in test runner (node:test + node:assert) that tests: 1. Health and public endpoints: - GET /health returns 200 with status "ok" - GET /tiers returns all 3 tiers with correct names 2. Authentication: - Unauthenticated requests to /api/* return 401 - Valid bearer token grants access - Invalid bearer token returns 401 3. Customer lifecycle: - Provision a new customer (tier 1) - Verify status shows correct tier - Upgrade to tier 2 - Verify tier change in status - Upgrade to tier 3 (enterprise) - Verify voice capability flag 4. Chat: - Send a message, get a response - Auto-capture works (send "my name is TestBot", verify memory stored) - Conversation history is maintained (multiple messages) 5. Memory: - Store a memory - Search for it - Store multiple memories with different categories - Verify importance-based search ordering - Verify tier memory limits 6. Customer isolation: - Provision 2 customers - Send different data to each - Verify no cross-contamination 7. Error handling: - Invalid tier number - Missing required fields - Empty message 8. Load (optional): - Provision 5 customers concurrently - Send 10 messages concurrently - Verify all responses received Include test utilities (base URL config, auth helpers, cleanup). Make it runnable with: npx tsx tests/e2e.test.ts Output ONLY the TypeScript test code. """, "output_file": "subaiva-do/tests/e2e.test.ts", }, { "id": 6, "name": "Customer_Onboarding_Flow", "model": "flash", "prompt": f"""You are a TypeScript expert building a customer onboarding flow for Cloudflare Workers. CONTEXT: When a new customer signs up (via Stripe checkout or admin provisioning), we need an automated onboarding flow. {SUBAIVA_TYPES} Write a COMPLETE TypeScript file `src/onboarding.ts` that: 1. OnboardingFlow class with methods: - startOnboarding(customerId, tier, email, businessName, industry) — kicks off the flow - sendWelcomeEmail(customerId, email, tier) — via Resend API (Cloudflare-friendly) - createDefaultMemories(customerId, businessName, industry) — seed the SubAIVA with business context - configureSystemPrompt(customerId, businessName, industry) — generate a tailored system prompt - scheduleFollowUp(customerId) — set a DO alarm for 24hr follow-up check 2. Default memory seeding per industry: - plumbing: common services, emergency protocols, pricing guidelines - electrical: safety warnings, common issues, licensing requirements - hvac: seasonal patterns, maintenance schedules, brand knowledge - general: business hours, contact info, FAQs 3. System prompt generation: - Include business name, industry context - Australian professional tone (warm, clear, NO slang like "G'day mate" or "Too easy") - Memory-aware instructions - Tier-appropriate capabilities description 4. Welcome email template: - Subject: "Your AI Assistant is Ready — [Business Name]" - HTML template with Sunaiva branding - Quick start guide (how to chat, what to teach your AI) - Tier-specific feature highlights 5. Error handling and retry logic Output ONLY the TypeScript code with comments. """, "output_file": "subaiva-do/src/onboarding.ts", }, { "id": 7, "name": "Continuous_Dev_Daemon", "model": "flash", "prompt": """You are a Python expert building a continuous development daemon for the Genesis system. CONTEXT: We have a SubAIVA Durable Objects project at subaiva-do/. We want Gemini agents to continuously: - Run tests against the deployed SubAIVA - Monitor logs and report anomalies - Generate and test new feature code - File bug reports as KG entities when tests fail - Suggest improvements based on usage patterns Write a Python script `scripts/cloudflare_continuous_dev.py` that: 1. Runs in a loop (configurable interval, default 30 minutes) 2. Each cycle: a. Check if SubAIVA is deployed (hit /health endpoint) b. If deployed: run smoke tests, collect metrics, report status c. If not deployed: run local dev tests instead d. Fire a Gemini agent to analyze test results and suggest improvements e. Write results to KG entities f. Optionally auto-fix simple issues (type errors, missing imports) 3. Uses GeminiExecutor from core.gemini_executor 4. Writes cycle reports to subaiva-do/reports/cycle_YYYYMMDD_HHMMSS.json 5. Sends a daily summary to a markdown file 6. Graceful shutdown on SIGTERM/SIGINT 7. PID file for process management 8. Configurable via CLI args (--interval, --url, --auto-fix, --verbose) Import pattern: ```python sys.path.insert(0, str(Path(__file__).parent.parent)) from core.gemini_executor import GeminiExecutor ``` Output ONLY the Python code with docstrings. """, "output_file": "scripts/cloudflare_continuous_dev.py", }, ] async def run_mission(executor: GeminiExecutor, mission: dict) -> dict: """Execute a single build mission.""" start = time.time() mission_id = mission["id"] name = mission["name"] output_file = Path("/mnt/e/genesis-system") / mission["output_file"] print(f"[Mission {mission_id}] Starting: {name}") try: result = executor.execute( mission["prompt"], model=executor.MODELS.get(mission["model"], mission["model"]), ) if result.success and result.response: output_file.parent.mkdir(parents=True, exist_ok=True) # For code files, strip markdown code fences if present content = result.response if content.startswith("```"): lines = content.split("\n") # Remove first line (```typescript or ```python) and last line (```) if lines[-1].strip() == "```": lines = lines[1:-1] elif lines[0].startswith("```"): lines = lines[1:] content = "\n".join(lines) # Add generation header as comment ext = output_file.suffix if ext in (".ts", ".js"): header = f"// Generated by Cloudflare Build Swarm — {name}\n// Date: {datetime.now().isoformat()}\n// Model: {result.model}\n// Tokens: {result.tokens_used}\n\n" elif ext in (".py",): header = f"# Generated by Cloudflare Build Swarm — {name}\n# Date: {datetime.now().isoformat()}\n# Model: {result.model}\n# Tokens: {result.tokens_used}\n\n" elif ext in (".sh",): header = f"#!/usr/bin/env bash\n# Generated by Cloudflare Build Swarm — {name}\n# Date: {datetime.now().isoformat()}\n# Model: {result.model}\n# Tokens: {result.tokens_used}\n\n" else: header = "" output_file.write_text(header + content, encoding="utf-8") elapsed = time.time() - start print(f"[Mission {mission_id}] COMPLETE: {name} ({elapsed:.1f}s, {result.tokens_used} tokens) → {output_file}") return { "id": mission_id, "name": name, "success": True, "tokens": result.tokens_used, "cost": result.cost_estimate, "elapsed": elapsed, "output_file": str(output_file), } else: print(f"[Mission {mission_id}] FAILED: {name} - {result.error}") return { "id": mission_id, "name": name, "success": False, "error": result.error, } except Exception as e: elapsed = time.time() - start print(f"[Mission {mission_id}] ERROR: {name} - {e}") return { "id": mission_id, "name": name, "success": False, "error": str(e), "elapsed": elapsed, } async def main(): """Run the Cloudflare Build Swarm.""" import argparse parser = argparse.ArgumentParser(description="Cloudflare Build Swarm (Phase 2)") parser.add_argument("--missions", default="all", help="Mission IDs (e.g., '1,3,5' or 'all')") parser.add_argument("--parallel", type=int, default=4, help="Max parallel missions") args = parser.parse_args() if args.missions == "all": selected = MISSIONS else: ids = [int(x.strip()) for x in args.missions.split(",")] selected = [m for m in MISSIONS if m["id"] in ids] print(f"=== CLOUDFLARE BUILD SWARM (Phase 2) ===") print(f"Missions: {len(selected)}") print(f"Parallel: {args.parallel}") print(f"Output: deployable code files") print(f"Started: {datetime.now().isoformat()}") print(f"========================================\n") executor = GeminiExecutor(default_model="flash", use_rate_maximizer=True) semaphore = asyncio.Semaphore(args.parallel) async def bounded_mission(mission): async with semaphore: return await run_mission(executor, mission) results = await asyncio.gather(*[bounded_mission(m) for m in selected]) # Summary print(f"\n=== BUILD SWARM RESULTS ===") total_tokens = 0 total_cost = 0.0 successes = 0 for r in results: status = "OK" if r.get("success") else "FAIL" tokens = r.get("tokens", 0) cost = r.get("cost", 0) total_tokens += tokens total_cost += cost if r.get("success"): successes += 1 out = r.get("output_file", "N/A") print(f" [{status}] Mission {r['id']}: {r['name']} ({tokens} tokens) → {out}") print(f"\nTotal: {successes}/{len(results)} succeeded") print(f"Total tokens: {total_tokens:,}") print(f"Total cost: ${total_cost:.4f}") # KG entry kg_path = Path("/mnt/e/genesis-system/KNOWLEDGE_GRAPH/entities/cloudflare_build_swarm_2026_02_26.jsonl") with open(kg_path, "a") as f: f.write(json.dumps({ "id": f"cf_build_{int(time.time())}", "type": "swarm_execution", "name": "Cloudflare Build Swarm Phase 2", "description": f"{successes}/{len(results)} build missions completed. {total_tokens:,} tokens. ${total_cost:.4f}. Produced: stripe webhook, admin API, metering, deploy scripts, E2E tests, onboarding flow, continuous dev daemon.", "created": datetime.now().isoformat(), "session": 108, "tags": ["cloudflare", "build", "swarm", "subaiva", "phase-2"], }) + "\n") return results if __name__ == "__main__": asyncio.run(main())