Workflow Engine Analysis¶

Date: December 2, 2025 Repository: cbworkflow (~/projects/nominate/cbworkflow) Status: Phases 1-3 Complete, Phases 4-5 Placeholder

Executive Summary¶

cbworkflow is a lightweight, JSON-backed workflow engine for CRM contact management. It provides:

Contact Management - CRM contacts with tags, organization, and metadata
Workflow Templates - Reusable blueprints with directed graph structure (nodes + transitions)
Workflow Instances - Live executions linked to contacts with state tracking
Event Timeline - Immutable audit log for every action

The architecture follows a clean layered pattern:

API Routes → Services → JsonStore → JSON Files (with file locking)

Core Entities¶

1. Contact¶

CRM entity representing a person to engage with.

Contact:
  id: str                      # 12-char UUID
  name: str                    # Required
  email: EmailStr | None
  phone: str | None
  county: str | None           # Geographic filter
  organization: str | None
  role: str | None
  workflow_instance_ids: list  # Links to active workflows
  tags: list[str]              # Filterable tags
  metadata: dict               # Flexible key-value store
  created_at, updated_at: datetime

2. WorkflowTemplate¶

Blueprint defining the workflow graph structure.

WorkflowTemplate:
  id: str
  name: str
  description: str | None
  version: str                 # Semantic versioning
  nodes: dict[str, Node]       # Node graph
  start_node: str              # Entry point
  tags: list[str]
  metadata: dict

Node Types: | Type | Purpose | Key Fields | |------|---------|------------| | ACTION | Task to perform | action, action_config | | DECISION | Branch logic | rule (heuristic/AI/manual) | | WAIT | Pause execution | wait_condition (duration/until/event) | | WEBHOOK | External call | webhook_url, webhook_method | | END | Terminal state | end_status |

Decision Rule Types: | Type | Description | |------|-------------| | HEURISTIC | Python expression evaluated against state | | AI | Claude API call with prompt template | | MANUAL | Wait for user input |

3. WorkflowInstance¶

Live execution of a template, linked to a contact.

WorkflowInstance:
  id: str
  contact_id: str              # Link to contact
  template_id: str             # Source template
  template_version: str        # Frozen at creation
  current_node: str            # Current position in graph
  status: InstanceStatus       # pending/active/paused/waiting/completed/failed/cancelled
  state: dict                  # Mutable state data
  nodes: dict[str, Node]       # Copy of template nodes (frozen)
  timeline: list[Event]        # Append-only audit log
  tags: list[str]
  metadata: dict
  created_at, updated_at, started_at, completed_at: datetime

4. Event¶

Immutable audit log entry.

Event:
  id: str
  type: EventType              # 17 event types
  timestamp: datetime
  node_id: str | None
  actor: str | None            # user_id, "system", "ai"
  data: dict
  message: str | None

Event Types Include: - Lifecycle: INSTANCE_CREATED, INSTANCE_STARTED, INSTANCE_COMPLETED - Navigation: NODE_ENTERED, NODE_EXITED - Actions: ACTION_STARTED, ACTION_COMPLETED, ACTION_FAILED - Decisions: DECISION_EVALUATED, DECISION_MANUAL_INPUT - State: STATE_UPDATED, TAG_ADDED, TAG_REMOVED - User: USER_NOTE, USER_ACTION

API Endpoints¶

Contacts API (`/api/contacts`)¶

Method	Endpoint	Description
GET	`/`	List contacts (filter: county, tag, organization)
GET	`/search?q=`	Search by name/email/phone
GET	`/{id}`	Get single contact
POST	`/`	Create contact
PATCH	`/{id}`	Update contact
DELETE	`/{id}`	Delete contact
POST	`/{id}/tags/{tag}`	Add tag
DELETE	`/{id}/tags/{tag}`	Remove tag

Workflows API (`/api/workflows`)¶

Method	Endpoint	Description
GET	`/`	List templates (filter: tag)
GET	`/{id}`	Get template
POST	`/`	Create template
PATCH	`/{id}`	Update template
DELETE	`/{id}`	Delete template
POST	`/{id}/clone`	Clone template
GET	`/{id}/validate`	Validate graph structure

Instances API (`/api/instances`)¶

Method	Endpoint	Description
GET	`/`	List instances (filter: contact_id, template_id, status, tag)
GET	`/{id}`	Get instance
POST	`/`	Create instance from template
DELETE	`/{id}`	Delete instance
POST	`/{id}/start`	Start workflow
POST	`/{id}/pause`	Pause workflow
POST	`/{id}/resume`	Resume workflow
POST	`/{id}/transition`	Manual transition with outcome
POST	`/{id}/state`	Update instance state
POST	`/{id}/notes`	Add user note
GET	`/{id}/timeline`	Get event history

Data Flow¶

Creating a Workflow Instance¶

1. Client creates Contact
   POST /api/contacts → Contact saved to data/contacts/{id}.json

2. Client creates Instance from Template
   POST /api/instances
   {
     "contact_id": "abc123",
     "template_id": "county-outreach-v1",
     "initial_state": {"preferred_contact": "phone"}
   }

3. Service:
   - Loads template from data/templates/
   - Creates WorkflowInstance with frozen copy of nodes
   - Links instance to contact (adds to workflow_instance_ids)
   - Saves to data/instances/{id}.json
   - Returns instance with status: PENDING

Executing a Transition¶

1. Client transitions instance
   POST /api/instances/{id}/transition
   {
     "outcome": "answered_interested",
     "state_updates": {"call_notes": "Very engaged"},
     "actor": "user123"
   }

2. Service:
   - Validates outcome exists in current node's transitions
   - Updates state if provided
   - Appends NODE_EXITED event
   - Moves to target node
   - Appends NODE_ENTERED event
   - If END node → status = COMPLETED
   - If WAIT node → status = WAITING

Storage Architecture¶

File-Based JSON Storage¶

data/
├── contacts/
│   ├── abc123.json
│   └── def456.json
├── instances/
│   ├── inst001.json
│   └── inst002.json
└── templates/
    └── county-outreach-v1.json

Features: - One JSON file per entity - File locking via portalocker for concurrency - Async I/O via aiofiles - Lock files: .{id}.lock (auto-cleaned)

Locked Update Pattern:

async with store.locked_update(id) as entity:
    entity.name = "new name"
    # Automatically saved on context exit

Example Workflow Template¶

The included county-outreach-v1 template demonstrates a multi-touch voter outreach workflow:

initial_call (ACTION: phone_call)
  ├── answered_interested → schedule_followup (WAIT: 3 days)
  │                              └── followup_text (ACTION: sms)
  │                                    ├── sent → end_success
  │                                    └── failed → send_postcard
  ├── answered_not_interested → end_declined
  ├── no_answer → schedule_door (ACTION: door_canvass)
  └── voicemail → schedule_door
                      ├── contacted_interested → door_decision (DECISION: AI)
                      │                               ├── text → followup_text
                      │                               └── mail → send_postcard
                      ├── contacted_not_interested → end_declined
                      └── not_home → send_postcard
                                       ├── sent → end_success
                                       └── failed → end_failed

Key Features: - Multi-channel: phone, door, SMS, mail - AI decision point using Claude to analyze door notes - Wait conditions for follow-up timing - Multiple end states (success, declined, failed)

Development Status¶

Phase	Description	Status
1	Foundation (models, storage, contacts)	Complete
2	Workflow templates (CRUD, validation)	Complete
3	Workflow instances (state, timeline)	Complete
4	Decision engine (heuristic, AI, manual)	Placeholder
5	Event system (pub/sub, cross-workflow)	Placeholder
6	Production hardening (Redis, error handling)	Not started

Placeholders: - app/engine/__init__.py - Empty (decision execution logic) - app/events/__init__.py - Empty (pub/sub system)

Integration Questions¶

1. Relationship to cbapp (Campaign Site)¶

Question: How should cbworkflow integrate with the per-tenant campaign sites?

Options: 1. Shared Database - Workflow contacts ARE the cbapp Person table 2. API Integration - cbapp calls cbworkflow API to manage workflows 3. Embedded - cbworkflow becomes a module within cbapp 4. Separate Service - cbworkflow runs independently, syncs data

My Understanding: The Contact model in cbworkflow looks similar to the Person model in cbapp. Are these the same entity? Should workflows be triggered when a person is imported into cbapp?

2. Multi-Tenancy¶

Question: How should cbworkflow handle multi-tenancy?

Current State: - No tenant_id in models - Single data/ directory for all data - NGINX config suggests workflow.nominate.ai

Options: 1. Tenant-per-instance - Each tenant gets their own cbworkflow deployment 2. Tenant-aware - Add tenant_id to all models, share single deployment 3. Integrated with cbtenant - cbtenant manages cbworkflow deployments like it does cbapp

My Understanding: Given the architecture of cbapp (one deployment per tenant), should cbworkflow follow the same pattern?

3. Action Execution¶

Question: Who/what executes the actions defined in workflow nodes?

Current State: - Action nodes have action (e.g., "phone_call", "sms", "door_canvass") - Action nodes have action_config (e.g., {"script_id": "intro_script_v1"}) - No execution logic implemented

Options: 1. External Systems - cbworkflow just tracks state, external systems do the work 2. Built-in Executors - cbworkflow has adapters for Twilio, SendGrid, etc. 3. Webhook-based - cbworkflow calls configured webhooks for each action 4. Manual - Users manually mark actions complete via API

My Understanding: The current design seems to expect manual transition via the /transition endpoint. Is this the intended long-term approach?

4. AI Decision Points¶

Question: How should AI decisions be executed?

Current State: - ANTHROPIC_API_KEY in config - anthropic_model: str = "claude-sonnet-4-20250514" configured - No execution logic implemented

Example from template:

{
  "type": "decision",
  "rule": {
    "type": "ai",
    "prompt": "Based on the canvasser notes: '{state.door_notes}', determine the best follow-up method..."
  }
}

Questions: - Should this be auto-evaluated when entering the node? - Should the UI present choices and let the user confirm? - How do we handle AI errors/timeouts?

5. Wait Conditions¶

Question: Who evaluates wait conditions and triggers transitions?

Current State: - Wait nodes define conditions (duration, until, event) - No scheduler or background job implemented

Options: 1. Background Scheduler - Celery/APScheduler checks and transitions 2. External Cron - Script runs periodically to check waits 3. On-demand - Check wait conditions when instance is accessed 4. Event-driven - External system sends events to trigger

6. Event System¶

Question: What's the intended use of the event system?

Current State: - Events are stored in instance timeline - No pub/sub implemented - app/events/__init__.py is empty

Potential Uses: - Cross-workflow communication - External system notifications - Real-time UI updates (WebSocket) - Analytics and reporting

7. Campaign Site Integration¶

Question: How do campaign staff interact with workflows?

Options: 1. Dedicated Workflow UI - Separate interface for workflow management 2. Integrated in cbapp - Workflow status shown on Person detail page 3. Both - Summary in cbapp, management in separate UI

My Understanding: Campaign volunteers need to see "who to call next" and record outcomes. Where does this happen?

8. Contact vs Person¶

Question: What's the relationship between cbworkflow Contact and cbapp Person?

cbworkflow Contact:

name, email, phone, county, organization, role,
workflow_instance_ids, tags, metadata

cbapp Person (from Session docs):

Appears to track voters with various attributes

Options: 1. Same Entity - Contact IS Person, shared database 2. Linked - Contact references Person by ID 3. Separate - Different use cases, no relationship 4. Sync - Copy between systems

Architecture Observations¶

Strengths¶

Clean Layered Design - API → Service → Storage separation
Type Safety - Pydantic models throughout
Audit Trail - Complete event timeline
Concurrent Safe - File locking prevents race conditions
Graph Validation - Detects unreachable nodes and dead ends
Template Versioning - Instances freeze template at creation

Current Limitations¶

No Authentication - API is completely open
No Multi-tenancy - Single data directory
No Background Jobs - Wait conditions not auto-triggered
No Action Execution - Manual transitions only
Linear Scaling - File-based storage won't scale to millions

Potential Enhancements¶

Authentication - JWT tokens (like cbtenant)
Database Backend - DuckDB or PostgreSQL option
Background Workers - Celery for async operations
Action Adapters - Twilio, SendGrid, Dialpad integrations
Real-time Updates - WebSocket for live timeline

Deployment Considerations¶

As Standalone Service¶

Domain: workflow.nominate.ai
Port: 8000 (default)
Storage: data/ directory

NGINX config exists at: nginx/workflow.nominate.ai

With Tenant Manager¶

If integrated with cbtenant: - Add to Site Launcher - Allocate port (32330+?) - Per-tenant deployment - Shared i360 linking?

As cbapp Module¶

If embedded in cbapp: - Merge models with existing Person - Add routes to cbapp API - Share database

Next Steps (Questions for You)¶

Summary¶

cbworkflow is a well-structured workflow engine with solid foundations (Phases 1-3). The core CRUD operations work, the graph model is sound, and the timeline provides complete audit capability.

What's Missing: - Actual execution logic (Phase 4-5) - Integration with campaign sites - Multi-tenancy support - Authentication

Ready To: - Add authentication - Build decision engine - Create action execution framework - Integrate with cbapp

Looking forward to your answers on the integration questions above!