core-pattern-service

Discovers behavioral patterns across interactions using a Python CrewAI multi-agent system running in Workers Containers. Triggered by cron schedules for daily, weekly, monthly, and yearly analysis periods. The TypeScript worker manages cron scheduling and D1 storage while the Python container performs the actual AI analysis.

Worker name: crow-core-pattern-service Domain (prod): internal.patterns.crowai.dev Domain (dev): dev.internal.patterns.crowai.dev Main entrypoint: src/worker.ts (not src/index.ts)

Schema

pattern_result

Column	Type	Notes
id	text PK
organization_id	text
period	text	daily, weekly, monthly, yearly
source_type	text	nullable, e.g. cctv for source-specific analysis
report	text	Full AI-generated analysis report
generated_at	timestamp

patterns

Column	Type	Notes
id	text PK
organization_id	text
type	text	Pattern classification
confidence	real	default 0.5
data	text	JSON pattern data
detected_at	integer	epoch
created_at	integer	epoch

Routes

Method	Path	Description
GET	/api/v1/patterns/organization/{orgId}	List patterns for org
GET	/api/v1/patterns/results/organization/{orgId}	List pattern analysis results
POST	/api/v1/patterns/analyze	Trigger ad-hoc analysis

Cron Schedules

Cron	Period	Description
0 2 * * *	daily	Every day at 02:00 UTC
0 3 * * 1	weekly	Every Monday at 03:00 UTC
0 4 1 * *	monthly	1st of month at 04:00 UTC
0 5 1 1 *	yearly	January 1st at 05:00 UTC

Environment Variables

Variable	Example
ENVIRONMENT	dev
API_GATEWAY_URL	https://dev.api.crowai.dev
AUTH_SERVICE_URL	https://dev.internal.auth-api.crowai.dev

Secrets

Secret	Purpose
SYSTEM_SECRET	Auth header for gateway calls (X-System-Token)
INTERNAL_GATEWAY_KEY	Gateway trust validation

Bindings

Binding	Type	Name
DB	D1	crow-core-pattern-service-db
AI	Workers AI	LLM inference
PATTERN_CONTAINER	Durable Object	PatternAnalyzerContainer (Workers Container)

Workers Container

• Class: PatternAnalyzerContainer (extends Container)
• Image: ./Dockerfile (Python FastAPI app)
• Max instances: 2
• Default port: 8080
• Sleep after: 10 minutes of inactivity
• Endpoint: POST /analyze with body { orgId, period, apiGatewayUrl, systemSecret }

Cron Worker Flow

1. Cron trigger fires (mapped to a period via cronPeriodMap)
2. Worker calls GET /api/v1/organizations on the gateway with X-System-Token header to fetch all org IDs
3. For each org, gets a container instance via getContainer(env.PATTERN_CONTAINER, orgId)
4. Sends POST /analyze to the container with org context
5. Container fetches interactions and products from the gateway API
6. Container runs CrewAI PatternCrew with the Cloudflare Workers AI LLM
7. Results are inserted into pattern_result table by the TypeScript worker

Python Container Architecture

src/
  main.py              -- FastAPI app with /analyze and /health endpoints
  crews/
    pattern_crew.py    -- CrewAI crew definition
  llm/
    cloudflare_workers_ai.py -- Custom LLM wrapper for Workers AI REST API

The container uses httpx to call the gateway API for interaction and product data, then runs a CrewAI crew for pattern analysis.

Dependencies

• Inbound: gateway (pattern queries from dashboard)
• Outbound: gateway API (interactions, products, organizations), auth service (JWKS for JWT verification)

Key Behaviors

• INTERNAL_GATEWAY_KEY guard: HTTP routes require the shared internal key
• JWT middleware: Accepts either JWT verification or X-Internal-Key + X-Organization-Id as gateway-trusted auth path
• JWKS cache TTL: 300 seconds (5 minutes)
• Source-type analysis: Each org gets two analysis runs per cron -- one general and one CCTV-specific