Temporal Workflows

Every long-running operation in DocuStore -- PDF parsing, embedding generation, LLM summarization, compound extraction -- is orchestrated by Temporal. This provides automatic retries, timeout handling, observability, and crash recovery without custom retry logic.

Architecture

The system runs three independent worker processes:

The pipeline worker detects domain events from the EventStoreDB subscription and starts the appropriate Temporal workflow. The Temporal worker polls for tasks and executes the actual processing logic.

# Start all three processes for local development
make run                # API server
make run-workflow-worker # Pipeline worker (EventStoreDB → Temporal)
make run-temporal-worker # Temporal worker (executes activities)

Event Pipeline

Domain events trigger workflows in a cascade. Each workflow completes by saving results to the aggregate, which emits a new event, which triggers the next workflow:

Artifact.Created
  └─ ArtifactProcessingWorkflow (PDF parse → page creation)

Page.Created
  └─ ExtractCompoundMentionsWorkflow (OCSR)

Page.TextMentionUpdated
  ├─ TextEmbeddingWorkflow
  │    └─ Page.TextEmbeddingGenerated
  │         └─ SummarizationWorkflow
  │              └─ Page.SummaryCandidateUpdated
  │                   ├─ TriggerArtifactSummarizationUseCase
  │                   └─ PageSummaryEmbeddingWorkflow
  └─ NERExtractionWorkflow
       └─ Page.TagMentionsUpdated
            └─ ArtifactTagAggregationWorkflow

Page.CompoundMentionsUpdated
  └─ SmilesEmbeddingWorkflow

Idempotency

Workflow IDs are derived deterministically from entity IDs:

All workflows use the ALLOW_DUPLICATE reuse policy, meaning that re-triggering the same workflow for the same entity is a safe no-op if the workflow is already running.

Port-Based Design

The application layer depends on an abstract WorkflowOrchestrator port:

class WorkflowOrchestrator(Protocol):
    async def start_embedding_workflow(self, page_id: UUID) -> None: ...
    async def start_compound_extraction_workflow(self, page_id: UUID) -> None: ...
    async def start_ner_extraction_workflow(self, page_id: UUID) -> None: ...
    async def start_artifact_summarization_workflow(self, artifact_id: UUID) -> None: ...
    # ... additional workflow methods

The TemporalWorkflowOrchestrator in the infrastructure layer implements this port using the Temporal Python SDK. The application layer never imports Temporal directly, so the orchestration backend could be swapped to Celery, AWS Step Functions, or any other system without changing business logic.

Task Queues

DocuStore uses two task queues to control concurrency:

Separating LLM activities onto their own queue prevents slow LLM calls from blocking fast embedding and parsing tasks. For local development with Ollama, TEMPORAL_MAX_CONCURRENT_LLM_ACTIVITIES=2 prevents memory exhaustion.

Retry Configuration

Activities are configured with appropriate timeouts and retry policies:

Configuration

Observability

Every workflow execution is visible in the Temporal UI at http://localhost:8080 (default). You can inspect:

• Workflow execution history (started, activities, completed/failed)
• Activity inputs, outputs, and retry attempts
• Pending activities and their timeouts
• Workflow search by ID pattern (e.g., all text-embedding-* workflows)