CIAgent/ARCHITECTURE.md

Architecture

Overview

apps/web (Next.js)  <---REST/JSON--->  apps/api (FastAPI)  <--->  PostgreSQL
                                             |    \
                                             |     ---> Redis (Celery broker + cache)
                                             |
                                        Celery worker(s) + Celery Beat
                                             |
                            collectors -> LLM analysis -> change detection -> notifications

Everything runs via docker-compose.yml: postgres, redis, api, worker, beat, web. Each backend concern (apps/api/app/*) is organized by responsibility, not by HTTP route, so services can be reused from Celery tasks and from route handlers alike.

Provider abstraction pattern

Every external dependency that costs money, requires a credential, or might later move to a different vendor (including a future Firebase/GCP migration) sits behind a small Protocol interface with a Mock/Console implementation that requires no configuration and is what the app uses by default:

Concern	Interface	Implementations
Auth	AuthProvider (app/auth/base.py)	LocalAuthProvider, JWTAuthProvider
LLM	LLMProvider (app/analysis/llm/base.py)	MockLLMProvider, AnthropicLLMProvider, OllamaLLMProvider, GeminiLLMProvider
Search	SearchProvider (app/search/base.py)	MockSearchProvider, BraveSearchProvider
Notifications	NotificationProvider (app/notifications/base.py)	ConsoleProvider, SmtpEmailProvider, TwilioSmsProvider
Source collection	SourceCollector (app/collectors/base.py)	WebsiteCollector, RssCollector (incl. Google News auto-discovery), CustomUrlCollector, SecEdgarCollector, GithubCollector, JobPostingCollector, GovContractCollector (USASpending.gov, free/keyless), PatentSourceCollector (real USPTO ODP call when USPTO_API_KEY is set, fixture fallback otherwise), ReviewSourceCollector (fixture)
Company enrichment	EnrichmentProvider (app/enrichment/base.py)	MockEnrichmentProvider, NinjaPearProvider (nubela.co, paid/per-credit, real call when NINJAPEAR_API_KEY is set)

Selection is via environment variables (LLM_PROVIDER, SEARCH_PROVIDER, AUTH_MODE, etc.), read once in app/core/config.py (a pydantic-settings Settings singleton) and resolved through small factory functions — never imported directly by call sites. This is what makes the Firebase/GCP migration path (see docs/FIREBASE_MIGRATION.md) tractable: swapping AuthProvider for a Firebase-backed one, or the DB session for a Firestore client, doesn't ripple through route handlers or Celery tasks.

Data flow: monitoring run

1. MonitorConfiguration.next_run is reached, or an individual Source has its own faster frequency_type override that's independently due (Beat), or a user hits "Run Now" (API) → a MonitoringRun row is created (status=queued) and a collection.run_monitoring Celery task is enqueued with the run ID.
2. The task loads active Source rows for the company. For a SCHEDULED trigger, only sources actually due right now are collected (a source with no override rides the company's own cadence; one with an override uses its own next_check, via SourceRepository.list_due_for_company) - a MANUAL "Run now" always collects every active source regardless of individual cadence. For each, the task calls the matching SourceCollector.collect(), going through app/core/http.safe_fetch() (SSRF-guarded) for anything network-bound.
3. Each successful collection produces a SourceDocument (raw extracted text + metadata) and updates the run's sources_successful / sources_failed counters. A Snapshot (structured summary + hash) is derived per source.
4. change_detection compares the new Snapshot against the most recent prior one for the same source, layer by layer (hash → structured fields → bounded text diff → optional LLM semantic check), producing a DetectedChange with a significance_score, confidence, and severity.
5. analysis (LLM) generates/updates the Report for the company from the accumulated evidence (SourceDocuments + DetectedChanges), never inventing facts outside that evidence.
6. For each DetectedChange above the user's severity threshold, an Alert is created (after dedup/cooldown checks) and notifications delivers it to each enabled NotificationDestination linked to that company (via the notification_destination_companies join table - a destination can be shared across several companies, e.g. one email registered once but linked to every company the user monitors), recording a NotificationDelivery. A destination with zero remaining company links is garbage-collected when the last linking company is deleted.
7. The frontend polls MonitoringRun status and then renders the Report/Alert once available.

Data flow: company discovery (onboarding)

POST /api/v1/companies/discover (app/services/discovery_service.py) turns a bare company name into a pre-filled, user-editable profile, and is deliberately kept separate from the two pipelines above:

• SearchProvider answers "where should I look" — it returns URLs/snippets, never facts about the company itself.
• SourceCollector (the same collectors the monitoring pipeline uses) answers "what's actually there" by fetching real pages.
• LLMProvider only ever analyzes evidence already fetched by the two above — it is never prompted with "tell me about company X" from its own training data, matching the evidence-grounded pattern every other analysis task in this app follows (app/prompts/*).

Steps: (1) resolve official_website from the user's hint or search.search(f"{name} official website"); (2) fetch that homepage via the existing fetch_with_retries/extract_readable_text helpers; (3) run a few targeted searches (headquarters, competitors, aliases) for more evidence snippets; (4) call app/prompts/company_profile.py::extract_company_profile once over all of it; (5) merge — any user-supplied hint wins over the discovered value; (6) call the website/github/sec_edgar/job_posting collectors' existing .discover() methods against a synthetic, unsaved CompanyContext to preview likely sources. Nothing is persisted by this endpoint — the wizard's Review step shows the result for editing, and only the final POST /companies call (unchanged) writes a Company row. Source rows themselves still come from the pre-existing lazy discovery gate in tasks/collection.py (first monitoring run, not onboarding) — discovery preview and source persistence intentionally stay two different code paths so a user can preview/abandon without any DB writes.

Data flow: company enrichment (paid, onboarding-only)

app/services/enrichment_service.py is a third, deliberately separate data-gathering path from the two above, for a paid third-party vendor (NinjaPear/nubela.co) that bills per-field per-request:

• Only ever runs once, right after a Company row is actually committed in company_service.create_company — never on a recurring schedule, and never during the discovery preview step (which would spend real credits on companies a user might abandon before creating). The enqueue itself (enrich_company.delay(...), routed to its own enrichment Celery queue) is skipped entirely unless NINJAPEAR_API_KEY is configured, so a user who never opts in gets zero extra background-task volume — same "gate the enqueue on the key, not just the provider" pattern as PatentSourceCollector.discover().
• A CompanyEnrichment row (1:1 with Company, same shape as MonitorConfiguration) is created with status=pending synchronously, in the same transaction as company creation, before the Celery task even starts — this gives the frontend something real to poll on (useCompany's refetchInterval), since without it "not yet enriched" and "never configured" would both look like enrichment: null.
• The task (app/tasks/enrichment.py) calls several independent NinjaPear endpoints (company details, funding, competitors, products, customer/updates listings, plus capped per-leadership-member work-email/profile lookups) through EnrichmentProvider; one failed call is recorded in errors and never sinks the others, same principle as tasks/collection.py's per-source loop. Overall status becomes complete/partial/failed depending on how many of those independent calls actually succeeded.
• Once present, CompanyEnrichment.data feeds into report generation as a company_enrichment evidence block (report_service.py, prompts/report_generation.py) exactly like company_profile does — real fetched data, never invented, and skipped entirely (falls back to an honest empty block) if enrichment never ran, is still pending, or failed outright.

Change significance scoring (documented, testable formula)

Implemented in app/change_detection/scoring.py. Each detected difference starts with a base weight from its category (e.g. leadership_change=0.9, pricing_change=0.6, wording_change=0.15), then is adjusted by:

significance = base_weight
             * source_trust_score          (0.3 - 1.0)
             * min(1.0, independent_sources / 2)   # corroboration bonus, caps at 2 sources
             * focus_match_multiplier       (1.3 if matches user's stated focus, else 1.0)
             * recency_multiplier           (1.0 if new, 0.5 if a repeat of a prior alert)
confidence   = f(extraction_confidence, source_trust_score, corroboration)

severity is then a deterministic bucket over significance * confidence (see SEVERITY_THRESHOLDS in the same module), with a hard rule that critical requires confidence >= 0.7 regardless of score — an uncorroborated single-source signal cannot be labeled Critical. Full unit tests live in apps/api/tests/unit/test_scoring.py.

Directory layout

See PLAN.md for the top-level file tree. Within apps/api/app, packages are: api/v1 (routers only — thin, no business logic), auth, core (config, logging, http/SSRF, security), db (session, base model), models (SQLAlchemy ORM), schemas (Pydantic request/response), repositories (query layer, one per aggregate), services (business logic orchestration), collectors, analysis (LLM provider + prompt tasks), change_detection, notifications, tasks (Celery task definitions, thin wrappers around services), prompts (prompt templates + response schemas, one module per analysis task).

Why FastAPI routes stay thin

Route handlers only: parse/validate input (Pydantic), call one service method, map the result/exception to an HTTP response. All ownership checks, business rules, and orchestration live in services/, which is what both the API and Celery tasks call — this avoids duplicating authorization or business logic between the two entry points.