Processing Architecture

This document describes the processing system architecture for FarmCove Delta, which enables intelligent document processing, data extraction, validation, and workflow automation through both AI-driven and manual validation steps.

Table of Contents

1. Overview
2. System Components
3. Processing Flow
4. Database Schema
5. Service Architecture
6. Execution Group-Based Callback Architecture
7. Dynamic Processor System
8. Enhanced Address Matching with AI
9. Integration Points
10. Prompt Resolution System
11. Template Variable Substitution
12. Status Tracking
13. Error Handling
14. Security Considerations
15. Key Architecture Principles
16. Future Enhancements
17. Implementation Examples
18. Idempotency and Duplicate Processing Prevention
19. Conclusion

Overview

The Processing system provides a flexible, template-based approach to processing various entity types (transactions, budgets, schedules) through multi-step workflows. It supports both AI-driven processing and custom validation steps, with parallel execution, conditional logic, issue tracking, and comprehensive status monitoring.

Key Features

• Template-Based Workflows: Define reusable processing templates for different entity types
• Hybrid Processing: Supports both AI-driven steps (OCR, classification) and manual validation steps (total checks, format validation)
• Dynamic Processor System: Automatically routes step results to appropriate service processors
• Issue Resolution: Manual validation steps can create issues requiring human resolution
• Parallel Execution: Steps with the same execution group run concurrently
• Conditional Processing: Steps can be skipped based on conditions
• Real-time Status Updates: Track progress of individual steps, overall job, and issue resolution
• Polymorphic Design: Single system handles multiple entity types
• Error Recovery: Optional steps don't block the whole job on failure, and a cron sweep detects and either restarts or terminates jobs whose pipeline has gone silent (see Stall detection and recovery)
• Extensible Architecture: Easy to add new processors without modifying core logic

System Components

1. App Layer (Next.js)

• Service Layer: /packages/app/src/services/processing/
• base.ts: Core processing functions and processor registry (CRUD operations only)
• server.ts: Server-side operations with AI integrations and external services
• client.ts: Client-side operations for React components
• actions.ts: Server Actions for Next.js App Router
• processors/transactionProcessors.ts: Transaction-specific processors

Service Layer Architecture: Services follow a strict layered architecture where:

• Base layer handles pure business logic and database operations
• Server layer manages external service integrations (AI, geocoding, etc.)

• See docs/development/SERVICE_PATTERNS.md for detailed patterns

• API Routes: /packages/app/src/app/api/ai-processing/

• callback/route.ts: Process step results and route to processors

• start-job/route.ts: Initiate processing jobs

• UI Components:

• ProcessingStatusIndicator: Visual status display
• Integration with existing transaction/budget/schedule UI

2. AI Hub (Genkit Service)

• Location: /packages/aiHub/
• Purpose: Executes AI processing tasks
• Key Features:
• OCR document processing
• Document classification
• Data extraction with structured output
• Budget linking and validation
• Vendor matching

3. Database Layer

• Templates: Define processing workflows
• Jobs: Track processing instances
• Steps: Individual processing tasks
• Integration: Links to existing entities (transactions, budgets, etc.)

Processing Flow

Processing Flow Orchestration

The AI processing system uses a two-tier function architecture to manage the flow from initial request to AI Hub execution:

High-Level Orchestration: createJobAndStartProcessing

Purpose: Entry point for AI processing that handles all setup and orchestration.

Location: packages/app/src/services/processing/server.ts

Responsibilities:

1. Validates the processable type
2. Creates the AI processing job in the database
3. Routes to type-specific processing based on entity type
4. Fetches required data (e.g., project ID for transactions)
5. Calls the appropriate service's processing function

Usage Example:

// From WhatsApp bot or UI
const result = await createJobAndStartProcessing(
  PROCESSABLE_TYPE.TRANSACTIONS,
  transactionId,
  userId,
  'Receipt Processing',
  { uploadedVia: 'whatsapp' },
  'application/pdf'
);

Call Flow:

WhatsApp/UI Upload
    ↓
createJobAndStartProcessing (creates job)
    ↓
Type-specific router (e.g., startTransactionProcessing)
    ↓
sendFirstGroupToAIHub (initiates AI Hub processing)

Low-Level Execution: sendFirstGroupToAIHub

Purpose: Sends the first execution group of steps to AI Hub after setup is complete.

Location: packages/app/src/services/processing/base.ts

Responsibilities:

1. Retrieves job steps from database
2. Prepares the first execution group
3. Resolves prompts with entity data
4. Sends initial request to AI Hub
5. Updates job status to IN_PROGRESS

Requirements:

• Job must already exist in database
• All entity data must be provided
• Storage path and file metadata required

Usage Example:

// Called by service-specific processing functions
const success = await sendFirstGroupToAIHub(
  PROCESSABLE_TYPE.TRANSACTIONS,
  transactionId,
  transactionEntity, // Full entity with relations
  TASK_TYPE.TRANSACTION_PROCESS,
  storagePath,
  mimeType,
  jobId, // Must already exist
  userId,
  projectId,
  sendToAIHub // Callback to AI Hub
);

Key Differences

Aspect	createJobAndStartProcessing	sendFirstGroupToAIHub
Purpose	Orchestration & Setup	AI Hub Communication
Job Creation	Creates new job	Expects existing job
Entity Data	Fetches from database	Must be provided
Type Routing	Routes by entity type	Generic for all types
Abstraction Level	High (business logic)	Low (technical execution)
Entry Point	External APIs/Actions	Internal service calls
Error Handling	User-friendly messages	Technical logging

1. Job Creation

// Example: Create processing job for a transaction
const job = await createProcessingJob(
  PROCESSABLE_TYPE.TRANSACTIONS,
  transactionId,
  'Transaction Import Processing',
  { fileUrl, mimeType }
);

2. Template Structure

Templates define the processing workflow:

-- Example: Transaction Import Processing Template
Template
├── OCR Step (Group 1, Order 1)
├── Classification Step (Group 2, Order 1, Depends on: OCR)
├── Data Extraction Step (Group 3, Order 1, Depends on: Classification)
├── Budget Linking Step (Group 4, Order 1, Depends on: Data Extraction, Optional)
├── Vendor Matching Step (Group 4, Order 2, Depends on: Data Extraction, Optional)
└── Finish Processing Step (Final Group, MANDATORY for all templates)

Mandatory Finish Processing Step

Important: Every processing template MUST include a FinishProcessing step as the final step. This is a base step type that:

• Applies to all task types (transactions, scripts, scenes, schedules, etc.)
• Runs after all other steps complete successfully
• Triggers final status updates in the application
• Does NOT require AI processing - it's a system step
• Enables task-specific completion logic through processor methods

Example configuration:

-- In any template's step configuration
INSERT INTO process_template_steps (
  process_template_id,
  step_key,
  display_name,
  execution_group,
  depends_on_steps,
  ai_hub_step_type,
  processor_method
) VALUES (
  template_id,
  'finish_processing',
  'Finish Processing',
  5, -- Last execution group
  ARRAY['all_previous_steps'], -- Depends on all prior steps
  'FinishProcessing',
  'finishTransactionProcessing' -- Task-specific finish method
);

Registering a NEW template (dispatch does not follow from the DB rows)

Inserting a process_templates row and its steps is necessary but NOT sufficient — jobs for an unregistered template are created and then sit at ready forever, or the start-job route rejects them outright. Every new template needs three app-side registrations:

1. createJobAndStartProcessing (services/processing/server.ts): add the template to the isReprocessing allowlist (otherwise the route demands document file data and 500s with "File data required for transaction processing") AND give it a dispatch branch. A template whose FIRST group is manual — or that mixes manual and AI groups — dispatches via advanceToNextExecutionGroup, the generic claim → prepare → dispatch driver; the single-purpose dispatchers (AI-payload senders, executeManualSteps) each handle only one shape.
2. TransactionProcessingTemplate enum (constants/transaction.ts): the member's value must match process_templates.name byte-for-byte — dispatch resolves the template by name.
3. TransactionReprocessingIndicator's template set if the transaction detail view should show an in-progress banner and drive the active-job lockout/invalidation for this template's runs.

Restart contract for manual-first templates: restartProcessingJob only resets the job and claims the new generation — it must never pre-resolve the next group's AI payload. getNextExecutionGroupSteps exists to build the AI-Hub dispatch payload, so it filters to AI steps and legitimately returns an empty set when the next group is manual; treating that empty set as "no steps to execute" (the historical guard) failed every manual-first template after the reset had already committed, stranding the job with a ready step nothing would drive. The caller (restartTransactionProcessing) re-drives group 1 through processNextExecutionGroup, which executes manual groups inline and hands AI groups to the dispatch path.

Transactions Processing template (execution groups)

The main transaction pipeline is the reference instance of the structure above. Its 16 steps run across 12 execution groups; steps sharing a group run in parallel, and every group is homogeneous (ai or manual, never both — enforced by a trigger, see database/AI_PROCESSING.md).

Group	Step key(s)	Type	Purpose
1	transaction_document_parsing	ai	Parse the source document
2	transaction_ocr_extraction	ai	Extract header fields and line items
3	transaction_currency_rate_check	manual	Gate foreign-currency transactions on a project rate (On Hold when absent)
4	transaction_check_totals	manual	Reconcile header totals against line evidence
5	transaction_entity_matching	ai	Match / create the supplier, customer or person
6	transaction_type_discovery	ai	Settle the transaction's type and direction
7	transaction_line_item_summarisation	manual	Apply the project's per-type line-item processing mode
8	transaction_justification_analysis, transaction_budget_item_matching	ai	Justification analysis; match line items to budget items
9	transaction_chart_of_accounts_matching, transaction_budget_daily_allocation	ai	Match line items to COA entries; allocate them across production days
10	transaction_tax_validation, transaction_company_validation, transaction_duplicate_check	manual	External validations and duplicate detection
11	transaction_payment_creation	manual	Create the payment record
12	transaction_finish_processing	manual	Mandatory finish step: final status, submit for approval

Group 7 sits where it does for a reason: the mode is configured per transaction type, so it cannot be resolved before transaction_type_discovery (group 6) has written the type; and it rewrites the item set every later matching step consumes, so it must precede them. transaction_budget_item_matching, transaction_justification_analysis and transaction_chart_of_accounts_matching each declare it in depends_on_steps — they must never match against a line-item set whose mode has not yet settled.

Line-item processing mode

A project chooses, per transaction type, whether the pipeline keeps every extracted document line or condenses them: projects.metadata.transaction.<key>.line_item_processing is 'detailed' or 'summary', and an absent map, key or value all read as detailed (resolveLineItemProcessingMode in utils/transaction.ts; the settings form always writes all six keys). The applied mode is stamped on transactions.line_item_mode — a line_item_processing_mode enum column where NULL means the transaction was processed before the mode system existed and is semantically detailed.

processLineItemSummarisation (services/transaction/base.ts) reads the mode at step time against a fresh transaction read — never the processing context cache, which may predate the type-discovery write the mode is keyed on.

• Detailed is a no-op on the items: the rows OCR already persisted are the detailed set, so the step only stamps the mode. Routing this through the replacement RPC would delete and re-insert every item for nothing, orphaning their matches.
• Summary re-derives the detailed lines from this job's transaction_ocr_extraction result_data — never from the mutable transaction_items rows, which carry downstream edits that must not fold back into a fresh derivation. Re-deriving from the immutable source is what makes a retry deterministic and idempotent.
• A document that produced no OCR lines is not a failure: there is nothing to consolidate, so the mode is stamped and the step passes.

Summarisation semantics (summariseTransactionLineItems, utils/transaction.ts) — pure, deterministic, no AI and no database. It produces the minimum number of items that preserve every distinct tax treatment on the document:

1. Lines carrying a real tax_rate_id merge into one item per distinct rate, so each rate keeps its own subtotal/tax pair.
2. Lines with no rate and no tax merge into a single collective no-tax item (tax_rate_id: null).
3. Lines with no rate but a non-zero tax_total pass through unmerged, one item each. Merging them would mean inventing or averaging a rate, and a tax figure with no identified rate is exactly where that guess is wrong. A summary item never carries a fabricated or averaged rate.

Output order is deterministic (rate groups in first-appearance order, then the no-tax group, then pass-through lines in input order) with line_number reassigned from 1. Merged items are pure sums: quantity collapses to 1 at a unit_price equal to the group's net subtotal, discount fields clear (the summed subtotals are already net of source-line discounts), and match-assigned fields (budget item, payroll classification, account and tracking references) clear — the consolidated item is a new subject for the matching steps that follow. The result is pinned to the transaction's unchanged document totals via reconcileLineItemsToTransactionTotals.

The swap itself goes through replace_transaction_line_items, a service-role-only INVOKER RPC that deletes the old items, drops any pipeline-created daily allocation the delete left with zero links, inserts the new set and stamps the mode — all in one transaction under the transaction's row lock. PostgREST has no cross-request transactions, so a service-layer delete-then-insert would leave the transaction with zero line items on any mid-flight failure. Contract: database/VIEWS_AND_FUNCTIONS.md.

Audit trail. The step writes its own record into its result_data before completing — prior and applied mode, whether the items were rewritten, the source job and OCR step ids, the produced items, the merge groups (each group's tax_rate_id, absorbed source line numbers, and whether it was merged or passed through) and the reconciliation residual. The regeneration step adds superseded_issues (the keys it dismissed, the ids actually ignored, the count, and any failure). The write is a guarded no-op IN_PROGRESS → IN_PROGRESS transition, so it stays fenced on the job's generation while the framework's later completion write COALESCE-preserves it. A refused audit write is benign and never fails the step: the diary entry losing a race must not turn a successful rewrite into a pipeline failure. The original OCR lines are never destroyed — they remain in the OCR step's own result_data, which is why a rebuild is always possible.

Regenerating line items. Changing a project's mode does not retroactively rewrite existing transactions. The targeted Transaction Regenerate Line Items template exists for that: transaction_line_item_regeneration → check totals → budget item matching → chart-of-accounts matching and daily allocation → finish processing. processLineItemRegeneration sources the latest completed transaction_ocr_extraction across all of the transaction's jobs (getLatestCompletedStepForProcessable, services/processing/base.ts — this job has no OCR step of its own), rebuilds the detailed lines, applies the currently configured mode and always calls the replacement RPC. Always rewriting is the point: in detailed mode the rebuild is exactly what discards an earlier summarisation. No usable source OCR extraction is a genuine failure here, not a benign no-op — reporting success for an unchanged item set would be a lie.

Once the replacement commits, the step also auto-ignores the pending issues the replacement invalidated — the PENDING processing_issues on the transaction whose issue_key is one the re-run steps raise: transaction_totals_mismatch (check totals), transaction_unmatched_budget_items (budget item matching), transaction_unmatched_chart_of_accounts (chart-of-accounts matching), and transaction_unmatched_budget_daily_allocation_dates / transaction_no_budget_daily_allocation_dates (daily allocation). Leaving them pending breaks two ways: the dedup in createProcessingIssues collapses a fresh raise into the surviving stale row, so the issue a user sees carries context data about line items that no longer exist; and an issue the rebuild genuinely fixed dangles forever because nothing revisits it. Dismissing is safe precisely because each re-run step raises a fresh issue if the problem persists against the new items. transaction_justification_needed is deliberately NOT in the set — the template does not run processJustificationAnalysis, so that verdict still stands. This is post-replacement hygiene, not a required side effect: the items are already committed, so a failed dismissal is captured in the step's superseded_issues audit payload (ignored ids, count, error) and never fails the step.

Dispatch is a permission-gated server action (derive-then-assert: the transaction's creator, or transaction:edit), offered from the transaction view when the mode applied to the record differs from the one currently configured for its type.

3. Step Execution

Execution Group-Based Processing

The system uses an execution group-based callback mechanism to ensure dependent steps always receive fresh entity data:

1. Group-by-Group Execution: Steps are processed in execution groups sequentially
2. Fresh Entity Data: After each group completes, a callback is sent to AI Hub with:
3. Updated entity data reflecting all completed steps' changes
4. Only the steps for the next execution group
5. Dependency Resolution: This ensures steps that depend on earlier steps see their database updates

Processing Flow

1. Initial Send: First execution group steps are sent to AI Hub with initial entity data
2. Group Completion: When all steps in a group complete (via webhook callbacks):
3. System fetches fresh entity data from database
4. Identifies next execution group's ready steps
5. Sends callback to AI Hub with updated entity and next steps
6. Parallel Execution: Steps within same group execute concurrently
7. Result Storage: Each step stores results for use by subsequent steps
8. Status Updates: Real-time progress tracking via webhooks
9. Finish Processing: Mandatory final step runs to complete the job
10. Completion Check: Job marked complete when finish step succeeds

Example Flow

Execution Group 1: OCR Extraction
  ↓ (completes, updates database)
Callback with fresh entity → AI Hub
  ↓
Execution Group 2: Classification (sees OCR results)
  ↓ (completes, updates database)
Callback with fresh entity → AI Hub
  ↓
Execution Group 3: Data Extraction (sees classification)
  ↓ (completes, updates database)
Callback with fresh entity → AI Hub
  ↓
Execution Group 4: Budget & Vendor Matching (parallel, both see extraction results)

4. Status Flow

PENDING → IN_PROGRESS → COMPLETED
                    ↘
                      → FAILED
                      → PARTIAL_SUCCESS

Database Schema

Core Tables

1. process_templates: Processing workflow definitions
2. service_name: Optional service name for processor routing
3. process_template_steps: Individual step definitions
4. process_template_id: Reference to template
5. processor_method: Optional method name for dynamic processor invocation
6. on_failure_action: Defines behavior when step fails (fail_job, skip_dependents, continue)
7. processing_jobs: Job instances
8. process_template_id: Reference to template
9. processing_job_steps: Step execution tracking
10. processing_job_id: Reference to job
11. process_template_step_id: Reference to template step
12. processing_issues: Issues requiring human resolution
13. processable_type: Type of entity with issue
14. processable_id: ID of entity with issue
15. processing_job_id: Optional reference to job
16. processing_job_step_id: Optional reference to step
17. issue_template_id: NOT NULL FK to the issue registry — the sole carrier of the issue key and its default copy/severity/flags
18. processing_issue_templates: The issue registry — every kind the pipeline can raise
19. project_issue_settings: Sparse per-project applicability overrides for configurable issue kinds
20. process_template_issue_actions: What a template does to a kind's pending issues at dispatch
21. ai_prompts: Reusable AI prompts with variable substitution

Key Relationships

process_templates
    ↓ (1:N)
process_template_steps → ai_prompts (optional)
    ↓ (1:N via job creation)
processing_jobs
    ↓ (1:N)
processing_job_steps
    ↓ (1:N)
processing_issues (optional, for manual validation)
    ↓ (N:1)
processing_issue_templates (the issue registry)

Issue raising: one chokepoint, DB-driven applicability

Every raise in the pipeline goes through createProcessingIssues (processing/base.ts) — no step writes processing_issues directly. It stamps the issue_template_id FK from the caller's registry key, applies applicability, and dedupes against existing PENDING rows for the same (record, template) pair. An unregistered key, or one registered for a different processable_type, throws: a new issue kind requires a processing_issue_templates seed row in a migration.

Applicability is data, not code. The template's applies_to_transaction_types (NULL = every type) is the default; for configurable templates a project's sparse project_issue_settings row for the exact (transaction_type, direction) pair overrides it. The chokepoint delegates the decision to the processable type's registered filterIssuesByApplicability processor — for transactions, filterTransactionIssuesByApplicability in transaction/server.ts — so applicability lives in exactly one place and the processing service never imports the owning service's base. A type with no registered filter raises everything.

Type resolution inside the filter is deliberately narrow: a processor that has just written the authoritative type passes it in and that wins; otherwise the transaction is fresh-read, never taken from the processing context cache, which can predate the type write the decision depends on. Direction always comes from the fresh read.

Per-template issue actions at dispatch

Before the first execution group runs, createJobAndStartProcessing calls applyTemplateIssueActions to apply the template's rows in process_template_issue_actions to the record's PENDING issues: ignore closes them out (status ignored, with a resolution_steps audit entry), supersede does that and clears is_current so they also leave the default list surfaces. This exists because the steps the template is about to re-run invalidate the verdicts they previously produced — leaving them pending would let the chokepoint's dedup collapse the fresh raise into the stale row.

It is hygiene, never a gate: per-issue failures are collected rather than thrown, and a failure of the hook itself is logged while the dispatch proceeds. When anything was applied, the outcome list ({issue_template_id, issue_key, action, ignored_issue_ids, failed_issue_ids}) is merged into processing_jobs.metadata under applied_issue_actions, readable via getJobIssueActionOutcomes(jobId). Full detail: ISSUE_RESOLUTION_SYSTEM.md.

Service Architecture

Client-Side Services

// packages/app/src/services/processing/client.ts

// Get job status with polling
const status = await getJobStatusWithPolling(jobId);

// Check if entity has active job
const hasActive = await hasActiveProcessingJob(
  PROCESSABLE_TYPE.TRANSACTIONS,
  transactionId
);

Server-Side Services

// packages/app/src/services/processing/server.ts

// High-level orchestration - creates job and starts processing
const result = await createJobAndStartProcessing(
  processableType,
  processableId,
  userId,
  title,
  metadata,
  fileType
);

// Low-level - send first group to AI Hub (job must exist)
const success = await sendFirstGroupToAIHub(
  processableType,
  processableId,
  entity,
  taskType,
  storagePath,
  mimeType,
  jobId,
  userId,
  projectId,
  sendToAIHub
);

// Update step status from webhook
await updateStepStatus(jobId, stepKey, status, resultData, errorMessage);

Transaction Service Integration

// packages/app/src/services/transaction/base.ts

// Start transaction processing
const success = await startTransactionProcessing(
  transactionId,
  fileDataUrl,
  mimeType,
  jobId,
  sendToAIHub
);

Status-bar read path (app layer)

The processing status bar and its details sheet read jobs through exactly two bounded reads. Neither ever fetches a whole project's job window.

• getInitialProcessingJobsWithLimits (services/processing/base.ts) is one get_processing_job_status_summary RPC call. It returns page one of all five status buckets plus exact per-bucket counts, as ProcessingJobStatusSummary (types/processing.ts). Each row is a ProcessingJobSummaryRow: identity, status, the two ordering timestamps, the template's display strings, and completed_ui_steps / total_ui_steps — the pre-aggregated progress a collapsed row renders. No step rows, no template rows, no request_data / result_data.
• The collapsed bar's five pills come from counts, never from row lengths. A bucket's rows are page one; its count is the whole bucket.
• Bucket names ARE the database overall_status values (ProcessingJobStatusBucket) — failed and partial_success are separate buckets with separate counts, so nothing merges them client-side.
• getMoreProcessingJobsByStatus loads ONE bucket's next page, keyset-paginated. The cursor pair must match the bucket's ordering — completed_at for the completed bucket, created_at for the rest — and the RPC rejects a mismatched pair rather than paging from the wrong column, so the mapping lives in exactly one place: buildProcessingJobBucketCursor (utils/processing.ts).
• getProcessingJobStepDetails(jobId) is the lazy per-job step read behind useProcessingJobStepsQuery, enabled ONLY while a card is expanded. A sheet of collapsed cards issues no step reads at all.

Load-more pages are generation-scoped. Accumulated pages live in ProcessingStatusBar state keyed to the summary query's dataUpdatedAt. Any refetch (poll tick, realtime invalidation, window refocus) re-derives every bucket's head from the server, so the accumulated pages are discarded wholesale and the reveal cursors in ProcessingDetailsView reset. This is not an optimisation — a cursor built from a row that a refetch removed, or a job that moved between buckets, would otherwise strand or duplicate a page. The next cursor is always the LAST row the bucket currently holds, so it can never point at a row the current generation dropped.

RLS is the boundary, not the client. The processing_jobs SELECT policy carries a record-visibility conjunct, so a job whose processable the caller cannot see is absent from both the page and the count. The former client-side compensation (dropJobsWithoutRelatedItem) has been removed.

Freshness policy: polling + authoritative refresh

Realtime is the primary freshness mechanism; polling is the backstop that covers what realtime structurally cannot. Three cadences, decided per tick rather than fixed at mount (React Query re-evaluates a function-valued refetchInterval whenever the query settles or its options change, and hands the callback the LIVE query object — so both counts and visibility are read at decision time):

State	Interval
Document hidden	no polling (false)
Visible, live work in flight	PROCESSING_STATUS_BAR_REFETCH_INTERVAL_MS (30s)
Visible, no live work	PROCESSING_STATUS_BAR_IDLE_REFETCH_INTERVAL_MS (5min)

"Live work" is read from the SERVER counts — pending + in_progress > 0 via hasActiveProcessingWork (utils/processing.ts) — never from loaded row lengths, which only ever describe page one. The predicate fails toward freshness: absent, null or non-numeric counts (no data yet, an errored query) count as ACTIVE, so an unknown state polls fast instead of latching into the idle cadence and going quiet. Returning false while hidden stops the timer outright; React Query's own refetchIntervalInBackground: false would keep firing it and merely skip the fetch.

Authoritative refresh. Three triggers mean "the client may have missed changes" and each forces one invalidation of the processing family:

1. A transition to visible (visibilitychange).
2. A window focus.
3. A realtime re-subscribe (onResubscribed) — events during the outage were never delivered and are not replayed.

The app's global QueryClient sets refetchOnWindowFocus: false, and query-core's focus manager observes only visibilitychange, so React Query contributes nothing on either transition — both listeners are the bar's own. Browsers commonly fire focus and visibilitychange together on a tab switch, so all three triggers funnel through one short coalescing window and produce exactly ONE refetch. That invalidation is also what re-arms the cadence: the interval callback is only re-evaluated when the query settles or its options change, so a visibility transition alone would otherwise leave the stopped timer stopped.

Accepted staleness window. A change that alters only a record's visibility — a sensitivity rule applied or lifted, an access grant revoked — writes nothing to processing_jobs and so emits no realtime event at all (RLS is evaluated per read). Such a record does not leave an open page until the next poll tick or the idle safety refresh, whichever comes first: at most 30s while work is in flight, and at most 5 minutes when idle. This is deliberate; closing it would require polling an otherwise-idle page at the active cadence.

Execution Group-Based Callback Architecture

Overview

The system implements an execution group-based callback mechanism that ensures each processing step works with the most current database state. This architecture provides data consistency across dependent steps while maintaining parallel processing efficiency.

Core Design Principles

1. Execution Groups: Steps are organized into sequential execution groups based on dependencies
2. Group-by-Group Processing: Each execution group is processed sequentially
3. Fresh Data Callbacks: After each group completes, fresh entity data is fetched from the database and sent with the next group
4. Parallel Within Groups: Steps within the same execution group run concurrently for optimal performance

Implementation Details

// When execution group completes
async function onExecutionGroupComplete(jobId: string) {
  // 1. Check if all steps in current group are done
  const currentGroupComplete = await checkGroupCompletion(jobId);

  if (currentGroupComplete) {
    // 2. Fetch fresh entity data (includes all DB changes)
    const freshEntity = await getEntityByJobId(jobId);

    // 3. Get next execution group steps
    const nextSteps = await getNextExecutionGroupSteps(jobId);

    if (nextSteps.length > 0) {
      // 4. Send callback to AI Hub with fresh data
      await sendToAIHub({
        entity: freshEntity, // Fresh data with all updates
        metadata: { steps: nextSteps },
      });
    }
  }
}

Key Benefits

1. Data Consistency: Dependent steps always work with the latest database state
2. Simplified Architecture: Clean separation between execution groups
3. Optimized Communication: Only relevant steps sent per callback to AI Hub
4. Parallel Performance: Steps within the same group execute concurrently
5. Clear Dependencies: Execution groups make step dependencies explicit and manageable

Dynamic Processor System

The dynamic processor system automatically routes AI processing results to appropriate service methods, eliminating the need for switch statements and making the system easily extensible.

Architecture

// Processor Registry
const processorRegistry: ProcessorRegistry = {
  transaction: {
    processDocumentParsing,
    processOcrExtraction,
    processProjectRelationshipMatching,
    processBudgetItemMatching,
    processJustificationAnalysis,
  },
  // Add more services and processors as needed
};

How It Works

1. Template Configuration: Each template step can specify:
2. processor_method: The method to call for processing results

3. Template service_name: The service containing the processors

4. Automatic Routing: When a step completes, the system:

// In processStepResult function
const processor = getProcessor(
  template.service_name,
  step.template_step.processor_method
);
if (processor) {
  const success = await processor(step);
}

1. Processor Implementation: Each processor follows a standard interface:

type ProcessorFunction = (
  step: ProcessingJobStepWithRelation
) => Promise<boolean>;

Adding New Processors

1. Create Processor Function:

export async function processNewStepType(
  step: ProcessingJobStepWithRelation
): Promise<boolean> {
  const { result_data, processing_job_id } = step;
  if (!result_data) return true;

  // Process the results
  // Update relevant tables

  return true;
}

1. Register in Service:

// In transactionProcessors.ts
export const transactionProcessors = {
  processDocumentParsing,
  processOcrExtraction,
  processNewStepType, // Add new processor
};

1. Configure in Database:

-- In template step configuration
UPDATE process_template_steps
SET processor_method = 'processNewStepType'
WHERE step_key = 'new_step_type';

Benefits

• No Code Changes: Add new processors without modifying core logic
• Type Safety: Full TypeScript support for processor functions
• Testability: Each processor is independently testable
• Maintainability: Clear separation of concerns
• Extensibility: Easy to add new entity types and processors

Example: Transaction Processing with Enhanced Address Matching

// Transaction processors with AI-enhanced operations
export const transactionProcessors = {
  // Document parsing extracts type and metadata
  processDocumentParsing: async (step) => {
    // Updates: document_type, handwritten_notes, etc.
  },

  // OCR extraction with address creation
  processOcrExtraction: async (step) => {
    const { result_data } = step;

    // Parse AI response from message field
    const aiResponse = JSON.parse(result_data.message);

    // Create transaction using base layer
    const transaction = await createTransactionBase(aiResponse);

    // Enhanced address creation with AI matching (server layer)
    if (aiResponse.address) {
      const address = await createAddressWithAIMatching({
        ...aiResponse.address,
        transactionId: transaction.id,
      });

      // Update transaction with matched/created address
      await updateTransaction(transaction.id, {
        address_id: address.id,
      });
    }

    // Return full transaction for context
    return transaction;
  },

  // Project matching links to projects
  processProjectRelationshipMatching: async (step) => {
    // Updates: project_relationship_id, type, etc.
  },

  // Finish processing - MANDATORY final step
  finishTransactionProcessing: async (step) => {
    // Updates transaction status to 'Ready'
    // Performs any final cleanup or notifications
    // This runs after ALL other steps complete successfully
  },

  // Budget linking for line items
  processBudgetItemMatching: async (step) => {
    // Updates: line item budget associations
  },

  // Justification analysis
  processJustificationAnalysis: async (step) => {
    // Updates: ai_justification field
  },
};

Enhanced Address Matching with AI

The system implements intelligent address deduplication using AI-powered semantic matching to prevent duplicate addresses and improve data quality.

Architecture

Address creation follows a multi-layer approach:

1. Transaction Service (server): Initiates address creation from OCR results
2. Address Service (server): Handles AI matching and geocoding
3. Address Service (base): Performs pure CRUD operations

AI Matching Process

// packages/app/src/services/address/server.ts (example integration)
export async function createAddressWithAIMatching(
  addressData: AddressCreateData
): Promise<ServiceResponse<Address>> {
  // 1. Fetch existing addresses for comparison
  const existingAddresses = await findExistingAddresses();

  // 2. Use AI to find semantic matches
  const prompt = await resolvePromptWithVariables({
    promptIdentifier: { name: 'AddressMatchingPrompt' },
    variables: {
      entity: { newAddress: addressData, existingAddresses },
      entityId: addressData.transactionId,
      entityType: 'address',
    },
  });

  const aiMatch = await aiHubClient.findMatch(prompt);

  // 3. Return existing if match found
  if (aiMatch?.matchId) {
    return findAddressById(aiMatch.matchId);
  }

  // 4. Geocode new address
  const geocoded = await geocodeAddress(addressData);

  // 5. Create new address
  return createAddressBase({ ...addressData, ...geocoded });
}

Benefits

• Deduplication: Prevents duplicate addresses with different formatting
• Data Quality: Maintains clean, consistent address database
• Smart Matching: Uses semantic understanding, not just string comparison
• Geocoding Integration: Enriches addresses with geographic data

Integration Points

1. WhatsApp Bot (Sana)

• Creates processing jobs when documents are uploaded
• Provides status updates during processing
• Displays results when complete

2. Web Application

• Shows processing status in transaction/budget/schedule views
• Allows manual retry of failed jobs (automatic recovery of a job that stalled without failing is handled by the stall sweeper, not the UI)
• Displays extracted data for review

3. AI Hub Communication

• Initial Request: App → SQS → AI Hub (first execution group only)
• Group Completion Callbacks: App → AI Hub (subsequent groups with fresh entity data)
• Status Updates: AI Hub → Webhook → App (for each step completion)
• File Transfer: Via S3 presigned URLs
• Prompt Resolution: Dynamic variable substitution in prompts using @delta/common

Callback Mechanism

The callback system ensures data consistency by sending fresh entity data with each execution group:

Callback Structure:

{
  type: 'TRANSACTION_PROCESS',
  entityId: transactionId,
  entity: freshEntityData, // Fetched after group completion
  metadata: {
    steps: nextGroupSteps,  // Only next group's steps
    jobId: jobId,
    executionGroup: nextGroup
  }
}

Key Features:

• Fresh entity data fetched from database after each group completes
• Only the next execution group's steps are sent
• Maintains execution group parallelism
• Ensures data consistency across the processing pipeline

Prompt Resolution System

The system provides flexible prompt resolution with dynamic variable substitution, supporting both prompt IDs and names for maximum flexibility.

Prompt Identifier Pattern

Prompts can be resolved using either:

• Prompt ID: Direct UUID reference for performance
• Prompt Name: Human-readable identifier for maintainability

// Resolution by ID
const prompt = await resolvePromptWithVariables({
  promptIdentifier: { id: 'uuid-123' },
  variables: { entity, entityId, entityType },
});

// Resolution by name
const prompt = await resolvePromptWithVariables({
  promptIdentifier: { name: 'TransactionOCRExtraction' },
  variables: { entity, entityId, entityType },
});

Server-Side Prompt Resolution

The server layer provides a wrapper for secure prompt resolution:

// packages/app/src/services/aiPrompt/server.ts (example)
export async function resolvePromptWithVariables(
  promptIdentifier: { name?: string; id?: string },
  variables: PromptVariables
): Promise<ServiceResponse<string>> {
  // Fetches prompt from database
  // Resolves variables using template engine
  // Returns processed prompt text
}

Variable Structure

Variables follow the PromptVariables type:

interface PromptVariables {
  entity: Record<string, any>; // The main entity data
  entityId: string; // Entity identifier
  entityType: string; // Entity type (transaction, budget, etc.)
}

Template Variable Substitution

The system uses the resolveTemplate function from @delta/common for variable substitution. This allows prompts to access entity data and format it appropriately for AI processing.

Variable Syntax

Single Field Access

Access simple fields using dot notation:

{!entity.field}
{!entity.nested.field}

Example:

Template: "Process transaction {!entity.transaction_code} from {!entity.submission_address}"
Entity: { transaction_code: "TXN-001", submission_address: "john@example.com" }
Result: "Process transaction TXN-001 from john@example.com"

Array Formatting

Extract and format specific fields from arrays:

{!Array:path.to.array|field1,field2,field3}

Example:

Template: "Tax rates: {!Array:entity.project.tax_scheme.tax_rates|id,label,description,percentage}"
Entity: {
  project: {
    tax_scheme: {
      tax_rates: [
        { id: "uuid-123", label: "Standard", description: "Standard VAT", percentage: 20 },
        { id: "uuid-456", label: "Reduced", description: "Reduced VAT", percentage: 5 }
      ]
    }
  }
}
Result: "Tax rates: id: uuid-123, label: Standard, description: Standard VAT, percentage: 20, id: uuid-456, label: Reduced, description: Reduced VAT, percentage: 5"

Object Field Extraction

Extract specific fields from an object:

{!Object:path.to.object|field1,field2}

Example:

Template: "Project details: {!Object:entity.project|id,title,status}"
Entity: {
  project: {
    id: "proj-001",
    title: "Summer Campaign",
    status: "active",
    created_at: "2024-01-01",
    budget: 50000
  }
}
Result: "Project details: id: proj-001, title: Summer Campaign, status: active"

Real-World Usage in Prompts

Document Parsing Prompt

-- In ai_prompts table
'Analyze this document for the following:
1. Document type classification
2. Tax eligibility based on: {!entity.project.tax_scheme.rules}
3. Submit any found codes from the document'

OCR Extraction Prompt

-- In ai_prompts table
'Extract transaction details from this receipt.
Valid tax rates for line items: {!Array:entity.project.tax_scheme.tax_rates|id,label,description,percentage}
Please use the tax rate ID from the list above for each line item.'

Project Matching Prompt

-- In ai_prompts table
'Match this transaction to the appropriate project.
Available projects: {!Array:entity.available_projects|id,project_code,title}
Current project context: {!Object:entity.project|id,title,type}'

Implementation Details

1. Variable Resolution: Happens in packages/app/src/services/processing/base.ts during prompt preparation
2. Entity Context: The entity object contains the full entity with all relations (e.g., TransactionWithRelations)
3. Null Handling: Missing or null values resolve to empty strings
4. Type Safety: All values are converted to strings for prompt injection
5. Generic Design: Works with any entity type and field structure

Status Tracking

Job Status

• PENDING: Job created, waiting to start
• IN_PROGRESS: At least one step is running
• COMPLETED: All steps finished successfully
• FAILED: All steps failed or critical step failed
• PARTIAL_SUCCESS: Some optional steps failed

Step Status

• PENDING: Step created, dependencies not met
• READY: Dependencies satisfied, ready to execute
• IN_PROGRESS: Currently executing
• COMPLETED: Finished successfully
• FAILED: Execution failed
• SKIPPED: Condition not met or dependency failed

Progress Calculation

-- Database function: calculate_job_progress
Progress = (Completed Steps / Total Steps) * 100

Error Handling

Step-Level Errors

• Error message stored in error_message field
• Optional steps don't block job completion
• Failed dependencies cause dependent steps to be skipped
• Steps can define failure action behavior via on_failure_action field

Failure Action Types

The on_failure_action field in ai_process_template_steps determines how the system responds when a step fails:

1. fail_job (Default)
2. The entire job is marked as FAILED
3. All dependent steps are automatically skipped
4. Use for critical steps where failure means the entire process cannot continue

5. Example: OCR extraction failing means no data to process

6. skip_dependents

7. The job continues running
8. Only steps that directly depend on this step are skipped
9. Other parallel or independent steps continue normally
10. Use for optional enrichment steps that have specific dependents

11. Example: Budget matching fails, so budget validation is skipped, but vendor matching continues

12. continue

13. The job continues as if the step succeeded
14. Dependent steps still run normally
15. The step is marked as FAILED but doesn't impact flow
16. Use for completely optional steps that don't affect downstream processing
17. Example: Sending a notification fails but shouldn't stop processing

Retry Configuration

The on_retry_action field in ai_process_template_steps allows fine-grained control over retry behavior:

Model Switching

Configure automatic model switching on retries to handle model-specific failures:

{
  "switch_models": ["claude-3-opus", "gpt-4-turbo"]
}

Behavior: Models cycle through the array plus the original model. With 5 max retries:

• Initial attempt: Original model from model_config
• Retry 1: claude-3-opus
• Retry 2: gpt-4-turbo
• Retry 3: Original model
• Retry 4: claude-3-opus
• Retry 5: gpt-4-turbo

Max Tokens Adjustment

Increase max output tokens on each retry to handle truncated JSON responses:

{
  "raise_max_tokens_by": 2000
}

Behavior: Max tokens increases by the specified amount per retry (no cap):

• Initial: 8000 (from config)
• Retry 1: 10000
• Retry 2: 12000
• Retry 3: 14000
• Retry 4: 16000

Combined Configuration

Both strategies can be used together:

{
  "switch_models": ["claude-3-opus"],
  "raise_max_tokens_by": 2000
}

This provides flexibility for handling different failure scenarios.

Job-Level Errors

• Job marked as FAILED if any step with fail_job action fails
• PARTIAL_SUCCESS if some optional steps fail but critical steps succeed
• Retry capability at job or step level with configurable retry actions
• A job whose AI dispatch is exhausted is given a job-level terminal status by markJobFailedAfterRetries, not just failed steps — a job left in_progress with no failing step is invisible to the user and to every status query

Callback Resilience

• Step writes are idempotent: a callback for an already-completed step skips the write but still re-enters the continuation, because a previous callback that completed the step and then failed to advance the job is exactly the case a retry must heal
• Stale callbacks from a superseded run are discarded by restart_count comparison, and every step write re-asserts that generation inside the database (see the guarded CAS below)
• Job and step existence are validated before any service-role write, and an unreadable job is never mistaken for an absent one

A lost callback is not retried by anything in the callback path itself. AI Hub retries only while its own process is alive and only for classified-retryable failures; if that process dies, or the retries are exhausted, nothing else in this path will ever drive the job again. That gap is what the stall sweeper below exists to close.

Stall detection and recovery

A processing job stalls when the actor driving it dies between steps: nothing is in flight, no callback is coming, and no status will ever change again. The job sits in in_progress with error_message NULL, which reads to the user as "still working" forever. The pieces below make that state impossible to reach silently.

Truthful reads: null means verified absence

Every reader in packages/app/src/services/processing/base.ts throws on a genuine database failure and returns null / [] only for a confirmed-absent row. A catch { return null } in a reader collapses "the database was unreachable" into "there is no such job", and every caller downstream then acts on a false absence — abandoning a legitimate result, or declaring a live job dead.

The wrapper layer preserves that distinction: getJobWithSteps and friends are wrapService consts in packages/app/src/services/processing/server.ts, so a thrown read arrives as { data: null, error } while a confirmed absence arrives as { data: null, error: null }. Callers branch on .error before interpreting .data.

isProcessingHaltedForJob is the sharpest case: a failed read returns an errored envelope, never false. Advancing a pipeline whose halt state is merely unknown can resurrect a cancelled or On Hold record; declining to advance is always recoverable by a retry.

Fenced execution-group claims

processing_jobs carries three claim columns (documented in database/AI_PROCESSING.md):

• current_processing_group — a monotonic high-water mark. It is never rewound, so a stale callback can never re-drive an already-advanced group.
• current_processing_group_claim_id — an opaque owner token minted per claim attempt.
• current_processing_group_claimed_at — when the claim was taken.

claimExecutionGroup grants a claim in exactly three states: no group claimed yet, the claimed group is below the target, or the claimed group is the target but carries no owner token (a claim released after a failed dispatch, so the group is free to be re-driven). A target group that is currently owned is never stolen.

releaseExecutionGroupClaim clears only the token, never the high-water mark, and is fenced on the claimId: presenting a stale id matches zero rows and is a no-op, so a slow worker waking up after its claim was superseded cannot unlock the group its successor now owns.

advanceToNextExecutionGroup is the single owner of claim → prepare → dispatch → release-on-failure. Splitting those across caller and callee is what allowed a claimed group to be left owned by a dispatch that never happened: the claim blocked every later attempt while nothing was in flight, and the job stalled with no signal. Any failure after a successful claim — a thrown error, an errored envelope, a handler reporting failure, or a missing continuation handler for a job that needs one — releases the claim before the failure propagates. A lost claim race is not a failure: another callback is already advancing the job, and this one returns benignly.

Staged step completion + generation asserts

A completed callback is written in three stages, all through the update_processing_step_guarded RPC (see database/VIEWS_AND_FUNCTIONS.md):

1. Stage the result while the step stays in_progress.
2. Run the processor against the staged result.
3. CAS to completed, asserting the step is still in_progress.

Writing completed up front and then running the processor leaves a step marked done whose side-effects never ran, and the retry short-circuits on the duplicate check — the result is permanently lost. With staging, a processor failure leaves in_progress + staged data, so a retry re-runs the processor alone without re-invoking the AI step.

The staging CAS admits any callback that finds the step ready/in_progress, so two concurrent duplicate callbacks can both stage and both run the processor before either reaches stage 3. Processor execution is at-least-once, not exactly-once: every processor reachable from this path is required to be idempotent under a re-run against the same staged result, and each was audited against that contract (with idempotency guards added where they were missing). Narrowing the window further would mean holding a lock across the processor's whole runtime, which is not acceptable for AI-length work.

Every guarded write asserts the job's restart_count (the generation the caller was dispatched for) under the parent job's row lock — the same lock claim_job_recovery and the four lifecycle RPCs take, which is what serialises a step write against a concurrent recovery sweep or lifecycle transition. It then asserts the parent job is still live: a job that reached a terminal status on this same generation refuses the write with job_terminal, because writing a step behind a written verdict reopens a settled job's step set and nothing recomputes it afterwards. It finally asserts the execution-group claim the caller was dispatched under, when one is supplied. Rejections come back as outcomes (stale_generation, job_terminal, claim_lost, status_conflict, not_found), never as silent overwrites.

Manual-step execution re-enters on the same outcomes as the AI callback. A claim CAS that conflicts because the step is already completed means a previous attempt ran the processor and wrote its completion but died before unblocking the dependents: the write and the processor are both skipped (the side-effects already happened) while the follow-ups — readiness recomputation and the job roll-up — do run, because they are exactly what went missing. That is the only path by which such a step heals, since a from-point retry leaves completed steps alone. A job_terminal outcome, by contrast, stands the group down without failing: the job's verdict is written, and only an explicit restart may revive it.

Atomic lifecycle transitions

Every lifecycle transition of a job — full restart, retry from a step, resume from an execution group, fail-out of a stall, finalize a halted run — mutates the job row AND its step rows together. Issued as separate PostgREST requests those writes auto-commit independently, so a concurrent transition can interleave between the job-row compare-and-swap that authorises the change and the step writes it authorises, leaving the winner's steps mutated by the loser. Each transition is therefore ONE RPC — restart_processing_job_atomic, resume_processing_job_atomic, fail_stalled_job_atomic, finalize_halted_job_atomic (see database/VIEWS_AND_FUNCTIONS.md) — running the whole transition in one transaction under the same job row lock the guarded step write and claim_job_recovery take. A rejected transition writes nothing; an accepted one is indivisible. The TypeScript lifecycle functions in services/processing/base.ts are thin wrappers that branch on the RPC's semantic outcome, never on a SQLSTATE.

Resume claims a NEW generation. Restart, retry AND resume all bump restart_count. Resume bumping it is a deliberate contract, not a side-effect: a resume resets every step at or after its group regardless of status and re-drives that group itself, so no legitimate in-flight result for it could survive the resume anyway. A straggler callback from the halted run therefore gets a confirmed-absent answer, which is correct, while a second concurrent resume — or a sweeper's fenced finalize arriving just behind the resume — loses cleanly instead of resetting the run a second time. The same bump is what allows the execution-group high-water mark to be rewound so the resumed group can be re-claimed. A superseded resume is surfaced to the user as benign ("already resumed"), not as an error.

Callers dispatch under the generation their own call won. Each RPC returns new_restart_count, and that value — never a fresh read of the job — is what the caller dispatches with. Between the transaction and a re-read a concurrent restart can establish a later generation, and adopting it would drive a group claim under someone else's run.

Dispatch re-asserts the claim PAIR. Before any group is dispatched, the pre-dispatch validation checks the generation AND that the live row still records current_processing_group === executionGroup with a matching current_processing_group_claim_id. Ownership of a group is the (pointer, token) pair and both can move without the generation moving — a resume that rewinds the pointer, or a competitor that took and re-minted the token, leaves the frame holding an id the job no longer records. A mismatch stands the frame down as claim_lost and hands the claim back; the release is itself claim-fenced, so returning an id that already moved on is a no-op.

Fenced processor domain commits

A processor's domain writes (a transaction's binding, its line items, its payment, its allocations) used to be sequential PostgREST requests outside any job-row transaction, so a lifecycle transition landing mid-flight rejected only the abandoned run's step-state write while its domain writes committed into the new run. Each of those commits now runs inside ONE fenced RPC.

The guarantee is linearization, not erasure. Every fenced commit takes the SAME processing_jobs row lock the lifecycle RPCs and the guarded step write take, so once a restart/retry/resume/recovery commits, no fenced write from the superseded generation can commit after it. Domain writes that committed BEFORE the transition are old-run work the new run supersedes by re-driving its reset steps — the idempotent-re-run contract above is unchanged, and processor execution stays at-least-once within a generation.

Two row locks, two distinct guarantees. The job row lock serializes generation, execution-group claim and lifecycle ordering; it does NOT serialize against record cancellation, which never touches the job row. The transaction row lock, taken after it (SELECT … FOR UPDATE), separately serializes record eligibility against a concurrent cancel/reject compare-and-swap — a plain status read races that CAS. Conflating the two is the mistake the design exists to prevent. Lock order is always job → record → children; nothing in the trigger graph these writes reach acquires them in reverse.

The five fenced commits (full signatures, outcome ladders and grants in database/VIEWS_AND_FUNCTIONS.md):

RPC	Processor	What it commits
create_transaction_payment_fenced	processPaymentCreation	Payment + its allocation in one transaction, with the convergence probe under both locks. All three success arms (created / repaired / converged) return the canonical id pair the caller must adopt — on converged and repaired the caller's own proposed payment id was never written. A supplied reuse_payment_id is locked FOR UPDATE and re-validated against the criteria the caller matched on; a candidate that moved returns reuse_conflict (zero-write) and the processor retries once WITHOUT the reuse id, creating a fresh payment
claim_transaction_for_approval_fenced	finishTransactionProcessing	The pre-approval status CAS. p_allowed_from_statuses is a POSITIVE in-list, so an unnamed status is refused rather than assumed claimable
bind_transaction_entities_fenced	processEntityMatching	The binding write (entity ids, relationship FKs, ai_*_logic), restricted to an allowlist; only supplied keys are written, so a partial binding never clears a column another step established
replace_transaction_items_fenced	OCR line-item write	Delete-all + batch insert, atomic as well as fenced — this also fixes the old two-request non-atomicity, where a failed insert left the record with no line items at all
insert_daily_allocations_fenced	processBudgetDailyAllocation	Allocations (find-or-create on their natural key) plus their line-item links, with every link remapped onto the canonical allocation id; a duplicate natural key or an unresolvable link raises and rolls the whole call back

Fence inputs are DISPATCHED values, never re-reads. ProcessingContext carries expectedRestartCount and claimId; the manual path threads the claim the orchestrator holds, and the AI path threads the claim signed into the callback token. A re-read would adopt whatever claim the live run owns now, which is exactly the run a superseded callback must not write into.

update_processing_step_guarded now asserts the claim too. A new p_expected_claim_id (asserted after the generation and job_terminal arms; NULL asserts nothing) refuses a step write from a worker whose execution-group claim was replaced on the SAME generation. claim_lost is a legitimate discard: the callback route answers 200 at all three stages, and the manual loop stands the group down.

A rejection lands in exactly one of three tiers, decided by what the guarded step write would do with the same job state. classifyFencedOutcome (services/processing/base.ts) is the single place this is decided:

Tier	Outcomes	Handling
Completion-safe supersession	not_found, stale_generation, job_terminal, claim_lost	Benign abort, processor reports success — the guarded write independently refuses each of these, so it stays the single decision point
Halt	record_terminal	ProcessingRecordTerminalError; step left un-completed (below)
Invariant failure	job_record_mismatch, record_missing	ProcessingFenceInvariantError so the STEP FAILS (below). The job's generation, liveness and claim are all still valid here too, so the guarded write would accept a completion for a commit that never happened — a benign abort would silently pass

Outcome handling also FAILS CLOSED: each wrapper passes its RPC's own success allowlist (FENCED_*_SUCCESS_OUTCOMES, constants/processing.ts) and any value in neither that list nor the rejection ladder throws. A renamed or newly added SQL arm can therefore never be read as success.

record_terminal never becomes a completed step. It is the one arm where the job's generation, liveness and claim are all still valid, so the guarded write that follows would happily record a completion for a commit that never happened. The processor therefore raises ProcessingRecordTerminalError (utils/processing.ts) instead of returning; both step-running loops leave the step un-completed and let the existing halt path (isProcessingHaltedForJob → finalize_halted_job_atomic) settle the job, marking the tail not_needed. The callback route answers 200 and skips stage 3 entirely. This is uniform across every fenced call site, including the approval claim in finishTransactionProcessing: declining a terminal record means the approval submission never ran, so the finish step must not be recorded as completed.

An invariant rejection fails the step and is answered 401, never 503. job_record_mismatch and record_missing are decided under the job row lock, so the refusal is VERIFIED, PERMANENT and zero-write — retrying it can only meet the identical answer until AI Hub's ladder is exhausted or the stall sweeper intervenes. classifyFencedOutcome therefore raises the typed ProcessingFenceInvariantError (utils/processing.ts), which processStepResult and handleStepCompletion carry as their own flag rather than collapsing into a generic processor failure. The callback route writes the step FAILED through the ordinary guarded CAS under the generation and claim the dispatch held (expected prior status in_progress, the invariant message as error_message), runs the dependent-skip cascade and the roll-up, and answers 401. The fence still rules that write: a supersession reported by it (stale_generation, claim_lost, job_terminal, not_found) falls through to the ordinary 200 discard, and only a genuine database error is 503. On the MANUAL path the same throw reaches executeManualSteps' generic catch, which already routes it through failManualStep — the identical fenced failure write plus cascade and roll-up — and re-throws to fail the job.

Resume is fenced against cancellation. resume_processing_job_atomic takes an optional p_blocked_processable_statuses; when supplied for a transaction job it reads the record FOR UPDATE after the job lock and refuses with record_terminal, zero-write. The lock, not the status read, is the fence — the app-side pre-read in resumeHeldTransactionProcessing remains only as a fast path. The rejection is a benign standdown, matching that fast path and the superseded arm beside it: the only caller is the queue worker (/api/ai-processing/start-job with resumeFromHold), whose route maps a returned error to a 500 and would have QStash retry a decision that can never change. The record's status is re-read for the log so a rejected record is not reported as cancelled.

Residuals — narrowed, not excluded. These are the parts of the mid-flight window this design does not close:

• Long-tail processors boundary re-assert. processDocumentParsing, processTypeDiscovery, processBudgetItemMatching, processChartOfAccountsMatching, processJustificationAnalysis, the currency-rate check's write branch, and the tax/company validation integrations call assertDispatchedGenerationStillCurrent(jobId, expectedRestartCount, claimId) at their write-phase boundary. That is ONE read: a lifecycle transition committing between it and the writes that follow is still admitted. It NARROWS the single-request TOCTOU window; idempotent re-drive of the reset step remains the backstop. Converting any of them to a fenced RPC is now a local change, since the context already carries the fence inputs.
• The storage move in type discovery cannot be DB-fenced. A storage object cannot participate in a database transaction. The move sits immediately behind its own re-assert, and its rename target is deterministic, so a re-drive converges on the same path rather than compounding.
• Entity/relationship find-or-create rows survive a superseded run. Those creations sit deliberately OUTSIDE the fence: a row a stale run created is a convergent-dedupe row the next run finds and reuses, whereas a BINDING written by a stale run would attach that run's conclusion to the live record. Only the binding is fenced.
• Approval crash-after-claim-before-submission is unchanged. A claim that committed before a lifecycle transition is old-run work; a crash between the claim and the submission leaves a record in IN_APPROVAL with no approval workflow attached. That is a pre-existing residual of the approval design, out of scope here and not repaired by this fence — the re-driven finish step is not asserted to fix it.
• Pre-deploy app instances keep writing unfenced until they drain. During a rolling deploy, instances still running the previous build issue the old sequential PostgREST writes; their in-flight callbacks do the same until they finish. That transient window is exactly the permanent exposure this change removes, so it is accepted rather than gated behind a flag or a two-phase rollout. It closes once those instances cycle out and their last in-flight callbacks retire — callback-token lifetime is bounded by the dispatched steps' timeouts, so the tail is bounded too.
• The nullable claim assertion is a rollout compatibility window. p_expected_claim_id accepts NULL so a callback whose token predates claim binding can still write. Callback-token lifetime is bounded by the dispatched steps' timeouts, so one release cycle after this ships no such token can still be in flight; the follow-up is to make the claim non-null-required end to end.

Callback status-code contract

packages/app/src/app/api/ai-processing/callback/route.ts answers with a deliberately narrow vocabulary, because AI Hub's retry decision is made from the status code alone:

Status	Meaning
200	The result landed, or was legitimately discarded (stale generation, superseded run, duplicate step write whose continuation re-ran)
400	The body failed schema validation — a malformed request no retry can fix
401	Confirmed rejection only: bad API key, failed callback-token verification, a job/step the app read successfully and found absent, a step write refused because the job already reached a terminal status (job_terminal — the run is over, so no retry can ever succeed), or a fenced domain write refused for a broken invariant (job_record_mismatch / record_missing — the step is FAILED under the dispatch's own fence first, then the callback is rejected)
500	Server misconfiguration (the callback signing secret is unset) or an unhandled error
503	Any read or continuation failure — the state is unknown, and a retry is the correct response

The job's own roll-up state is recomputed explicitly on this path. The guarded CAS touches only the step row, so unlike the legacy updateStepStatus write nothing else moves progress_percentage or overall_status. continueAfterStepCompletion therefore calls recomputeJobRollup (progress from the visible steps, plus the terminal-status + completed_at write once every step is terminal) before the group-terminal check — the "no next group" branch returns without finalizing anything, so a job whose last step is an AI step would otherwise never reach a terminal status, and the progress bar would sit frozen for the whole AI phase. Both halves derive from fresh reads, so the call is idempotent.

The 503-vs-401 split is the whole point: answering 401 to a transient database failure tells AI Hub the result is unauthorised and must be abandoned, permanently losing a legitimate step result. Every gate in continueAfterStepCompletion therefore fails closed to 503 — an unreadable halt state, an unresolvable execution group, and a failed advance all retry rather than guessing.

On the AI Hub side, packages/aiHub/src/services/api/index.ts classifies failures via shouldRetry: transport/timeout errors (no HTTP status reached us) and 5xx responses are retryable; every 4xx — including 429 — is terminal, since the app answers 4xx only for a request it has confirmed invalid. httpClient throws HttpStatusError on !response.ok so the status is available to that classifier.

The stall sweeper

The cron at packages/app/src/app/api/cron/sweep-stalled-processing-jobs/route.ts is the only thing that notices a job nobody is driving. It calls sweepStalledProcessingJobs (processing/server.ts), which walks live jobs least-recently-touched first (updated_at ascending, SWEEP_BATCH_SIZE). Ordering by updated_at both matches the partial index idx_processing_jobs_live_updated_at that serves the scan and keeps the batch rotating: a job that made progress sinks behind the ones that have not.

Staleness predicate — both bounds must hold, so a long-running AI step is never swept out from under a worker that is still working:

• The step watermark (latest processing_job_steps.updated_at, falling back to processing_jobs.created_at) is older than STALL_WATERMARK_MS; and
• every in_progress step is past started_at + timeout_seconds + STEP_TIMEOUT_BUFFER_MS, using the step's own process_template_steps.timeout_seconds (or DEFAULT_STEP_TIMEOUT_SECONDS when it declares none).

Claim ladder — every candidate goes through the claim_job_recovery RPC, which re-derives the watermark under the job's row lock and re-checks the sweeper's observation there. That is what makes acting on a stale read safe. Outcomes: not_found → terminal → superseded (restart_count moved) → progressed (a callback landed between read and claim) all leave the job alone; exhausted fails it out; only claimed authorises recovery. Attempts are fingerprint-scoped — the fingerprint is computed over terminal steps and their max execution group, so a fingerprint change means real progress and resets sweep_attempts to 1, while an unchanged pipeline walks the ladder to MAX_RECOVERY_ATTEMPTS and stops.

Targeted reset — the reset rides claim_job_recovery itself (p_perform_reset), so the claim, the generation bump and the step reset are one transaction under the job row lock. Only in_progress steps return to ready (clearing started_at and error_message); completed/failed/skipped/not-needed steps and every step's result_data survive, so recovery re-drives from the lowest incomplete execution group without discarding finished work. restart_count is always bumped — callback tokens bind { jobId, restartCount }, so the bump is what invalidates the abandoned attempt's tokens and stops a late worker writing into the generation this recovery is about to dispatch. The whole execution-group claim is cleared here, group pointer included. Rewinding the pointer is safe only under the recovery fence — the claim plus the restart bump establish that no live worker holds the previous generation — and would be wrong on the normal callback path, where the pointer must stay monotonic so a stale callback cannot re-drive an advanced group. Without the rewind, a job whose pointer had run ahead of its lowest incomplete group (the orchestrator's recursion through empty groups advances it) could never re-claim that group, and the sweep would dispatch nothing while reporting the job recovered.

Recovery checks the halt state first (resurrecting a cancelled or On Hold record is the one unrecoverable mistake available here, and an unreadable halt state is a reason not to proceed, not a reason to guess). The claim RPC then clears the dead worker's fenced claim, bumps the generation and resets the stalled steps in its single transaction, and the sweeper re-dispatches under the generation that call returned.

advanceToNextExecutionGroup reports what it achieved, not merely that it did not error: dispatched (AI steps sent or a manual group executed), claim_lost (another worker owns the group), or no_more_groups (nothing left to run, or the job was finalized). The sweeper counts a job as recovered ONLY on dispatched — the other outcomes are legitimate no-ops, and counting them would report the sweeper healing jobs it never touched while the retry ladder quietly burned down to a fail-out.

Fail-out — failStalledJob distinguishes two shapes of stall, because only one is a failure:

• Missed final write: every step already reached a terminal status but the job's own status write never landed. Nothing about the run is wrong, so the terminal status is recomputed from the steps and no error_message is recorded. Forcing FAILED here would destroy a successful run's result. The reported outcome always reflects the status actually written, never which shape was taken: only an all-successful recompute is completed_recomputed; an all-terminal set containing failures reports failed / partial_success and therefore still notifies.
• Genuine death: non-terminal steps remain. They are marked failed carrying the reason, and the recomputed job status is failed or partial_success.

Either way the job leaves the live statuses (so the sweeper cannot pick it up again) and the claim token is cleared, while current_processing_group is preserved as the pipeline's high-water mark. The whole transition is the fail_stalled_job_atomic RPC: the still-unfinished step set is derived under the job row lock, never from the sweeper's earlier snapshot (a snapshot can name steps that have since reached a terminal verdict, and marking those failed would destroy a genuine result), and the terminal job write and the step writes commit together. The generation the sweeper observed is a mandatory argument — a restart since the observation returns superseded with nothing written, so a fail-out can never kill a live run. A job whose processable_type has no registered continuation handler is failed out rather than resurrected into the same silence — there is nothing that could drive it forward.

Halted, not stalled — a swept job whose record turns out to be halted (cancelled, On Hold) is settled rather than recovered, through finalize_halted_job_atomic fenced on the generation the sweep observed. Every non-finalized outcome (superseded, already_terminal, no_steps, not_found) is a benign logged no-op: another actor already owns the job. These jobs get their own haltedFinalized bucket in the sweep summary rather than being folded into skipped — they WERE acted on, and counting the sweeper's most consequential write as an untouched job hides it.

Notification — only a genuine failure (failed / partial_success) sends the PROCESSING_STALLED notification to the job's initiator (created_by_user_id); a recomputed success would be crying wolf about work that actually succeeded. Channels are left to sendNotification to derive from the template (in-app mandatory, WhatsApp opt-in), and a send failure is logged and swallowed — the terminal status is already written, and losing the courtesy alert must not turn the sweep into a failure. A system-ingested job (email/Sana) has no created_by_user_id, so there is nobody to tell: that is logged at error level rather than passed silently to sendNotification, since the failure is invisible to any user by construction and ops alerting is the only thing that can surface it. Follow-up: give those jobs a fallback recipient (e.g. the project's processing owners) rather than only an ops log.

Cron wiring — scheduled every 15 minutes in packages/app/vercel.json (per-minute schedules require the Pro plan). The route authenticates itself with a Bearer ${CRON_SECRET} check, which means it must also be in the serviceApiRoutes allowlist in packages/app/src/lib/database/middleware.ts (via API_URLS.CRON_SWEEP_STALLED_PROCESSING_JOBS) — otherwise updateSession 401s the request before the route's own check ever runs.

Per-job failures are contained: one unrecoverable job never prevents the rest of the batch from being swept.

The unprocessed-transaction sweeper (dispatch loss)

The stall sweeper repairs jobs that exist and stopped moving. It cannot see the window before a job exists: a transaction is persisted with its document, the queue publish that would create its processing job is lost in transport (or dead-lettered after its retries), and the record sits in Received forever with nothing driving it and no job to sweep. Because pre-pipeline transactions are hidden from every list surface, nobody sees it either.

The cron at packages/app/src/app/api/cron/sweep-unprocessed-transactions/route.ts closes that window. It calls sweepUnprocessedTransactions (processing/server.ts), which runs a two-step diff in TypeScript — deliberately, for the first iteration; an RPC is the follow-up if candidate volumes ever grow beyond the near-zero expected for a rare-loss backstop.

Candidate predicate — status IN PRE_PROCESSING_TRANSACTION_STATUSES (Received, Processing) AND created_at older than SWEEP_UNPROCESSED_AGE_MS (30 min) AND attachment_id IS NOT NULL, oldest-first, capped at SWEEP_UNPROCESSED_BATCH_SIZE (20 — 20 × the 10s worst-case publish deadline stays inside the route's 300s budget). Three things about that predicate carry weight:

• On Hold is excluded. A held transaction is parked awaiting an exchange rate and is owned by the resume flow; re-dispatching one would resurrect work the pipeline deliberately paused.
• The age bound exceeds the 15-minute cron interval, so a publish still legitimately in flight is never mistaken for a lost one (the same invariant as STALL_WATERMARK_MS).
• An attachment is required. The pipeline is document-driven: createTransaction only reports shouldStartProcessing once an attachment exists, so a transaction whose attachment creation failed is legitimately jobless and re-dispatching it could only fail.

Set difference — surviving candidates are those with no job of the INITIAL template (TransactionProcessingTemplate.TRANSACTION_PROCESSING, resolved once per run and matched on processing_jobs.process_template_id). Filtering on "no job at all" would be wrong: an unrelated later job (a budget-activation reprocess landing on a never-OCR'd transaction) would mask a first run that never happened.

Dedup id — each re-dispatch publishes with sweep-unprocessed:${transactionId}:${bucket} where bucket = Math.floor(Date.now() / SWEEP_UNPROCESSED_AGE_MS). The time bucket makes the id an attempt marker: overlapping sweep runs inside one bucket cannot double-enqueue, while a later bucket can genuinely re-attempt. It deliberately differs from the upload id below — QStash retains deduplication ids for 90 days including for dead-lettered deliveries, so reusing the upload's id would make QStash swallow the sweep publish for exactly the lost-delivery case the sweep exists to repair.

Failure accounting — startReprocessingForTransaction returns an error envelope on publish failure but can still throw on unexpected ones, so the call is .catch-normalized and branched on .error; neither surface escapes the per-transaction accounting. A failure logs and continues to the next candidate. Each run ends with a summary log { scanned, jobless, enqueued, failed }, which the route returns as { success: true, ...summary }.

Escalation — a candidate that is BOTH older than SWEEP_UNPROCESSED_ESCALATION_AGE_MS (2h) AND whose enqueue failed in this run has demonstrably stopped recovering on its own. Two things fire:

• An error-level log under the stable event name UNPROCESSED_SWEEP_ESCALATION_EVENT (processing.sweep_unprocessed_escalation), carrying transaction/project ids and age. This is the ops alerting hook — do not rename it without moving the alert.
• An in-app notification (TEMPLATES.TRANSACTION_PROCESSING_NOT_STARTED) to every ACTIVE holder of the project:owner role on the project, resolved via getProjectUserIdsWithRoleKey. Owner-lookup and send failures are logged and swallowed — the sweep's job is re-dispatching work, not guaranteeing delivery of an alert about it — and a project with no owner is logged at error level rather than passed silently to sendNotification.

A processing issue is deliberately NOT raised: issues surface on transaction/processing UI surfaces, and a pre-pipeline transaction is hidden from all of them, so the issue row would reach nobody. Notifications carry no idempotency key, so the send is restricted to the single sweep window in which a candidate first crosses the 2h threshold (age >= 2h AND age < 2h + SWEEP_UNPROCESSED_AGE_MS) — one notification per transaction, rather than a fresh alert every 15 minutes. The log still fires on every failing run.

Cron wiring — every 15 minutes in packages/app/vercel.json, with the same hardened Bearer ${CRON_SECRET} guard as the stall sweeper (an unset secret refuses with 500 rather than comparing against the interpolated "Bearer undefined", which any caller could send verbatim) and the matching serviceApiRoutes entry via API_URLS.CRON_SWEEP_UNPROCESSED_TRANSACTIONS.

Bounded publish transport + stable upload dedup ids

Two changes upstream of both sweepers reduce how often either has to act:

• The publish is deadline-bounded. enqueueJob (packages/app/src/lib/queue.ts) constructs the QStash client with retry: { retries: 2 } (the SDK's default 5-retry ladder is unabortable once started) and races publishJSON against a timer defaulting to QSTASH_PUBLISH_DEADLINE_MS (10s), overridable per call via QueueJobOptions.publishDeadlineMs for publishers inside a tighter external ACK window. Note options.timeout is a different thing — QStash's destination-call timeout. The race bounds the caller's wait but cannot cancel the in-flight publish, so a timed-out publish may still land; the loser of the race carries a no-op .catch so a late rejection is not an unhandled-rejection crash, and the sweep's no-job predicate plus the dedup ids tolerate the duplicate.
• Upload publishes carry stable dedup ids. createTransactionFromDocument and createTransactionFromStagedDocument both publish with buildInitialProcessingDeduplicationId(transactionId) = `${DatabaseTable.TRANSACTIONS}:${transactionId}:${TransactionProcessingTemplate.TRANSACTION_PROCESSING}`, derivable identically at both sites because the transaction row is persisted before the publish. A retried creation, or a publish the caller timed out on that nevertheless landed, collapses onto one QStash message instead of starting a second initial pipeline run. Dedup ids are deliberately NOT added to startReprocessingForTransaction's ordinary callers or to startResumeForHeldTransaction: a genuine later re-run within QStash's 90-day retention would be swallowed.

Security Considerations

Authentication

• Service role for system operations
• User authentication for job creation
• Row-Level Security on all tables

Data Access

• Users can only view their own jobs
• Organisation-based access control
• Audit trail via created_by fields

File Security

• S3 presigned URLs with expiration
• Encrypted file transfer
• No permanent storage of sensitive documents

Key Architecture Principles

Service Layer Separation

The system strictly enforces separation between service layers:

• Base Services: Pure business logic, no external calls
• Server Services: External integrations (AI, geocoding, email)
• Client Services: React component data fetching
• Sana Services: WhatsApp bot specific logic

For detailed patterns, see docs/development/SERVICE_PATTERNS.md

AI Response Parsing

AI responses are consistently parsed from the message field:

// Correct parsing pattern
const aiResponse = JSON.parse(result_data.message);
// NOT: JSON.parse(result_data.response)

Idempotent Design

All processing steps are designed to be idempotent:

• Check completion status before processing
• Return success with skipped: true for completed steps
• Atomic database updates prevent partial state

Future Enhancements

1. Batch Processing

• Process multiple documents in single job
• Bulk status updates
• Parallel document processing

2. Custom Workflows

• User-defined templates
• Drag-and-drop workflow builder
• Custom AI prompts per organisation (partially implemented with prompt templates)

3. Advanced Analytics

• Processing time metrics
• Success rate tracking
• Cost optimization insights

4. Integration Expansion

• Email document ingestion
• Direct API uploads
• Third-party system webhooks

5. Enhanced Processing Features

• Conditional processor selection based on result data
• Multi-step transactions with rollback support
• Real-time processing progress via WebSockets
• Per-step-type tuning of the recovery strategy (the generic stall sweep is built — see Stall detection and recovery — but staleness thresholds and the retry ladder are global rather than per step type)

Implementation Examples

Creating a Processing Job

// In WhatsApp bot handler
const job = await createTransactionProcessingJob(transactionId, {
  fileUrl: s3Url,
  mimeType: 'application/pdf',
  uploadedBy: userId,
});

// Start processing - only sends first execution group
await startTransactionProcessing(
  transactionId,
  fileUrl,
  mimeType,
  job.id,
  sendToAIHub // Callback that sends to AI Hub
);

// Subsequent groups sent automatically via callbacks
// as each execution group completes

Displaying Status

// In React component
const ProcessingStatus = ({ transactionId }) => {
  const { data: status } = useJobStatus(transactionId);

  return (
    <ProcessingStatusIndicator
      job={status?.job}
      steps={status?.steps}
      onRetry={handleRetry}
    />
  );
};

Webhook Handler

// API route handler for status updates from AI Hub
export async function POST(request: Request) {
  const { jobId, stepKey, status, resultData } = await request.json();

  // Update step status in database
  const success = await updateProcessingStepStatus(
    jobId,
    stepKey,
    status,
    resultData
  );

  // Check if execution group is complete
  if (success && status === 'COMPLETED') {
    await checkAndSendNextGroup(jobId);
  }

  return Response.json({ success });
}

Callback Handler with Idempotency

// API route handler with duplicate processing prevention
export async function POST(request: Request) {
  const { jobId, stepKey, status, resultData, errorMessage } =
    await request.json();

  // Check if step is already completed (idempotency)
  const stepStatus = await getStepStatus(jobId, stepKey);
  if (stepStatus === PROCESSING_STEP_STATUS.COMPLETED) {
    return Response.json({
      success: true,
      skipped: true,
      message: 'Step already completed',
    });
  }

  // Update step status
  const updated = await updateProcessingStepStatus(
    jobId,
    stepKey,
    status,
    resultData,
    errorMessage
  );

  if (!updated) {
    return Response.json({ success: false, error: 'Failed to update status' });
  }

  // Process results if step completed successfully
  if (status === PROCESSING_STEP_STATUS.COMPLETED && resultData) {
    const processed = await processStepResult(jobId, stepKey);

    // Check if current execution group is complete
    const groupComplete = await isExecutionGroupComplete(jobId, stepKey);

    if (groupComplete) {
      // Fetch fresh entity data
      const freshEntity = await getEntityByJobId(jobId);

      // Get next execution group steps
      const nextSteps = await getNextExecutionGroupSteps(jobId);

      if (nextSteps.length > 0) {
        // Send callback to AI Hub with fresh data
        await sendToAIHub({
          type: getTaskType(jobId),
          entityId: getEntityId(jobId),
          entity: freshEntity,
          metadata: {
            steps: nextSteps,
            jobId,
            executionGroup: nextSteps[0].execution_group,
          },
        });
      }
    }

    return Response.json({ success: processed });
  }

  return Response.json({ success: true });
}

Idempotency and Duplicate Processing Prevention

The system implements comprehensive idempotency to handle duplicate messages from SQS and ensure data integrity.

Key Features

1. Step Status Checking: Before processing, the system checks if a step is already completed
2. Atomic Updates: Database operations use transactions to prevent partial updates
3. Message Deduplication: Handles SQS message redelivery gracefully
4. Skipped Response: Returns skipped: true flag for already-processed steps. Note that only the step write is skipped — the callback still re-enters the continuation, because a previous callback that completed the step and then failed to advance the job is exactly the stall a retry must heal. The advance is itself idempotent (the claim mutex admits one advancer), so re-entering costs nothing when the job did progress.

Implementation

// Check step status before processing
export async function getStepStatus(
  jobId: string,
  stepKey: string
): Promise<string | null> {
  const step = await findOne(DatabaseTable.PROCESSING_JOB_STEPS, {
    processing_job_id: jobId,
    step_key: stepKey,
  });
  return step?.status || null;
}

// Idempotent processor
export async function processStep(step: ProcessingJobStep) {
  // Check if already processed
  if (step.status === PROCESSING_STEP_STATUS.COMPLETED) {
    return { success: true, skipped: true };
  }

  // Process the step
  const result = await performProcessing(step);

  // Atomic status update
  await updateStepStatus(step.id, PROCESSING_STEP_STATUS.COMPLETED);

  return { success: true, skipped: false };
}

Benefits

• Data Integrity: Prevents duplicate processing and data corruption
• Resilience: Handles network failures and message redelivery
• Auditability: Clear tracking of skipped vs processed steps
• Performance: Avoids unnecessary reprocessing

Conclusion

The AI Processing Architecture provides a robust, scalable foundation for intelligent document processing in FarmCove Delta. Its template-based approach allows easy extension to new entity types and processing workflows while maintaining consistency and reliability.