Document Python worker sessions and rebuilding local state

docs/features/worker-sessions.md

@@ -81,6 +81,10 @@ on a `schedule_activity` workflow-task command under `worker_session`.
 }
 ```
 
+## Python service workers {#python-service-workers}
+
+The [Python worker-session example](../polyglot/python.md#worker-sessions) creates `WorkerSessionOptions`, schedules two sequential remote activities, declares the holder capability and handles an exhausted activity failure. It also shows rebuilding disposable worker-local state after holder loss and explains why explicit close or TTL expiry needs a fresh session id. The recipe is checked with published Python SDK 2.9.0 and Server 2.11.0, including the Server's scheduler/maintenance role.
+
 ## Lifecycle
 
 Session creation is lazy. The matching layer creates or reacquires a session

docs/polyglot/python.md

@@ -1279,6 +1279,84 @@ Use `PassthroughWorkerInterceptor` as a base class when you only need one
 hook. Implement `WorkerInterceptor` directly when you want type checkers to
 force all hooks to be present.
 
+### Worker sessions {#worker-sessions}
+
+Use `WorkerSessionOptions` when sequential remote activities need one worker's local resource. This example is checked with Python SDK **2.9.0** and Server **2.11.0**; see the [portable affinity version requirements](./portable-worker-affinity.md). Use a unique `job_id` for each workflow run so a new run does not reuse a terminal session id.
+
+The cache below is disposable. Each activity can rebuild it from the durable business input, so a replacement holder does not depend on a previous process's memory:
+
+```python
+from durable_workflow import ActivityFailed, WorkerSessionOptions, activity, workflow
+
+# Disposable process-local state. Rebuild it from durable business input.
+documents = {}
+
+def ensure_document(job_id):
+    rebuilt = job_id not in documents
+    if rebuilt:
+        documents[job_id] = {"text": "document:" + job_id}
+    return documents[job_id], rebuilt
+
+@activity.defn(name="prepare_document")
+async def prepare_document(job_id):
+    document, rebuilt = ensure_document(job_id)
+    return {"holder": activity.context().info.worker_id, "rebuilt": rebuilt}
+
+@activity.defn(name="inspect_document")
+async def inspect_document(job_id):
+    document, rebuilt = ensure_document(job_id)
+    return {"holder": activity.context().info.worker_id,
+            "rebuilt": rebuilt, "text": document["text"]}
+
+@workflow.defn(name="inspect_document_workflow")
+class InspectDocumentWorkflow:
+    def run(self, ctx, job_id):
+        session = WorkerSessionOptions(
+            session_id="documents:" + job_id,
+            requirements=("fs:documents",),
+            lease_seconds=5,
+            ttl_seconds=60,
+            max_concurrent_activities=1,
+            allow_reacquire_after_failure=True,
+        )
+        options = dict(worker_session=session,
+                       start_to_close_timeout=5,
+                       schedule_to_start_timeout=12,
+                       schedule_to_close_timeout=20,
+                       retry_policy={"max_attempts": 1})
+        try:
+            prepared = yield ctx.schedule_activity("prepare_document", [job_id], **options)
+            inspected = yield ctx.schedule_activity("inspect_document", [job_id], **options)
+        except ActivityFailed as error:
+            if error.activity_type not in {"prepare_document", "inspect_document"}:
+                raise
+            return {"status": "resource_unavailable", "job_id": job_id}
+        return {"status": "inspected", "prepared": prepared, "inspected": inspected}
+```
+
+Register both activities and declare the required capability on the worker:
+
+```python
+from durable_workflow import Worker
+
+worker = Worker(
+    client,
+    task_queue="documents",
+    workflows=[InspectDocumentWorkflow],
+    activities=[prepare_document, inspect_document],
+    capabilities=["fs:documents"],
+)
+await worker.run()
+```
+
+Start `inspect_document_workflow` on the `documents` task queue with `[job_id]` as input. The first activity lazily acquires the session; the second uses the same holder while its lease remains valid. `max_concurrent_activities=1` bounds simultaneous activities in that session. Queue routing and capability checks still apply. These short leases/timeouts demonstrate bounded behavior; size them for your real work and heartbeat well before expiry.
+
+After holder loss or lease expiry, `allow_reacquire_after_failure=True` permits another eligible worker to reacquire the session. Its local resource must be rebuilt. Already committed activity results replay, while uncommitted effects can repeat: use durable resource identifiers and idempotent external operations. A session does not migrate memory, files or open handles.
+
+Explicit close and absolute TTL expiry are terminal for that session id. A later activity cannot reopen it and can remain unadmitted until its configured timeout is enforced. Run the Server's scheduler/maintenance role from the [published deployment recipe](https://github.com/durable-workflow/server/blob/2.11.0/docker-compose.published.yml); an API-only process is insufficient for timeout enforcement. The workflow handles exhausted activity failure with `ActivityFailed` and returns `resource_unavailable`; the application can then choose a fresh session and rebuild from the business input. This example does not promise an immediate session-closed exception or automatically reopen a terminal session.
+
+For explicit holder lifecycle operations, use `worker.worker_session(options)` and its `create()`, `renew()` and `close(reason)` methods. Graceful worker shutdown closes its held sessions. See the [shared lifecycle contract](../features/worker-sessions.md#lifecycle).
+
 ## Logging
 
 The SDK uses Python's standard `logging` module with structured logger names:

i18n/de/docusaurus-plugin-content-docs/current/features/worker-sessions.md

@@ -79,6 +79,10 @@ JSON-Form unter `worker_session` eines Workflow-Aufgabenbefehls
 }
 ```
 
+## Python-Service-Worker {#python-service-workers}
+
+Das [Python-Beispiel für Worker-Sessions](../polyglot/python.md#worker-sessions) erstellt `WorkerSessionOptions`, plant zwei aufeinanderfolgende Remote-Activities, deklariert die Capability des Inhabers und behandelt einen endgültigen Activity-Fehler. Es zeigt außerdem den Neuaufbau verwerfbaren lokalen Zustands nach dem Verlust des Inhabers und erklärt, warum explizites Schließen oder TTL-Ablauf eine neue Session-ID erfordert. Das Rezept wurde mit den veröffentlichten Versionen Python SDK 2.9.0 und Server 2.11.0 einschließlich der Scheduler-/Wartungsrolle des Servers geprüft.
+
 ## Lebenszyklus {#lifecycle}
 
 Sitzungen werden bei Bedarf erstellt. Die Zuordnungsschicht erstellt oder

i18n/de/docusaurus-plugin-content-docs/current/polyglot/python.md

@@ -1257,6 +1257,84 @@ Verwenden Sie `PassthroughWorkerInterceptor` als Basisklasse für nur einen
 Hook. Implementieren Sie `WorkerInterceptor` direkt, wenn Typprüfer das
 Vorhandensein aller Hooks erzwingen sollen.
 
+### Worker-Sessions {#worker-sessions}
+
+Verwenden Sie `WorkerSessionOptions`, wenn aufeinanderfolgende Remote-Activities eine lokale Ressource desselben Workers brauchen. Dieses Beispiel wurde mit Python SDK **2.9.0** und Server **2.11.0** geprüft; siehe die [Versionsanforderungen für portable Affinität](./portable-worker-affinity.md). Verwenden Sie pro Workflow-Ausführung eine eindeutige `job_id`, damit ein neuer Lauf keine Session-ID im Endzustand wiederverwendet.
+
+Der folgende Cache ist verwerfbar. Jede Activity kann ihn aus dauerhaft gespeicherten Geschäftseingaben neu aufbauen. Ein neuer Inhaber hängt daher nicht vom Speicher des vorherigen Prozesses ab:
+
+```python
+from durable_workflow import ActivityFailed, WorkerSessionOptions, activity, workflow
+
+# Disposable process-local state. Rebuild it from durable business input.
+documents = {}
+
+def ensure_document(job_id):
+    rebuilt = job_id not in documents
+    if rebuilt:
+        documents[job_id] = {"text": "document:" + job_id}
+    return documents[job_id], rebuilt
+
+@activity.defn(name="prepare_document")
+async def prepare_document(job_id):
+    document, rebuilt = ensure_document(job_id)
+    return {"holder": activity.context().info.worker_id, "rebuilt": rebuilt}
+
+@activity.defn(name="inspect_document")
+async def inspect_document(job_id):
+    document, rebuilt = ensure_document(job_id)
+    return {"holder": activity.context().info.worker_id,
+            "rebuilt": rebuilt, "text": document["text"]}
+
+@workflow.defn(name="inspect_document_workflow")
+class InspectDocumentWorkflow:
+    def run(self, ctx, job_id):
+        session = WorkerSessionOptions(
+            session_id="documents:" + job_id,
+            requirements=("fs:documents",),
+            lease_seconds=5,
+            ttl_seconds=60,
+            max_concurrent_activities=1,
+            allow_reacquire_after_failure=True,
+        )
+        options = dict(worker_session=session,
+                       start_to_close_timeout=5,
+                       schedule_to_start_timeout=12,
+                       schedule_to_close_timeout=20,
+                       retry_policy={"max_attempts": 1})
+        try:
+            prepared = yield ctx.schedule_activity("prepare_document", [job_id], **options)
+            inspected = yield ctx.schedule_activity("inspect_document", [job_id], **options)
+        except ActivityFailed as error:
+            if error.activity_type not in {"prepare_document", "inspect_document"}:
+                raise
+            return {"status": "resource_unavailable", "job_id": job_id}
+        return {"status": "inspected", "prepared": prepared, "inspected": inspected}
+```
+
+Registrieren Sie beide Activities und deklarieren Sie die erforderliche Capability am Worker:
+
+```python
+from durable_workflow import Worker
+
+worker = Worker(
+    client,
+    task_queue="documents",
+    workflows=[InspectDocumentWorkflow],
+    activities=[prepare_document, inspect_document],
+    capabilities=["fs:documents"],
+)
+await worker.run()
+```
+
+Starten Sie `inspect_document_workflow` in der Task-Queue `documents` mit `[job_id]` als Eingabe. Die erste Activity erwirbt die Session bei Bedarf; die zweite nutzt denselben Inhaber, solange die Lease gültig ist. `max_concurrent_activities=1` begrenzt gleichzeitig laufende Activities in dieser Session. Queue-Routing und Capability-Prüfungen gelten weiterhin. Diese kurzen Leases und Timeouts zeigen begrenztes Verhalten; passen Sie sie an die tatsächliche Arbeit an und senden Sie Heartbeats deutlich vor Ablauf.
+
+Nach dem Verlust des Inhabers oder Ablauf der Lease erlaubt `allow_reacquire_after_failure=True` einem anderen geeigneten Worker, die Session erneut zu erwerben. Er muss die lokale Ressource neu aufbauen. Bereits bestätigte Activity-Ergebnisse werden wiedergegeben, während unbestätigte Effekte sich wiederholen können: Verwenden Sie dauerhafte Ressourcenkennungen und idempotente externe Operationen. Eine Session migriert weder Speicher noch Dateien oder offene Handles.
+
+Explizites Schließen und der Ablauf der absoluten TTL sind für diese Session-ID endgültig. Eine spätere Activity kann sie nicht erneut öffnen und bleibt möglicherweise bis zur Durchsetzung ihres konfigurierten Timeouts unzugelassen. Führen Sie die Scheduler-/Wartungsrolle des Servers aus dem [veröffentlichten Deployment-Rezept](https://github.com/durable-workflow/server/blob/2.11.0/docker-compose.published.yml) aus; ein reiner API-Prozess reicht zur Timeout-Durchsetzung nicht aus. Der Workflow behandelt den endgültigen Activity-Fehler mit `ActivityFailed` und gibt `resource_unavailable` zurück. Die Anwendung kann dann eine neue Session wählen und die Ressource aus den Geschäftseingaben neu aufbauen. Das Beispiel verspricht weder eine sofortige Ausnahme wegen einer geschlossenen Session noch ein automatisches Wiederöffnen einer endgültig beendeten Session.
+
+Für explizite Operationen am Lebenszyklus des Inhabers verwenden Sie `worker.worker_session(options)` und dessen Methoden `create()`, `renew()` und `close(reason)`. Beim geordneten Herunterfahren schließt der Worker seine gehaltenen Sessions. Siehe den [gemeinsamen Lebenszyklusvertrag](../features/worker-sessions.md#lifecycle).
+
 ## Logging {#logging}
 
 Das SDK verwendet Pythons Standardmodul `logging` mit strukturierten Logger-Namen:

i18n/de/source-hashes.json

@@ -55,7 +55,7 @@
   "features/updates.md": "ecc24a4ae764edf0281430583216d47d4b75fd9f58d183f7dd5bc9794cbd1c06",
   "features/versioning.md": "3686d25c27aa8a4f576b1fb01c554c6d58f8fa91363d64314830a325872a1ed3",
   "features/webhooks.md": "edcdfc05ccb0d34a77c51594f85bc3cab343e787f8e0e08d2cd4ddebd048e8c8",
-  "features/worker-sessions.md": "2b08a0cdaee486970e099ea8bcb91d7f6daf135b0bc98629ed47f3885e64c77d",
+  "features/worker-sessions.md": "5065368fb5b93599dbd08d97cb9ac312576eb42fee14f081db663aa9f78fb68c",
   "how-it-works.md": "90454c37f4244297d5c11bcc59549859acf59f88d15ecd26f0a6951c115f86eb",
   "installation.md": "1e2e04a8f23414d0e81c72b166700fd6350129ab81b2f22a6205856c23741967",
   "introduction.md": "06bf96d164d402d4e376c6308b20f96d324110b2e0cf96c2a1a67b39906072e3",
@@ -82,7 +82,7 @@
   "polyglot/polling-patterns.md": "30875099a23731c6c471b481a771378acc6ff57eed2fda4405a509057534c989",
   "polyglot/portable-worker-affinity.md": "40973c6a31a76845b4f95e9a505e68ad76abd5174bbdccf37993cdebc8904959",
   "polyglot/prerelease-history-migration.md": "a57f05f492228a8a67791f9152648a1e5f80856325b78777cf63b6ef49b7413c",
-  "polyglot/python.md": "8776584d0bb4b5667f719e698cea654f53f16cf927265fae27c2930779188648",
+  "polyglot/python.md": "b0cddc8757690c94087f1d91b0edf2820e2673b834d43e8fcbbc1cf5bd6ea028",
   "polyglot/rust-cloud-quickstart.md": "2303434fc9302a81a8381ad4836423ef9952eb3201a4574c432254e3cb68c5f3",
   "polyglot/rust.md": "0bb9929cafe2751e1fde615ee033e127c5f6c6eaa8d4bbae941b4e1e86b06d58",
   "polyglot/server-api-reference.md": "e94af77a13b341d6fdc9941d4ee94e5053ab932d2e77069b238eb33351414e83",

i18n/es/docusaurus-plugin-content-docs/current/features/worker-sessions.md

@@ -71,6 +71,10 @@ La instantánea de opciones guarda el mismo contrato en `activity_options.worker
 }
 ```
 
+## Workers de servicio en Python {#python-service-workers}
+
+El [ejemplo de sesiones de worker en Python](../polyglot/python.md#worker-sessions) crea `WorkerSessionOptions`, programa dos actividades remotas secuenciales, declara la capacidad del titular y maneja el fallo definitivo de actividad. También muestra cómo reconstruir el estado local desechable tras perder al titular y explica por qué el cierre explícito o la expiración del TTL requieren un identificador de sesión nuevo. La receta se ha comprobado con Python SDK 2.9.0 y Server 2.11.0 publicados, incluido el rol de planificador/mantenimiento del Server.
+
 ## Ciclo de vida {#lifecycle}
 
 La creación es diferida. La capa de asignación crea o readquiere la sesión cuando la primera tarea de actividad de la sesión se admite en un worker capaz. Un worker también puede crearla explícitamente antes de consultar tareas mediante `POST /api/worker/sessions`.

i18n/es/docusaurus-plugin-content-docs/current/polyglot/python.md

@@ -1226,6 +1226,84 @@ worker = Worker(
 Usa `PassthroughWorkerInterceptor` como base si solo necesitas un hook. Implementa
 `WorkerInterceptor` directamente para que la comprobación de tipos exija todos.
 
+### Sesiones de worker {#worker-sessions}
+
+Use `WorkerSessionOptions` cuando varias actividades remotas secuenciales necesiten un recurso local del mismo worker. Este ejemplo se ha comprobado con Python SDK **2.9.0** y Server **2.11.0**; consulte los [requisitos de versión de afinidad portable](./portable-worker-affinity.md). Use un `job_id` único por ejecución del workflow para que una ejecución nueva no reutilice el identificador de una sesión terminal.
+
+La caché siguiente es desechable. Cada actividad puede reconstruirla a partir de la entrada de negocio duradera, de modo que el nuevo titular no dependa de la memoria del proceso anterior:
+
+```python
+from durable_workflow import ActivityFailed, WorkerSessionOptions, activity, workflow
+
+# Disposable process-local state. Rebuild it from durable business input.
+documents = {}
+
+def ensure_document(job_id):
+    rebuilt = job_id not in documents
+    if rebuilt:
+        documents[job_id] = {"text": "document:" + job_id}
+    return documents[job_id], rebuilt
+
+@activity.defn(name="prepare_document")
+async def prepare_document(job_id):
+    document, rebuilt = ensure_document(job_id)
+    return {"holder": activity.context().info.worker_id, "rebuilt": rebuilt}
+
+@activity.defn(name="inspect_document")
+async def inspect_document(job_id):
+    document, rebuilt = ensure_document(job_id)
+    return {"holder": activity.context().info.worker_id,
+            "rebuilt": rebuilt, "text": document["text"]}
+
+@workflow.defn(name="inspect_document_workflow")
+class InspectDocumentWorkflow:
+    def run(self, ctx, job_id):
+        session = WorkerSessionOptions(
+            session_id="documents:" + job_id,
+            requirements=("fs:documents",),
+            lease_seconds=5,
+            ttl_seconds=60,
+            max_concurrent_activities=1,
+            allow_reacquire_after_failure=True,
+        )
+        options = dict(worker_session=session,
+                       start_to_close_timeout=5,
+                       schedule_to_start_timeout=12,
+                       schedule_to_close_timeout=20,
+                       retry_policy={"max_attempts": 1})
+        try:
+            prepared = yield ctx.schedule_activity("prepare_document", [job_id], **options)
+            inspected = yield ctx.schedule_activity("inspect_document", [job_id], **options)
+        except ActivityFailed as error:
+            if error.activity_type not in {"prepare_document", "inspect_document"}:
+                raise
+            return {"status": "resource_unavailable", "job_id": job_id}
+        return {"status": "inspected", "prepared": prepared, "inspected": inspected}
+```
+
+Registre ambas actividades y declare la capacidad requerida en el worker:
+
+```python
+from durable_workflow import Worker
+
+worker = Worker(
+    client,
+    task_queue="documents",
+    workflows=[InspectDocumentWorkflow],
+    activities=[prepare_document, inspect_document],
+    capabilities=["fs:documents"],
+)
+await worker.run()
+```
+
+Inicie `inspect_document_workflow` en la cola de tareas `documents` con `[job_id]` como entrada. La primera actividad adquiere la sesión de forma diferida; la segunda usa el mismo titular mientras su concesión siga vigente. `max_concurrent_activities=1` limita las actividades simultáneas de esa sesión. El enrutamiento de cola y las comprobaciones de capacidad siguen aplicándose. Estas concesiones y timeouts breves demuestran un comportamiento acotado; ajústelos al trabajo real y envíe heartbeats con suficiente antelación a la expiración.
+
+Tras perder al titular o expirar la concesión, `allow_reacquire_after_failure=True` permite que otro worker apto readquiera la sesión. Debe reconstruir el recurso local. Los resultados de actividad ya confirmados se reproducen, pero los efectos no confirmados pueden repetirse: use identificadores duraderos e interacciones externas idempotentes. Una sesión no migra memoria, archivos ni descriptores abiertos.
+
+El cierre explícito y la expiración del TTL absoluto son terminales para ese identificador de sesión. Una actividad posterior no puede reabrirlo y puede permanecer sin admisión hasta que se aplique el timeout configurado. Ejecute el rol de planificador/mantenimiento del Server de la [receta de despliegue publicada](https://github.com/durable-workflow/server/blob/2.11.0/docker-compose.published.yml); un proceso solo de API no basta para aplicar los timeouts. El workflow maneja el fallo definitivo de actividad con `ActivityFailed` y devuelve `resource_unavailable`; la aplicación puede elegir una sesión nueva y reconstruir a partir de la entrada de negocio. Este ejemplo no promete una excepción inmediata de sesión cerrada ni reabre automáticamente una sesión terminal.
+
+Para gestionar explícitamente el ciclo de vida del titular, use `worker.worker_session(options)` y sus métodos `create()`, `renew()` y `close(reason)`. El apagado ordenado del worker cierra sus sesiones. Consulte el [contrato de ciclo de vida compartido](../features/worker-sessions.md#lifecycle).
+
 ## Registros {#logging}
 
 El SDK usa `logging` estándar con nombres estructurados: