ARC — Pipeline Design — PIP-ARC-01

ARC Pipeline Design

Every automated and user-triggered data flow in ARC described as a staged pipeline. Each pipeline stage shows the node type, the operation performed, its inputs, its outputs, and whether execution is automatic, conditional, or requires human action. Source references to C:/repos/arc/ accompany every pipeline.

Document ID

PIP-ARC-01

Pipelines

Source

C:/repos/arc/

Status

Active

Legend for "Execution" column in stage tables:

CRON Scheduled background job USER Triggered by user action EVENT Triggered by system event AI Claude API call

ARC-PIPE-01 — Gmail Email Ingest

Scheduled cron. Indexes client emails from Gmail every 15 minutes.

ARC-PIPE-01 Gmail Email Ingest Trigger: Cron every 15 min (env configurable) | Source: arc/agent/app/scheduler.py

Fetches all new emails from Gmail since the last sync timestamp. Matches each email to a known client by exact address or wildcard pattern. Stores the email record and triggers sentiment analysis.

Inputs

Gmail OAuth2 credentials (DB)
last_synced_at timestamp (state)
Client email patterns (DB: client_emails)

Outputs

Email records (DB: emails)
"UV-Indexed" Gmail label applied
last_synced_at updated

#	Node Type	Operation	Source	Exec
1	Scheduler / Trigger	APScheduler fires email_sync job at configured interval. Passes last_synced_at timestamp.	`arc/agent/app/scheduler.py`	CRON
2	Gmail OAuth Fetch	Authenticates via stored OAuth2 refresh token. Calls Gmail API with search query: messages after last_synced_at. Paginates until complete. Fetches: message_id, thread_id, from, to, subject, date, body_plain.	`arc/agent/app/google_sync.py :: sync_emails()`	EVENT
3	Deduplication Check	For each message_id, queries DB for existing record. Skips if already stored. Prevents duplicate indexing on re-run.	`arc/agent/app/google_sync.py`	EVENT
4	Client Pattern Match	For each new email: iterates all client_emails patterns. Exact match on from/to address. Wildcard match using fnmatch. Sets client_id on email record. Emails with no match stored with client_id=null (visible as "unassigned" in UI).	`arc/agent/app/google_sync.py` `arc/backend/app/models/client.py`	EVENT
5	Email Store	Writes Email record to DB. Sets direction: sent if from address matches user's Gmail account, received otherwise. Sets is_indexed=True.	`arc/backend/app/models/email.py`	EVENT
6	Gmail Label Apply	Calls Gmail API labels.modify to add "UV-Indexed" label to the message. Creates label if it does not exist yet.	`arc/agent/app/google_sync.py`	EVENT
7	State Update	Updates last_synced_at to the most recent email date processed. Logging records email count synced.	`arc/agent/app/scheduler.py`	EVENT

ARC-PIPE-02 — Email Sentiment Analysis

Scheduled cron. Analyzes all unscored emails via Claude AI every 30 minutes.

ARC-PIPE-02 Email Sentiment Analysis Trigger: Cron every 30 min | Source: arc/agent/app/scheduler.py, arc/agent/app/agents.py

Queries all indexed emails lacking a sentiment score. For each batch, calls Claude with the SENTIMENT_SYSTEM prompt. Stores the result back on the email record. Also updates the email_sentiment_trend on the linked project.

Inputs

Email records where sentiment IS NULL
SENTIMENT_SYSTEM prompt (DB override or default)
Claude API key (env)

Outputs

Email.sentiment updated: positive|neutral|negative|urgent
Project sentiment trend updated

#	Node Type	Operation	Source	Exec
1	Scheduler Trigger	APScheduler fires sentiment_analyzer at 30 min interval.	`arc/agent/app/scheduler.py`	CRON
2	Query Unscored Emails	SELECT * FROM emails WHERE sentiment IS NULL LIMIT 50 (batch to control API cost).	`arc/agent/app/agents.py`	EVENT
3	Prompt Resolution	Calls _get_prompt("SENTIMENT_SYSTEM", DEFAULT_SENTIMENT_PROMPT). Returns DB override if user has customized it, otherwise returns hardcoded default.	`arc/agent/app/agents.py`	EVENT
4	Claude API Call	For each email: POST to Claude with system prompt + email body text. Response is a JSON object: {sentiment: "positive"\|"neutral"\|"negative"\|"urgent", confidence: 0.0-1.0, reasoning: string}.	`arc/agent/app/agents.py :: analyze_email()` `arc/agent/app/claude_client.py`	AI
5	Sentiment Store	Writes sentiment label to Email record. Updates project.sentiment_trend (rolling average of recent emails). Logs token usage.	`arc/agent/app/agents.py`	EVENT

ARC-PIPE-03 — Action Priority Scoring

Scheduled cron. Recalculates priority scores for all pending actions every hour via Claude.

ARC-PIPE-03 Action Priority Scoring Trigger: Cron (env configurable) | Source: arc/agent/app/agents.py :: score_actions()

Inputs

Actions where is_completed=False
Project metadata (status, waiting_on, estimated_value, deadline)
Email sentiment trend
Days since last email
ACTION_SCORING_SYSTEM prompt

Outputs

Action.priority_score updated (1-100)
Dashboard feed reflects new order

#	Node Type	Operation	Source	Exec
1	Scheduler Trigger	APScheduler fires action_scoring at configured interval.	`arc/agent/app/scheduler.py`	CRON
2	Context Assembly	For each pending action, assembles context: action description, project status, waiting_on, estimated_value (PLN), days since last email with client, most recent email sentiment, deadline days remaining.	`arc/agent/app/agents.py`	EVENT
3	Claude Scoring	Sends all contexts to Claude with ACTION_SCORING_SYSTEM prompt. Response: array of {action_id, priority_score: 1-100, reasoning}.	`arc/agent/app/agents.py :: score_actions()`	AI
4	Score Write	Updates Action.priority_score for each returned result. Dashboard reads the new scores on next load.	`arc/agent/app/agents.py`	EVENT

ARC-PIPE-04 — Calendar Sync (Bidirectional)

Pulls from Google Calendar every 30 min. Pushes immediately on user create/edit.

ARC-PIPE-04 Calendar Bidirectional Sync Trigger: Cron 30 min (pull) + User action (push) | Source: arc/agent/app/google_sync.py

Two sub-flows: pull (Google Calendar to DB) runs on cron; push (DB event to Google Calendar) fires immediately on create/edit from UI.

4A: Pull from Google Calendar (Cron)

#	Node Type	Operation	Exec
1	Cron Trigger	APScheduler fires calendar_sync every 30 minutes.	CRON
2	GCal API Fetch	Calls Google Calendar API for events from timeMin=now-1day to timeMax=now+90days. Returns all events for the user's calendar.	EVENT
3	Upsert to DB	For each event: upsert to CalendarEvent table using gcal_event_id as unique key. New events inserted, changed events updated, cancelled events marked inactive.	EVENT

4B: Push to Google Calendar (User-Triggered)

#	Node Type	Operation	Exec
1	User Action	User creates or edits a calendar event (from Calendar page or Project Deadlines section).	USER
2	DB Write	Backend writes CalendarEvent to DB. Returns the DB record.	EVENT
3	GCal Push	Backend calls agent: POST /gcal/event (create) or PATCH /gcal/event/:gid (update). Agent calls Google Calendar API. Returns gcal_event_id. Backend updates CalendarEvent.gcal_event_id.	EVENT

ARC-PIPE-05 — Email-to-Calendar AI Parsing

Runs every 60 min or on-demand. Claude reads emails and creates/removes calendar events.

ARC-PIPE-05 Email-to-Calendar AI Parse Trigger: Cron 60 min + User button "Parse from emails" | Source: arc/agent/app/agents.py :: parse_calendar_events()

Inputs

Email threads from last 7 days
PARSE_CALENDAR_SYSTEM prompt (DB overridable)
Existing CalendarEvent titles (for dedup)

Outputs

New events created in GCal + DB
Removed events deleted from GCal + DB
No duplicates

#	Node Type	Operation	Exec
1	Trigger	Cron fires or user clicks "Parse from emails" button in Calendar page header.	CRON / USER
2	Thread Aggregation	Queries emails from last 7 days. Groups by thread_id. Formats as conversation blocks for Claude.	EVENT
3	Claude Parse	Sends conversation blocks + PARSE_CALENDAR_SYSTEM prompt to Claude. Response JSON: {add: [{title, date_start, date_end, type}], remove_titles: [string]}.	AI
4	Deduplication	For each event in "add": checks if CalendarEvent with same title and date already exists. Skips if found.	EVENT
5	GCal + DB Create	For new events: creates CalendarEvent in DB, calls agent POST /gcal/event, stores returned gcal_event_id.	EVENT
6	Removal	For each title in "remove_titles": finds matching CalendarEvent, calls agent DELETE /gcal/event/:gid, removes DB record.	EVENT

ARC-PIPE-06 — Offer Creation with AI Steps

User-initiated. Creates a project offer with optional AI-suggested phases.

ARC-PIPE-06 Offer Creation with AI Step Suggestions Trigger: User creates offer | Source: arc/frontend/src/app/offers/page.tsx, arc/agent/app/agents.py

Inputs

Project context (name, description, estimated_value)
Historical offer steps from similar projects
OFFER_STEPS_SYSTEM prompt
User edits to step table

Outputs

Offer record (DB: offers)
OfferStep records (up to 15, DB: offer_steps)
OfferVersion snapshot (DB: offer_versions)

#	Node Type	Operation	Source	Exec
1	User Opens Form	User navigates to /offers/new?projectId=:id. Form pre-fills project and client name. Empty step table shown.	`arc/frontend/src/app/offers/page.tsx`	USER
2	AI Suggest Trigger	User clicks "AI Suggest Steps". Frontend calls POST /agent/suggest-offer-steps with project_id.	`arc/frontend/src/app/offers/page.tsx`	USER
3	Context Fetch	Backend assembles: project description, client name, historical OfferSteps from up to 5 past similar offers. Calls agent POST /suggest-offer-steps.	`arc/backend/app/routers/agent.py`	EVENT
4	Claude Step Generation	Claude returns JSON array of 5-7 steps: [{step_number, name, description, hours, hourly_rate}]. Populates step table in UI.	`arc/agent/app/agents.py :: suggest_offer_steps()`	AI
5	User Editing	User reviews, edits any field, adds or removes rows. Row total = hours x hourly_rate. Grand total is auto-calculated sum of all row totals. Maximum 15 rows enforced.	`arc/frontend/src/app/offers/page.tsx`	USER
6	Save & Version	User saves. Backend creates Offer record + OfferStep records. Creates OfferVersion snapshot with full JSON representation. Version number increments on each subsequent save.	`arc/backend/app/routers/offers.py`	EVENT

ARC-PIPE-07 — Invoice Creation with AI Analysis

User-initiated. Creates invoice from offer, with AI analysis side panel.

ARC-PIPE-07 Invoice Creation with AI Analysis Trigger: User creates invoice | Source: arc/backend/app/routers/invoices.py, arc/agent/app/agents.py

#	Node Type	Operation	Source	Exec
1	Creation Trigger	User clicks "Create Invoice from Offer" on offer detail page, or "Create Invoice" from project page.	`arc/frontend/src/app/invoices/`	USER
2	Pre-fill from Offer	When linked to offer: maps OfferSteps to InvoiceLineItems. Auto-generates invoice number in configured format.	`arc/backend/app/routers/invoices.py`	EVENT
3	User Editing	User reviews and edits all fields. Sets due_date, payment terms.	`arc/frontend/src/app/invoices/`	USER
4	Save Invoice	Backend writes Invoice + InvoiceLineItem records. Status: Draft.	`arc/backend/app/routers/invoices.py`	EVENT
5	AI Analysis (On Demand)	User clicks "Analyze" in the side panel. Backend calls agent POST /analyze-invoice with invoice data + linked offer data. Claude returns array of insights. Displayed in side panel immediately, not stored.	`arc/agent/app/agents.py :: analyze_invoice()`	AI (on demand)
6	PDF Export	User selects template, clicks Export PDF. Backend renders invoice HTML with WeasyPrint, returns PDF file for download.	`arc/backend/app/services/pdf_service.py`	USER

ARC-PIPE-08 — Bank CSV Import & Invoice Matching

User uploads CSV. AI matches transactions to invoices automatically.

ARC-PIPE-08 Bank CSV Import and Invoice Matching Trigger: User uploads CSV | Source: arc/backend/app/routers/bank.py, arc/agent/app/agents.py

#	Node Type	Operation	Source	Exec
1	CSV Upload	User uploads bank statement CSV on Finance / Import page. Multipart file POST to backend.	`arc/frontend/src/app/bank/`	USER
2	CSV Parse & Preview	Backend uses pandas to detect format (Wise / PKO / mBank) by column header inspection. Parses rows into preview: date, amount, currency, description, reference. Displays preview table to user.	`arc/backend/app/services/bank_service.py`	EVENT
3	User Confirm	User reviews preview and clicks "Import". Backend writes all rows to BankTransaction table in a single transaction. Import is append-only.	`arc/backend/app/routers/bank.py`	USER
4	Matching Trigger	On successful import, backend immediately calls agent POST /match-transactions with the new batch's transaction IDs.	`arc/backend/app/routers/bank.py`	EVENT
5	AI Matching	Agent queries all open invoices and the new transactions. Matching logic: (a) reference field substring match, (b) amount within 1% tolerance, (c) currency match. Claude handles fuzzy cases. Returns matched pairs and unmatched transaction IDs.	`arc/agent/app/agents.py :: match_transactions()`	AI
6	Status Update	For each matched pair: sets Invoice.status = "paid", BankTransaction.matched_invoice_id = invoice_id. Unmatched: BankTransaction.match_status = "unmatched" (visible in Finance / Transactions view).	`arc/backend/app/routers/bank.py`	EVENT

ARC-PIPE-09 — Gmail Backfill & Client Auto-Discovery

User-initiated one-time historical import with 2-pass AI address assessment.

ARC-PIPE-09 Gmail Backfill and Client Auto-Discovery Trigger: User action from Settings / Backfill | Source: arc/agent/app/backfill.py

#	Node Type	Operation	Source	Exec
1	Range Selection	User selects import range: 1, 3, 6, or 12 months. Clicks "Start Backfill". Creates BackfillJob record with status=running.	`arc/frontend/src/app/settings/`	USER
2	Gmail Historical Fetch	Agent calls Gmail API with date range query. Paginates through all matching messages. Fetches message metadata and body_plain. Skips messages already in DB (dedup by gmail_id).	`arc/agent/app/backfill.py`	EVENT
3	Email Store	Writes Email records. Sets client_id using same pattern matching as regular sync. Increments BackfillJob.email_count. Updates progress status.	`arc/agent/app/backfill.py`	EVENT
4	Pass 1: Heuristic Filter	Extracts all unique email addresses from the fetched emails. Heuristic rules eliminate: noreply@, no-reply@, automated@, newsletter, notifications@, @accounts.google.com, @bounce., etc. Remaining candidates passed to Pass 2.	`arc/agent/app/backfill.py`	EVENT
5	Pass 2: Claude Batch Assessment	Remaining candidate addresses sent to Claude in a single batch with DISCOVERY_SYSTEM prompt. Claude returns per-address: {is_business_contact: bool, suggested_client_name: string, confidence: 0.0-1.0}. Token usage logged.	`arc/agent/app/backfill.py` `arc/agent/app/agents.py`	AI
6	Review UI Presentation	Results stored in DiscoveredAddress table. BackfillJob.status = "review_ready". UI shows grouped-by-domain review screen with confidence scores, user can edit suggested client names, check/uncheck addresses, mark "Merge with existing".	`arc/frontend/src/app/settings/backfill/`	USER REVIEW
7	Batch Client Creation	User clicks "Create Clients". Backend creates Client + ClientEmail records for all approved groups. Retroactively sets client_id on all historical email records matching the new client patterns.	`arc/backend/app/routers/clients.py`	USER

Efficiency note: The 2-pass approach (heuristic first, then single Claude batch) is intentional to minimize token cost. Only ambiguous business addresses that pass heuristic screening are sent to Claude.

ARC-PIPE-10 — AI Email Reply Draft

User-initiated per email thread. Claude generates a style-matched reply draft.

ARC-PIPE-10 AI Email Reply Draft Generation Trigger: User clicks "Suggest Reply" on email thread | Source: arc/agent/app/agents.py :: suggest_email_reply()

Inputs

Full email thread context
Project status + waiting_on
User's sent email history (writing style sample)
EMAIL_REPLY_SYSTEM prompt

Outputs

Draft text in editable UI area
Optional: saved as draft to Gmail
Optional: sent immediately via Gmail API

#	Node Type	Operation	Source	Exec
1	User Request	User clicks "Suggest Reply" on email thread or on a project action.	`arc/frontend/src/app/emails/`	USER
2	Style Profile Fetch	Backend queries the last 50 sent emails (direction=sent) from the user as a writing style corpus. Passed to agent.	`arc/backend/app/routers/agent.py`	EVENT
3	Claude Draft	Agent calls Claude with EMAIL_REPLY_SYSTEM prompt + full thread + style corpus + project context. Returns draft text in the appropriate language (PL/EN inferred from thread). Draft appears immediately in editable UI text area.	`arc/agent/app/agents.py :: suggest_email_reply()`	AI
4	User Edit	User edits the draft freely in the text area before taking any action.	`arc/frontend/src/app/emails/`	USER
5a	Send via Gmail	User clicks "Send". Backend calls Gmail API messages.send with draft content as reply to original thread. Draft is also stored in Email table with direction=sent.	`arc/backend/app/services/gmail_service.py`	USER
5b	Save as Draft	Alternative: User clicks "Save Draft". Draft stored in Email table, not yet sent to Gmail.	`arc/backend/app/routers/emails.py`	USER

Cron Schedule Summary

All scheduled background jobs in the ARC agent container

Pipeline	Job Name	Interval	Can Also Trigger	Source
ARC-PIPE-01	email_sync	Env variable (default 15 min)	Manual "Sync" button on /emails	`arc/agent/app/scheduler.py`
ARC-PIPE-02	sentiment_analyzer	30 min	Triggered after email_sync completes	`arc/agent/app/scheduler.py`
ARC-PIPE-03	action_scoring	Env variable (default 1 hr)	Not user-triggerable	`arc/agent/app/scheduler.py`
ARC-PIPE-04A	calendar_sync	30 min	Not user-triggerable	`arc/agent/app/scheduler.py`
ARC-PIPE-05	calendar_from_emails	60 min	"Parse from emails" button on /calendar	`arc/agent/app/scheduler.py`
-	status_inferrer	2 hours	Not user-triggerable	`arc/agent/app/scheduler.py`