Tax Practice AI - Architecture Document¶

Version: 1.8 Last Updated: 2025-12-28 Status: Draft

Table of Contents¶

Design Principles
Mixed Architecture Overview
Project Structure
Service Centralization
Core Services → SERVICES.md
Java Components
Database Layer
External Integrations → SERVICES.md
Workflow Orchestration
Configuration Management
Code Patterns and Conventions → PATTERNS.md
Implementation Details → IMPLEMENTATION.md
Frontend Architecture → frontend.md
Cost Summary
Development Velocity Assumptions

Architecture Subdocuments¶

Document	Purpose
architecture/SERVICES.md	Core services (Aurora, S3, Email) and external integrations (SmartVault, SurePrep, Stripe)
architecture/PATTERNS.md	Code patterns: async-first, DI, Pydantic, transactions, repository pattern, testing
architecture/IMPLEMENTATION.md	Implementation details by sequence (S2-S11), component tables, flow diagrams
architecture/frontend.md	Frontend architecture: React + Vite, Client Portal, Staff App, shared components

1. Design Principles¶

1.1 Service Centralization (Single Point of Access)¶

All external service access MUST go through centralized service modules.

This is the foundational architectural principle. Benefits: - Single place to modify connection logic, credentials, retry policies - Consistent error handling across all callers - Simplified testing via mock injection - Audit logging in one location - Connection pooling and resource management

Anti-pattern (DO NOT DO):

# BAD: Direct service access scattered throughout codebase
def some_business_logic():
    conn = snowflake.connector.connect(
        account=os.environ['SNOWFLAKE_ACCOUNT'],
        user=os.environ['SNOWFLAKE_USER'],
        # ... credentials scattered everywhere
    )
    cursor = conn.cursor()
    cursor.execute("SELECT ...")

Required pattern:

# GOOD: All access through centralized service
from src.services.snowflake_service import SnowflakeService

def some_business_logic():
    sf = SnowflakeService.get_instance()
    results = sf.execute_query("SELECT ...")

1.2 Additional Principles¶

Principle	Description
Configuration Isolation	All config in one place (src/config/), never hardcoded
Explicit Dependencies	Services declare dependencies via constructor injection
Fail Fast	Validate configuration at startup, not at first use
Audit Everything	All data access and mutations logged automatically
Immutable Records	Tax data uses append-only patterns with version history

2. Mixed Architecture Overview¶

This project uses both Java and Python, each for their strengths:

2.1 Technology Selection¶

┌─────────────────────────────────────────────────────────────────┐
│                        TAX PRACTICE AI                          │
├─────────────────────────────────────────────────────────────────┤
│  PYTHON                          │  JAVA                        │
│  ──────                          │  ────                        │
│  • FastAPI (REST API)            │  • Document extraction/OCR   │
│  • Lambda functions              │  • Tax calculation engine    │
│  • Bedrock/AI integration        │  • High-volume batch ops     │
│  • Configuration management      │  • Performance-critical I/O  │
│  • Workflow logic                │                              │
├─────────────────────────────────────────────────────────────────┤
│                    SHARED: config.yaml                          │
│        Python: PyYAML loader    |    Java: SnakeYAML loader     │
└─────────────────────────────────────────────────────────────────┘

2.2 Why Mixed Architecture?¶

Concern	Language	Rationale
API Layer	Python	FastAPI is productive, async-native, great for REST
AI Integration	Python	Bedrock SDK, prompt engineering, rapid iteration
Orchestration	AWS-native	Step Functions + EventBridge (serverless)
Document Parsing	Java	Performance for OCR/PDF processing at scale
Tax Calculations	Java	Complex business logic, type safety, existing libraries
Batch Processing	Java	Memory efficiency, parallelism for large datasets

2.3 Shared Configuration¶

Both languages read from the same config.yaml:

# config.yaml - Single source of truth for Python AND Java

# =============================================================================
# SNOWFLAKE CONNECTION
# =============================================================================
snowflake:
  # account: Snowflake account identifier
  # Format: <org>-<account> or <locator>.<region>
  account: ${SNOWFLAKE_ACCOUNT}

  # database: Target database (environment-specific)
  # NO DEFAULT - must be explicitly set
  database: ${SNOWFLAKE_DATABASE}

  # warehouse: Compute warehouse
  warehouse: ${SNOWFLAKE_WAREHOUSE}

  # role: Access role (optional, defaults to SYSADMIN for dev)
  role: ${SNOWFLAKE_ROLE:-SYSADMIN}

Python loading:

from src.config import get_config
config = get_config()
print(config.snowflake.account)

Java loading:

Config config = ConfigLoader.load("config.yaml");
String account = config.getSnowflake().getAccount();

2.4 Communication Patterns¶

Pattern	Use Case
Direct DB	Both access Aurora/Snowflake via centralized services
Message Queue	SQS for async handoff (Python → Java batch jobs)
REST API	Java components call Python API for workflow updates
Shared Storage	S3 for document handoff between components

3. Project Structure¶

ali-ai-acctg/
│
├── src/
│   │
│   ├── config/                     # === CONFIGURATION (centralized) ===
│   │   ├── __init__.py
│   │   ├── settings.py             # All application settings
│   │   ├── environments.py         # Environment-specific overrides
│   │   └── secrets.py              # Secrets management (pulls from env/vault)
│   │
│   ├── services/                   # === EXTERNAL SERVICE ACCESS (centralized) ===
│   │   ├── __init__.py
│   │   ├── base_service.py         # Abstract base for all services
│   │   ├── aurora_service.py       # Aurora PostgreSQL access
│   │   ├── s3_service.py           # Document storage
│   │   ├── email_service.py        # SES/SendGrid
│   │   ├── sms_service.py          # Twilio
│   │   ├── google_service.py       # Google Workspace (Meet, Calendar, Drive, E-signatures)
│   │   ├── persona_service.py      # Identity verification (S2-003)
│   │   ├── template_service.py     # Engagement template rendering (S3-001)
│   │   ├── audit_service.py        # Audit logging service
│   │   ├── malware_service.py      # ClamAV malware scanning (S4-003)
│   │   ├── classification_service.py  # AI document classification (S4-004)
│   │   ├── smartvault_service.py   # SmartVault sync (S4-005)
│   │   ├── sureprep_service.py     # SurePrep OCR/extraction (S4-006)
│   │   ├── email_intake_service.py # Email document intake (S4-002)
│   │   ├── stripe_service.py        # Stripe payments (S9-003, S10-002, S11)
│   │   # Note: E-filing via UltraTax CS (no separate service needed)
│   │   # Future: snowflake_service.py (analytics), bedrock_service.py (AI)
│   │
│   ├── repositories/               # === DATA ACCESS LAYER ===
│   │   ├── __init__.py
│   │   ├── base_repository.py      # Abstract base with common CRUD
│   │   ├── client_repository.py    # Client records (Aurora)
│   │   ├── email_verification_repository.py  # Email verification tokens (S2-001)
│   │   ├── login_attempt_repository.py  # Login attempt tracking (S2-002)
│   │   ├── magic_link_repository.py     # Magic link tokens (S2-002)
│   │   ├── identity_verification_repository.py  # Phone/ID/Persona verification (S2-003)
│   │   ├── engagement_repository.py  # Engagement/template/fee/consent (S3-001/S3-003)
│   │   ├── document_repository.py    # Document upload/metadata (S4-001)
│   │   ├── extraction_repository.py  # Document extraction data (S4-004/S4-006)
│   │   ├── checklist_repository.py   # Document checklists (S4-007)
│   │   ├── messaging_repository.py   # Messaging/notifications (S8)
│   │   ├── delivery_repository.py    # Delivery packages/signatures/payments (S9)
│   │   ├── efiling_repository.py     # E-file submissions/rejections (S10)
│   │   ├── invoice_repository.py     # Invoices/payments/aging (S11)
│   │   # Future: return_repository.py (tax returns), workflow_repository.py, analytics_repository.py
│   │
│   ├── domain/                     # === BUSINESS DOMAIN MODELS ===
│   │   ├── __init__.py
│   │   ├── client.py               # Client entity (with registration fields)
│   │   ├── registration.py         # Registration entities (S2-001)
│   │   ├── authentication.py       # Authentication entities (S2-002)
│   │   ├── identity_verification.py  # Identity verification entities (S2-003)
│   │   ├── engagement.py           # Engagement letter (S3-001)
│   │   ├── document.py             # Document entities (S4-001)
│   │   ├── extraction.py           # Document extraction (S4-004/S4-006)
│   │   ├── checklist.py            # Document checklists (S4-007)
│   │   ├── workflow.py             # Tax return workflow (S5-S7)
│   │   ├── messaging.py            # Messaging/notifications (S8)
│   │   ├── delivery.py             # Delivery packages/signatures/payments (S9)
│   │   ├── efiling.py              # E-file submissions/rejections (S10)
│   │   ├── invoice.py              # Invoice/Payment/AgingReport (S11)
│   │   ├── audit.py                # Audit logging entities
│   │   # Future: estimated_tax.py
│   │
│   ├── workflows/                  # === BUSINESS LOGIC / USE CASES ===
│   │   ├── __init__.py
│   │   ├── intake/                 # Client intake workflow
│   │   │   ├── __init__.py
│   │   │   ├── registration_workflow.py     # Client self-registration (S2-001)
│   │   │   ├── client_authentication.py     # Returning client auth (S2-002)
│   │   │   ├── identity_verification.py     # Identity verification (S2-003)
│   │   │   ├── engagement_workflow.py       # Engagement letters (S3-001/S3-002)
│   │   │   └── conflict_check.py
│   │   ├── consent_workflow.py              # Form 7216 consent (S3-003)
│   │   ├── documents/              # Document processing workflow (S4)
│   │   │   ├── __init__.py
│   │   │   ├── upload_workflow.py           # Document upload (S4-001)
│   │   │   ├── email_intake_workflow.py     # Email document intake (S4-002)
│   │   │   ├── scan_workflow.py             # Malware scanning (S4-003)
│   │   │   ├── classification_workflow.py   # AI classification (S4-004)
│   │   │   ├── smartvault_sync_workflow.py  # SmartVault sync (S4-005)
│   │   │   ├── sureprep_extraction_workflow.py # SurePrep OCR (S4-006)
│   │   │   ├── checklist_workflow.py        # Document checklists (S4-007)
│   │   │   └── correction_workflow.py       # Extraction correction (S4-008)
│   │   ├── preparation/            # Tax preparation workflow
│   │   │   ├── __init__.py
│   │   │   ├── preliminary_analysis.py
│   │   │   ├── preparer_review.py
│   │   │   └── final_review.py
│   │   ├── delivery/               # Client delivery workflow (S9)
│   │   │   ├── __init__.py
│   │   │   ├── package_workflow.py        # Tax package generation (S9-001)
│   │   │   ├── signature_workflow.py      # Google Workspace signatures (S9-002)
│   │   │   └── payment_workflow.py        # Stripe payment authorization (S9-003)
│   │   ├── filing/                 # E-filing workflow (S10)
│   │   │   ├── __init__.py
│   │   │   └── efiling_workflow.py        # E-file status/ready/rejections (S10-001 to S10-004)
│   │   ├── billing/                # Billing workflow (S11)
│   │   │   ├── __init__.py
│   │   │   ├── invoice_workflow.py        # Invoice generation/sending/payment (S11-002, S11-003)
│   │   │   └── reminder_workflow.py       # Payment reminders/aging (S11-004)
│   │   └── estimated_tax/          # Estimated tax workflow
│   │       ├── __init__.py
│   │       ├── voucher_generation.py
│   │       └── reminder_scheduler.py
│   │
│   ├── ai/                         # === AI CAPABILITIES ===
│   │   ├── __init__.py
│   │   ├── prompts/                # Prompt templates
│   │   │   ├── document_extraction.py
│   │   │   ├── preliminary_analysis.py
│   │   │   └── qa_assistant.py
│   │   ├── skills/                 # Knowledge modules for AI
│   │   │   ├── federal/            # Federal tax knowledge by year
│   │   │   │   └── 2024/
│   │   │   ├── state/              # State tax knowledge by year
│   │   │   │   └── 2024/
│   │   │   ├── firm/               # Firm-specific guidelines
│   │   │   └── integrations/       # External system knowledge
│   │   │       ├── smartvault/     # SmartVault API understanding
│   │   │       │   ├── api_reference.md
│   │   │       │   ├── folder_structures.md
│   │   │       │   └── mappings.py
│   │   │       ├── sureprep/       # SurePrep data interpretation
│   │   │       │   ├── api_reference.md
│   │   │       │   ├── extraction_fields.md
│   │   │       │   ├── confidence_scoring.md
│   │   │       │   └── mappings.py
│   │   │       └── ultratax/       # UltraTax field knowledge (no API)
│   │   │           ├── field_mappings.md
│   │   │           └── form_structures.md
│   │   └── extractors/             # Document-specific extractors
│   │       ├── w2_extractor.py
│   │       ├── 1099_extractor.py
│   │       └── k1_extractor.py
│   │
│   ├── api/                        # === API LAYER ===
│   │   ├── __init__.py
│   │   ├── main.py                 # FastAPI application
│   │   ├── routes/
│   │   │   ├── __init__.py
│   │   │   ├── clients.py
│   │   │   ├── registration.py     # Public registration endpoints (S2-001)
│   │   │   ├── client_auth.py      # Client authentication endpoints (S2-002)
│   │   │   ├── verification.py     # Identity verification endpoints (S2-003)
│   │   │   ├── documents.py
│   │   │   ├── returns.py
│   │   │   ├── workflow.py
│   │   │   ├── review.py           # Preparer/reviewer interface (S7)
│   │   │   ├── messaging.py        # Client messaging (S8)
│   │   │   ├── delivery.py         # Package/signature/payment (S9)
│   │   │   ├── efiling.py          # E-file status/ready/rejections (S10)
│   │   │   ├── invoices.py         # Invoice/payment/aging (S11)
│   │   │   └── webhooks.py         # Stripe, Persona, Google, SurePrep callbacks
│   │   ├── middleware/
│   │   │   ├── __init__.py
│   │   │   ├── auth.py
│   │   │   ├── audit_logging.py
│   │   │   └── rate_limiting.py
│   │   └── schemas/                # Pydantic request/response models
│   │       ├── __init__.py
│   │       ├── client_schemas.py
│   │       ├── registration_schemas.py  # Registration API schemas (S2-001)
│   │       ├── client_auth_schemas.py   # Authentication API schemas (S2-002)
│   │       ├── verification_schemas.py  # Verification API schemas (S2-003)
│   │       ├── document_schemas.py
│   │       ├── return_schemas.py
│   │       ├── review_schemas.py        # Preparer/reviewer schemas (S7)
│   │       ├── messaging_schemas.py     # Messaging schemas (S8)
│   │       ├── delivery_schemas.py      # Package/signature/payment schemas (S9)
│   │       ├── efiling_schemas.py       # E-file status/ready/rejection schemas (S10)
│   │       └── invoice_schemas.py       # Invoice/payment/aging schemas (S11)
│   │
│   ├── portal/                     # === CLIENT PORTAL (if separate) ===
│   │   └── (TBD - may be separate frontend repo)
│   │
│   └── utils/                      # === SHARED UTILITIES ===
│       ├── __init__.py
│       ├── encryption.py           # Field-level encryption (SSN, account numbers)
│       ├── validation.py           # Common validators (SSN, EIN, phone, email)
│       ├── date_utils.py           # Tax calendar utilities
│       └── formatting.py           # Currency, SSN masking, etc.
│
├── tests/                          # Python tests
│   ├── unit/
│   ├── integration/
│   └── fixtures/
│
├── java/                           # === JAVA COMPONENTS ===
│   ├── common/                     # Shared Java utilities
│   │   ├── pom.xml
│   │   └── src/main/java/co/aliunde/common/
│   │       ├── config/             # Config loading (SnakeYAML)
│   │       ├── db/                 # Database access (centralized)
│   │       └── util/               # Shared utilities
│   │
│   ├── document-extractor/         # Document OCR/extraction
│   │   ├── pom.xml
│   │   └── src/main/java/co/aliunde/extractor/
│   │       ├── W2Extractor.java
│   │       ├── Form1099Extractor.java
│   │       └── ...
│   │
│   └── tax-engine/                 # Tax calculation integration
│       ├── pom.xml
│       └── src/main/java/co/aliunde/tax/
│
├── scripts/                        # Operational scripts
│   ├── local_api.py               # ⚠️ DEV-ONLY: Monolithic API for local testing
│   ├── bootstrap.py               # ⚠️ DEV-ONLY: Database setup for local dev
│   ├── setup_dev_environment.sh
│   ├── launch_claude.sh
│   └── db_migrations/
│
├── infrastructure/                 # IaC (Terraform/CDK)
│   ├── terraform/
│   ├── step-functions/             # Step Function state machines (JSON/YAML)
│   ├── lambda/                     # Lambda function definitions
│   └── docker/
│
├── config.yaml                     # Shared config (Python + Java)
├── .env                            # Local environment (not committed)
├── .env.example                    # Template for .env
├── .gitignore
├── pyproject.toml                  # Python project config
├── requirements.txt                # Python dependencies
├── pom.xml                         # Parent POM for Java modules
├── ARCHITECTURE.md                 # This file
├── CLAUDE.md                       # Claude Code instructions
├── RUNBOOK.md                      # Operational procedures
└── backlog.md                      # Priority items and debt

4. Service Centralization¶

4.1 Service Registry Pattern¶

All external services are accessed through a centralized registry that: - Initializes services once at application startup - Validates configuration before first use - Provides consistent access patterns - Enables dependency injection for testing

# src/services/__init__.py

from src.services.snowflake_service import SnowflakeService
from src.services.aurora_service import AuroraService
from src.services.s3_service import S3Service
from src.services.bedrock_service import BedrockService
# ... other services

class ServiceRegistry:
    """
    Centralized access point for all external services.

    Usage:
        from src.services import services

        # Access any service
        sf = services.snowflake
        aurora = services.aurora
        s3 = services.s3
    """

    _instance = None
    _initialized = False

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def initialize(self, config: 'AppConfig') -> None:
        """
        Initialize all services with configuration.
        Called once at application startup.
        Validates all connections before proceeding.
        """
        if self._initialized:
            return

        self._snowflake = SnowflakeService(config.snowflake)
        self._aurora = AuroraService(config.aurora)
        self._s3 = S3Service(config.s3)
        self._bedrock = BedrockService(config.bedrock)
        # ... initialize other services

        # Validate connections at startup (fail fast)
        self._validate_connections()

        self._initialized = True

    @property
    def snowflake(self) -> SnowflakeService:
        self._ensure_initialized()
        return self._snowflake

    @property
    def aurora(self) -> AuroraService:
        self._ensure_initialized()
        return self._aurora

    # ... other service properties

    def _ensure_initialized(self) -> None:
        if not self._initialized:
            raise RuntimeError(
                "ServiceRegistry not initialized. "
                "Call services.initialize(config) at startup."
            )

    def _validate_connections(self) -> None:
        """Verify all services are reachable at startup."""
        # Each service implements a health_check() method
        pass

# Singleton instance for import
services = ServiceRegistry()

4.2 Base Service Contract¶

All services inherit from a base class ensuring consistent behavior:

# src/services/base_service.py

from abc import ABC, abstractmethod
from typing import TypeVar, Generic
import logging

ConfigT = TypeVar('ConfigT')

class BaseService(ABC, Generic[ConfigT]):
    """
    Base class for all external service wrappers.

    Provides:
    - Consistent initialization pattern
    - Health check interface
    - Logging setup
    - Retry policy hooks
    """

    def __init__(self, config: ConfigT):
        self._config = config
        self._logger = logging.getLogger(self.__class__.__name__)
        self._initialize()

    @abstractmethod
    def _initialize(self) -> None:
        """Set up connections/clients. Called once during construction."""
        pass

    @abstractmethod
    def health_check(self) -> bool:
        """
        Verify service is reachable and credentials are valid.
        Returns True if healthy, raises exception with details if not.
        """
        pass

    @property
    def logger(self) -> logging.Logger:
        return self._logger

5. Core Services¶

Full Details: architecture/SERVICES.md

Core services provide centralized access to infrastructure:

Service	Purpose
AuroraService	Primary database (all operational data)
S3Service	Document storage with KMS encryption
EmailService	Transactional email via SES

All services follow the BaseService pattern with health checks, connection pooling, and consistent error handling.

6. Java Components¶

6.1 Overview¶

Java components handle performance-critical processing. They follow the same centralization principles as Python:

Centralized database access via common library
Shared configuration from config.yaml
Consistent logging and error handling

6.2 Common Library (java/common)¶

Provides shared infrastructure for all Java modules:

// co.aliunde.common.config.ConfigLoader
public class ConfigLoader {
    /**
     * Load configuration from config.yaml with environment variable substitution.
     *
     * Environment variable syntax in YAML:
     *   ${VAR_NAME}           - Required, fails if not set
     *   ${VAR_NAME:-default}  - Optional with default value
     */
    public static AppConfig load(String configPath) {
        // Uses SnakeYAML for parsing
        // Substitutes environment variables after loading
    }
}

// co.aliunde.common.db.SnowflakeClient - Centralized Snowflake Access
public class SnowflakeClient {
    private static SnowflakeClient instance;
    private final SnowflakeConfig config;

    /**
     * All Snowflake queries in Java components go through this client.
     * No direct JDBC connection creation elsewhere.
     */
    public static SnowflakeClient getInstance() {
        if (instance == null) {
            instance = new SnowflakeClient(ConfigLoader.load().getSnowflake());
        }
        return instance;
    }

    public List<Map<String, Object>> executeQuery(String sql, Object... params) {
        // Connection management, logging, error handling
    }
}

6.3 Document Extractor (java/document-extractor)¶

Extracts structured data from tax documents (W-2s, 1099s, etc.):

// co.aliunde.extractor.W2Extractor
public class W2Extractor implements DocumentExtractor {

    /**
     * Extract W-2 data from document image/PDF.
     * Uses OCR and pattern matching to identify fields.
     */
    public W2Data extract(byte[] documentContent) {
        // OCR processing
        // Field identification
        // Validation
        return w2Data;
    }
}

Build and run:

cd java/document-extractor
mvn clean package

java -jar target/document-extractor-1.0.0.jar \
  --input /path/to/documents \
  --output /path/to/extracted \
  --batch-id 123

6.4 Tax Engine Integration (java/tax-engine)¶

Integrates with third-party tax software (Drake, Lacerte, etc.):

// co.aliunde.tax.TaxEngineClient
public interface TaxEngineClient {
    /**
     * Submit return data to tax software for calculation.
     */
    CalculationResult calculate(TaxReturnData data);

    /**
     * Generate final forms for e-filing.
     */
    List<TaxForm> generateForms(TaxReturnData data);
}

7. Database Layer¶

7.1 Record Retention Requirements¶

Tax preparers must retain records per federal and state regulations:

Requirement	Period	Source
IRS minimum (preparers)	3 years	26 CFR § 1.6107-1
IRS extended (underreporting)	6 years	IRS guidelines
SEC workpapers	7 years	SEC rules
State requirements (e.g., NY)	7 years	State boards
AICPA recommendation	7 years	Industry best practice
Basis records (assets)	Indefinite	Until sold + 7 years

Design requirement: System must retain all records for minimum 7 years.

7.2 Architecture Decision: Start Simple¶

Principle: Add complexity only when you hit actual pain.

Aurora PostgreSQL can handle: - 7+ years of audit logs (with table partitioning) - Hundreds of thousands of return records - All dashboard and reporting needs for small/medium firms - Full-text search, JSON queries, time-series data

Why not start with Snowflake/Athena? - Adds ETL jobs, schema sync, query routing complexity - Two systems to maintain instead of one - Cost overhead not justified at small scale - Developers must know which system to query

7.3 Recommended Architecture¶

┌─────────────────────────────────────────────────────────────────┐
│                     STARTING ARCHITECTURE                        │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│   ┌─────────────────┐                                           │
│   │  Aurora         │  ← All operational data                   │
│   │  PostgreSQL     │    Clients, returns, workflow, audit      │
│   │                 │    7+ years with table partitioning       │
│   └─────────────────┘                                           │
│                                                                  │
│   ┌─────────────────┐                                           │
│   │  S3             │  ← Documents only                         │
│   │                 │    PDFs, images, signed forms             │
│   │                 │    (Not a query layer)                    │
│   └─────────────────┘                                           │
│                                                                  │
│   ┌─────────────────┐                                           │
│   │  Bedrock        │  ← AI via API                             │
│   │  (Claude)       │    Document analysis, Q&A                 │
│   └─────────────────┘                                           │
│                                                                  │
│   Estimated cost: $150-300/month at small scale                 │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

7.4 When to Add Complexity¶

Pain Point	Solution
Audit queries slow on old data	Add table partitioning by year
Still slow after partitioning	Archive to S3 + Athena
Need cross-client analytics	Evaluate Snowflake
Heavy ML/AI feature engineering	Evaluate Snowflake or Databricks

Trigger thresholds: - 10,000+ returns/year - Query response > 5 seconds on indexed data - Complex analytics spanning multiple years

7.5 Future Architecture (If Needed)¶

┌─────────────────────────────────────────────────────────────────┐
│                     SCALED ARCHITECTURE                          │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│   ┌─────────────────┐                                           │
│   │  Aurora         │  ← Hot data (current + 1 year)            │
│   │  PostgreSQL     │    API serving, workflow                  │
│   └────────┬────────┘                                           │
│            │ CDC or nightly archive                              │
│            ▼                                                     │
│   ┌─────────────────┐                                           │
│   │  Snowflake      │  ← Historical analytics                   │
│   │  (or S3+Athena) │    7+ year queries                        │
│   └─────────────────┘    Cross-client reporting                 │
│                                                                  │
│   Add only when Aurora partitioning is insufficient             │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

7.6 Local Development¶

Aurora is cloud-only. Use PostgreSQL in Docker for local development (Aurora is PostgreSQL-compatible).

# docker-compose.yml
services:
  postgres:
    image: postgres:15
    environment:
      POSTGRES_DB: tax_practice
      POSTGRES_USER: dev_user
      POSTGRES_PASSWORD: dev_password
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data:

Environment-based config switching:

# config.yaml
database:
  # host: localhost for dev, Aurora endpoint for AWS
  host: ${DB_HOST:-localhost}
  port: ${DB_PORT:-5432}
  database: ${DB_NAME:-tax_practice}
  user: ${DB_USER:-dev_user}
  password: ${DB_PASSWORD}

Environment	DB_HOST
Local dev	localhost (Docker)
AWS dev	tax-practice-dev.cluster-xxx.us-east-1.rds.amazonaws.com
AWS prod	tax-practice-prod.cluster-xxx.us-east-1.rds.amazonaws.com

Development-Only Scripts (Never Deploy to Production):

The following scripts provide a simplified development experience but must never be used in production:

Script	Purpose	Why Dev-Only
`scripts/local_api.py`	Monolithic API server (1,730 lines)	No connection pooling, hardcoded config, duplicates src/ functionality
`scripts/bootstrap.py`	Database schema + seed data	Creates tables directly, bypasses migrations

For production, use: - src/api/ - Proper FastAPI application with connection pooling - infrastructure/db_migrations/ - Versioned schema migrations

The production architecture in src/ follows proper layering (Services → Repositories → Domain) while the dev scripts prioritize convenience for local testing and demos.

7.7 Aurora Tables¶

For complete entity definitions, attributes, and relationships, see DATA_MODEL.md.

Summary of core tables (singular naming convention per PostgreSQL best practices):

Table	Purpose
client	Client master records
client_contact	Multiple contacts per client
tax_return	Tax return header records
document	Document metadata (content in S3)
workflow_state	Current workflow state per return
workflow_history	Audit trail of state transitions
engagement	Engagement letters
invoice	Billing records
estimated_tax_payment	Quarterly payment tracking
users	Staff user accounts (plural to avoid reserved word)
audit_log	All data access and modifications

See DATA_MODEL.md for complete entity definitions (35 entities across 9 domain groups).

7.8 S3 Document Storage¶

S3 is used for document content only (not as a query layer):

s3://tax-practice-documents/
├── clients/
│   └── {client_id}/
│       └── {year}/
│           ├── w2_upload_20241215.pdf
│           ├── 1099_upload_20241216.pdf
│           └── signed_8879_20241220.pdf
├── returns/
│   └── {return_id}/
│       ├── final_return.pdf
│       └── supporting_docs/
└── engagements/
    └── {engagement_id}/
        └── signed_engagement_letter.pdf

Document metadata lives in Aurora (document table). S3 stores the actual files.

8. External Integrations¶

Full Details: architecture/SERVICES.md

External integrations use a dual approach: - Services (src/services/) - API calls, auth, connection management - Skills (src/ai/skills/integrations/) - AI context for data interpretation

Service	Purpose
SmartVaultService	Client document portal
SurePrepService	Document OCR/extraction, UltraTax bridge
GoogleService	Google Workspace (Meet, Calendar, E-signatures)
StripeService	Payment processing (S9, S10, S11)
PersonaService	Identity verification (S2-003)

Note: UltraTax has no API - e-filing handled via SurePrep CS Connect.

9. Workflow Orchestration (Airflow)¶

9.1 Overview¶

Workflow orchestration uses self-hosted Apache Airflow on a dedicated EC2 instance.

┌─────────────────────────────────────────────────────────────────┐
│                    ORCHESTRATION ARCHITECTURE                    │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│   ┌─────────────────┐                                           │
│   │  Airflow        │  ← Workflow orchestration                 │
│   │  (EC2 t3.med)   │    DAGs, scheduling, retries, monitoring  │
│   │  PostgreSQL     │    Metadata DB (local or Aurora)          │
│   └────────┬────────┘                                           │
│            │                                                     │
│            ▼                                                     │
│   ┌─────────────────┐                                           │
│   │  Task Execution │  ← Compute                                │
│   │  - Local        │    Simple tasks on Airflow VM             │
│   │  - Lambda       │    Serverless tasks via boto3             │
│   │  - Fargate      │    Long-running containers                │
│   └─────────────────┘                                           │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

9.2 Why Self-Hosted Airflow¶

Factor	Self-Hosted Airflow	MWAA (Managed)	Step Functions
Cost	~$20-35/mo (EC2+EBS)	$360+/mo	Pay per transition
Control	Full	Limited	Limited
UI	Yes (monitoring, logs)	Yes	Basic
DAG complexity	Python (full power)	Python	JSON/YAML
Cold start	Always warm	Minutes	Instant
Maintenance	Self-managed	Managed	Zero

Decision Rationale: Self-hosted Airflow on t3.medium provides full orchestration capabilities at ~$20-35/month vs $360+/month for MWAA, acceptable for a small practice with predictable workloads.

9.3 Infrastructure Specification¶

Component	Specification	Cost (On-Demand)	Cost (Reserved 1yr)
EC2 Instance	t3.medium (2 vCPU, 4GB)	~$30/mo	~$19/mo
EBS Storage	50GB gp3	~$4/mo	~$4/mo
Total		~$34/mo	~$23/mo

Upgrade Path: If running document parsers on the same VM, upgrade to t3.large (2 vCPU, 8GB) for ~$29/mo reserved.

9.4 Airflow Configuration¶

# airflow/airflow.cfg (key settings)

[core]
# executor: LocalExecutor for single-node deployment
executor = LocalExecutor

# dags_folder: DAG definitions
dags_folder = /opt/airflow/dags

# parallelism: max active tasks across all DAGs
parallelism = 8

[scheduler]
# dag_dir_list_interval: how often to scan for new DAGs (seconds)
dag_dir_list_interval = 300

[webserver]
# web_server_port: Airflow UI port
web_server_port = 8080

# authenticate: enable authentication
authenticate = True

[database]
# sql_alchemy_conn: PostgreSQL connection (local or Aurora)
sql_alchemy_conn = postgresql://airflow:password@localhost/airflow

9.5 DAG Examples¶

Document Processing DAG¶

# dags/document_processing.py

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.operators.lambda_function import LambdaInvokeFunctionOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'tax-practice-ai',
    'depends_on_past': False,
    'email_on_failure': True,
    'email': ['alerts@example.com'],
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'document_processing',
    default_args=default_args,
    description='Process uploaded tax documents',
    schedule_interval=None,  # Triggered by S3 event
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['documents', 'core'],
) as dag:

    classify = LambdaInvokeFunctionOperator(
        task_id='classify_document',
        function_name='classify-document',
        payload='{{ dag_run.conf }}',
    )

    extract = LambdaInvokeFunctionOperator(
        task_id='extract_data',
        function_name='extract-document-data',
    )

    validate = PythonOperator(
        task_id='validate_extraction',
        python_callable=validate_extraction_confidence,
    )

    store = LambdaInvokeFunctionOperator(
        task_id='store_results',
        function_name='store-extraction',
    )

    classify >> extract >> validate >> store

Scheduled Tasks DAG¶

# dags/scheduled_tasks.py

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

with DAG(
    'daily_tasks',
    default_args=default_args,
    description='Daily scheduled operations',
    schedule_interval='0 9 * * *',  # Daily at 9 AM UTC
    start_date=datetime(2024, 1, 1),
    catchup=False,
    tags=['scheduled'],
) as dag:

    check_deadlines = PythonOperator(
        task_id='check_estimated_tax_deadlines',
        python_callable=check_upcoming_deadlines,
    )

    send_reminders = PythonOperator(
        task_id='send_reminder_emails',
        python_callable=send_deadline_reminders,
    )

    poll_efile = PythonOperator(
        task_id='poll_efile_status',
        python_callable=check_efile_status_updates,
    )

    check_deadlines >> send_reminders
    poll_efile  # Runs in parallel

9.6 Use Cases¶

Task	DAG	Trigger	Notes
Document processing	document_processing	S3 event → API → Airflow	Classify, extract, validate, store
Estimated tax reminders	daily_tasks	Schedule (9 AM)	Check deadlines, send emails
E-file status check	daily_tasks	Schedule (hourly)	Poll SurePrep for UltraTax e-file status
Nightly reports	nightly_reports	Schedule (2 AM)	Generate summaries, archive logs
Client onboarding	client_onboarding	API trigger	Multi-step with approval waits
Bookkeeping monthly	bookkeeping_monthly	Schedule (1st of month)	Statement reminders, reconciliation

9.7 Webhook-to-Airflow Integration¶

External webhooks (SurePrep, Stripe, Persona, Google) integrate with Airflow using an event-driven pattern. Note: E-file status comes via SurePrep CS Connect (UltraTax handles actual transmission).

┌─────────────────────────────────────────────────────────────────────┐
│                    WEBHOOK INTEGRATION FLOW                         │
├─────────────────────────────────────────────────────────────────────┤
│                                                                     │
│   External Service                                                  │
│        │                                                            │
│        ▼                                                            │
│   POST /v1/webhooks/{provider}  ─────► FastAPI Webhook Handler     │
│        │                                                            │
│        ├── 1. Verify signature (HMAC)                              │
│        ├── 2. Parse event payload                                   │
│        ├── 3. Update database state                                 │
│        ├── 4. Trigger Airflow DAG (if workflow action needed)      │
│        │         │                                                  │
│        │         ▼                                                  │
│        │   Airflow REST API: POST /api/v1/dags/{dag_id}/dagRuns    │
│        │         │                                                  │
│        │         ▼                                                  │
│        │   DAG executes multi-step workflow                        │
│        │                                                            │
│        └── 5. Return 200 OK to external service                    │
│                                                                     │
└─────────────────────────────────────────────────────────────────────┘

Webhook-to-DAG Mapping:

Webhook Event	Provider	Airflow DAG	Action
`binder.processing_complete`	SurePrep	document_processing	Import extracted data
`extraction.ready`	SurePrep	document_processing	Validate and store fields
`payment_intent.succeeded`	Stripe	payment_processing	Update invoice, send receipt
`inquiry.completed`	Persona	identity_verification	Advance verification tier
`signature.completed`	Google	signature_processing	Update engagement/consent
`efile.accepted`	SurePrep	efile_processing	Update return status, notify client
`efile.rejected`	SurePrep	efile_rejection	Parse error, notify preparer

Implementation Pattern:

# src/api/routes/webhooks.py

from airflow_client.client import Client as AirflowClient

async def trigger_airflow_dag(dag_id: str, conf: dict):
    """Trigger Airflow DAG via REST API."""
    client = AirflowClient(
        host=config.airflow.api_url,
        auth=(config.airflow.username, config.airflow.password)
    )
    client.trigger_dag(
        dag_id=dag_id,
        conf=conf,
        execution_date=datetime.utcnow().isoformat()
    )

@router.post("/webhooks/sureprep")
async def sureprep_webhook(request: Request):
    # 1. Verify signature
    verify_sureprep_signature(request)

    payload = await request.json()
    event = payload["event"]

    # 2. Update database
    await update_binder_status(payload)

    # 3. Trigger Airflow if workflow action needed
    if event in ["binder.processing_complete", "extraction.ready"]:
        await trigger_airflow_dag(
            dag_id="document_processing",
            conf={
                "binder_id": payload["binder_id"],
                "event": event,
                "documents": payload.get("data", {}).get("documents", [])
            }
        )

    return {"status": "ok"}

Key Design Decisions:

Immediate DB update: Webhook handler updates database state synchronously for consistency
Async DAG trigger: Airflow handles long-running multi-step workflows asynchronously
Idempotency: DAGs check current state before acting (handles duplicate webhooks)
Return quickly: Webhook handlers return 200 before DAG completion to avoid timeouts

9.8 Monitoring & Alerting¶

Airflow provides built-in monitoring:

Web UI: DAG status, task logs, execution history
Email alerts: On task failure (configurable)
Health checks: /health endpoint for uptime monitoring

Additional monitoring:

# CloudWatch Agent on Airflow EC2
# Metrics: CPU, memory, disk, task queue depth
# Alarms: High CPU, disk full, scheduler down

9.9 Deployment¶

# infrastructure/airflow/deploy.sh

# Install on EC2
sudo apt update
sudo apt install python3-pip postgresql

# Install Airflow
pip install apache-airflow[postgres,amazon]

# Initialize database
airflow db init

# Create admin user
airflow users create --username admin --role Admin

# Start services (via systemd)
sudo systemctl start airflow-webserver
sudo systemctl start airflow-scheduler

10. Configuration Management¶

10.1 Configuration Hierarchy¶

Environment Variables (.env / AWS Secrets Manager)
         ↓
    secrets.py (loads and validates secrets)
         ↓
    settings.py (application configuration)
         ↓
    environments.py (dev/staging/prod overrides)
         ↓
    ServiceRegistry (injects config into services)

10.2 Settings Structure¶

# src/config/settings.py

from dataclasses import dataclass
from src.services.snowflake_service import SnowflakeConfig
from src.services.aurora_service import AuroraConfig
from src.services.s3_service import S3Config
# ... other configs

@dataclass
class AppConfig:
    """
    Complete application configuration.
    All service configs centralized here.
    """
    environment: str  # "development", "staging", "production"

    # Database configurations
    snowflake: SnowflakeConfig
    aurora: AuroraConfig
    s3: S3Config

    # External service configurations
    bedrock: BedrockConfig
    google: GoogleConfig
    stripe: StripeConfig
    persona: PersonaConfig
    # Note: E-filing via UltraTax - no separate config needed
    email: EmailConfig
    sms: SMSConfig
    google: GoogleConfig

    @classmethod
    def from_environment(cls) -> 'AppConfig':
        """Build configuration from environment variables."""
        # Implementation loads all config from env/secrets
        pass

11. Code Patterns and Conventions¶

Full Details: architecture/PATTERNS.md Implementation Details: architecture/IMPLEMENTATION.md

This codebase follows consistent patterns:

Pattern	Purpose
Async-First	I/O-bound operations use async/await throughout
Dependency Injection	Container pattern for service instantiation
Pydantic Validation	Request/response schemas with automatic validation
Transaction Boundaries	Atomic operations per workflow
Repository Pattern	Data access abstraction with entity mapping
Layered Architecture	API → Workflows → Repositories → Services

Key Libraries: FastAPI, asyncpg, httpx, aiobotocore, Pydantic v2

12. Implementation Details¶

Full Details: architecture/IMPLEMENTATION.md

Implementation progress by sequence:

Sequence	Description	Status	Tests
S1	Project Setup	Complete	-
S2	Client Onboarding	Complete	187
S3	Engagement & Consent	Complete	95
S4	Document Management	Complete	120
S5	AI Analysis	Complete	85
S6	Tax Return Workflow	Complete	90
S7	Preparer/Reviewer Interface	Complete	145
S8	Client Communication	Complete	122
S9	Client Delivery	Complete	135
S10	E-Filing Status	Complete	165
S11	Billing & Payments	Complete	78
Total		1522 tests

13. Frontend Architecture¶

Full Details: architecture/frontend.md

Tax Practice AI uses a two-application frontend architecture:

Application	Purpose	Users
Client Portal	Document upload, status checking, signing, payments	Tax clients
Staff App	Workflow queues, review UI, AI Q&A, billing	Preparers, reviewers, admins

Tech Stack: React 18, Vite, TypeScript, Tailwind CSS, shadcn/ui, React Query, Zustand

Both apps share a common component library (@tax-practice/ui) with document viewer, signature pad, and upload components.

14. Cost Summary¶

See COST_DETAIL.md for full details.

Category	Monthly Cost
AWS Infrastructure	$135
Third-Party Services	$285
Project Total	$420/mo

Existing client software (UltraTax, SurePrep, SmartVault): ~$10,000-40,000/year - not project costs

15. Development Velocity Assumptions¶

This section documents the assumptions underlying effort estimates. Use these to recalculate if team composition or tooling changes.

15.1 Current Model: AI-Assisted Development¶

Role	Responsibility
Human Architect	Design decisions, requirements, code review, approval gates
Claude (AI)	All implementation, testing, documentation

Force Multiplier: AI-assisted development is estimated at 5-10x velocity compared to traditional junior/mid developer implementation.

15.2 Baseline Estimates¶

Scope	Estimate	Notes
Tax Practice AI MVP	11 weeks	Full system: intake through e-filing
Individual service integration	2-3 days	e.g., SmartVault service + skill
Complex service integration	3-5 days	e.g., SurePrep service + skill
Full integration layer	1-2 weeks	All external services connected
New workflow module	3-5 days	e.g., estimated tax reminders
AI skill (tax knowledge)	1-2 days	e.g., state-specific rules
AI skill (integration)	1-2 days	e.g., SurePrep field mappings

15.3 Key Assumptions¶

Assumption	Impact if Wrong
Architect available for decisions	Delays if blocked on approvals
Claude Code has repo context	Slower if frequent context rebuilding
APIs behave as documented	Add 50-100% buffer for undocumented quirks
No major requirement changes mid-sprint	Rework costs are real even with AI
Testing integrated into implementation	No separate QA phase needed

15.4 Recalculation Guide¶

If assumptions change, adjust estimates:

Change	Adjustment
Add junior developer (instead of AI)	Multiply estimates by 5-10x
Add senior developer (instead of AI)	Multiply estimates by 2-3x
Architect unavailable (async only)	Add 50% for decision latency
New/undocumented API	Add 100% buffer for discovery
Regulatory compliance review required	Add 1-2 weeks per review cycle

15.5 Velocity Tracking¶

Track actual vs estimated to refine future estimates:

Feature	Estimated	Actual	Variance	Notes
(To be filled as features complete)

16. V1 Companion Deployment¶

16.1 Overview¶

V1 deploys Tax Practice AI as a back-office AI companion for Tax Season 2025. Staff use AI analysis capabilities alongside existing workflows without disrupting client-facing processes.

Philosophy: Augment, don't replace. Zero client disruption.

16.2 Deployment Model¶

┌─────────────────────────────────────────────────────────────────┐
│                     V1 COMPANION ARCHITECTURE                    │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│   EXISTING SYSTEMS (unchanged)        TAX PRACTICE AI (V1)      │
│   ┌─────────────┐                    ┌─────────────────────┐   │
│   │  SmartVault │──── manual ───────►│   Staff App         │   │
│   │  (uploads)  │     download       │   • Quick client    │   │
│   └─────────────┘                    │   • Upload docs     │   │
│                                       │   • AI analysis     │   │
│   ┌─────────────┐                    │   • Annotations     │   │
│   │  UltraTax   │                    │   • Worksheet       │   │
│   │  (prep)     │                    └─────────┬───────────┘   │
│   └─────────────┘                              │               │
│                                                 ▼               │
│   ┌─────────────┐                    ┌─────────────────────┐   │
│   │   Legacy    │◄── link ──────────│   AWS Cloud         │   │
│   │   System    │    account #       │   • S3 (docs)       │   │
│   └─────────────┘                    │   • RDS (data)      │   │
│                                       │   • Anthropic API   │   │
│                                       └─────────────────────┘   │
└─────────────────────────────────────────────────────────────────┘

16.3 What Changes vs Stays Same¶

Component	V1 Behavior
Staff App	Deployed and used daily
AI Analysis	Fully operational via Anthropic API
Document Storage	S3 with graceful offline fallback
Client Records	Linked to legacy via account number
Worksheet Export	PDF/Excel with source citations

Component	Existing Behavior (unchanged)
Client Portal	SmartVault
Tax Preparation	UltraTax CS
E-Filing	Via UltraTax
Client Communication	Existing processes
Source of Truth	Legacy system

16.4 Key V1 Features¶

Feature	Description
Quick Client Entry	Minimal form: name, tax year, optional legacy account
Drag-Drop Upload	Upload documents directly into viewer
Folder Import	Point to local folder, import all documents
AI Classification	Automatic document type identification
AI Analysis	Prior year comparison, anomaly detection, missing docs
Q&A Assistant	Ask questions during review with source citations
Annotations	Notes, flags, questions on documents
Worksheet Export	PDF/Excel with full source citations
Offline Fallback	AI works when S3 unavailable

16.5 Account Numbering¶

New clients get system-generated account numbers: - Format: Prefix + sequence (e.g., A10001, A10002) - Prefix configurable (default: 'A') - Legacy account number stored separately for cross-reference

Document	Description
V1_COMPANION_REQUIREMENTS.md	Full requirements specification
V1_USE_CASES.md	Detailed use cases
V1_UI_CHANGES.md	UI component specifications

Document History¶

Full version history available in architecture/IMPLEMENTATION.md.

Version	Date	Author	Changes
1.3	2024-12-24	Don McCarty	S9: Client Delivery. 135 tests.
1.4	2024-12-24	Don McCarty	S10: E-Filing Status. StripeService, payment gate. 1340 tests.
1.5	2024-12-24	Don McCarty	S11: Billing & Payments. Invoice generation, payment collection, reminders. 1522 tests.
1.6	2024-12-24	Don McCarty	Split ARCHITECTURE.md into subdocs: SERVICES.md, PATTERNS.md, IMPLEMENTATION.md.
1.7	2024-12-24	Don McCarty	Added frontend.md: React + Vite architecture, Client Portal, Staff App.
1.8	2024-12-24	Don McCarty	V1 Companion deployment model for Tax Season 2025.

Document	Purpose
COST_DETAIL.md	Full cost breakdown: AWS, AI models, token usage, vendor pricing
DATA_MODEL.md	Logical data model: 35+ entities, attributes, relationships, ER diagrams
PROCESS_FLOWS.md	State machines and process flows: Tax Return (17 states), Document, E-Filing, Identity Verification
tax_practice_ai_requirements.md	Full requirements specification with business rules
RUNBOOK.md	Operational procedures
backlog.md	Priority items and technical debt