Add durabletask.extensions.history_export (#147)

* History export extension - V1

* Strict typing, type correctness

* PR feedback

* Use relative import for _test_helpers (CI fix)

The absolute 'from tests.durabletask...' import worked locally because pip install -e .[dev] adds the repo root to sys.path, but CI runs pytest without that and pytest's rootdir doesn't include a 'tests' top-level package. Switch to a relative import inside the history_export test package (which already has __init__.py).

* PR Feedback 2

* Schema back to 1.0, drop PENDING state

* PR Feedback 3

* Hashed blob names

* Address latest PR review comments from berndverst

Author: andystaples

CHANGELOG.md

@@ -5,6 +5,54 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project
 adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## Unreleased
+
+ADDED
+
+- Added `durabletask.extensions.history_export` for exporting the event history of
+  terminal orchestrations to an external destination. Includes
+  `ExportHistoryClient`, a per-job `ExportHistoryJobClient` returned by
+  `get_job_client(...)`, and `list_jobs(...)` for enumerating jobs by status
+  or last-modified window. Ships with a bundled `AzureBlobHistoryExportWriter`
+  (installed with `pip install durabletask[history-export-azure]`) and a
+  `HistoryWriter` protocol for plugging in custom destinations. Supports both
+  `ExportMode.BATCH` (export a window and complete) and `ExportMode.CONTINUOUS`
+  (tail terminal instances indefinitely until stopped via `delete_job`).
+  Exported blobs are self-describing: each blob carries an explicit
+  `schema_version`, the orchestration's `OrchestrationState` metadata, and
+  the full ordered event list. Blob names are a lowercase-hex SHA-256 of
+  ``{last_updated_at}|{instance_id}`` with the format extension appended
+  (matches the .NET `ExportInstanceHistoryActivity` naming scheme), so
+  re-exporting an instance after a later terminal update lands at a new
+  blob path rather than overwriting the previous one, and instance IDs
+  that differ only by `/` no longer collide. Each exported blob also
+  carries `{"instance_id": <id>}` as destination-side metadata (the Azure
+  writer persists this as Azure Blob metadata) so consumers can scan a
+  container without parsing each blob body. The export workflow retries each instance up
+  to 3 times with exponential backoff (15s/30s/60s), retries failed batches
+  up to 3 times, caps in-flight exports via `max_parallel_exports`
+  (default 32), continues-as-new every 5 page cycles to bound orchestrator
+  history while preserving cumulative totals across continue-as-new segments,
+  and re-fetches entity state at the top of every page loop so
+  external delete or mark-failed signals stop the orchestrator cleanly.
+  Empty-page BATCH checkpoints no longer reset the persisted resume cursor,
+  and duplicate `mark_failed` signals are now idempotent no-ops when a job
+  is already failed to reduce transition-noise logs.
+  `delete_job` actively tears the job down: it clears the entity state,
+  terminates the driving orchestrator, waits briefly for it to settle, and
+  purges its orchestration history so a re-created job with the same ID
+  starts from a clean slate. Per-instance exports refuse to write a blob
+  when the target instance has been purged or has re-entered a non-terminal
+  state, surfacing the skipped instance as a per-batch failure.
+  Job state lives in a durable entity with an explicit state-transition
+  matrix (ACTIVE / COMPLETED / FAILED); invalid transitions raise
+  `ExportJobInvalidTransitionError`. Persisted entity state uses a
+  versioned, schema-stable JSON shape (`STATE_SCHEMA_VERSION`) with no
+  embedded Python type metadata. Each export job's driving orchestrator
+  uses a deterministic instance ID (`export-job-{job_id}`, exposed via
+  `orchestrator_instance_id_for(...)`) so callers can correlate a job ID
+  with its orchestrator for logging, monitoring, and restart.
+
 ## v1.5.0
 
 BREAKING CHANGES (type-level only — no runtime impact for typical users)

docs/features.md

@@ -412,6 +412,165 @@ class MyPayloadStore(PayloadStore):
 
 See the [large payload example](../examples/large_payload/) for a complete working sample.
 
+### Orchestration history export
+
+The optional `durabletask.extensions.history_export` package provides a workflow for exporting the
+full event history of terminal orchestrations to an external destination (for example Azure Blob
+Storage). It is modeled after the .NET SDK's `ExportHistory` package.
+
+An export job scans a time window of terminal instances, fetches each instance's history through
+the standard client API, serializes it, and writes it through a pluggable `HistoryWriter`. Job
+state is owned by a durable entity so progress survives worker restarts.
+
+#### Installation
+
+The core extension has no extra dependencies beyond the SDK. The bundled Azure Blob writer requires
+an optional dependency:
+
+```bash
+pip install durabletask[history-export-azure]
+```
+
+#### Configuring an export job
+
+```python
+from datetime import datetime, timedelta, timezone
+
+from durabletask import client, worker
+from durabletask.extensions.history_export import (
+    ExportDestination,
+    ExportFormat,
+    ExportFormatKind,
+    ExportHistoryClient,
+    ExportJobCreationOptions,
+    ExportMode,
+)
+from durabletask.extensions.history_export.azure_blob import (
+    AzureBlobHistoryExportWriter,
+    AzureBlobHistoryExportWriterOptions,
+)
+
+writer = AzureBlobHistoryExportWriter(
+    AzureBlobHistoryExportWriterOptions(
+        container_name="orchestration-history",
+        connection_string="DefaultEndpointsProtocol=https;...",
+    )
+)
+dt_client = client.TaskHubGrpcClient(host_address="localhost:4001")
+export_client = ExportHistoryClient(dt_client, writer)
+
+with worker.TaskHubGrpcWorker(host_address="localhost:4001") as w:
+    export_client.register_worker(w)
+    w.start()
+
+    now = datetime.now(timezone.utc)
+    desc = export_client.create_job(ExportJobCreationOptions(
+        mode=ExportMode.BATCH,
+        completed_time_from=now - timedelta(days=1),
+        completed_time_to=now,
+        destination=ExportDestination(container="orchestration-history", prefix="2026-05"),
+        format=ExportFormat(kind=ExportFormatKind.JSONL_GZIP),
+        max_instances_per_batch=100,
+    ))
+    final = export_client.wait_for_job(desc.job_id, timeout=600)
+    print(final.status, final.exported_instances, final.failed_instances)
+```
+
+#### Output formats
+
+| `ExportFormatKind` | Per-instance blob extension | Content-Type | Content-Encoding |
+|---|---|---|---|
+| `JSON` | `.json` | `application/json` | (none) |
+| `JSONL_GZIP` | `.jsonl.gz` | `application/x-ndjson` | `gzip` |
+
+The JSONL format prepends a metadata line and writes one event per line, which streams well for
+large histories.
+
+#### Modes
+
+Two `ExportMode` values are supported:
+
+- `BATCH` exports a fixed time window (`completed_time_from` .. `completed_time_to`) and then
+  marks the job `Completed`. This is the default and is appropriate for one-off backfills.
+- `CONTINUOUS` tails terminal instances indefinitely, sleeping between empty pages. The job
+  has no natural completion; stop it by calling `export_client.delete_job(job_id)` (or signalling
+  `mark_failed`). The orchestrator re-reads entity state at the top of each page loop, so the
+  next iteration after the delete observes the missing entity and exits cleanly.
+
+#### Listing and managing jobs
+
+Use `list_jobs(ExportJobQuery(...))` to enumerate existing jobs, optionally filtered by status
+or last-modified window:
+
+```python
+from durabletask.extensions.history_export import ExportJobQuery, ExportJobStatus
+
+for desc in export_client.list_jobs(
+    ExportJobQuery(status=[ExportJobStatus.FAILED])
+):
+    print(desc.job_id, desc.last_error)
+```
+
+Use `get_job_client(job_id)` for a per-job convenience wrapper that exposes `describe()`,
+`wait(timeout=...)`, and `delete()` directly:
+
+```python
+job_client = export_client.get_job_client(desc.job_id)
+final = job_client.wait(timeout=600)
+print(final.status.value, final.exported_instances)
+job_client.delete()
+```
+
+#### Custom destinations
+
+The Azure Blob writer is one implementation of the
+`HistoryWriter` extension point. Implement the protocol (no
+inheritance required — it's a `typing.Protocol`) to send exports to
+any destination (S3, GCS, SFTP, local filesystem, a database, etc.):
+
+```python
+from durabletask.extensions.history_export import HistoryWriter
+
+
+class LocalFileSystemHistoryWriter:
+    def __init__(self, root_dir: str) -> None:
+        self._root = root_dir
+
+    def write(
+        self,
+        *,
+        instance_id: str,
+        container: str,
+        blob_name: str,
+        payload: bytes,
+        content_type: str,
+        content_encoding: str | None,
+    ) -> None:
+        import os
+        # ``container`` is the destination's logical container name
+        # (an ExportDestination.container).  Per-job routing writers
+        # combine it with ``blob_name``; writers that pin to a fixed
+        # location at construction time may ignore it.
+        path = os.path.join(self._root, container, blob_name)
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "wb") as fp:
+            fp.write(payload)
+
+
+export_client = ExportHistoryClient(
+    dt_client, LocalFileSystemHistoryWriter("/var/exports")
+)
+```
+
+> [!TIP]
+> The bundled `AzureBlobHistoryExportWriter` lives in the optional
+> `durabletask.extensions.history_export.azure_blob` submodule and
+> requires `pip install durabletask[history-export-azure]`. The
+> core history-export package has no third-party runtime
+> dependencies — only the bundled destination does. Future
+> first-party destinations (S3, GCS, etc.) will be packaged as
+> additional optional extras using the same pattern.
+
 ### Logging configuration
 
 Both the TaskHubGrpcWorker and TaskHubGrpcClient (as well as DurableTaskSchedulerWorker and

durabletask/extensions/history_export/__init__.py

@@ -0,0 +1,80 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Orchestration history export for Durable Task.
+
+This optional extension package provides a workflow for exporting
+orchestration history from terminal instances to a configured
+destination, modeled after the durabletask-dotnet ``ExportHistory``
+package.
+
+The core building blocks (models, durable entity, activities,
+orchestrator, and the public client) live in this package and have
+no required runtime dependencies beyond the core SDK.  Specific
+destinations (for example Azure Blob Storage) may require optional
+dependencies; see the destination module documentation for details.
+"""
+
+from durabletask.extensions.history_export._constants import (
+    ENTITY_NAME,
+    ORCHESTRATOR_NAME,
+    orchestrator_instance_id_for,
+)
+from durabletask.extensions.history_export.activities import (
+    HistoryExportContext,
+)
+from durabletask.extensions.history_export.client import (
+    ExportHistoryClient,
+    ExportHistoryJobClient,
+)
+from durabletask.extensions.history_export.entity import ExportJobEntity
+from durabletask.extensions.history_export.exceptions import (
+    ExportJobError,
+    ExportJobInvalidTransitionError,
+    ExportJobNotFoundError,
+)
+from durabletask.extensions.history_export.models import (
+    STATE_SCHEMA_VERSION,
+    ExportCheckpoint,
+    ExportDestination,
+    ExportFailure,
+    ExportFilter,
+    ExportFormat,
+    ExportFormatKind,
+    ExportJobConfiguration,
+    ExportJobCreationOptions,
+    ExportJobDescription,
+    ExportJobQuery,
+    ExportJobState,
+    ExportJobStatus,
+    ExportMode,
+)
+from durabletask.extensions.history_export.writer import HistoryWriter
+
+__all__ = [
+    "ENTITY_NAME",
+    "ORCHESTRATOR_NAME",
+    "STATE_SCHEMA_VERSION",
+    "ExportCheckpoint",
+    "ExportDestination",
+    "ExportFailure",
+    "ExportFilter",
+    "ExportFormat",
+    "ExportFormatKind",
+    "ExportHistoryClient",
+    "ExportHistoryJobClient",
+    "ExportJobConfiguration",
+    "ExportJobCreationOptions",
+    "ExportJobDescription",
+    "ExportJobEntity",
+    "ExportJobError",
+    "ExportJobInvalidTransitionError",
+    "ExportJobNotFoundError",
+    "ExportJobQuery",
+    "ExportJobState",
+    "ExportJobStatus",
+    "ExportMode",
+    "HistoryExportContext",
+    "HistoryWriter",
+    "orchestrator_instance_id_for",
+]

durabletask/extensions/history_export/_constants.py

@@ -0,0 +1,34 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Stable, cross-cutting constants for the history-export extension.
+
+These constants are imported by both the entity and the orchestrator
+modules.  Keeping them in their own module avoids the circular import
+that would arise if either of those modules imported the other.
+"""
+
+from __future__ import annotations
+
+ENTITY_NAME = "ExportJobEntity"
+"""Logical name of the export-job durable entity."""
+
+ORCHESTRATOR_NAME = "export_job_orchestrator"
+"""Function-derived name of the export-job orchestrator."""
+
+ORCHESTRATOR_INSTANCE_ID_PREFIX = "export-job-"
+"""Prefix applied to deterministic orchestrator instance IDs."""
+
+
+def orchestrator_instance_id_for(job_id: str) -> str:
+    """Return the deterministic orchestrator instance ID for *job_id*.
+
+    All export-job orchestrators share a stable instance-ID pattern so
+    that public clients can reliably correlate a job ID with the
+    orchestrator driving it (for logs, monitoring, restart, etc.).
+    Matches the .NET ``ExportHistoryConstants.GetOrchestratorInstanceId``
+    pattern.
+    """
+    if not job_id:
+        raise ValueError("job_id must be a non-empty string")
+    return f"{ORCHESTRATOR_INSTANCE_ID_PREFIX}{job_id}"

durabletask/extensions/history_export/_internal.py

@@ -0,0 +1,38 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Package-internal helpers shared across the history-export modules.
+
+Names here have no leading underscore so they can be imported by sibling
+modules without tripping pyright's ``reportPrivateUsage`` check.  They
+remain package-private by convention: nothing in this module is exported
+from :mod:`durabletask.extensions.history_export.__init__`.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+
+def dt_to_iso(value: datetime | None) -> str | None:
+    """Normalize *value* to a UTC ISO-8601 string (or ``None``)."""
+    if value is None:
+        return None
+    if value.tzinfo is None:
+        value = value.replace(tzinfo=timezone.utc)
+    else:
+        value = value.astimezone(timezone.utc)
+    return value.isoformat()
+
+
+def dt_from_iso(value: str | None) -> datetime | None:
+    """Parse *value* as an ISO-8601 timestamp, defaulting naive values to UTC."""
+    if value is None:
+        return None
+    parsed = datetime.fromisoformat(value)
+    if parsed.tzinfo is None:
+        parsed = parsed.replace(tzinfo=timezone.utc)
+    return parsed
+
+
+__all__ = ["dt_from_iso", "dt_to_iso"]

durabletask/extensions/history_export/_logging.py

@@ -0,0 +1,19 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Shared logger for the history-export extension.
+
+Submodules that emit log records should import :data:`logger` from
+this module rather than calling :func:`logging.getLogger` themselves.
+This keeps every emit attributed to the same logger name so that
+callers can configure / filter the extension's output in one place.
+"""
+
+from __future__ import annotations
+
+import logging
+
+logger = logging.getLogger("durabletask.extensions.history_export")
+"""Module-wide logger for the history-export extension."""
+
+__all__ = ["logger"]