Version: 1.2 Date: March 9, 2026 Status: Active Aligned with: PRD v2.2
This document describes the complete technical architecture for Agent-X. It is derived from and fully aligned with the PRD v2.2. Where technical decisions are referenced, they match the PRD exactly.
| Version | Date | Changes |
|---|---|---|
| 1.2 | 2026-03-09 | Fixed X API metrics fetch to include non_public_metrics with Basic tier note; added Docker credential volume mount (credentials.json:ro); added server-side MIME validation (python-magic) to image upload endpoint; corrected pgvector extension naming to vector; made weekly improvement schedule configurable via IMPROVEMENT_SCHEDULE_DAY/IMPROVEMENT_SCHEDULE_HOUR env vars (default Sunday 06:00); added security controls for file upload validation. |
| 1.1 | 2026-03-08 | Replaced ASCII diagrams with Mermaid; clarified V1 JWT auth flow; fixed async compute_embedding; added publish_post validation; added cosine similarity inline comment; added adapter factory pattern section; added performance and fallback notes; added mobile responsiveness guidance; timezone consistency pass (America/Los_Angeles everywhere); general consistency and quality sweep. |
| 1.0 | 2026-03-08 | Initial architecture specification aligned with PRD v2.2. |
- Architecture Overview
- System Architecture Diagram
- Component Architecture
- Data Architecture
- API Architecture
- Integration Architecture
- Security Architecture
- Deployment Architecture
- Scalability and Future-Proofing
- Performance and Fallback Behavior
- Development Workflow
- Appendix: FastAPI Application Entry Point
Agent-X is a semi-autonomous, human-in-the-loop AI content engine for X/Twitter. The system automates the research-to-draft pipeline (topic discovery, ranking, content drafting, uniqueness checking) while requiring explicit human approval for every post before publishing.
The system comprises three main deployable units:
- Backend (Python FastAPI): Async API server handling business logic, LLM orchestration, scheduling, and external service integration.
- Frontend (Next.js): Web dashboard for human approval, source management, analytics, and system configuration.
- Reverse Proxy (Nginx): Routes traffic, terminates TLS, and serves as the single entry point.
All persistent state lives in Supabase (hosted Postgres with the vector extension (commonly known as pgvector) and Storage). There is no local state that cannot survive a container restart.
| Principle | Description |
|---|---|
| Modularity | Each system component (research, ranking, drafting, publishing, analytics) is a standalone module with a well-defined interface. |
| Abstraction | External services (LLM, publisher, email) are accessed through abstract interfaces (Ports). Concrete implementations (Adapters) can be swapped without changing business logic. |
| Testability | The core/ layer has zero framework dependencies. All external dependencies are injected, making unit testing straightforward with mocks. |
| Restart-Safety | All state is persisted to Supabase. Scheduler jobs are stored in the database. On restart, the system recovers missed jobs and resumes normal operation. |
| Human-in-the-Loop | No post is ever published without explicit human approval. This is enforced at the code level, not just the UI level. |
| Graceful Degradation | If one source fails, others continue. If the LLM fails, the cycle logs the error and retries. No single failure brings down the system. |
Ports & Adapters (Hexagonal Architecture) -- for integrations
External services are accessed through abstract base classes (Ports). Each concrete implementation (Adapter) can be replaced independently. This applies to:
- LLM providers (Vertex AI today, Groq/OpenAI tomorrow)
- Publishing platforms (X/Twitter today, LinkedIn/Medium in V3)
- Email providers (Resend today, alternatives later)
Layered Architecture -- for the backend
The backend is organized into strict layers with unidirectional dependencies:
graph TD
API["API Layer (api/)\nHandles HTTP, auth, request/response serialization"]
Core["Core Business Logic (core/)\nPure Python, no framework imports, receives deps via params"]
Integrations["Integrations (integrations/)\nAdapters for external services behind abstract interfaces"]
API --> Core
Core --> Integrations
The API layer depends on Core. Core depends on abstract interfaces only. Integrations implement those interfaces. Core never imports from API or from concrete integration implementations.
Repository Pattern -- for data access
All database operations go through the Supabase client wrapper. The core layer never directly calls Supabase; instead, data access functions are passed as parameters or injected via the dependency system.
graph TD
Users["Internet / Users"]
Users -->|HTTPS| Nginx["Nginx Reverse Proxy (TLS)"]
Users -->|HTTPS| XAPI["X/Twitter API (Publishing)"]
Nginx -->|"/*"| Frontend["Next.js Frontend (SSR)"]
Nginx -->|"/api/*"| Backend["FastAPI Backend (Async)"]
Backend -->|tweepy| XAPI
Backend --> VertexAI["Vertex AI Studio (LLM + Embeddings)"]
Backend --> Supabase["Supabase (PostgreSQL + vector extension + Storage)"]
Backend --> Resend["Resend (Email)"]
Backend --> HNAPI["HN API (HTTP)"]
Backend --> RSSFeeds["RSS Feeds (HTTP GET)"]
flowchart LR
A["1. RESEARCH\nHN API + RSS Feeds\nFilter by categories"] --> B["2. IDEA GEN\nRanked topics sent to LLM\nfor angle expansion\n2-4 angles/topic"]
B --> C["3. DRAFTING\nTopic-angle pairs sent to\nLLM with style config\n280-char enforcement"]
C --> D["4. UNIQUENESS CHECK\nEmbedding cosine similarity\n+ hash check\nReject dupes"]
D --> E["5. APPROVAL (Dashboard)\nApprove / Reject / Edit\nNeeds Image\nEmail notification"]
E --> F["6. PUBLISHING\nScheduled with jitter\nPre-pub checks\nTweet via X API v2\nLog result"]
F --> G["7. ANALYTICS\nManual entry or\non-demand fetch\nPer-post + aggregate views"]
G --> H["8. IMPROVEMENT ENGINE\nWeekly LLM analysis\nStyle config proposals\nfor human review"]
graph TD
subgraph Browser
NextJS["Next.js App (App Router)\nPages: Login, Dashboard,\nApprovals, Posts, Sources,\nAnalytics, Settings"]
end
subgraph Server
FastAPI["FastAPI Backend (Async Python)\nLayers: api/, core/,\nintegrations/, scheduler/,\nmodels/, utils/"]
end
NextJS <-->|"fetch(/api/*)\nJSON + JWT"| FastAPI
FastAPI --> Supabase["Supabase\nPostgreSQL + vector extension + Storage"]
FastAPI --> VertexAI["Vertex AI Studio"]
FastAPI --> XAPI["X API"]
FastAPI --> Resend["Resend"]
FastAPI --> HNAPI["HN API"]
The backend follows a strict layered architecture. Each layer has clear responsibilities and dependency rules.
The API layer is the HTTP boundary. It handles request parsing, authentication, response serialization, and delegates all business logic to the core/ layer.
Route Handlers:
| File | Responsibility |
|---|---|
auth.py |
Login endpoint, JWT token creation, password verification |
posts.py |
Post listing, approval actions, image upload, metrics |
sources.py |
Source CRUD, health status |
analytics.py |
Aggregate analytics, category performance |
categories.py |
Category CRUD |
style.py |
Style config read/update |
improvements.py |
Improvement proposal listing, approve/reject |
system.py |
System status, manual cycle triggers |
dependencies.py |
Shared FastAPI dependencies (auth verification, Supabase client) |
Authentication Middleware:
# backend/api/dependencies.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, JWTError
from datetime import datetime, timedelta
security = HTTPBearer()
SECRET_KEY: str # Loaded from config
ALGORITHM = "HS256"
TOKEN_EXPIRE_HOURS = 24
def create_access_token(data: dict) -> str:
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(hours=TOKEN_EXPIRE_HOURS)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
async def verify_token(
credentials: HTTPAuthorizationCredentials = Depends(security),
) -> dict:
try:
payload = jwt.decode(
credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM]
)
if payload.get("sub") != "owner":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token",
)
return payload
except JWTError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or expired token",
)CORS Configuration:
# In main.py
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=[FRONTEND_URL], # e.g., "https://agentx.yourdomain.com"
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)Login Rate Limiting:
# backend/api/auth.py
from collections import defaultdict
from datetime import datetime, timedelta
login_attempts: dict[str, list[datetime]] = defaultdict(list)
MAX_ATTEMPTS = 5
WINDOW_MINUTES = 5
async def check_rate_limit(client_ip: str) -> None:
now = datetime.utcnow()
cutoff = now - timedelta(minutes=WINDOW_MINUTES)
login_attempts[client_ip] = [
t for t in login_attempts[client_ip] if t > cutoff
]
if len(login_attempts[client_ip]) >= MAX_ATTEMPTS:
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail=f"Too many login attempts. Try again in {WINDOW_MINUTES} minutes.",
)
login_attempts[client_ip].append(now)Request/Response Models:
All request and response bodies use Pydantic models defined in models/. The API layer imports these models and uses them for automatic validation and serialization.
CRITICAL RULE: This layer has ZERO framework dependencies. No FastAPI, no Supabase client, no tweepy, no google-cloud imports. Every external dependency is received via function parameters or dependency injection.
This ensures:
- Unit tests can run with simple mocks
- Business logic is portable across frameworks
- Coupling to external services is eliminated
Module: research.py
# backend/core/research.py
"""
Research engine: fetches topics from HN and RSS, filters by category.
No direct imports of httpx, feedparser, or supabase.
"""
from typing import Protocol, Any
class TopicFetcher(Protocol):
async def fetch_hn_top_stories(self, limit: int) -> list[dict]: ...
async def fetch_rss_feed(self, url: str) -> list[dict]: ...
class TopicStore(Protocol):
async def get_enabled_sources(self) -> list[dict]: ...
async def get_enabled_categories(self) -> list[str]: ...
async def get_recent_posts(self, hours: int) -> list[dict]: ...
async def update_source_health(self, source_id: str, success: bool) -> None: ...
async def run_research_cycle(
fetcher: TopicFetcher,
store: TopicStore,
hn_story_limit: int = 30,
) -> list[dict]:
"""
Fetch topics from all enabled sources, filter by categories,
and return raw topic list for ranking.
"""
sources = await store.get_enabled_sources()
categories = await store.get_enabled_categories()
topics = []
for source in sources:
try:
if source["type"] == "hackernews":
hn_topics = await fetcher.fetch_hn_top_stories(hn_story_limit)
topics.extend(hn_topics)
elif source["type"] == "rss":
rss_topics = await fetcher.fetch_rss_feed(source["url"])
topics.extend(rss_topics)
await store.update_source_health(source["id"], success=True)
except Exception:
await store.update_source_health(source["id"], success=False)
continue # Graceful degradation: skip failed source
# Filter by enabled categories
filtered = [t for t in topics if t.get("category") in categories]
return filteredModule: topic_ranker.py
# backend/core/topic_ranker.py
"""
Scores and ranks topics. Deduplicates against recent posts (72h window).
"""
from typing import Protocol
class LLMProvider(Protocol):
async def generate_json(self, prompt: str, schema: dict) -> dict: ...
class RecentPostStore(Protocol):
async def get_recent_posts(self, hours: int) -> list[dict]: ...
async def rank_topics(
raw_topics: list[dict],
llm: LLMProvider,
store: RecentPostStore,
recent_window_hours: int = 72,
) -> list[dict]:
"""
Score topics on trend potential, engagement likelihood, and novelty.
Dedup against recent posts. Return ranked list.
"""
recent_posts = await store.get_recent_posts(recent_window_hours)
recent_titles = [p["topic"] for p in recent_posts]
# Build prompt with topics and recent posts for dedup
prompt = _build_ranking_prompt(raw_topics, recent_titles)
schema = {"type": "array", "items": {"type": "object"}}
ranked = await llm.generate_json(prompt, schema)
return rankedModule: content_generator.py
# backend/core/content_generator.py
"""
LLM-based post drafting with style config, 280-char enforcement, retry logic.
"""
from typing import Protocol
class LLMProvider(Protocol):
async def generate_text(self, prompt: str, max_tokens: int) -> str: ...
class StyleStore(Protocol):
async def get_active_style_config(self) -> dict: ...
async def get_recent_rejections(self, days: int) -> list[dict]: ...
async def generate_drafts(
topic_angles: list[dict],
llm: LLMProvider,
style_store: StyleStore,
max_retries: int = 2,
char_limit: int = 280,
) -> list[dict]:
"""
Generate draft posts for each topic-angle pair.
Enforces 280-char limit with retry logic.
"""
style_config = await style_store.get_active_style_config()
recent_rejections = await style_store.get_recent_rejections(days=7)
drafts = []
for ta in topic_angles:
prompt = _build_drafting_prompt(ta, style_config, recent_rejections)
draft_text = None
for attempt in range(1 + max_retries):
text = await llm.generate_text(prompt, max_tokens=150)
if len(text) <= char_limit:
draft_text = text
break
# Retry with explicit shortening instruction
prompt = _build_shorten_prompt(text, char_limit)
if draft_text:
drafts.append({
"content": draft_text,
"topic": ta["topic"],
"angle": ta["angle_type"],
"category": ta["category"],
"source_url": ta.get("source_url"),
})
# If still over limit after retries, skip and log
return draftsModule: uniqueness_checker.py
# backend/core/uniqueness_checker.py
"""
Embedding-based cosine similarity + hash checking for deduplication.
"""
import hashlib
from typing import Protocol
class EmbeddingProvider(Protocol):
async def compute_embedding(self, text: str) -> list[float]: ...
class EmbeddingStore(Protocol):
async def get_all_embeddings(self) -> list[dict]: ...
async def get_recent_hashes(self, days: int) -> list[str]: ...
def compute_content_hash(text: str) -> str:
"""Normalized hash of first 80 chars: lowercase, whitespace-stripped."""
normalized = text[:80].lower().strip()
normalized = "".join(normalized.split()) # Remove all whitespace
return hashlib.sha256(normalized.encode()).hexdigest()
def cosine_similarity(a: list[float], b: list[float]) -> float:
"""Compute cosine similarity between two embedding vectors.
Used with a default threshold of 0.85: posts with similarity >= 0.85
are considered duplicates and will be rejected from the approval queue.
"""
dot = sum(x * y for x, y in zip(a, b))
norm_a = sum(x * x for x in a) ** 0.5
norm_b = sum(x * x for x in b) ** 0.5
if norm_a == 0 or norm_b == 0:
return 0.0
return dot / (norm_a * norm_b)
async def check_uniqueness(
drafts: list[dict],
embedding_provider: EmbeddingProvider,
store: EmbeddingStore,
similarity_threshold: float = 0.85,
hash_window_days: int = 7,
) -> tuple[list[dict], list[dict]]:
"""
Check each draft for uniqueness. Returns (unique_drafts, rejected_drafts).
"""
existing_embeddings = await store.get_all_embeddings()
recent_hashes = await store.get_recent_hashes(hash_window_days)
unique = []
rejected = []
for draft in drafts:
content_hash = compute_content_hash(draft["content"])
# Hash check
if content_hash in recent_hashes:
draft["rejection_reason"] = "hash_duplicate"
rejected.append(draft)
continue
# Embedding check
embedding = await embedding_provider.compute_embedding(draft["content"])
max_sim = 0.0
similar_post_id = None
for existing in existing_embeddings:
sim = cosine_similarity(embedding, existing["embedding"])
if sim > max_sim:
max_sim = sim
similar_post_id = existing["post_id"]
# Posts with similarity >= 0.85 are considered duplicates
if max_sim > similarity_threshold:
draft["rejection_reason"] = f"embedding_similarity:{max_sim:.3f}"
draft["similar_post_id"] = similar_post_id
rejected.append(draft)
else:
draft["embedding"] = embedding
draft["content_hash"] = content_hash
unique.append(draft)
return unique, rejectedModule: publisher.py
# backend/core/publisher.py
"""
Publish queue management, pre-publish checks, post-publish logging.
"""
from typing import Protocol
from datetime import datetime
class PublishTarget(Protocol):
async def publish_post(
self, text: str, media_path: str | None
) -> dict: ...
class PublishStore(Protocol):
async def get_daily_post_count(self, date: datetime) -> int: ...
async def get_monthly_post_count(self, year: int, month: int) -> int: ...
async def mark_post_published(
self, post_id: str, tweet_id: str, published_at: datetime
) -> None: ...
async def mark_post_failed(self, post_id: str, error: str) -> None: ...
class AlertSender(Protocol):
async def send_alert(self, subject: str, body: str) -> None: ...
async def publish_approved_post(
post: dict,
publisher: PublishTarget,
store: PublishStore,
alert_sender: AlertSender,
daily_limit: int = 5,
monthly_limit: int = 500,
) -> bool:
"""
Run pre-publish checks, then publish. Returns True on success.
"""
now = datetime.utcnow()
# Pre-publish checks
daily_count = await store.get_daily_post_count(now)
if daily_count >= daily_limit:
await alert_sender.send_alert(
"Daily post limit reached",
f"Already published {daily_count} posts today.",
)
return False
monthly_count = await store.get_monthly_post_count(now.year, now.month)
if monthly_count >= monthly_limit:
await alert_sender.send_alert(
"Monthly post limit reached",
f"Already published {monthly_count} posts this month.",
)
return False
# Budget alerts
if monthly_count >= int(monthly_limit * 0.95):
await alert_sender.send_alert(
"CRITICAL: 95% of monthly post budget used",
f"{monthly_count}/{monthly_limit} posts used.",
)
elif monthly_count >= int(monthly_limit * 0.80):
await alert_sender.send_alert(
"WARNING: 80% of monthly post budget used",
f"{monthly_count}/{monthly_limit} posts used.",
)
if len(post["content"]) > 280:
await store.mark_post_failed(post["id"], "Content exceeds 280 chars")
return False
# Publish
try:
result = await publisher.publish_post(
text=post["content"],
media_path=post.get("media_url"),
)
await store.mark_post_published(
post["id"], result["tweet_id"], now
)
return True
except Exception as e:
# Retry once after failure
try:
result = await publisher.publish_post(
text=post["content"],
media_path=post.get("media_url"),
)
await store.mark_post_published(
post["id"], result["tweet_id"], now
)
return True
except Exception as retry_error:
await store.mark_post_failed(post["id"], str(retry_error))
await alert_sender.send_alert(
"Publishing failed",
f"Post {post['id']} failed after retry: {retry_error}",
)
return FalseModule: analytics_tracker.py
# backend/core/analytics_tracker.py
"""
Metrics aggregation and performance analysis.
"""
from typing import Protocol
from datetime import datetime
class MetricsStore(Protocol):
async def get_post_metrics(self, post_id: str) -> dict | None: ...
async def get_metrics_in_range(
self, start: datetime, end: datetime
) -> list[dict]: ...
async def upsert_metrics(self, post_id: str, metrics: dict) -> None: ...
class MetricsFetcher(Protocol):
async def fetch_metrics(self, tweet_id: str) -> dict: ...
async def compute_aggregate_metrics(
store: MetricsStore,
start: datetime,
end: datetime,
) -> dict:
"""Compute aggregate engagement metrics over a time window."""
metrics_list = await store.get_metrics_in_range(start, end)
total_impressions = sum(m.get("impressions", 0) for m in metrics_list)
total_likes = sum(m.get("likes", 0) for m in metrics_list)
total_reposts = sum(m.get("reposts", 0) for m in metrics_list)
total_replies = sum(m.get("replies", 0) for m in metrics_list)
post_count = len(metrics_list)
engagement_rate = 0.0
if total_impressions > 0:
engagement_rate = (total_likes + total_replies) / total_impressions
return {
"post_count": post_count,
"total_impressions": total_impressions,
"total_likes": total_likes,
"total_reposts": total_reposts,
"total_replies": total_replies,
"engagement_rate": round(engagement_rate, 4),
}Module: improvement_engine.py
# backend/core/improvement_engine.py
"""
Weekly analysis of performance data. Generates style config proposals.
"""
from typing import Protocol
class LLMProvider(Protocol):
async def generate_json(self, prompt: str, schema: dict) -> dict: ...
class PerformanceStore(Protocol):
async def get_published_posts_with_metrics(self, days: int) -> list[dict]: ...
async def get_rejected_posts_with_feedback(self, days: int) -> list[dict]: ...
async def get_active_style_config(self) -> dict: ...
async def store_improvement_proposal(self, proposal: dict) -> None: ...
async def run_weekly_analysis(
llm: LLMProvider,
store: PerformanceStore,
analysis_window_days: int = 7,
) -> dict:
"""
Analyze recent performance, generate style improvement proposal.
"""
published = await store.get_published_posts_with_metrics(analysis_window_days)
rejected = await store.get_rejected_posts_with_feedback(analysis_window_days)
current_style = await store.get_active_style_config()
prompt = _build_improvement_prompt(published, rejected, current_style)
schema = {
"type": "object",
"properties": {
"analysis_summary": {"type": "string"},
"proposed_changes": {"type": "object"},
"rationale": {"type": "string"},
},
}
proposal = await llm.generate_json(prompt, schema)
await store.store_improvement_proposal(proposal)
return proposalEach integration implements an abstract interface (Port). Concrete implementations are Adapters that can be swapped without modifying business logic.
LLM Interface (integrations/llm/base.py)
# backend/integrations/llm/base.py
from abc import ABC, abstractmethod
class BaseLLM(ABC):
"""Abstract interface for LLM providers (Port)."""
@abstractmethod
async def generate_text(
self,
prompt: str,
max_tokens: int = 512,
temperature: float = 0.7,
) -> str:
"""Generate text completion from a prompt."""
...
@abstractmethod
async def generate_json(
self,
prompt: str,
schema: dict,
max_tokens: int = 1024,
temperature: float = 0.3,
) -> dict:
"""Generate structured JSON output matching the given schema."""
...
@abstractmethod
async def compute_embedding(
self,
text: str,
) -> list[float]:
"""Compute a text embedding vector."""
...Vertex AI Implementation (integrations/llm/vertex_ai.py)
# backend/integrations/llm/vertex_ai.py
import json
from google.cloud import aiplatform
from vertexai.generative_models import GenerativeModel
from vertexai.language_models import TextEmbeddingModel
from .base import BaseLLM
class VertexAILLM(BaseLLM):
"""Vertex AI Studio adapter (Adapter)."""
def __init__(self, project_id: str, location: str = "us-central1"):
aiplatform.init(project=project_id, location=location)
self._text_model = GenerativeModel("gemini-1.5-flash")
self._embedding_model = TextEmbeddingModel.from_pretrained(
"text-embedding-004"
)
async def generate_text(
self,
prompt: str,
max_tokens: int = 512,
temperature: float = 0.7,
) -> str:
response = await self._text_model.generate_content_async(
prompt,
generation_config={
"max_output_tokens": max_tokens,
"temperature": temperature,
},
)
return response.text.strip()
async def generate_json(
self,
prompt: str,
schema: dict,
max_tokens: int = 1024,
temperature: float = 0.3,
) -> dict:
json_prompt = (
f"{prompt}\n\nRespond ONLY with valid JSON matching this schema: "
f"{json.dumps(schema)}"
)
response = await self._text_model.generate_content_async(
json_prompt,
generation_config={
"max_output_tokens": max_tokens,
"temperature": temperature,
"response_mime_type": "application/json",
},
)
return json.loads(response.text)
async def compute_embedding(self, text: str) -> list[float]:
embeddings = await self._embedding_model.get_embeddings_async([text])
return embeddings[0].valuesPublisher Interface (integrations/publishers/base.py)
# backend/integrations/publishers/base.py
from abc import ABC, abstractmethod
from dataclasses import dataclass
@dataclass
class PublishResult:
post_id: str # Platform-specific post ID (e.g., tweet_id)
url: str # URL to the published post
published_at: str # ISO timestamp
@dataclass
class PostMetrics:
impressions: int
likes: int
reposts: int
replies: int
fetched_at: str # ISO timestamp
class BasePublisher(ABC):
"""Abstract interface for publishing platforms (Port)."""
@abstractmethod
async def publish_post(
self,
text: str,
media_path: str | None = None,
) -> PublishResult:
"""Publish a post to the platform. Returns publish result."""
...
@abstractmethod
async def fetch_metrics(
self,
post_id: str,
) -> PostMetrics:
"""Fetch engagement metrics for a published post."""
...X/Twitter Publisher (integrations/publishers/x_publisher.py)
# backend/integrations/publishers/x_publisher.py
import tweepy
from datetime import datetime, timezone
from .base import BasePublisher, PublishResult, PostMetrics
class XPublisher(BasePublisher):
"""X/Twitter adapter using tweepy (Adapter)."""
def __init__(
self,
api_key: str,
api_secret: str,
access_token: str,
access_token_secret: str,
):
# v2 Client for tweet creation
self._client = tweepy.Client(
consumer_key=api_key,
consumer_secret=api_secret,
access_token=access_token,
access_token_secret=access_token_secret,
)
# v1.1 API for media uploads
auth = tweepy.OAuth1UserHandler(
api_key, api_secret, access_token, access_token_secret
)
self._api = tweepy.API(auth)
async def publish_post(
self,
text: str,
media_path: str | None = None,
) -> PublishResult:
# Validate post content before attempting to publish
if not text or not text.strip():
raise ValueError("Post content must not be empty.")
if len(text) > 280:
raise ValueError(
f"Post content exceeds 280 characters ({len(text)} chars). "
"Cannot publish."
)
media_ids = None
if media_path:
# Upload media via v1.1
media = self._api.media_upload(filename=media_path)
media_ids = [media.media_id]
# Create tweet via v2
response = self._client.create_tweet(
text=text,
media_ids=media_ids,
)
tweet_id = str(response.data["id"])
return PublishResult(
post_id=tweet_id,
url=f"https://twitter.com/i/status/{tweet_id}",
published_at=datetime.now(timezone.utc).isoformat(),
)
async def fetch_metrics(self, post_id: str) -> PostMetrics:
tweet = self._client.get_tweet(
post_id,
tweet_fields=["public_metrics", "non_public_metrics"],
)
public = tweet.data.public_metrics or {}
non_public = tweet.data.get("non_public_metrics") or {} if hasattr(tweet.data, "get") else getattr(tweet.data, "non_public_metrics", None) or {}
# impression_count lives in non_public_metrics (requires X API Basic tier $100/mo).
# Fallback to 0 on Free tier where non_public_metrics is unavailable.
impressions = non_public.get("impression_count", 0) or public.get("impression_count", 0)
return PostMetrics(
impressions=impressions,
likes=public.get("like_count", 0),
reposts=public.get("retweet_count", 0),
replies=public.get("reply_count", 0),
fetched_at=datetime.now(timezone.utc).isoformat(),
)
# Note: `impression_count` requires `non_public_metrics` which is only available
# with X API Basic tier ($100/mo) or higher. On Free tier, impressions will default to 0.Supabase Client (integrations/supabase_client.py)
# backend/integrations/supabase_client.py
"""
Supabase Python client wrapper.
All DB operations go through this module.
Uses Supabase REST API (not direct Postgres connection).
"""
from supabase import create_client, Client
class SupabaseClient:
"""Centralized Supabase client wrapper."""
def __init__(self, url: str, service_role_key: str):
self._client: Client = create_client(url, service_role_key)
@property
def client(self) -> Client:
return self._client
# --- Posts ---
async def get_posts(
self, status: str | None = None, limit: int = 50, offset: int = 0
) -> list[dict]:
query = self._client.table("posts").select("*")
if status:
query = query.eq("status", status)
result = query.order("created_at", desc=True).range(
offset, offset + limit - 1
).execute()
return result.data
async def create_post(self, post_data: dict) -> dict:
result = self._client.table("posts").insert(post_data).execute()
return result.data[0]
async def update_post(self, post_id: str, updates: dict) -> dict:
result = (
self._client.table("posts")
.update(updates)
.eq("id", post_id)
.execute()
)
return result.data[0]
# --- Sources ---
async def get_sources(self, enabled_only: bool = False) -> list[dict]:
query = self._client.table("sources").select("*")
if enabled_only:
query = query.eq("enabled", True)
return query.execute().data
# --- Metrics ---
async def upsert_metrics(self, post_id: str, metrics: dict) -> dict:
data = {"post_id": post_id, **metrics}
result = self._client.table("post_metrics").upsert(data).execute()
return result.data[0]
# --- Style Config ---
async def get_active_style_config(self) -> dict:
result = (
self._client.table("style_config")
.select("*")
.eq("is_active", True)
.limit(1)
.execute()
)
return result.data[0] if result.data else {}
# --- Storage ---
async def upload_image(self, file_path: str, file_bytes: bytes) -> str:
self._client.storage.from_("post-images").upload(
file_path, file_bytes
)
return self._client.storage.from_("post-images").get_public_url(
file_path
)Resend Email Client (integrations/resend_client.py)
# backend/integrations/resend_client.py
import resend
class ResendClient:
"""Resend email API wrapper."""
def __init__(self, api_key: str, from_email: str, to_email: str):
resend.api_key = api_key
self._from_email = from_email
self._to_email = to_email
async def send_email(
self, subject: str, html_body: str
) -> dict:
params = {
"from": self._from_email,
"to": [self._to_email],
"subject": f"[Agent-X] {subject}",
"html": html_body,
}
return resend.Emails.send(params)
async def send_batch_ready_notification(
self, batch_id: str, draft_count: int
) -> dict:
return await self.send_email(
subject=f"New draft batch ready: {batch_id}",
html_body=(
f"<h2>New Draft Batch Ready</h2>"
f"<p>Batch <strong>{batch_id}</strong> has "
f"<strong>{draft_count}</strong> drafts ready for review.</p>"
f"<p><a href='https://your-domain.com/approvals'>"
f"Review drafts now</a></p>"
),
)
async def send_alert(self, subject: str, body: str) -> dict:
return await self.send_email(
subject=subject,
html_body=f"<h2>Alert</h2><p>{body}</p>",
)RSS Parser (integrations/rss_parser.py)
# backend/integrations/rss_parser.py
import feedparser
from datetime import datetime
class RSSParser:
"""RSS feed parser wrapper using feedparser."""
async def parse_feed(self, url: str) -> list[dict]:
feed = feedparser.parse(url)
items = []
for entry in feed.entries:
published = None
if hasattr(entry, "published_parsed") and entry.published_parsed:
published = datetime(*entry.published_parsed[:6]).isoformat()
items.append({
"title": entry.get("title", ""),
"url": entry.get("link", ""),
"summary": entry.get("summary", ""),
"published_at": published,
"source_type": "rss",
})
return items# backend/scheduler/jobs.py
"""
APScheduler configuration with Supabase job persistence.
"""
import random
from datetime import datetime, timedelta, timezone
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.date import DateTrigger
class AgentXScheduler:
"""Manages all scheduled jobs for Agent-X."""
def __init__(
self,
timezone: str = "America/Los_Angeles",
jitter_min: int = 8,
jitter_max: int = 12,
):
self._scheduler = AsyncIOScheduler(timezone=timezone)
self._timezone = timezone
self._jitter_min = jitter_min
self._jitter_max = jitter_max
def start(self) -> None:
"""Start the scheduler."""
self._scheduler.start()
def shutdown(self, wait: bool = True) -> None:
"""Graceful shutdown."""
self._scheduler.shutdown(wait=wait)
def schedule_research_cycle(
self, callback, hour: int = 7, minute: int = 0
) -> None:
"""Schedule the daily research cycle."""
self._scheduler.add_job(
callback,
trigger=CronTrigger(
hour=hour, minute=minute, timezone=self._timezone
),
id="daily_research_cycle",
replace_existing=True,
)
def schedule_weekly_improvement(
self, callback, day_of_week: str = "sun", hour: int = 6
) -> None:
"""Schedule weekly improvement analysis.
Default: Sunday 06:00 America/Los_Angeles (~7 PM PKT, reasonable for owner review).
Configurable via IMPROVEMENT_SCHEDULE_DAY and IMPROVEMENT_SCHEDULE_HOUR env vars.
"""
self._scheduler.add_job(
callback,
trigger=CronTrigger(
day_of_week=day_of_week,
hour=hour,
timezone=self._timezone,
),
id="weekly_improvement",
replace_existing=True,
)
def schedule_post_publish(
self, callback, post_id: str, base_time: datetime
) -> str:
"""
Schedule a post for publishing with random jitter.
Returns the job ID.
"""
jitter = random.randint(
-self._jitter_max, self._jitter_max
)
# Ensure minimum magnitude of jitter
if abs(jitter) < self._jitter_min:
jitter = self._jitter_min if jitter >= 0 else -self._jitter_min
scheduled_time = base_time + timedelta(minutes=jitter)
job_id = f"publish_{post_id}"
self._scheduler.add_job(
callback,
trigger=DateTrigger(run_date=scheduled_time),
args=[post_id],
id=job_id,
replace_existing=True,
)
return job_id
def trigger_now(self, job_id: str) -> None:
"""Manually trigger a job immediately."""
job = self._scheduler.get_job(job_id)
if job:
job.modify(next_run_time=datetime.now(timezone.utc))# backend/models/post.py
from pydantic import BaseModel, Field
from datetime import datetime
from enum import Enum
from typing import Optional
class PostStatus(str, Enum):
PENDING = "pending"
APPROVED = "approved"
REJECTED = "rejected"
NEEDS_IMAGE = "needs_image"
PUBLISHED = "published"
FAILED = "failed"
class PostBase(BaseModel):
content: str = Field(..., max_length=280)
topic: str
angle: str
category: str
source_url: Optional[str] = None
class PostCreate(PostBase):
batch_id: str
embedding: Optional[list[float]] = None
content_hash: Optional[str] = None
class PostUpdate(BaseModel):
status: Optional[PostStatus] = None
content: Optional[str] = Field(None, max_length=280)
rejection_feedback: Optional[str] = None
media_url: Optional[str] = None
class PostResponse(PostBase):
id: str
batch_id: str
status: PostStatus
original_content: Optional[str] = None
media_url: Optional[str] = None
tweet_id: Optional[str] = None
scheduled_at: Optional[datetime] = None
published_at: Optional[datetime] = None
rejection_feedback: Optional[str] = None
created_at: datetime
updated_at: datetime
class PostListResponse(BaseModel):
posts: list[PostResponse]
total: int
page: int
page_size: int# backend/models/source.py
from pydantic import BaseModel, Field
from datetime import datetime
from enum import Enum
from typing import Optional
class SourceType(str, Enum):
RSS = "rss"
HACKERNEWS = "hackernews"
class SourceCreate(BaseModel):
type: SourceType
name: str
url: str
category: str
enabled: bool = True
class SourceUpdate(BaseModel):
name: Optional[str] = None
url: Optional[str] = None
category: Optional[str] = None
enabled: Optional[bool] = None
class SourceResponse(BaseModel):
id: str
type: SourceType
name: str
url: str
enabled: bool
category: str
last_fetched_at: Optional[datetime] = None
consecutive_errors: int = 0
created_at: datetime# backend/models/metrics.py
from pydantic import BaseModel
from datetime import datetime
from typing import Optional
class MetricsInput(BaseModel):
impressions: int = 0
likes: int = 0
reposts: int = 0
replies: int = 0
class MetricsResponse(BaseModel):
id: str
post_id: str
impressions: int
likes: int
reposts: int
replies: int
fetched_at: datetime
class AggregateMetrics(BaseModel):
post_count: int
total_impressions: int
total_likes: int
total_reposts: int
total_replies: int
engagement_rate: float
period_start: datetime
period_end: datetime# backend/models/common.py
from pydantic import BaseModel
class LoginRequest(BaseModel):
password: str
class LoginResponse(BaseModel):
token: str
expires_in: int # seconds
class ErrorResponse(BaseModel):
error: str
detail: str | None = None
code: str | None = None
class HealthResponse(BaseModel):
status: str
daily_post_count: int
monthly_post_count: int
monthly_post_limit: int
scheduler_running: bool
sources_healthy: int
sources_total: int# backend/utils/logger.py
import logging
import json
import sys
from datetime import datetime, timezone
class JSONFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
log_entry = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"level": record.levelname,
"module": record.module,
"message": record.getMessage(),
}
if record.exc_info:
log_entry["exception"] = self.formatException(record.exc_info)
return json.dumps(log_entry)
def setup_logger(name: str, level: str = "INFO") -> logging.Logger:
logger = logging.getLogger(name)
logger.setLevel(getattr(logging, level.upper()))
# Stdout handler (for Docker logs)
stdout_handler = logging.StreamHandler(sys.stdout)
stdout_handler.setFormatter(JSONFormatter())
logger.addHandler(stdout_handler)
# Rotating file handler
from logging.handlers import RotatingFileHandler
file_handler = RotatingFileHandler(
"logs/agent-x.log",
maxBytes=5 * 1024 * 1024, # 5MB
backupCount=3,
)
file_handler.setFormatter(JSONFormatter())
logger.addHandler(file_handler)
return logger# backend/utils/jitter.py
import random
def generate_jitter(min_minutes: int = 8, max_minutes: int = 12) -> int:
"""Generate random jitter in minutes. Can be positive or negative."""
magnitude = random.randint(min_minutes, max_minutes)
return magnitude if random.random() > 0.5 else -magnitude
def generate_posting_times(
count: int,
start_hour: int = 9,
end_hour: int = 23,
min_gap_minutes: int = 30,
) -> list[int]:
"""
Generate 'count' random posting minutes within the posting window.
Ensures at least min_gap_minutes between consecutive posts.
Returns sorted list of minutes-since-midnight.
"""
start_min = start_hour * 60
end_min = end_hour * 60
times = []
for _ in range(count * 10): # Try up to 10x to find valid slots
if len(times) >= count:
break
candidate = random.randint(start_min, end_min)
if all(abs(candidate - t) >= min_gap_minutes for t in times):
times.append(candidate)
times.sort()
return times[:count]# backend/utils/rate_limiter.py
import asyncio
from datetime import datetime, timedelta
from collections import defaultdict
class RateLimiter:
"""Simple rate limiter for external API calls."""
def __init__(self):
self._timestamps: dict[str, list[datetime]] = defaultdict(list)
self._lock = asyncio.Lock()
async def check_and_record(
self, key: str, max_calls: int, window_seconds: int
) -> bool:
"""
Check if a call is allowed under the rate limit.
If allowed, record the call and return True.
If not allowed, return False.
"""
async with self._lock:
now = datetime.utcnow()
cutoff = now - timedelta(seconds=window_seconds)
self._timestamps[key] = [
t for t in self._timestamps[key] if t > cutoff
]
if len(self._timestamps[key]) >= max_calls:
return False
self._timestamps[key].append(now)
return True
async def wait_if_needed(
self, key: str, max_calls: int, window_seconds: int
) -> None:
"""Block until a call is allowed under the rate limit."""
while not await self.check_and_record(key, max_calls, window_seconds):
await asyncio.sleep(1)frontend/src/app/
layout.tsx # Root layout: sidebar, header, auth check
page.tsx # Dashboard home / system overview
login/
page.tsx # Login page (public)
approvals/
page.tsx # Approval queue (pending drafts)
posts/
page.tsx # Post history (all statuses, search, filter)
sources/
page.tsx # Source management (CRUD)
analytics/
page.tsx # Analytics overview + category breakdown
settings/
page.tsx # Style config, categories, notification prefs
| Page | Description | Key Features |
|---|---|---|
| Login | Password-based login (issues JWT token) | Password input, error display, rate limit feedback. Single-user, no registration. |
| Dashboard Home | System overview | Daily/monthly post counts, API budget, recent activity, source health, quick links |
| Approvals | Approval queue | Draft cards with Approve/Reject/Edit/Needs Image actions, batch info, char count, feedback input |
| Posts | Post history | All posts with status filter, date range, search, pagination. Shows tweet link for published posts. |
| Sources | Source management | List sources with health indicators, add/edit/delete/enable/disable, test feed button |
| Analytics | Engagement metrics | Per-post metrics table, aggregate charts (7d/30d/all), category breakdown, approval rate trend |
| Settings | Configuration | Style config editor (JSON), category management, notification preferences, manual cycle trigger |
frontend/src/components/
ApprovalCard.tsx # Single draft card with action buttons
PostCard.tsx # Post display card (history view)
MetricsChart.tsx # Chart component for analytics (recharts or chart.js)
SourceList.tsx # Source list with health indicators
ImageUploader.tsx # Image upload with preview
StatusBadge.tsx # Color-coded status pill
FeedbackInput.tsx # Text input for rejection feedback
CharCounter.tsx # Character count display (color changes near 280)
Layout/
Sidebar.tsx # Navigation sidebar
Header.tsx # Page header with title and actions
- React Server Components (RSC): Used for data fetching on server side where possible (post lists, analytics data, source lists). Reduces client-side JavaScript.
- Client-side state: Used for interactive components (approval actions, form inputs, image upload, real-time character counting).
- No global state library in V1. React
useStateanduseReducerare sufficient for single-user dashboard. - Optimistic updates: Approval/reject actions update UI immediately, then confirm with the server.
The Next.js dashboard must be fully mobile-responsive. The owner (Anas Aqeel) needs to review and approve drafts on a phone, so mobile usability is a core requirement, not an afterthought.
Implementation approach:
- Use Tailwind CSS responsive utilities (
sm:,md:,lg:breakpoints) for all layout decisions. - The default layout targets mobile-first; larger breakpoints add multi-column layouts.
Key mobile considerations:
| Area | Mobile Behavior |
|---|---|
| Approval queue | Must be fully usable on phone. Approve/Reject/Edit actions accessible with one tap. |
| Post cards | Cards stack vertically on small screens (< sm:). No horizontal scrolling required. |
| Navigation | Sidebar collapses to a hamburger menu on mobile. |
| Tap targets | All interactive elements (buttons, links, toggles) have a minimum tap target size of 44x44px per Apple HIG / WCAG guidelines. |
| Forms | Feedback input and edit text areas use full-width on mobile. Character counter remains visible. |
| Charts | Analytics charts resize responsively. Legends collapse or move below the chart on small screens. |
| Image upload | Image uploader supports camera capture on mobile devices (accept="image/*" with capture attribute). |
Breakpoint strategy (Tailwind defaults):
| Breakpoint | Min Width | Target |
|---|---|---|
| Default | 0px | Mobile phones (portrait) |
sm: |
640px | Large phones (landscape) |
md: |
768px | Tablets |
lg: |
1024px | Laptops / desktops |
xl: |
1280px | Large desktops |
// frontend/src/lib/api.ts
const API_BASE = process.env.NEXT_PUBLIC_API_URL || "/api";
async function fetchAPI<T>(
endpoint: string,
options: RequestInit = {}
): Promise<T> {
const token = getAuthToken(); // from cookie or localStorage
const response = await fetch(`${API_BASE}${endpoint}`, {
...options,
headers: {
"Content-Type": "application/json",
...(token ? { Authorization: `Bearer ${token}` } : {}),
...options.headers,
},
});
if (response.status === 401) {
// Redirect to login
window.location.href = "/login";
throw new Error("Unauthorized");
}
if (!response.ok) {
const error = await response.json();
throw new Error(error.detail || "API error");
}
return response.json();
}
// Typed API methods
export const api = {
// Auth
login: (password: string) =>
fetchAPI<{ token: string }>("/auth/login", {
method: "POST",
body: JSON.stringify({ password }),
}),
// Posts
getPosts: (params?: { status?: string; page?: number }) =>
fetchAPI<PostListResponse>(`/posts?${new URLSearchParams(params as any)}`),
approvePost: (id: string) =>
fetchAPI<PostResponse>(`/posts/${id}`, {
method: "PATCH",
body: JSON.stringify({ status: "approved" }),
}),
rejectPost: (id: string, feedback: string) =>
fetchAPI<PostResponse>(`/posts/${id}`, {
method: "PATCH",
body: JSON.stringify({ status: "rejected", rejection_feedback: feedback }),
}),
editAndApprove: (id: string, content: string) =>
fetchAPI<PostResponse>(`/posts/${id}`, {
method: "PATCH",
body: JSON.stringify({ status: "approved", content }),
}),
uploadImage: (id: string, file: File) => {
const formData = new FormData();
formData.append("image", file);
return fetchAPI<PostResponse>(`/posts/${id}/image`, {
method: "POST",
headers: {}, // Let browser set Content-Type for FormData
body: formData,
});
},
// Sources
getSources: () => fetchAPI<SourceResponse[]>("/sources"),
createSource: (data: SourceCreate) =>
fetchAPI<SourceResponse>("/sources", {
method: "POST",
body: JSON.stringify(data),
}),
// Analytics
getOverview: (period?: string) =>
fetchAPI<AggregateMetrics>(`/analytics/overview?period=${period || "7d"}`),
// System
getStatus: () => fetchAPI<HealthResponse>("/status"),
triggerCycle: () =>
fetchAPI<{ message: string }>("/cycle/trigger", { method: "POST" }),
};sequenceDiagram
participant User
participant Layout as Next.js Layout
participant Login as /login Page
participant API as FastAPI Backend
User->>Layout: Navigate to any page
Layout->>Layout: Check for JWT token (localStorage or httpOnly cookie)
alt No token found
Layout->>Login: Redirect to /login
User->>Login: Enter password
Login->>API: POST /api/auth/login { password }
API->>API: Verify password hash (bcrypt)
API-->>Login: { token: "jwt...", expires_in: 86400 }
Login->>Login: Store JWT token
end
User->>API: All subsequent API calls include Authorization: Bearer <token>
alt API returns 401
API-->>Layout: 401 Unauthorized
Layout->>Login: Redirect to /login
end
| Service | Image | Ports | Purpose |
|---|---|---|---|
backend |
Custom Python image | 8000 (internal) | FastAPI application |
frontend |
Custom Node.js image | 3000 (internal) | Next.js application |
nginx |
nginx:alpine | 80, 443 (external) | Reverse proxy, TLS termination |
Routes /api/* to the backend container and all other routes to the frontend container. Handles TLS termination via Let's Encrypt certificates (or Caddy as alternative).
All services use restart: unless-stopped to ensure automatic recovery from crashes. The backend persists all state to Supabase, so restarts are safe.
- Backend:
GET /api/statusreturns system health (HTTP 200 if healthy) - Frontend:
GET /returns page (HTTP 200 if healthy) - Nginx: Docker health check with
curlto localhost
-- Enable the `vector` extension (commonly known as pgvector) for embedding storage and similarity search.
-- Note: Supabase uses the official PostgreSQL extension name `vector` (the project is commonly known as pgvector).
CREATE EXTENSION IF NOT EXISTS vector;
-- Enable UUID generation
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
-- Custom ENUM types
CREATE TYPE post_status AS ENUM (
'pending',
'approved',
'rejected',
'needs_image',
'published',
'failed'
);
CREATE TYPE source_type AS ENUM (
'rss',
'hackernews'
);
CREATE TYPE proposal_status AS ENUM (
'pending',
'approved',
'rejected'
);
CREATE TYPE job_status AS ENUM (
'pending',
'running',
'completed',
'failed'
);-- ============================================================
-- POSTS: Core content table
-- ============================================================
CREATE TABLE posts (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
batch_id TEXT NOT NULL,
content TEXT NOT NULL CHECK (char_length(content) <= 280),
original_content TEXT,
status post_status NOT NULL DEFAULT 'pending',
topic TEXT NOT NULL,
angle TEXT NOT NULL,
category TEXT NOT NULL,
source_url TEXT,
media_url TEXT,
tweet_id TEXT,
scheduled_at TIMESTAMPTZ,
published_at TIMESTAMPTZ,
rejection_feedback TEXT,
embedding vector(768), -- Vertex AI text-embedding-004 dimension
content_hash TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
-- Indexes for posts
CREATE INDEX idx_posts_status ON posts (status);
CREATE INDEX idx_posts_batch_id ON posts (batch_id);
CREATE INDEX idx_posts_created_at ON posts (created_at DESC);
CREATE INDEX idx_posts_category ON posts (category);
CREATE INDEX idx_posts_content_hash ON posts (content_hash);
CREATE INDEX idx_posts_published_at ON posts (published_at DESC);
-- HNSW index for fast cosine similarity search on embeddings
CREATE INDEX idx_posts_embedding ON posts
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);
-- Auto-update updated_at timestamp
CREATE OR REPLACE FUNCTION update_updated_at()
RETURNS TRIGGER AS $$
BEGIN
NEW.updated_at = NOW();
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
CREATE TRIGGER trigger_posts_updated_at
BEFORE UPDATE ON posts
FOR EACH ROW
EXECUTE FUNCTION update_updated_at();
-- ============================================================
-- POST_METRICS: Engagement metrics per post
-- ============================================================
CREATE TABLE post_metrics (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
post_id UUID NOT NULL REFERENCES posts(id) ON DELETE CASCADE,
impressions INTEGER NOT NULL DEFAULT 0,
likes INTEGER NOT NULL DEFAULT 0,
reposts INTEGER NOT NULL DEFAULT 0,
replies INTEGER NOT NULL DEFAULT 0,
fetched_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_post_metrics_post_id ON post_metrics (post_id);
CREATE INDEX idx_post_metrics_fetched_at ON post_metrics (fetched_at DESC);
-- ============================================================
-- SOURCES: Research sources (RSS feeds, HN config)
-- ============================================================
CREATE TABLE sources (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
type source_type NOT NULL,
name TEXT NOT NULL,
url TEXT NOT NULL,
enabled BOOLEAN NOT NULL DEFAULT TRUE,
category TEXT NOT NULL,
last_fetched_at TIMESTAMPTZ,
consecutive_errors INTEGER NOT NULL DEFAULT 0,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_sources_enabled ON sources (enabled);
CREATE INDEX idx_sources_type ON sources (type);
-- ============================================================
-- STYLE_CONFIG: Style configuration versions
-- ============================================================
CREATE TABLE style_config (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
config JSONB NOT NULL,
is_active BOOLEAN NOT NULL DEFAULT FALSE,
proposed_by TEXT NOT NULL DEFAULT 'human',
approved_at TIMESTAMPTZ,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_style_config_is_active ON style_config (is_active)
WHERE is_active = TRUE;
-- ============================================================
-- IMPROVEMENT_PROPOSALS: Weekly improvement suggestions
-- ============================================================
CREATE TABLE improvement_proposals (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
analysis TEXT NOT NULL,
proposed_changes JSONB NOT NULL,
status proposal_status NOT NULL DEFAULT 'pending',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_improvement_proposals_status
ON improvement_proposals (status);
-- ============================================================
-- SCHEDULER_JOBS: Persistent job tracking
-- ============================================================
CREATE TABLE scheduler_jobs (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
job_type TEXT NOT NULL,
scheduled_at TIMESTAMPTZ NOT NULL,
status job_status NOT NULL DEFAULT 'pending',
payload JSONB,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_scheduler_jobs_status ON scheduler_jobs (status);
CREATE INDEX idx_scheduler_jobs_scheduled_at
ON scheduler_jobs (scheduled_at);
-- ============================================================
-- CATEGORIES: Content categories
-- ============================================================
CREATE TABLE categories (
id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
name TEXT NOT NULL UNIQUE,
enabled BOOLEAN NOT NULL DEFAULT TRUE,
description TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_categories_enabled ON categories (enabled);For V1 (single user, service role key), RLS is minimal but established for future multi-user expansion:
-- Enable RLS on all tables
ALTER TABLE posts ENABLE ROW LEVEL SECURITY;
ALTER TABLE post_metrics ENABLE ROW LEVEL SECURITY;
ALTER TABLE sources ENABLE ROW LEVEL SECURITY;
ALTER TABLE style_config ENABLE ROW LEVEL SECURITY;
ALTER TABLE improvement_proposals ENABLE ROW LEVEL SECURITY;
ALTER TABLE scheduler_jobs ENABLE ROW LEVEL SECURITY;
ALTER TABLE categories ENABLE ROW LEVEL SECURITY;
-- V1: Allow full access via service role key
-- (Service role key bypasses RLS by default in Supabase)
-- These policies allow authenticated access via anon key as fallback:
CREATE POLICY "Allow all for authenticated users" ON posts
FOR ALL USING (true) WITH CHECK (true);
CREATE POLICY "Allow all for authenticated users" ON post_metrics
FOR ALL USING (true) WITH CHECK (true);
CREATE POLICY "Allow all for authenticated users" ON sources
FOR ALL USING (true) WITH CHECK (true);
CREATE POLICY "Allow all for authenticated users" ON style_config
FOR ALL USING (true) WITH CHECK (true);
CREATE POLICY "Allow all for authenticated users" ON improvement_proposals
FOR ALL USING (true) WITH CHECK (true);
CREATE POLICY "Allow all for authenticated users" ON scheduler_jobs
FOR ALL USING (true) WITH CHECK (true);
CREATE POLICY "Allow all for authenticated users" ON categories
FOR ALL USING (true) WITH CHECK (true);-- Default categories (from PRD Section 13.2)
INSERT INTO categories (name, enabled, description) VALUES
('AI tools', TRUE, 'AI tools and products'),
('AI startups', TRUE, 'AI startups and funding'),
('Engineering insights', TRUE, 'Engineering insights and best practices'),
('Research papers', TRUE, 'Research papers (simplified explanations)'),
('Founder lessons', TRUE, 'Founder lessons and startup advice'),
('Open source', TRUE, 'Open source projects'),
('AI trends', TRUE, 'AI industry trends and predictions'),
('Developer tools', TRUE, 'Developer tools and productivity'),
('Tech careers', TRUE, 'Tech career advice'),
('Contrarian takes', TRUE, 'Contrarian takes on popular tech topics');
-- Default style config
INSERT INTO style_config (config, is_active, proposed_by, approved_at) VALUES (
'{
"tone": "Knowledgeable, conversational tech voice. Slightly opinionated, not corporate.",
"cta_frequency": 0.7,
"emoji_range": [0, 4],
"hashtag_range": [0, 2],
"example_posts": [],
"hook_styles": [
"question",
"bold_claim",
"contrarian_take",
"experience_framing"
]
}'::JSONB,
TRUE,
'human',
NOW()
);
-- Default HN source
INSERT INTO sources (type, name, url, category, enabled) VALUES
('hackernews', 'Hacker News Top Stories',
'https://hacker-news.firebaseio.com/v0/topstories.json',
'AI tools', TRUE);flowchart TD
Sched["Scheduler (07:00 America/Los_Angeles)"] --> Research["research.py"]
Research --> HN["HN API (GET /topstories.json)\nStory details: title, score, url"]
Research --> RSS["RSS Feeds (GET feed URLs)\nFeed items: title, link, summary"]
HN --> Filter["Filter by enabled categories"]
RSS --> Filter
Filter --> Ranker["topic_ranker.py\nLLM: rank topics, expand angles\nRanked topic-angle pairs (JSON)"]
Ranker --> Generator["content_generator.py\nLLM: draft posts\nRaw drafts (text)"]
Generator --> Enforce["280-char enforcement + retry"]
Enforce --> Unique["uniqueness_checker.py\nLLM: compute embeddings\nEmbedding vectors"]
Unique --> Compare["Supabase: compare against stored embeddings"]
Compare --> Store["Store unique drafts in Supabase (status=pending)"]
Store --> Notify["Send batch notification via Resend"]
flowchart TD
Open["Owner opens Dashboard"] --> Fetch["GET /api/posts?status=pending"]
Fetch --> Queue["Approval Queue renders draft cards"]
Queue --> Approve["Approve\nPATCH /api/posts/{id}\nstatus: approved"]
Queue --> Reject["Reject\nPATCH /api/posts/{id}\nstatus: rejected + feedback"]
Queue --> EditApprove["Edit + Approve\nPATCH /api/posts/{id}\nstatus: approved + new content"]
Queue --> NeedsImage["Needs Image\nPATCH /api/posts/{id}\nstatus: needs_image"]
Approve --> Schedule["Schedule post (APScheduler + jitter)"]
Reject --> StoreFeedback["Store feedback in Supabase\n(used in future prompts)"]
EditApprove --> StoreOriginal["Store original_content, update content\nSchedule post"]
NeedsImage --> Upload["POST /api/posts/{id}/image (multipart)\nUpload to Supabase Storage\nUpdate media_url\nOwner can then Approve"]
flowchart TD
Fire["APScheduler fires publish job\n(scheduled_at + jitter)"] --> Checks["publisher.py Pre-publish checks:\nDaily count < 5?\nMonthly count < 500?\nContent <= 280 chars?\nStatus == approved?"]
Checks -->|FAIL| Hold["Hold post, alert via dashboard"]
Checks -->|PASS| Media{"Has media_url?"}
Media -->|YES| Download["Download image from Supabase Storage\ntweepy API v1.1: media_upload()\nObtain media_id"]
Media -->|NO| Tweet
Download --> Tweet["tweepy Client v2:\ncreate_tweet(text, media_ids?)"]
Tweet -->|SUCCESS| StoreResult["Store tweet_id, published_at in Supabase\nIncrement daily/monthly counters\nCheck budget thresholds (80%, 95%)"]
Tweet -->|FAIL| Retry["Wait 60s, retry once"]
Retry -->|SUCCESS| StoreResult
Retry -->|FAIL| Failed["Mark status=failed, alert owner"]
flowchart TD
subgraph ManualEntry["Manual Entry"]
ME1["Dashboard analytics page"] --> ME2["POST /api/posts/{id}/metrics"] --> ME3["Upsert into post_metrics table"]
end
subgraph OnDemand["On-Demand Fetch"]
OD1["Dashboard 'Fetch Metrics' button"] --> OD2["POST /api/posts/fetch-metrics"] --> OD3["For each recent published post:\ntweepy Client v2: get_tweet(metrics)\nUpsert into post_metrics table"]
end
subgraph WeeklyAnalysis["Weekly Analysis"]
WA1["Scheduler (Sunday 06:00 America/Los_Angeles, configurable)"] --> WA2["improvement_engine.py"]
WA2 --> WA3["Query: published posts + metrics (7 days)\nQuery: rejected posts + feedback (7 days)\nQuery: current style config"]
WA3 --> WA4["LLM: generate improvement proposal"]
WA4 --> WA5["Store proposal (status=pending)"]
WA5 --> WA6["Owner reviews in Dashboard"]
WA6 -->|Approve| WA7["Update style_config"]
WA6 -->|Reject| WA8["Log and discard"]
end
All endpoints are prefixed with /api. All endpoints except /api/auth/login and /api/status require a valid JWT token in the Authorization: Bearer <token> header.
| Method | Path | Auth | Rate Limit | Description |
|---|---|---|---|---|
POST |
/api/auth/login |
None | 5 per 5 min per IP | Authenticate and get JWT token |
Request:
{
"password": "string"
}Response (200):
{
"token": "eyJhbGciOiJIUzI1NiIs...",
"expires_in": 86400
}Response (401):
{
"error": "unauthorized",
"detail": "Invalid password"
}Response (429):
{
"error": "rate_limited",
"detail": "Too many login attempts. Try again in 5 minutes."
}| Method | Path | Auth | Description |
|---|---|---|---|
GET |
/api/posts |
Required | List posts with filtering |
GET |
/api/posts/{id} |
Required | Get single post |
PATCH |
/api/posts/{id} |
Required | Update post (approve, reject, edit) |
POST |
/api/posts/{id}/image |
Required | Upload image for a post |
GET |
/api/posts/{id}/metrics |
Required | Get metrics for a post |
POST |
/api/posts/fetch-metrics |
Required | Trigger on-demand metrics fetch |
GET /api/posts
Query parameters:
status(optional): Filter by status (pending,approved,rejected,needs_image,published,failed)category(optional): Filter by categorybatch_id(optional): Filter by batchstart_date(optional): Filter by date range start (ISO 8601)end_date(optional): Filter by date range end (ISO 8601)search(optional): Full-text search on contentpage(default: 1): Page numberpage_size(default: 20): Items per page
Response (200):
{
"posts": [
{
"id": "uuid",
"batch_id": "Batch 2026-03-08 #1",
"content": "OpenAI's new model cuts inference costs by 40%...",
"original_content": null,
"status": "pending",
"topic": "OpenAI GPT-5 launch",
"angle": "prediction",
"category": "AI tools",
"source_url": "https://news.ycombinator.com/item?id=...",
"media_url": null,
"tweet_id": null,
"scheduled_at": null,
"published_at": null,
"rejection_feedback": null,
"created_at": "2026-03-08T07:15:00Z",
"updated_at": "2026-03-08T07:15:00Z"
}
],
"total": 45,
"page": 1,
"page_size": 20
}PATCH /api/posts/{id}
Request body (all fields optional):
{
"status": "approved | rejected | needs_image",
"content": "Edited content text (max 280 chars)",
"rejection_feedback": "Feedback text when rejecting"
}Logic:
- If
status=approvedandcontentis provided: store current content asoriginal_content, update content, set status to approved. - If
status=rejected:rejection_feedbackis required. - If
status=needs_image: marks post as awaiting image upload.
POST /api/posts/{id}/image
Multipart form data:
image: Image file (JPEG, PNG, GIF, WebP, max 5MB)
Server-side validation (mandatory):
- Validate MIME type using the
python-magiclibrary (reads file header bytes, not just extension) - Allowed MIME types:
image/jpeg,image/png,image/gif,image/webp - Reject files exceeding 5MB with HTTP 413 status
- Reject non-image files (e.g.,
.exerenamed to.jpg) with HTTP 422INVALID_IMAGE_FORMAT - Client-side validation alone is insufficient; server-side checks prevent bypass via
curlor API tools
Response (200):
{
"id": "uuid",
"media_url": "https://your-project.supabase.co/storage/v1/object/public/post-images/...",
"status": "needs_image"
}| Method | Path | Auth | Description |
|---|---|---|---|
GET |
/api/sources |
Required | List all sources |
POST |
/api/sources |
Required | Add a new source |
PATCH |
/api/sources/{id} |
Required | Update a source |
DELETE |
/api/sources/{id} |
Required | Delete a source |
POST /api/sources
{
"type": "rss",
"name": "TechCrunch AI",
"url": "https://techcrunch.com/category/artificial-intelligence/feed/",
"category": "AI startups",
"enabled": true
}PATCH /api/sources/{id}
{
"name": "Updated Name",
"url": "https://new-url.com/feed",
"category": "AI tools",
"enabled": false
}| Method | Path | Auth | Description |
|---|---|---|---|
GET |
/api/analytics/overview |
Required | Aggregate metrics |
GET |
/api/analytics/categories |
Required | Per-category performance |
GET /api/analytics/overview
Query parameters:
period(default:7d): Time period (7d,30d,all)
Response (200):
{
"post_count": 28,
"total_impressions": 45000,
"total_likes": 890,
"total_reposts": 120,
"total_replies": 67,
"engagement_rate": 0.0213,
"approval_rate": 0.72,
"posts_per_day_avg": 4.0,
"period_start": "2026-03-01T00:00:00Z",
"period_end": "2026-03-08T00:00:00Z"
}GET /api/analytics/categories
Response (200):
{
"categories": [
{
"name": "AI tools",
"post_count": 12,
"avg_engagement_rate": 0.032,
"total_likes": 450,
"total_replies": 34
}
]
}| Method | Path | Auth | Description |
|---|---|---|---|
GET |
/api/style-config |
Required | Get active style config |
PUT |
/api/style-config |
Required | Update style config |
PUT /api/style-config
{
"config": {
"tone": "Knowledgeable, conversational tech voice.",
"cta_frequency": 0.7,
"emoji_range": [0, 4],
"hashtag_range": [0, 2],
"example_posts": ["Example tweet 1", "Example tweet 2"],
"hook_styles": ["question", "bold_claim", "contrarian_take"]
}
}| Method | Path | Auth | Description |
|---|---|---|---|
GET |
/api/improvements |
Required | List proposals |
PATCH |
/api/improvements/{id} |
Required | Approve or reject a proposal |
PATCH /api/improvements/{id}
{
"status": "approved | rejected"
}When approved, the system creates a new style_config entry with is_active = TRUE and deactivates the previous one.
| Method | Path | Auth | Description |
|---|---|---|---|
GET |
/api/status |
None | System health check |
POST |
/api/cycle/trigger |
Required | Manually trigger research cycle |
GET /api/status
Response (200):
{
"status": "healthy",
"daily_post_count": 3,
"monthly_post_count": 67,
"monthly_post_limit": 500,
"scheduler_running": true,
"sources_healthy": 5,
"sources_total": 6
}| Method | Path | Auth | Description |
|---|---|---|---|
GET |
/api/categories |
Required | List all categories |
POST |
/api/categories |
Required | Add a category |
PATCH |
/api/categories/{id} |
Required | Update a category |
POST /api/categories
{
"name": "AI regulation",
"description": "AI policy and regulation news",
"enabled": true
}All errors return a consistent JSON structure:
{
"error": "error_code",
"detail": "Human-readable description of what went wrong",
"code": "DOMAIN_SPECIFIC_CODE"
}| Status Code | Usage |
|---|---|
| 200 | Successful GET, PATCH, PUT |
| 201 | Successful POST (resource created) |
| 400 | Invalid request body, validation errors |
| 401 | Missing or invalid auth token |
| 403 | Valid token but insufficient permissions |
| 404 | Resource not found |
| 409 | Conflict (e.g., duplicate resource) |
| 422 | Unprocessable entity (valid JSON but invalid data) |
| 429 | Rate limit exceeded |
| 500 | Internal server error |
| Code | Description |
|---|---|
POST_NOT_FOUND |
Post ID does not exist |
POST_ALREADY_PUBLISHED |
Cannot modify a published post |
POST_CONTENT_TOO_LONG |
Content exceeds 280 characters |
REJECTION_FEEDBACK_REQUIRED |
Feedback text required when rejecting |
DAILY_LIMIT_REACHED |
Daily post limit reached |
MONTHLY_LIMIT_REACHED |
Monthly post budget exhausted |
SOURCE_FETCH_FAILED |
Failed to fetch/parse source |
LLM_ERROR |
LLM provider returned an error |
PUBLISH_FAILED |
Failed to publish to X/Twitter |
IMAGE_TOO_LARGE |
Uploaded image exceeds 5MB |
INVALID_IMAGE_FORMAT |
Image is not JPEG, PNG, or GIF |
| Aspect | Details |
|---|---|
| Library | tweepy v4.14+ |
| Auth method | OAuth 1.0a (API Key, API Secret, Access Token, Access Token Secret) |
| Tweet creation | X API v2 via tweepy.Client.create_tweet() |
| Media upload | X API v1.1 via tweepy.API.media_upload() |
| Metrics fetch | X API v2 via tweepy.Client.get_tweet(tweet_fields=["public_metrics", "non_public_metrics"]). Note: impression_count requires non_public_metrics (X API Basic tier $100/mo or higher). On Free tier, impressions default to 0. |
| Rate limits | Free tier: 500 posts/month, 50 requests/15 min for tweet creation |
| Rate limit strategy | Track monthly/daily counts in Supabase. Alerts at 80% and 95%. Hard block at limit. |
| Retry strategy | On failure: wait 60s, retry once. If retry fails, mark as "failed" and alert. |
| Error handling | Log full error context. Handle 403 (suspended) as critical alert requiring manual intervention. |
| Aspect | Details |
|---|---|
| Library | google-cloud-aiplatform v1.40+ |
| Auth method | Google Cloud service account credentials (JSON key file) |
| Text generation | Gemini 1.5 Flash (cost-effective for drafting) |
| Embeddings | text-embedding-004 (768 dimensions) |
| Rate limits | Generous for paid tier. Monitor credit usage. |
| Rate limit strategy | Track LLM call count and estimated cost per session. Cache embeddings. |
| Retry strategy | On transient failure: retry with exponential backoff (1s, 2s, 4s). Max 3 retries. |
| Fallback | If credits exhaust, swap adapter to Groq (Llama 3.1 70B free tier) via LLM interface. |
| Aspect | Details |
|---|---|
| Protocol | HTTP REST (Firebase) |
| Auth method | None (public API) |
| Endpoints | GET /v0/topstories.json, GET /v0/item/{id}.json |
| Rate limits | No official limits. Self-imposed: max 1 request/second. |
| Rate limit strategy | Use asyncio.sleep(1) between story detail fetches. |
| Retry strategy | On failure: skip HN source, continue with RSS feeds. |
| Error handling | Log error, increment source error counter. If HN is down, research cycle proceeds with RSS only. |
| Aspect | Details |
|---|---|
| Library | resend v0.7+ |
| Auth method | API key |
| Endpoint | POST https://api.resend.com/emails |
| Rate limits | Free tier: 100 emails/day, 3,000/month |
| Rate limit strategy | Batch notifications where possible. Typical usage: 1-3 emails/day. |
| Retry strategy | On failure: log and skip. Email is non-critical; dashboard is primary notification channel. |
| Error handling | Log error. Never block the main workflow for email failures. |
| Aspect | Details |
|---|---|
| Library | supabase-py v2.0+ |
| Auth method | Service role key (bypasses RLS) |
| Access pattern | REST API (not direct Postgres connection) |
| Free tier limits | 500MB database, 1GB storage, 2GB bandwidth/month |
| Rate limit strategy | Monitor row counts and storage. Supabase free tier is generous for V1 scale. |
| Retry strategy | On connection loss: retry with backoff (1s, 2s, 4s). Queue operations in memory if down > 5 seconds. Alert if down > 5 minutes. |
| Error handling | Log full error. Critical alerts for persistent connection failures. |
sequenceDiagram
participant Browser
participant FastAPI as FastAPI Backend
Browser->>FastAPI: POST /api/auth/login { password: "..." }
FastAPI->>FastAPI: Verify password against hashed DASHBOARD_PASSWORD
FastAPI-->>Browser: { token: "jwt...", expires_in: 86400 }
Browser->>Browser: Store JWT token (localStorage or httpOnly cookie)
Browser->>FastAPI: GET /api/posts (Authorization: Bearer <token>)
FastAPI->>FastAPI: verify_token(): Decode JWT, check expiry, check sub="owner"
FastAPI-->>Browser: 200 OK (JSON response)
V1 Authentication Model:
Agent-X V1 uses a simple password-based login that issues a JWT token. This is a single-user system with no registration flow. The authentication model works as follows:
- Password storage: The
DASHBOARD_PASSWORDenvironment variable holds the plaintext password. The backend hashes it at startup using bcrypt and compares hashed values during login. - Token issuance: On successful password verification, the backend issues a JWT token (HS256, 24-hour expiry) with
sub: "owner". - API authentication: All protected endpoints require a valid JWT token in the
Authorization: Bearer <token>header. The system uses JWT tokens for API authentication, not session cookies. - Single-user: There is no registration endpoint. The system is designed for exactly one user (the owner). The only way to authenticate is by providing the correct
DASHBOARD_PASSWORD. - V2 upgrade path: In V2, this simple password + JWT mechanism will be replaced with Supabase Auth, which provides OAuth, email/password registration, and built-in JWT management.
| Control | Implementation |
|---|---|
| CORS | Backend accepts requests only from the configured FRONTEND_URL. |
| Secret management | All secrets in .env file. Never committed to git. .env.example with placeholder values provided. |
| HTTPS | Enforced via Nginx with Let's Encrypt certificates. HTTP redirects to HTTPS. |
| Login rate limiting | 5 attempts per 5 minutes per IP address. Returns HTTP 429 on exceeded. |
| JWT expiry | Tokens expire after 24 hours. No refresh token mechanism in V1. |
| Input validation | All request bodies validated by Pydantic models. Content limited to 280 chars. Image uploads limited to 5MB. |
| SQL injection | Prevented by using Supabase REST API (parameterized queries). No raw SQL from user input. |
| XSS | Next.js auto-escapes rendered content. No dangerouslySetInnerHTML usage. |
| File upload | Only JPEG, PNG, GIF, WebP accepted. Size limit (5MB) enforced server-side. MIME type validated server-side using python-magic (reads file header, not just extension). Malformed files (e.g., .exe renamed to .jpg) are rejected. Stored in Supabase Storage (not local filesystem). |
# docker-compose.yml
version: "3.8"
services:
backend:
build:
context: ./backend
dockerfile: Dockerfile
container_name: agentx-backend
restart: unless-stopped
ports:
- "8000:8000"
env_file:
- .env
environment:
- GOOGLE_APPLICATION_CREDENTIALS=/app/credentials.json
volumes:
- backend-logs:/app/logs
- ${GOOGLE_APPLICATION_CREDENTIALS}:/app/credentials.json:ro
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/api/status"]
interval: 30s
timeout: 10s
retries: 3
start_period: 15s
deploy:
resources:
limits:
memory: 4G
cpus: "1.5"
frontend:
build:
context: ./frontend
dockerfile: Dockerfile
container_name: agentx-frontend
restart: unless-stopped
ports:
- "3000:3000"
environment:
- NEXT_PUBLIC_API_URL=/api
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:3000"]
interval: 30s
timeout: 10s
retries: 3
start_period: 15s
deploy:
resources:
limits:
memory: 2G
cpus: "0.5"
nginx:
image: nginx:alpine
container_name: agentx-nginx
restart: unless-stopped
ports:
- "80:80"
- "443:443"
volumes:
- ./nginx/nginx.conf:/etc/nginx/nginx.conf:ro
- ./nginx/certs:/etc/nginx/certs:ro
- nginx-logs:/var/log/nginx
depends_on:
backend:
condition: service_healthy
frontend:
condition: service_healthy
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:80/health"]
interval: 30s
timeout: 10s
retries: 3
volumes:
backend-logs:
nginx-logs:# backend/Dockerfile
FROM python:3.12-slim
WORKDIR /app
# Install system dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
curl \
&& rm -rf /var/lib/apt/lists/*
# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# Copy application code
COPY . .
# Create log directory
RUN mkdir -p /app/logs
# Expose port
EXPOSE 8000
# Run with uvicorn
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "1"]Note: Single worker because APScheduler runs in-process. Multiple workers would create duplicate scheduled jobs.
# frontend/Dockerfile
FROM node:18-alpine AS base
# Install dependencies
FROM base AS deps
WORKDIR /app
COPY package.json package-lock.json* ./
RUN npm ci
# Build the application
FROM base AS builder
WORKDIR /app
COPY --from=deps /app/node_modules ./node_modules
COPY . .
RUN npm run build
# Production image
FROM base AS runner
WORKDIR /app
ENV NODE_ENV=production
RUN addgroup --system --gid 1001 nodejs
RUN adduser --system --uid 1001 nextjs
COPY --from=builder /app/public ./public
COPY --from=builder --chown=nextjs:nodejs /app/.next/standalone ./
COPY --from=builder --chown=nextjs:nodejs /app/.next/static ./.next/static
USER nextjs
EXPOSE 3000
ENV PORT=3000
ENV HOSTNAME="0.0.0.0"
CMD ["node", "server.js"]# nginx/nginx.conf
worker_processes auto;
events {
worker_connections 1024;
}
http {
# Logging
log_format main '$remote_addr - $remote_user [$time_local] '
'"$request" $status $body_bytes_sent '
'"$http_referer" "$http_user_agent"';
access_log /var/log/nginx/access.log main;
error_log /var/log/nginx/error.log warn;
# Gzip compression
gzip on;
gzip_types text/plain text/css application/json application/javascript;
# Rate limiting for login endpoint
limit_req_zone $binary_remote_addr zone=login:10m rate=5r/m;
# Redirect HTTP to HTTPS
server {
listen 80;
server_name your-domain.com;
location /health {
return 200 "OK";
add_header Content-Type text/plain;
}
location / {
return 301 https://$server_name$request_uri;
}
}
# HTTPS server
server {
listen 443 ssl;
server_name your-domain.com;
ssl_certificate /etc/nginx/certs/fullchain.pem;
ssl_certificate_key /etc/nginx/certs/privkey.pem;
ssl_protocols TLSv1.2 TLSv1.3;
ssl_ciphers HIGH:!aNULL:!MD5;
# Security headers
add_header X-Frame-Options DENY;
add_header X-Content-Type-Options nosniff;
add_header X-XSS-Protection "1; mode=block";
add_header Strict-Transport-Security "max-age=31536000; includeSubDomains";
# API routes -> backend
location /api/ {
proxy_pass http://backend:8000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
# Increase timeout for LLM calls
proxy_read_timeout 120s;
proxy_send_timeout 120s;
# Login rate limiting
location /api/auth/login {
limit_req zone=login burst=5 nodelay;
proxy_pass http://backend:8000;
}
}
# Image upload: increase body size
location /api/posts/ {
client_max_body_size 10M;
proxy_pass http://backend:8000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
# Everything else -> frontend
location / {
proxy_pass http://frontend:3000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
}
}All configuration is loaded from a single .env file at the project root. The backend uses python-dotenv to load variables. The frontend uses Next.js built-in env support.
Required environment variables (see PRD Appendix B for full list):
| Variable | Service | Description |
|---|---|---|
SUPABASE_URL |
Backend | Supabase project URL |
SUPABASE_SERVICE_ROLE_KEY |
Backend | Service role key (bypasses RLS) |
X_API_KEY |
Backend | X/Twitter API key |
X_API_SECRET |
Backend | X/Twitter API secret |
X_ACCESS_TOKEN |
Backend | X/Twitter access token |
X_ACCESS_TOKEN_SECRET |
Backend | X/Twitter access token secret |
GOOGLE_CLOUD_PROJECT |
Backend | GCP project ID |
GOOGLE_APPLICATION_CREDENTIALS |
Backend | Path to service account JSON (on host; mounted as /app/credentials.json inside container via Docker volume) |
RESEND_API_KEY |
Backend | Resend email API key |
NOTIFICATION_EMAIL |
Backend | Owner's email address |
DASHBOARD_PASSWORD |
Backend | Dashboard login password |
DAILY_POST_LIMIT |
Backend | Max posts per day (default: 5) |
MONTHLY_POST_LIMIT |
Backend | Max posts per month (default: 500) |
SIMILARITY_THRESHOLD |
Backend | Cosine similarity threshold (default: 0.85) |
RESEARCH_SCHEDULE_HOUR |
Backend | Hour for daily research (default: 7) |
RESEARCH_SCHEDULE_TIMEZONE |
Backend | Timezone (default: America/Los_Angeles) |
POSTING_HOURS_START |
Backend | Start of posting window (default: 9) |
POSTING_HOURS_END |
Backend | End of posting window (default: 23) |
JITTER_MINUTES_MIN |
Backend | Min jitter magnitude (default: 8) |
JITTER_MINUTES_MAX |
Backend | Max jitter magnitude (default: 12) |
IMPROVEMENT_SCHEDULE_DAY |
Backend | Day of week for weekly improvement cycle (default: sun; options: mon, tue, wed, thu, fri, sat, sun) |
IMPROVEMENT_SCHEDULE_HOUR |
Backend | Hour for weekly improvement cycle (default: 6, i.e. 06:00 America/Los_Angeles = ~7 PM PKT) |
LOG_LEVEL |
Backend | Logging level (default: INFO) |
LLM_PROVIDER |
Backend | LLM adapter to use (default: vertex_ai; options: vertex_ai, groq, mock) |
PUBLISH_PLATFORM |
Backend | Publisher adapter to use (default: x_twitter; options: x_twitter, mock) |
FRONTEND_URL |
Backend | Frontend URL for CORS (e.g., https://agentx.yourdomain.com) |
NEXT_PUBLIC_API_URL |
Frontend | Backend API URL (default: /api) |
- Backend: Structured JSON logs to stdout (captured by Docker) and rotating file handler (
/app/logs/agent-x.log, 5MB, 3 backups). - Nginx: Access and error logs in
/var/log/nginx/(volume-mounted). - Frontend: Next.js server logs to stdout.
All Docker container logs can be accessed via docker logs <container> or aggregated with a log driver.
| What | How |
|---|---|
| Container health | Docker health checks + restart policies |
| Application health | GET /api/status endpoint (checks scheduler, source health, post counts) |
| Error alerts | Email via Resend for critical errors (publish failures, API budget warnings) |
| Performance | Structured logs with duration fields for LLM calls, research cycles, publish operations |
| Resource usage | docker stats for CPU/memory monitoring |
| Data | Backup Method | Frequency |
|---|---|---|
| Database | Supabase automated backups (free tier) | Daily (managed by Supabase) |
| Database export | Manual pg_dump via Supabase dashboard | Weekly (manual) |
| Media files | Supabase Storage (cloud-hosted) | Continuous |
| Application logs | Volume-mounted, rotated | Continuous |
| Configuration | .env file backed up to secure location |
On change |
| Code | Git repository | On every push |
Adding a new LLM provider (e.g., Groq, OpenAI):
- Create
backend/integrations/llm/groq.py - Implement the
BaseLLMabstract class - Update the factory/config to instantiate the new adapter
- Zero changes to
core/business logic
# backend/integrations/llm/groq.py
from .base import BaseLLM
class GroqLLM(BaseLLM):
def __init__(self, api_key: str, model: str = "llama-3.1-70b-versatile"):
self._api_key = api_key
self._model = model
async def generate_text(self, prompt: str, max_tokens: int = 512, temperature: float = 0.7) -> str:
# Groq-specific implementation
...
async def generate_json(self, prompt: str, schema: dict, max_tokens: int = 1024, temperature: float = 0.3) -> dict:
# Groq-specific implementation
...
async def compute_embedding(self, text: str) -> list[float]:
# Use alternative embedding provider
...Adapters are instantiated via a factory function that reads configuration and returns the appropriate implementation. This centralizes adapter creation, makes swapping implementations trivial, and enables clean test mocking.
# backend/integrations/factories.py
"""
Factory functions for creating adapter instances.
Reads configuration and returns the appropriate implementation.
"""
from integrations.llm.base import BaseLLM
from integrations.publishers.base import BasePublisher
def create_llm_adapter(config: dict) -> BaseLLM:
"""
Create an LLM adapter based on the configured provider.
Enables easy swapping between Vertex AI, Groq, mock, etc.
"""
provider = config.get("LLM_PROVIDER", "vertex_ai")
if provider == "vertex_ai":
from integrations.llm.vertex_ai import VertexAILLM
return VertexAILLM(
project_id=config["GOOGLE_CLOUD_PROJECT"],
location=config.get("VERTEX_AI_LOCATION", "us-central1"),
)
elif provider == "groq":
from integrations.llm.groq import GroqLLM
return GroqLLM(api_key=config["GROQ_API_KEY"])
elif provider == "mock":
from integrations.llm.mock import MockLLM
return MockLLM()
raise ValueError(f"Unknown LLM provider: {provider}")
def create_publisher_adapter(config: dict) -> BasePublisher:
"""
Create a publisher adapter based on the configured platform.
"""
platform = config.get("PUBLISH_PLATFORM", "x_twitter")
if platform == "x_twitter":
from integrations.publishers.x_publisher import XPublisher
return XPublisher(
api_key=config["X_API_KEY"],
api_secret=config["X_API_SECRET"],
access_token=config["X_ACCESS_TOKEN"],
access_token_secret=config["X_ACCESS_TOKEN_SECRET"],
)
elif platform == "mock":
from integrations.publishers.mock import MockPublisher
return MockPublisher()
raise ValueError(f"Unknown publish platform: {platform}")Usage in application startup (main.py):
# In the lifespan function, replace direct instantiation with factory calls:
from integrations.factories import create_llm_adapter, create_publisher_adapter
app.state.llm = create_llm_adapter(settings.__dict__)
app.state.publisher = create_publisher_adapter(settings.__dict__)This pattern means:
- Testing: Set
LLM_PROVIDER=mockin test config to get aMockLLMthat returns deterministic responses without calling any external API. - Fallback: If Vertex AI credits exhaust, change
LLM_PROVIDER=groqin.envand restart. Zero code changes needed. - New providers: Implement
BaseLLM, add anelifbranch in the factory, and the new provider is available.
Adding a new publishing platform (e.g., LinkedIn):
- Create
backend/integrations/publishers/linkedin_publisher.py - Implement the
BasePublisherabstract class - Register it as an available publisher in config
- Zero changes to
core/publisher.py-- it only calls theBasePublisherinterface
Switching from Supabase to self-hosted Postgres:
- Update
backend/integrations/supabase_client.pyto useasyncpgorSQLAlchemyinstead of the Supabase Python client - Update connection string in
.env - Run the same DDL migrations on the new Postgres instance
- Zero changes to
core/-- it never imports the Supabase client directly
Adding agent frameworks (V2):
- The
core/modules already represent individual "agent" responsibilities - Wrap them in CrewAI or LangGraph agents that call the same functions
- The pure-Python core functions become tool implementations for the agent framework
- The scheduler coordinates agents instead of directly calling core functions
To add multi-user support, the following changes would be needed:
| Component | Change Required |
|---|---|
| Auth | Replace simple password with Supabase Auth (OAuth, email/password) |
| Database | Add user_id column to all tables. Update RLS policies per-user. |
| API | Extract user_id from JWT. Filter all queries by user_id. |
| Frontend | User registration, profile management, team invites |
| Config | Per-user style configs, sources, categories |
| Scheduler | Per-user schedules, separate job namespaces |
To add multi-platform publishing:
| Component | Change Required |
|---|---|
posts table |
Add platform column (enum: twitter, linkedin, medium, blog) |
| Content generator | Platform-specific char limits and formatting (Twitter: 280, LinkedIn: 3000, Medium: no limit) |
| Publisher | New adapters per platform, all implementing BasePublisher |
| Scheduler | Per-platform scheduling rules and rate limits |
| Dashboard | Platform selector in approval queue, per-platform analytics |
| Operation | Expected Duration | Notes |
|---|---|---|
| LLM text generation (single draft) | 2-8 seconds | Vertex AI Gemini 1.5 Flash; varies by prompt length |
| LLM JSON generation (ranking/analysis) | 3-10 seconds | Structured output may require longer generation |
| Embedding computation (single text) | 0.5-2 seconds | Vertex AI text-embedding-004 |
| X API tweet creation | 1-3 seconds | Network latency to X API servers |
| X API media upload | 2-10 seconds | Depends on image size (up to 5MB) |
| Supabase read query | 50-200 ms | REST API over HTTPS; indexed queries |
| Supabase write operation | 100-300 ms | Insert/update via REST API |
| Full research cycle (end-to-end) | 2-5 minutes | Fetching sources + ranking + drafting + uniqueness checks |
| Dashboard page load | < 2 seconds | Server-side rendering with Next.js |
| Resend email delivery | 1-5 seconds | API call; actual email delivery is async |
When external services are unavailable, Agent-X degrades gracefully rather than failing completely.
| Service | Failure Mode | Fallback Behavior |
|---|---|---|
| Vertex AI Studio | API errors, credit exhaustion, timeout | Retry with exponential backoff (1s, 2s, 4s; max 3 retries). If credits exhausted, swap to Groq adapter via LLM_PROVIDER config change. Log error, skip current cycle, retry on next scheduled run. |
| X API | Rate limit (429), auth failure (403), server error (5xx) | On rate limit: queue post for later, respect Retry-After header. On 403 (suspended): stop all publishing immediately, send critical alert, require manual intervention. On 5xx: retry once after 60 seconds with exponential backoff. |
| Supabase | Connection timeout, REST API error | Retry with exponential backoff (1s, 2s, 4s). Queue operations in memory temporarily (up to 5 seconds). If unreachable for > 5 minutes, send critical email alert. Scheduler pauses until connection restores. |
| Resend (Email) | API error, rate limit | Log error and skip. Email is non-critical; dashboard is the primary notification channel. Never block the main workflow for email failures. |
| Hacker News API | Timeout, server error | Skip HN source entirely. Continue research cycle with RSS feeds only. Log the failure, increment source error counter. |
| RSS Feeds | Malformed XML, timeout, DNS failure | Skip the failed feed. Continue with remaining feeds. Increment consecutive error counter; auto-disable source after 5 consecutive failures. |
All external service calls follow an exponential backoff pattern:
# Retry with exponential backoff (conceptual pattern used throughout)
import asyncio
from typing import TypeVar, Callable, Awaitable
T = TypeVar("T")
async def retry_with_backoff(
func: Callable[..., Awaitable[T]],
max_retries: int = 3,
base_delay: float = 1.0,
max_delay: float = 30.0,
) -> T:
"""
Retry an async function with exponential backoff.
Delays: 1s, 2s, 4s, 8s, ... capped at max_delay.
"""
last_exception = None
for attempt in range(max_retries + 1):
try:
return await func()
except Exception as e:
last_exception = e
if attempt < max_retries:
delay = min(base_delay * (2 ** attempt), max_delay)
await asyncio.sleep(delay)
raise last_exceptionFor services that may experience extended outages, Agent-X uses a lightweight circuit breaker to avoid overwhelming a failing service with retries:
| State | Behavior |
|---|---|
| Closed (normal) | Requests pass through. Failures are counted. |
| Open (tripped) | After N consecutive failures (default: 5), the circuit opens. All requests immediately fail without calling the service. A timer starts (default: 60 seconds). |
| Half-Open (testing) | After the timer expires, one test request is allowed through. If it succeeds, the circuit closes. If it fails, the circuit re-opens. |
This pattern is applied to:
- Vertex AI API calls (threshold: 5 failures, reset: 60s)
- X API publish calls (threshold: 3 failures, reset: 120s)
- Supabase connections (threshold: 5 failures, reset: 30s)
Agent-X is designed so that no single external service failure brings down the entire system:
- Research cycle: If HN fails, RSS continues (and vice versa). If the LLM fails, the cycle is skipped and retried next schedule.
- Publishing: If a single post fails to publish, it is marked as "failed" and the remaining queue continues. The owner is alerted.
- Analytics: If metrics fetch fails, manual entry remains available. Analytics are non-blocking.
- Email: All email failures are logged and swallowed. Dashboard notifications serve as the backup channel.
- Scheduler: Jobs are persisted in Supabase. On process restart, missed jobs are detected and executed.
# 1. Clone the repository
git clone https://github.com/your-org/agent-x.git
cd agent-x
# 2. Copy environment template
cp .env.example .env
# Edit .env with your credentials
# 3. Backend setup
cd backend
python -m venv venv
source venv/bin/activate # or venv\Scripts\activate on Windows
pip install -r requirements.txt
uvicorn main:app --reload --port 8000
# 4. Frontend setup (separate terminal)
cd frontend
npm install
npm run dev
# Frontend runs at http://localhost:3000
# 5. Or use Docker Compose for full stack
docker compose up --build| Type | Framework | Target | Location |
|---|---|---|---|
| Unit tests | pytest | core/ business logic (>=85% coverage) |
backend/tests/unit/ |
| Integration tests | pytest + httpx TestClient | All API endpoints | backend/tests/integration/ |
| Frontend tests | Vitest + React Testing Library | Key dashboard flows | frontend/src/__tests__/ |
| E2E tests | Playwright | Full approval-to-publish flow | e2e/ |
| Load tests | Locust | 10 concurrent approvals | tests/load/ |
Unit test example:
# backend/tests/unit/test_uniqueness_checker.py
import pytest
from core.uniqueness_checker import compute_content_hash, cosine_similarity
def test_content_hash_normalization():
"""Hash should be identical for same content with different whitespace."""
hash1 = compute_content_hash("Hello World this is a test")
hash2 = compute_content_hash("hello world this is a test")
assert hash1 == hash2
def test_cosine_similarity_identical():
"""Identical vectors should have similarity of 1.0."""
vec = [1.0, 0.0, 1.0]
assert cosine_similarity(vec, vec) == pytest.approx(1.0)
def test_cosine_similarity_orthogonal():
"""Orthogonal vectors should have similarity of 0.0."""
vec_a = [1.0, 0.0]
vec_b = [0.0, 1.0]
assert cosine_similarity(vec_a, vec_b) == pytest.approx(0.0)Integration test example:
# backend/tests/integration/test_posts_api.py
import pytest
from httpx import AsyncClient
from main import app
@pytest.fixture
async def client():
async with AsyncClient(app=app, base_url="http://test") as ac:
yield ac
@pytest.fixture
async def auth_token(client):
response = await client.post(
"/api/auth/login",
json={"password": "test-password"},
)
return response.json()["token"]
async def test_list_posts(client, auth_token):
response = await client.get(
"/api/posts",
headers={"Authorization": f"Bearer {auth_token}"},
)
assert response.status_code == 200
data = response.json()
assert "posts" in data
assert "total" in data# .github/workflows/ci.yml
name: CI
on:
push:
branches: [main, develop]
pull_request:
branches: [main]
jobs:
backend-tests:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python 3.12
uses: actions/setup-python@v5
with:
python-version: "3.12"
- name: Install dependencies
run: |
cd backend
pip install -r requirements.txt
pip install pytest pytest-asyncio pytest-cov
- name: Run unit tests
run: |
cd backend
pytest tests/unit/ -v --cov=core --cov-report=term-missing
- name: Run integration tests
env:
SUPABASE_URL: ${{ secrets.TEST_SUPABASE_URL }}
SUPABASE_SERVICE_ROLE_KEY: ${{ secrets.TEST_SUPABASE_KEY }}
DASHBOARD_PASSWORD: test-password
run: |
cd backend
pytest tests/integration/ -v
- name: Lint with Ruff
run: |
cd backend
pip install ruff
ruff check .
- name: Format check with Black
run: |
cd backend
pip install black
black --check .
frontend-tests:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Node.js 18
uses: actions/setup-node@v4
with:
node-version: "18"
- name: Install dependencies
run: |
cd frontend
npm ci
- name: Run tests
run: |
cd frontend
npm run test
- name: Lint
run: |
cd frontend
npm run lint
- name: Type check
run: |
cd frontend
npx tsc --noEmit
- name: Build
run: |
cd frontend
npm run build
e2e-tests:
needs: [backend-tests, frontend-tests]
runs-on: ubuntu-latest
if: github.ref == 'refs/heads/main'
steps:
- uses: actions/checkout@v4
- name: Start services
run: docker compose -f docker-compose.test.yml up -d
- name: Run Playwright tests
run: |
npx playwright install
npx playwright test
- name: Stop services
if: always()
run: docker compose -f docker-compose.test.yml down| Tool | Language | Config File | Purpose |
|---|---|---|---|
| Black | Python | pyproject.toml |
Code formatting (line length: 88) |
| Ruff | Python | pyproject.toml |
Linting (replaces flake8, isort) |
| Prettier | TypeScript/CSS | .prettierrc |
Code formatting |
| ESLint | TypeScript | .eslintrc.json |
Linting |
Python config (backend/pyproject.toml):
[tool.black]
line-length = 88
target-version = ["py312"]
[tool.ruff]
line-length = 88
target-version = "py312"
select = ["E", "F", "I", "N", "W", "UP"]
[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]Frontend config (frontend/.prettierrc):
{
"semi": true,
"trailingComma": "es5",
"singleQuote": false,
"tabWidth": 2,
"printWidth": 100
}gitGraph
commit id: "main (production)"
branch develop
commit id: "develop (integration)"
branch feature/research-pipeline
commit id: "research-pipeline"
checkout develop
branch feature/content-generator
commit id: "content-generator"
checkout develop
branch feature/dashboard-approvals
commit id: "dashboard-approvals"
checkout develop
branch fix/publish-retry-logic
commit id: "publish-retry-fix"
checkout develop
branch chore/update-dependencies
commit id: "update-deps"
- main: Production-ready code. Deployed to VPS.
- develop: Integration branch. All features merge here first.
- feature/*: Individual feature branches from develop.
- fix/*: Bug fix branches.
- chore/*: Non-functional changes (deps, config, docs).
Pull requests required for all merges to develop and main. CI must pass before merge.
# backend/main.py
"""
FastAPI application entry point.
Configures lifespan events, middleware, and route registration.
"""
import signal
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from config import settings
from api import auth, posts, sources, analytics, categories, style, improvements, system
from integrations.supabase_client import SupabaseClient
from integrations.factories import create_llm_adapter, create_publisher_adapter
from integrations.resend_client import ResendClient
from scheduler.jobs import AgentXScheduler
from utils.logger import setup_logger
logger = setup_logger("agent-x", settings.LOG_LEVEL)
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Startup and shutdown lifecycle."""
# --- Startup ---
logger.info("Starting Agent-X backend...")
# Initialize integrations (using adapter factories for LLM and publisher)
app.state.supabase = SupabaseClient(
settings.SUPABASE_URL, settings.SUPABASE_SERVICE_ROLE_KEY
)
app.state.llm = create_llm_adapter(settings.__dict__)
app.state.publisher = create_publisher_adapter(settings.__dict__)
app.state.email = ResendClient(
settings.RESEND_API_KEY,
"noreply@your-domain.com",
settings.NOTIFICATION_EMAIL,
)
# Initialize and start scheduler
app.state.scheduler = AgentXScheduler(
timezone=settings.RESEARCH_SCHEDULE_TIMEZONE,
jitter_min=settings.JITTER_MINUTES_MIN,
jitter_max=settings.JITTER_MINUTES_MAX,
)
app.state.scheduler.start()
# Check for missed jobs on restart
logger.info("Checking for missed scheduler jobs...")
# (recover missed jobs from Supabase)
logger.info("Agent-X backend started successfully.")
yield
# --- Shutdown ---
logger.info("Shutting down Agent-X backend...")
app.state.scheduler.shutdown(wait=True)
logger.info("Scheduler stopped. Agent-X backend shut down gracefully.")
app = FastAPI(
title="Agent-X API",
version="1.1.0",
lifespan=lifespan,
)
# CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=[settings.FRONTEND_URL],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Register route handlers
app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
app.include_router(posts.router, prefix="/api/posts", tags=["posts"])
app.include_router(sources.router, prefix="/api/sources", tags=["sources"])
app.include_router(analytics.router, prefix="/api/analytics", tags=["analytics"])
app.include_router(categories.router, prefix="/api/categories", tags=["categories"])
app.include_router(style.router, prefix="/api/style-config", tags=["style"])
app.include_router(improvements.router, prefix="/api/improvements", tags=["improvements"])
app.include_router(system.router, prefix="/api", tags=["system"])End of Architecture Specification
This document is aligned with PRD v2.2 and serves as the complete technical reference for implementing Agent-X. All code examples represent the intended architecture and interfaces; actual implementations may include additional error handling, logging, and edge case management as specified in the PRD.