feat: add Phase 7 execution read models

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

@@ -154,7 +154,7 @@ for operator review and never used to infer workflow position.
 
 ## Phase 7 — Execution read models and operations
 
-PR: #329, rebased onto Phase 6. Status: partial.
+PR: #329, rebased onto Phase 6. Status: complete.
 
 Purpose: answer where work is, why it is waiting, and what happened without reconstructing state from Jira labels or logs.
 
@@ -169,6 +169,15 @@ Work:
 
 Exit gate: operators can diagnose and recover an execution using persisted records and APIs alone.
 
+Completion evidence: `docs/architecture/phase-7-read-models-plan.md` records the
+implementation evidence for all six work items, including durable timeline storage,
+deterministic projection rebuilds, authenticated/paginated APIs, Org Pulse's versioned
+contract, bounded operational metrics, and the read-only architecture guard. Retention
+and rollback procedures are documented there. The full local stack suite, integration
+suite, focused Ruff checks, and targeted mypy checks pass. The local Zensical build
+remains unverified because its file watcher hit the environment's `EMFILE` open-file
+limit.
+
 ## Phase 8 — Compatibility removal and final cutover
 
 PR: #330. Status: partial and intentionally last.

docs/architecture/phase-7-read-models-plan.md

@@ -0,0 +1,80 @@
+# Phase 7 implementation plan: process and execution read models
+
+**Status:** Complete.
+
+**Depends on:** Versioned process definitions, station outcomes and durable effects
+
+**Goal:** Answer operator questions from durable execution records rather than Jira
+labels or worker logs. Read models are projections only: they cannot advance a workflow
+or execute an effect.
+
+## Delivery slices
+
+1. **Execution projection.** Combine checkpoint position, pinned definition, permitted
+   commands, waiting reason, station history, external observation metadata and effects
+   into one versioned response.
+2. **Pinned-definition visibility.** Retain the canonical declarative definition with
+   checkpoints so inspection never substitutes a newer Jira project property for the
+   revision an instance actually runs.
+3. **Operator API.** Expose the projection by workflow/ticket identity with explicit
+   unavailable fields for legacy checkpoints.
+4. **Durable decision and observation history.** Persist command decisions and normalized
+   observations, including ignored/stale reasons, then include them in the timeline.
+5. **Org Pulse and metrics.** Consume the API for dashboards and measure waiting age,
+   retries, blocked causes, stale observations and migration incompatibilities.
+
+## Completion evidence
+
+The Phase 7 work items are implemented in the following boundaries:
+
+1. `src/forge/read_models/timeline.py` provides idempotent in-memory and Redis
+   timeline stores. `project_execution` rebuilds observations, command decisions,
+   transitions, station attempts, effect attempts/results, migrations, and operator
+   actions into a deterministic timeline. Coverage is in
+   `tests/unit/read_models/test_timeline_store.py` and the read-model tests.
+2. `project_execution` exposes the pinned definition, position, permitted commands,
+   waits/blocks, stale/conflicting observations, effects, recovery options, and
+   evaluated rule explanations. Legacy checkpoints expose unavailable fields rather
+   than consulting Jira.
+3. `GET /api/v1/workflows/{ticket_key}/execution` and its authenticated timeline
+   endpoint are the stable operator surface. Timeline pagination is bounded to 200
+   entries and returns a deterministic cursor. Authentication and contract behavior
+   are covered by `tests/unit/api/routes/test_executions.py`.
+4. The Org Pulse contract is `GET /api/v1/org-pulse/workflows/{ticket_key}` and the
+   versioned `OrgPulseExecution` model in `src/forge/integrations/org_pulse.py`.
+   Contract and authentication coverage is in
+   `tests/unit/integrations/test_org_pulse.py` and
+   `tests/unit/api/routes/test_org_pulse.py`.
+5. Read-model latency and waiting-age histograms plus bounded-label gauges for
+   sampled retry count, drift, blocking, and migration eligibility are defined in
+   `src/forge/api/routes/metrics.py`; recording is covered by
+   `tests/unit/api/routes/test_metrics.py`. Sampled-state gauges are deliberately
+   not counters, so repeated Org Pulse GETs do not inflate event totals. Event
+   counters remain owned by their actual decision/transition writers.
+6. `rebuild_execution_timeline` and restart-style loader coverage prove deterministic
+   reconstruction from durable checkpoint, ledger, timeline, and effect records.
+
+## Operations, retention, and rollback
+
+Read models and operator routes are inspection-only: they do not advance checkpoints,
+execute effects, or issue provider mutations. The architecture guard in
+`tests/unit/architecture/test_read_model_boundaries.py` prevents mutation calls and
+effect-execution imports from returning to those boundaries.
+
+Timeline retention is exposed as the explicit `purge_before` operation on timeline
+stores; terminal effect retention remains the explicit
+`EffectService.purge_terminal_before` operation. Pending and running effects are not
+eligible for terminal retention. Retention is therefore an operator/deployment
+operation, not an implicit action during reads, and its deletion is irreversible
+without a backup.
+
+The API and Org Pulse payloads carry `schema_version` (`1.0`). Consumers must tolerate
+additive fields and treat absent/`null` legacy fields as unavailable. A read-model
+rollback deploys the prior application version; it does not rewrite checkpoints or
+effects. If a persisted timeline format changes, take a backup and use an explicit
+rebuild/migration before re-enabling the new reader.
+
+The full local stack suite, integration suite, focused Ruff checks, and targeted mypy
+checks pass. The documentation build remains unverified because the local Zensical file
+watcher hit the environment's `Too many open files` (`EMFILE`) limit; this is recorded
+as an environment limitation, not evidence that the documentation is invalid.

docs/reference/api.md

@@ -92,6 +92,50 @@ Exposes Prometheus-format metrics for the API server.
 | `forge_ci_fix_attempts_total` | Counter | CI fix attempts |
 | `forge_agent_duration_seconds` | Histogram | Agent execution time |
 
+### Operator execution API
+
+Execution inspection is a read-only API protected by the bearer token configured
+as `FORGE_OPERATOR_TOKEN`. Requests without a configured token return `503`; an
+invalid or missing bearer token returns `401`. The token is never accepted as a
+query parameter.
+
+```http
+GET /api/v1/workflows/{ticket_key}/execution
+GET /api/v1/workflows/{ticket_key}/execution/timeline?cursor=0&limit=50
+```
+
+Execution responses are versioned with `schema_version` (`1.0`). The timeline
+uses a deterministic integer cursor and returns `next_cursor` until the end;
+clients should treat cursors as opaque offsets and request no more than 200
+entries at a time. The response is a projection of durable Forge records and
+does not consult current Jira labels.
+
+The compact contract intended for Org Pulse is:
+
+```http
+GET /api/v1/org-pulse/workflows/{ticket_key}
+```
+
+It returns the execution status, current position, waiting/blocking information,
+retry count, observation freshness/conflict state, and migration eligibility.
+Org Pulse must preserve `schema_version`, tolerate additive fields, and treat
+`null` as “not available” (for example, legacy checkpoints have no migration
+decision). This endpoint is read-only and uses the same operator token.
+
+Timeline and terminal effect records are subject to the deployment's retention
+policy. Retention must not remove pending or running effects; consumers should
+not assume an old timeline event is available forever.
+
+**Operational metrics:** `forge_read_model_latency_seconds` measures API
+latency; `forge_execution_waiting_age_seconds`,
+`forge_execution_retry_count`, `forge_execution_drift_state`,
+`forge_execution_blocked_state`, and `forge_execution_migration_eligibility`
+expose waiting age, sampled retry count, drift, blocking codes, and migration
+eligibility. The retry, drift, blocked, and migration metrics are gauges for the
+most recently sampled execution; they are not event counters and repeated GETs
+do not inflate totals. `forge_read_model_latency_seconds` and waiting age are
+request/sample histograms by design.
+
 Worker metrics are available separately at `http://localhost:8001/metrics`.
 
 ## Webhook Configuration

src/forge/api/routes/__init__.py

@@ -1,15 +1,19 @@
 """API route modules."""
 
+from forge.api.routes.executions import router as executions_router
 from forge.api.routes.github import router as github_router
 from forge.api.routes.health import router as health_router
 from forge.api.routes.jira import router as jira_router
 from forge.api.routes.metrics import router as metrics_router
+from forge.api.routes.org_pulse import router as org_pulse_router
 
 __all__ = [
+    "executions_router",
     "github_router",
     "effects_router",
     "health_router",
     "jira_router",
     "metrics_router",
+    "org_pulse_router",
 ]
 from forge.api.routes.effects import router as effects_router

src/forge/api/routes/effects.py

@@ -2,12 +2,15 @@
 
 import secrets
 from collections.abc import Sequence
+from datetime import UTC, datetime
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, Header, HTTPException, status
 
 from forge.config import get_settings
+from forge.domain import stable_identity
 from forge.effects import EffectRecord, EffectService, create_default_effect_service
+from forge.read_models import RedisExecutionTimelineStore, TimelineEntry
 
 router = APIRouter(prefix="/api/v1/effects", tags=["effects"])
 
@@ -34,6 +37,14 @@ def authorize_operator(authorization: str | None) -> None:
 EffectServiceDep = Annotated[EffectService, Depends(get_effect_service)]
 
 
+def get_timeline_store() -> RedisExecutionTimelineStore:
+    """Build the durable operator timeline adapter for mutation auditing."""
+    return RedisExecutionTimelineStore()
+
+
+TimelineStoreDep = Annotated[RedisExecutionTimelineStore, Depends(get_timeline_store)]
+
+
 @router.get("/workflow/{run_id}", response_model=list[EffectRecord])
 async def list_workflow_effects(
     run_id: str, service: EffectServiceDep, authorization: OperatorAuth = None
@@ -55,14 +66,49 @@ async def get_effect(
 
 @router.post("/{idempotency_key}/replay", response_model=EffectRecord)
 async def replay_effect(
-    idempotency_key: str, service: EffectServiceDep, authorization: OperatorAuth = None
+    idempotency_key: str,
+    service: EffectServiceDep,
+    timeline_store: TimelineStoreDep,
+    authorization: OperatorAuth = None,
 ) -> EffectRecord:
     authorize_operator(authorization)
     try:
-        return await service.replay(idempotency_key)
+        replayed = 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
+    # The timeline write happens only after the effect journal has accepted the
+    # replay mutation.  Unauthorized, missing, and rejected replays therefore
+    # cannot manufacture operator-action evidence.
+    await timeline_store.append(
+        replayed.command.workflow.run_id,
+        TimelineEntry(
+            event_id=stable_identity(
+                "operator-action",
+                {
+                    "run_id": replayed.command.workflow.run_id,
+                    "action": "effect-replay",
+                    "effect_id": replayed.command.effect_id,
+                    "replay_count": replayed.replay_count,
+                },
+            ),
+            kind="operator_action",
+            occurred_at=replayed.updated_at or datetime.now(UTC),
+            status="accepted",
+            summary="Effect replay accepted",
+            details={
+                "action": "effect-replay",
+                "effect_id": replayed.command.effect_id,
+                "idempotency_key": replayed.command.idempotency_key,
+                "operation": replayed.command.operation,
+                "target": replayed.command.target.external_id,
+                "result": replayed.status.value,
+                "result_status": replayed.status.value,
+                "replay_count": replayed.replay_count,
+            },
+        ),
+    )
+    return replayed