Recover interrupted workflows and reject stale writes

docs/architecture/control-plane.md

@@ -72,6 +72,14 @@ Planning and review run on the host. Implementation runs in a rootless Podman
 container. Containers do not receive Jira, Redis, or source-control
 credentials; provider writes return through effects.
 
+### Durable processing and recovery
+
+Observation acceptance and workflow application are separate durable facts. A
+processing record retains unfinished work and its command/checkpoint handoff.
+Canonical ticket locks serialize execution, while ownership checks at Redis
+write boundaries reject expired writers. See [Delivery recovery and
+ownership](recovery.md) for recovery paths and their limits.
+
 ### Durable effects and read models
 
 Nodes request external writes through `EffectCommand` values. The effect journal

docs/architecture/index.md

@@ -9,6 +9,7 @@ For workflow details, see the [Feature](../guide/feature-workflow.md), [Bug](../
 |------|----------|
 | [Control-plane architecture](control-plane.md) | Ownership boundaries and safe project/core extension paths |
 | [System and components](overview.md) | Control-plane structure and component responsibilities |
+| [Delivery recovery and ownership](recovery.md) | Processing handoffs, lease fencing, recovery, and limits |
 | [Runtime internals](internals.md) | State authority, reconciliation, stations, effects, and security |
 | [Reference](reference.md) | Architectural decisions, known limitations, workflow lifecycles |
 | [Structured model output](structured-output.md) | Typed model responses and provider fallback behavior |

docs/architecture/internals.md

@@ -19,6 +19,9 @@ as a command, validated, and applied through the selected workflow's transition
 a checkpoint does not claim ownership of Jira issue content, pull-request state, or CI results; a
 new provider revision can cause those facts to be reconciled and re-evaluated.
 
+See [Delivery recovery and ownership](recovery.md) for the processing handoff,
+write fencing, and recovery guarantees.
+
 ## Delivery and concurrency
 
 Queue delivery is at least once. The observation `delivery_identity` makes equivalent webhook and

docs/architecture/recovery.md

@@ -0,0 +1,84 @@
+# Delivery recovery and ownership
+
+Forge separates observation acceptance from workflow execution. Acceptance means
+that provider evidence passed reconciliation; it does not establish that a
+command ran. Each newly accepted observation has a durable processing record,
+created in the same transaction as acceptance. Redelivery consults that record
+and resumes unfinished work using its original command evidence.
+
+```mermaid
+flowchart LR
+    Streams[Redis Streams] --> Prepare[Resolve canonical ticket]
+    Pending[Pending delivery recovery] --> Prepare
+    Retry[Scheduled retry] --> Prepare
+    Prepare --> Capacity[Bounded execution]
+    Capacity --> Lease[Ticket ownership]
+    Lease --> Accept[Observation and processing record]
+    Accept --> Handoff[Prepared command and checkpoint handoff]
+    Handoff --> Graph[Pinned workflow]
+    Graph --> Wait[Durable wait or terminal boundary]
+    Wait --> Applied[Finalize processing]
+    Applied --> Ack[Acknowledge delivery]
+    Graph --> Effects[Owned effect attempts]
+```
+
+## Ticket writes
+
+Ticket identity is resolved before taking the execution lock. Events correlated
+through a pull-request index therefore contend on the same key as Jira events
+for that ticket. Missing correlation is retried rather than running under an
+empty ticket lock.
+
+Lease loss cancels local execution. Cancellation alone is insufficient: a paused
+process can wake after another worker has acquired the ticket. The checkpoint
+Redis client also checks ownership inside a `WATCH`/`MULTI` transaction for each
+write or library-created pipeline. Redis key expiry or a replacement owner
+invalidates the transaction. This protects checkpoint JSON, pending writes,
+indexes, and latest-checkpoint pointers. A successful lock renewal may require
+retrying the transaction, but does not invalidate the owner's token.
+
+Ownership travels through asynchronous task context. An operation without ticket
+context, such as an operator maintenance command, must follow its own exclusion
+protocol. Stop workers before maintenance that rewrites checkpoints. The guarded
+checkpoint client supports Forge's single Redis deployment; it does not add
+Redis Cluster support.
+
+## Recovery boundaries
+
+A command's prepared decision and checkpoint handoff distinguish work that has
+not started from a transition already applied to the workflow. Recovery must not
+reinterpret an old approval against the workflow's newer position. Legitimate
+waits, explicit ignores, and terminal results are durable boundaries; receiving
+an observation or finding any checkpoint for a ticket is not sufficient evidence
+of completion.
+
+Blocked observations remain in the pending processing index. Later observations
+for the same ticket wait until the blocked decision is explicitly resolved;
+acknowledging its transport delivery does not remove that ordering barrier.
+
+Pending stream recovery and scheduled retries use the same bounded dispatch and
+execution protocol as new messages. Retry schedules and ownership remain
+transport state, separate from command application. A backlog remains in Redis
+when dispatch capacity is exhausted.
+
+Effects have independent, renewable claim tokens. Completing or rescheduling an
+effect requires its current token, so an expired executor cannot overwrite its
+replacement's result. The recovery service claims work immediately before
+execution rather than leasing a batch that will wait locally.
+
+## Limits
+
+These mechanisms depend on Redis retaining its durable records. Configure and
+test persistence, backups, and restoration for the required recovery objective;
+a persistent volume alone does not define the acknowledgement durability policy.
+
+A provider request already in flight cannot be revoked by losing a local lease.
+Provider idempotency or evidence-based reconciliation is still required. An
+ambiguous operation that cannot be safely reconciled needs operator attention.
+Uncheckpointed model work can repeat after a crash; the protocol does not promise
+exactly-once model execution.
+
+Legacy accepted observations have no reliable command-application marker. Treat
+missing processing records as ambiguous historical state, not automatic approval
+to replay them. Preserve the old observation and compare command, checkpoint,
+effect, and provider evidence before recovering an individual ticket.

docs/operations.md

@@ -69,6 +69,7 @@ GET  /api/v1/org-pulse/workflows/{ticket_key}
 GET  /api/v1/effects/workflow/{run_id}
 GET  /api/v1/effects/{idempotency_key}
 POST /api/v1/effects/{idempotency_key}/replay
+POST /api/v1/effects/{idempotency_key}/reconcile
 ```
 
 Start with the execution view, then use the timeline to find the accepted
@@ -88,12 +89,74 @@ and retains attempt and provider evidence under a stable idempotency key.
   closed while a required effect remains unresolved.
 - **Terminal/precondition failure:** correct the underlying provider,
   repository, or configuration problem before recovery.
+- **Indeterminate:** the provider may have accepted a write, but Forge cannot
+  safely establish the outcome. Inspect the provider before taking action;
+  automatic execution and ordinary replay do not repeat this write.
 - **Replay:** `POST .../replay` requeues an eligible terminal effect. It repeats
   only that provider mutation; it does not rerun an agent or advance the graph.
 
 Use replay only after confirming the intended write and correcting its cause.
 Do not use it as a substitute for inspecting the effect history.
 
+For an indeterminate effect, `POST .../reconcile` requires the same operator token
+and an explicit provider-side determination. Its JSON body includes `decision`
+(`applied` or `not_applied`), a required `note` describing the evidence, and an
+optional `provider_reference`. `applied` records success; `not_applied` returns
+the effect to pending execution. The journal retains reconciliation history and
+accepts this operation only for an unclaimed indeterminate record. Confirm that
+any prior request has settled before choosing `not_applied`.
+
+```json
+{
+  "decision": "applied",
+  "note": "Verified the attachment exists on the issue with the expected content.",
+  "provider_reference": "attachment-12345"
+}
+```
+
+## Upgrading the recovery protocol
+
+Drain and stop all old workers before deploying the ownership-aware worker.
+Mixed versions are unsafe: an old worker cannot enforce new claim tokens. Keep
+Redis and provider evidence intact throughout the cutover. Gateway delivery may
+continue while workers are stopped if Redis has adequate capacity.
+
+1. Snapshot Redis and record pending stream deliveries, retry schedules, running
+   effect leases, and active workflow checkpoints.
+2. Allow legacy running effect leases to expire before the new worker recovers
+   them. Retain completed effect records and their stable idempotency identities.
+3. Inspect accepted observations lacking processing records. They are ambiguous;
+   do not bulk replay them or mark them complete merely because a checkpoint
+   exists. Reconcile individual cases against command, checkpoint, and provider
+   evidence.
+4. Start the new worker at low concurrency and verify pending/retry recovery,
+   effect outcomes, and lease-loss logs using a controlled test ticket.
+5. Increase concurrency after recovery converges and outstanding work remains
+   within the configured dispatch bound.
+
+For rollback, stop the new workers first and preserve their records. Do not run
+an old writer against new active ownership records. Restore only from a reviewed
+snapshot/reconciliation procedure that accounts for provider writes made since
+that snapshot.
+
+Monitor `forge_queue_active_deliveries`, `forge_queue_buffered_deliveries`,
+`forge_queue_reclaimed_deliveries_total`, and `forge_queue_lease_losses_total`
+alongside stream pending counts. Dispatch gauges describe admitted local work;
+the backlog remains in Redis.
+
+`forge_observation_processing_pending` reports the accepted work still awaiting
+completion. `forge_observation_processing_oldest_age_seconds` advances with time
+between ledger updates, making stalled pending work visible.
+
+The counter `forge_ticket_stale_writes_rejected_total` identifies rejected stale
+checkpoint operations. Inspect ownership-loss logs alongside delivery recovery
+and effect history; a rejection protects state and does not itself establish
+that the underlying provider operation failed.
+
+See [Delivery recovery and ownership](architecture/recovery.md) for the actual
+boundaries. Recovery requires durable Redis data and can repeat model work that
+had not yet been checkpointed.
+
 ## Managing blocked workflows
 
 When Forge cannot safely continue, it adds `forge:blocked` and posts a Jira

docs/reference/config.md

@@ -503,3 +503,18 @@ Evaluate skill outputs against gold standards. See [Skill Evaluation](../dev/ski
 | `--results-dir DIR` | Batch mode | Runner output directory |
 | `--mlflow URI` | No | MLflow tracking URI |
 | `--mlflow-experiment NAME` | No | MLflow experiment name |
+
+## Queue recovery capacity
+
+`QUEUE_MAX_CONCURRENT_TASKS` controls executing handlers. The additional
+`QUEUE_MAX_OUTSTANDING_DELIVERIES` setting bounds active plus buffered deliveries
+(default 40). New stream reads reserve capacity before claiming work, leaving
+capacity for retries and recovery.
+
+`QUEUE_PENDING_IDLE_MS` sets the minimum idle time for recovering abandoned
+stream deliveries (default 60000 milliseconds). Active deliveries refresh their
+ownership; increasing this value delays recovery after a crash.
+
+`QUEUE_SHUTDOWN_TIMEOUT_SECONDS` bounds graceful draining (default 30 seconds).
+Unfinished work remains recoverable when that deadline expires. Choose the
+container termination grace period to allow this drain plus process cleanup.

src/forge/api/routes/effects.py

@@ -3,9 +3,10 @@
 import secrets
 from collections.abc import Sequence
 from datetime import UTC, datetime
-from typing import Annotated
+from typing import Annotated, Literal
 
 from fastapi import APIRouter, Depends, Header, HTTPException, status
+from pydantic import BaseModel, ConfigDict, Field
 
 from forge.config import get_settings
 from forge.domain import stable_identity
@@ -45,6 +46,14 @@ def get_timeline_store() -> RedisExecutionTimelineStore:
 TimelineStoreDep = Annotated[RedisExecutionTimelineStore, Depends(get_timeline_store)]
 
 
+class EffectReconciliationRequest(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    decision: Literal["applied", "not_applied"]
+    note: str = Field(min_length=1, max_length=2000)
+    provider_reference: str | None = None
+
+
 @router.get("/workflow/{run_id}", response_model=list[EffectRecord])
 async def list_workflow_effects(
     run_id: str, service: EffectServiceDep, authorization: OperatorAuth = None
@@ -112,3 +121,57 @@ async def replay_effect(
         ),
     )
     return replayed
+
+
+@router.post("/{idempotency_key}/reconcile", response_model=EffectRecord)
+async def reconcile_effect(
+    idempotency_key: str,
+    request: EffectReconciliationRequest,
+    service: EffectServiceDep,
+    timeline_store: TimelineStoreDep,
+    authorization: OperatorAuth = None,
+) -> EffectRecord:
+    """Record an operator's provider-side determination for an ambiguous attempt."""
+    authorize_operator(authorization)
+    try:
+        reconciled = await service.reconcile(
+            idempotency_key,
+            applied=request.decision == "applied",
+            note=request.note,
+            provider_reference=request.provider_reference,
+        )
+    except KeyError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND, detail="Effect not found"
+        ) from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc
+
+    await timeline_store.append(
+        reconciled.command.workflow.run_id,
+        TimelineEntry(
+            event_id=stable_identity(
+                "operator-action",
+                {
+                    "run_id": reconciled.command.workflow.run_id,
+                    "action": "effect-reconcile",
+                    "effect_id": reconciled.command.effect_id,
+                    "reconciliation_count": len(reconciled.reconciliation_history),
+                },
+            ),
+            kind="operator_action",
+            occurred_at=reconciled.updated_at,
+            status="accepted",
+            summary="Effect outcome reconciled",
+            details={
+                "action": "effect-reconcile",
+                "effect_id": reconciled.command.effect_id,
+                "idempotency_key": reconciled.command.idempotency_key,
+                "decision": request.decision,
+                "note": request.note,
+                "provider_reference": request.provider_reference,
+                "result_status": reconciled.status.value,
+            },
+        ),
+    )
+    return reconciled

src/forge/api/routes/executions.py

@@ -48,22 +48,44 @@ async def load_execution_read_model(
     production_defaults = checkpointer is None
     saver = checkpointer or await get_checkpointer()
     raw = await saver.aget({"configurable": {"thread_id": ticket_key}})
-    if raw is None:
+    checkpoint = dict(raw.get("channel_values", raw)) if raw is not None else {}
+    ledger = observation_ledger
+    if ledger is None and production_defaults:
+        ledger = RedisObservationLedger()
+    processing = ()
+    if ledger is not None:
+        list_processing = getattr(ledger, "processing_for_ticket", None)
+        if list_processing is not None:
+            processing = tuple(await list_processing(ticket_key))
+    if raw is None and not processing:
         return None
-    checkpoint = raw.get("channel_values", raw)
+    if raw is None:
+        latest = processing[-1]
+        pin = latest.workflow_pin or {}
+        checkpoint = {
+            "ticket_key": ticket_key,
+            "thread_id": ticket_key,
+            "current_node": "entry",
+            "is_blocked": latest.phase.value == "blocked",
+            "last_error": latest.last_error,
+            "workflow_name": pin.get("name"),
+            "workflow_definition_revision": pin.get("revision"),
+            "workflow_definition_digest": pin.get("digest"),
+            "workflow_definition": pin.get("definition"),
+        }
     definition_value = checkpoint.get("workflow_definition")
     manifest = None
     if isinstance(definition_value, dict):
         definition = load_workflow_value(definition_value)
-        if definition.digest != checkpoint.get("workflow_digest"):
+        expected_digest = checkpoint.get("workflow_digest") or checkpoint.get(
+            "workflow_definition_digest"
+        )
+        if definition.digest != expected_digest:
             raise ValueError("Pinned workflow definition digest does not match checkpoint")
         manifest = build_process_manifest(definition)
     journal = effect_journal or RedisEffectJournal()
     effects = await journal.list_for_workflow(str(checkpoint.get("thread_id") or ticket_key))
     run_id = str(checkpoint.get("thread_id") or ticket_key)
-    ledger = observation_ledger
-    if ledger is None and production_defaults:
-        ledger = RedisObservationLedger()
     decisions = ()
     if ledger is not None:
         history_for_run = getattr(ledger, "history_for_run", None)
@@ -86,6 +108,7 @@ async def load_execution_read_model(
         observation_decisions=_deduplicate_observation_decisions(
             [*checkpoint.get("observation_history", ()), *decisions]
         ),
+        observation_processing=processing,
         timeline_entries=persisted_timeline,
     )
 

src/forge/config.py

@@ -601,6 +601,30 @@ def ignored_ci_checks(self) -> list[str]:
         default=20,
         description="Maximum number of in-flight message processing tasks in the queue consumer. Prevents resource exhaustion during message bursts.",
     )
+    queue_max_outstanding_deliveries: int = Field(
+        default=40,
+        ge=2,
+        description=(
+            "Maximum number of active and buffered stream, retry, and recovered "
+            "deliveries per consumer. Stream reads reserve capacity before fetching."
+        ),
+    )
+    queue_pending_idle_ms: int = Field(
+        default=60_000,
+        ge=1,
+        description=(
+            "Minimum Redis pending-entry idle time before another consumer may "
+            "reclaim a delivery. Active consumers refresh the pending idle time."
+        ),
+    )
+    queue_shutdown_timeout_seconds: float = Field(
+        default=30.0,
+        gt=0,
+        description=(
+            "Maximum seconds to drain active queue deliveries during shutdown "
+            "before cancelling them and leaving their stream entries pending."
+        ),
+    )
 
     # Worker Metrics Configuration
     worker_metrics_port: int = Field(