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 as several independent processes:

The pipeline worker detects domain events from the EventStoreDB subscription and starts the appropriate Temporal workflow. The two Temporal workers poll their task queues and execute the actual processing logic.

# Start each process individually for local development
make run                  # API server
make run-read-models      # Read worker (EventStoreDB → MongoDB)
make run-workflow-worker  # Pipeline worker (EventStoreDB → Temporal)
make run-temporal-worker  # Temporal worker (artifact_processing queue)
make run-llm-worker       # LLM worker (llm_processing queue)
make run-plugin-consumer  # Plugin consumer (Kafka → plugin workflows)

# Or start all six together
make run-all

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
  └─ ParseArtifactWorkflow (Docling parse → pages + page text)
       ├─ Page.Created (each page)
       ├─ Page.TextMentionUpdated (each page with text)
       └─ Artifact.PagesAdded

Page.Created
  └─ ExtractCompoundMentionsWorkflow (OCSR)
       └─ Page.CompoundMentionsUpdated
            ├─ EmbedCompoundSmilesWorkflow (ChemBERTa → compound_embeddings)
            └─ ReconcileCompoundLabelsWorkflow

Page.TextMentionUpdated
  ├─ PageSummarizationWorkflow
  │    └─ Page.SummaryCandidateUpdated
  │         ├─ PageSummaryEmbeddingWorkflow (→ summary_embeddings)
  │         └─ once every page of the artifact has a summary:
  │              ├─ ArtifactSummarizationWorkflow
  │              │    └─ Artifact.SummaryCandidateUpdated
  │              │         └─ ArtifactSummaryEmbeddingWorkflow (→ summary_embeddings)
  │              └─ BatchReEmbedArtifactPagesWorkflow (all chunks → page_embeddings)
  ├─ NERExtractionWorkflow
  │    └─ Page.TagMentionsUpdated
  │         ├─ ArtifactTagAggregationWorkflow
  │         │    └─ Artifact.TagMentionsUpdated (→ Qdrant payload sync)
  │         ├─ page tags → Qdrant payload sync
  │         └─ ReconcileCompoundLabelsWorkflow
  └─ DocumentMetadataExtractionWorkflow (triggered by the first page)
       └─ Artifact.TitleMentionUpdated / AuthorMentionsUpdated / PresentationDateUpdated
            (authors and date → Qdrant payload sync)

Text chunks are embedded once per artifact, after all page summaries exist, so that every chunk's context header includes its page summary. GeneratePageEmbeddingWorkflow embeds a single page and is used for manual re-embedding through the API. ReconcileCompoundLabelsWorkflow runs from both the compound and the NER branch, whichever finishes last, to reconcile OCSR labels against NER compound names.

Idempotency

Workflow IDs are derived deterministically from entity IDs:

Workflows use Temporal's ALLOW_DUPLICATE ID reuse policy, so a finished workflow can be re-run under the same ID. Starting a workflow whose ID is still running is rejected by Temporal; the orchestrator logs the rejection and moves on, so duplicate triggers are a safe no-op.

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:

Each queue is served by its own worker process. 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:

Activity errors go through the retry policy above; errors that a retry cannot fix (not found, validation, not ready) are marked non-retryable. Embedding activities treat "nothing to embed" (a blank page, a deleted artifact) as a completed run, but fail the workflow on embedding provider or internal errors, so the failure shows in the workflow status endpoints instead of passing as success.

Configuration

Observability

Every workflow execution is visible in the Temporal UI at http://localhost:8233 (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 ner-extraction-* workflows)