feat: add Phase 3 durable effect execution

.env.example

@@ -270,6 +270,10 @@ WORKER_METRICS_ENABLED=true
 # API server uses port 8000, worker uses 8001 for metrics
 WORKER_METRICS_PORT=8001
 
+# Enables authenticated durable-effect inspection and replay endpoints.
+# Leave unset to keep /api/v1/effects disabled.
+# EFFECT_OPERATOR_TOKEN=replace-with-a-long-random-secret
+
 # =============================================================================
 # Application Configuration
 # =============================================================================

docs/architecture/internals.md

@@ -18,7 +18,10 @@ Gateway and Worker communicate only through Redis and can be deployed on separat
 
 **Idempotency:** A `DeduplicationService` exists but is not yet wired into the webhook routes. Branch creation and label operations are naturally idempotent; Jira comment posting is not.
 
-**Consistency caveat:** Checkpoint writes and external side effects (Jira comments, GitHub PRs) are not transactional. A crash between a side effect and its checkpoint write can cause duplicate actions on retry.
+**Consistency boundary:** Workflow mutations are persisted as stable effect intents before
+provider execution. Provider-specific recovery evidence and idempotent ref updates close
+the crash window between provider success and checkpoint acknowledgement. Workflow state
+and effect state remain separate durable records, joined by the workflow run identity.
 
 ## Failure and Recovery
 

docs/architecture/option-b-completion-plan.md

@@ -54,7 +54,7 @@ Exit gate: all new control-plane boundaries can be expressed with Phase 1 contra
 
 ## Phase 2 — Event interpretation and authoritative commands
 
-PR: #325. Status: partial.
+PR: #325. Status: complete.
 
 Purpose: make provider events inputs to pure interpretation, not direct selectors of graph nodes.
 
@@ -71,7 +71,7 @@ Exit gate: no provider event directly chooses a workflow node; every event produ
 
 ## Phase 3 — Durable effects
 
-PR: #326. Status: partial.
+PR: #326. Status: complete.
 
 Purpose: make every external mutation recoverable, idempotent, and observable across crashes.
 

docs/architecture/phase-3-durable-effects-plan.md

@@ -0,0 +1,60 @@
+# Phase 3 implementation plan: durable external effects
+
+**Status:** Complete
+
+**Depends on:** Phase 1 effect contracts and Phase 2 event/command boundary
+
+**Goal:** Persist external intent before calling a provider, execute it through narrow
+provider adapters, and record a durable result so recovery never requires rerunning an
+agent station merely to repeat an external write.
+
+## Delivery slices
+
+1. **Journal and leasing.** Store `EffectCommand` records by stable idempotency key,
+   index them by workflow run, atomically claim due work, recover expired leases and
+   retain terminal results.
+2. **Executor runtime.** Resolve provider-neutral operations through a registry, apply
+   bounded exponential retry, and turn exceptions into structured `EffectResult`
+   records.
+3. **First end-to-end effect.** Route Jira resume acknowledgements through the journal.
+   Embed the idempotency key in the provider object so a crash after the provider write
+   but before the result write is recoverable without duplication.
+4. **Effect migration.** Convert remaining direct Jira and source-control writes by
+   operation family, preserving their existing behavior and preconditions.
+5. **Operational surface.** Add pending/retry/terminal metrics, administrative replay,
+   retention policy and workflow-level effect history to the operator API.
+
+## Correctness rules
+
+- A command is durable before an executor is called.
+- One logical write has one stable idempotency key across retries and duplicate events.
+- Provider calls either support native idempotency or leave searchable recovery evidence.
+- Executors do not advance workflow position; reducers consume successful results.
+- A retry resumes the effect, not the station that produced it.
+- Terminal and precondition failures remain inspectable and are never silently replayed.
+
+## Completion evidence
+
+- Workflow Jira and source-control writes pass through provider-neutral durable ports;
+  architecture tests reject imports or registry access that bypass those ports.
+- Repository ref pushes are journalled using the intended commit SHA. Recovery treats an
+  already-pushed ref as success and prevents an older pending effect from overwriting a
+  newer local commit.
+- The production worker binds one Redis-backed effect service and pinned workflow
+  identity around every graph invocation. Local station execution uses the same ports
+  with an isolated in-memory conformance journal.
+- Jira comments, labels, descriptions, fields, attachments, transitions, issue creation,
+  links, archival, error notices, source-control branches/files/change requests/comments,
+  and repository pushes have idempotent executors and stable identities.
+- Required mutations fail closed before the workflow invocation can commit its next
+  checkpoint. Attempt history, retry state, replay count, provider references, and
+  terminal failures remain durable and inspectable.
+- `GET /api/v1/effects/{idempotency_key}`, `GET
+  /api/v1/effects/workflow/{run_id}`, and `POST
+  /api/v1/effects/{idempotency_key}/replay` provide the operational surface. They are
+  disabled unless `EFFECT_OPERATOR_TOKEN` is configured and require its bearer token.
+- Prometheus reports effect attempts, results by status, and operator replays. Terminal
+  retention is exposed by the journal/service API and does not delete pending work.
+- Crash-window tests cover expired leases, duplicate submission, provider-success
+  recovery, stale push supersession, retry history, explicit replay, and terminal
+  retention.

src/forge/api/routes/__init__.py

@@ -7,7 +7,9 @@
 
 __all__ = [
     "github_router",
+    "effects_router",
     "health_router",
     "jira_router",
     "metrics_router",
 ]
+from forge.api.routes.effects import router as effects_router

src/forge/api/routes/effects.py

@@ -0,0 +1,68 @@
+"""Authenticated operational API for durable external effects."""
+
+import secrets
+from collections.abc import Sequence
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Header, HTTPException, status
+
+from forge.config import get_settings
+from forge.effects import EffectRecord, EffectService, create_default_effect_service
+
+router = APIRouter(prefix="/api/v1/effects", tags=["effects"])
+
+
+def get_effect_service() -> EffectService:
+    return create_default_effect_service()
+
+
+def authorize_operator(authorization: str | None) -> None:
+    configured = get_settings().effect_operator_token
+    if configured is None:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="Effect operator API is disabled",
+        )
+    scheme, _, supplied = (authorization or "").partition(" ")
+    if scheme.lower() != "bearer" or not secrets.compare_digest(
+        supplied, configured.get_secret_value()
+    ):
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Unauthorized")
+
+
+OperatorAuth = Annotated[str | None, Header(alias="Authorization")]
+EffectServiceDep = Annotated[EffectService, Depends(get_effect_service)]
+
+
+@router.get("/workflow/{run_id}", response_model=list[EffectRecord])
+async def list_workflow_effects(
+    run_id: str, service: EffectServiceDep, authorization: OperatorAuth = None
+) -> Sequence[EffectRecord]:
+    authorize_operator(authorization)
+    return await service.journal.list_for_workflow(run_id)
+
+
+@router.get("/{idempotency_key}", response_model=EffectRecord)
+async def get_effect(
+    idempotency_key: str, service: EffectServiceDep, authorization: OperatorAuth = None
+) -> EffectRecord:
+    authorize_operator(authorization)
+    record = await service.journal.get(idempotency_key)
+    if record is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Effect not found")
+    return record
+
+
+@router.post("/{idempotency_key}/replay", response_model=EffectRecord)
+async def replay_effect(
+    idempotency_key: str, service: EffectServiceDep, authorization: OperatorAuth = None
+) -> EffectRecord:
+    authorize_operator(authorization)
+    try:
+        return await service.replay(idempotency_key)
+    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

src/forge/api/routes/metrics.py

@@ -148,6 +148,24 @@
     buckets=[1, 5, 10, 30, 60, 120, 300, 600],  # Same as AGENT_DURATION
 )
 
+EFFECT_ATTEMPTS = Counter(
+    "forge_effect_attempts_total",
+    "Durable external effect attempts",
+    ["operation"],
+)
+
+EFFECT_RESULTS = Counter(
+    "forge_effect_results_total",
+    "Durable external effect results",
+    ["operation", "status"],
+)
+
+EFFECT_REPLAYS = Counter(
+    "forge_effect_replays_total",
+    "Operator-requested durable effect replays",
+    ["operation"],
+)
+
 
 @router.get("/metrics")
 async def metrics() -> Response:
@@ -228,6 +246,18 @@ def record_revision_requested(stage: str) -> None:
     REVISIONS_REQUESTED.labels(stage=stage).inc()
 
 
+def record_effect_attempt(operation: str) -> None:
+    EFFECT_ATTEMPTS.labels(operation=operation).inc()
+
+
+def record_effect_result(operation: str, status: str) -> None:
+    EFFECT_RESULTS.labels(operation=operation, status=status).inc()
+
+
+def record_effect_replay(operation: str) -> None:
+    EFFECT_REPLAYS.labels(operation=operation).inc()
+
+
 def record_proposal_review_decision(artifact_type: str, disposition: str) -> None:
     """Record one semantic proposal-review thread decision."""
     PROPOSAL_REVIEW_DECISIONS.labels(

src/forge/config.py

@@ -602,6 +602,13 @@ def ignored_ci_checks(self) -> list[str]:
         default=True,
         description="Enable Prometheus metrics endpoint in worker",
     )
+    effect_operator_token: SecretStr | None = Field(
+        default=None,
+        description=(
+            "Bearer token for durable-effect inspection and replay endpoints. "
+            "The endpoints remain disabled when unset."
+        ),
+    )
 
     # OpenTelemetry Configuration
     otlp_endpoint: str = Field(

src/forge/effects/__init__.py

@@ -0,0 +1,20 @@
+"""Durable execution boundary for external side effects."""
+
+from forge.effects.defaults import create_default_effect_service
+from forge.effects.executors import EffectExecutor, EffectExecutorRegistry
+from forge.effects.journal import EffectJournal, InMemoryEffectJournal, RedisEffectJournal
+from forge.effects.models import EffectRecord, EffectRecordStatus
+from forge.effects.service import EffectService, RequiredEffectError
+
+__all__ = [
+    "EffectExecutor",
+    "EffectExecutorRegistry",
+    "EffectJournal",
+    "EffectRecord",
+    "EffectRecordStatus",
+    "EffectService",
+    "RequiredEffectError",
+    "InMemoryEffectJournal",
+    "RedisEffectJournal",
+    "create_default_effect_service",
+]

src/forge/effects/defaults.py

@@ -0,0 +1,16 @@
+"""Default durable effect runtime wiring."""
+
+from forge.effects.executors import EffectExecutorRegistry
+from forge.effects.jira import register_jira_executors
+from forge.effects.journal import RedisEffectJournal
+from forge.effects.repository import register_repository_executors
+from forge.effects.service import EffectService
+from forge.effects.source_control import register_source_control_executors
+
+
+def create_default_effect_service() -> EffectService:
+    registry = EffectExecutorRegistry()
+    register_jira_executors(registry)
+    register_source_control_executors(registry)
+    register_repository_executors(registry)
+    return EffectService(RedisEffectJournal(), registry)

src/forge/effects/executors.py

@@ -0,0 +1,29 @@
+"""Provider executor contracts and operation registry."""
+
+from __future__ import annotations
+
+from typing import Protocol
+
+from forge.domain import EffectCommand, EffectResult
+
+
+class EffectExecutor(Protocol):
+    operation: str
+
+    async def execute(self, command: EffectCommand) -> EffectResult: ...
+
+
+class EffectExecutorRegistry:
+    def __init__(self) -> None:
+        self._executors: dict[str, EffectExecutor] = {}
+
+    def register(self, executor: EffectExecutor) -> None:
+        if executor.operation in self._executors:
+            raise ValueError(f"Executor already registered for {executor.operation}")
+        self._executors[executor.operation] = executor
+
+    def resolve(self, operation: str) -> EffectExecutor:
+        try:
+            return self._executors[operation]
+        except KeyError as exc:
+            raise ValueError(f"No effect executor registered for {operation}") from exc