From 63a979ff2683d75c6f6833634e8a0e6d556e1508 Mon Sep 17 00:00:00 2001 From: eshulman2 Date: Thu, 27 Aug 2026 21:02:03 +0300 Subject: [PATCH 1/5] feat: add workflow execution read model API --- docs/architecture/phase-7-read-models-plan.md | 31 +++ src/forge/api/routes/__init__.py | 2 + src/forge/api/routes/executions.py | 49 +++++ src/forge/main.py | 2 + src/forge/read_models/__init__.py | 6 + src/forge/read_models/execution.py | 185 ++++++++++++++++++ src/forge/read_models/models.py | 85 ++++++++ tests/unit/api/routes/test_executions.py | 54 +++++ tests/unit/read_models/__init__.py | 1 + tests/unit/read_models/test_execution.py | 155 +++++++++++++++ 10 files changed, 570 insertions(+) create mode 100644 docs/architecture/phase-7-read-models-plan.md create mode 100644 src/forge/api/routes/executions.py create mode 100644 src/forge/read_models/__init__.py create mode 100644 src/forge/read_models/execution.py create mode 100644 src/forge/read_models/models.py create mode 100644 tests/unit/api/routes/test_executions.py create mode 100644 tests/unit/read_models/__init__.py create mode 100644 tests/unit/read_models/test_execution.py diff --git a/docs/architecture/phase-7-read-models-plan.md b/docs/architecture/phase-7-read-models-plan.md new file mode 100644 index 00000000..780daf64 --- /dev/null +++ b/docs/architecture/phase-7-read-models-plan.md @@ -0,0 +1,31 @@ +# Phase 7 implementation plan: process and execution read models + +**Status:** In progress + +**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. + +## Current slice + +This PR implements slices 1–3 and the read-side contracts required by slices 4–5. It +uses existing checkpoint and Phase 3 effect records. Legacy checkpoints remain readable; +they explicitly report when their canonical definition or observation history predates +the read model instead of guessing from current Jira state. diff --git a/src/forge/api/routes/__init__.py b/src/forge/api/routes/__init__.py index 2f606fa9..4184c292 100644 --- a/src/forge/api/routes/__init__.py +++ b/src/forge/api/routes/__init__.py @@ -1,11 +1,13 @@ """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 __all__ = [ + "executions_router", "github_router", "effects_router", "health_router", diff --git a/src/forge/api/routes/executions.py b/src/forge/api/routes/executions.py new file mode 100644 index 00000000..6825a509 --- /dev/null +++ b/src/forge/api/routes/executions.py @@ -0,0 +1,49 @@ +"""Operator read API for durable workflow execution state.""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, HTTPException + +from forge.effects import RedisEffectJournal +from forge.orchestrator.checkpointer import get_checkpointer +from forge.read_models import ExecutionReadModel, project_execution +from forge.workflow.declarative.loader import load_workflow_value +from forge.workflow.declarative.manifest import build_process_manifest + +router = APIRouter(prefix="/api/v1/workflows", tags=["workflows"]) + + +async def load_execution_read_model( + ticket_key: str, + *, + checkpointer: Any = None, + effect_journal: Any = None, +) -> ExecutionReadModel | None: + saver = checkpointer or await get_checkpointer() + raw = await saver.aget({"configurable": {"thread_id": ticket_key}}) + if raw is None: + return None + checkpoint = raw.get("channel_values", raw) + 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"): + 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)) + return project_execution(checkpoint, effects=effects, manifest=manifest) + + +@router.get("/{ticket_key}/execution", response_model=ExecutionReadModel) +async def get_execution(ticket_key: str) -> ExecutionReadModel: + try: + model = await load_execution_read_model(ticket_key) + except ValueError as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + if model is None: + raise HTTPException(status_code=404, detail=f"Workflow {ticket_key} was not found") + return model diff --git a/src/forge/main.py b/src/forge/main.py index 75c7d0fc..c206d51d 100644 --- a/src/forge/main.py +++ b/src/forge/main.py @@ -14,6 +14,7 @@ from forge.api.middleware.correlation import CorrelationIdMiddleware from forge.api.routes import ( effects_router, + executions_router, github_router, health_router, jira_router, @@ -147,6 +148,7 @@ def create_app() -> FastAPI: app.include_router(effects_router) app.include_router(jira_router) app.include_router(github_router) + app.include_router(executions_router) return app diff --git a/src/forge/read_models/__init__.py b/src/forge/read_models/__init__.py new file mode 100644 index 00000000..2979b97d --- /dev/null +++ b/src/forge/read_models/__init__.py @@ -0,0 +1,6 @@ +"""Operator-facing projections over durable workflow records.""" + +from forge.read_models.execution import project_execution +from forge.read_models.models import ExecutionReadModel + +__all__ = ["ExecutionReadModel", "project_execution"] diff --git a/src/forge/read_models/execution.py b/src/forge/read_models/execution.py new file mode 100644 index 00000000..b18ef38c --- /dev/null +++ b/src/forge/read_models/execution.py @@ -0,0 +1,185 @@ +"""Pure projection from durable execution records to an operator view.""" + +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from datetime import UTC, datetime, timedelta +from typing import Any + +from forge.domain import Observation +from forge.effects import EffectRecord +from forge.read_models.models import ( + DefinitionView, + EffectView, + ExecutionReadModel, + ExecutionStatus, + MigrationView, + NextTransitionView, + ObservationView, + StationAttemptView, + WaitingView, +) +from forge.workflow.declarative.manifest import ProcessChangeImpact, ProcessManifest + + +def project_execution( + checkpoint: Mapping[str, Any], + *, + effects: Sequence[EffectRecord] = (), + manifest: ProcessManifest | None = None, + last_observation: Observation | None = None, + migration: ProcessChangeImpact | None = None, + now: datetime | None = None, + stale_after: timedelta = timedelta(hours=1), +) -> ExecutionReadModel: + """Build a read-only explanation without consulting Jira labels or logs.""" + now = now or datetime.now(UTC) + ticket_key = str(checkpoint.get("ticket_key") or checkpoint.get("thread_id") or "unknown") + run_id = str(checkpoint.get("thread_id") or ticket_key) + position = str(checkpoint.get("current_node") or "entry") + status = _status(checkpoint, position) + waiting = _waiting(checkpoint, status) + permitted = _permitted_commands(status, position) + transitions = tuple( + NextTransitionView(outcome=item.outcome, target=item.target) + for item in (manifest.transitions if manifest else ()) + if item.source == position + ) + definition = DefinitionView( + name=str(checkpoint.get("workflow_name") or checkpoint.get("ticket_type") or "legacy"), + revision=int(checkpoint.get("workflow_revision") or 1), + digest=checkpoint.get("workflow_digest"), + available=manifest is not None, + manifest=manifest.model_dump(mode="json") if manifest else None, + ) + observation_view = _observation(last_observation, checkpoint, now, stale_after) + return ExecutionReadModel( + run_id=run_id, + ticket_key=ticket_key, + status=status, + current_position=position, + definition=definition, + permitted_commands=permitted, + next_transitions=transitions, + waiting=waiting, + last_observation=observation_view, + station_attempts=_station_attempts(checkpoint), + effects=tuple(_effect(record) for record in effects), + migration=MigrationView( + eligible=migration.compatible_for_in_flight if migration else None, + incompatibilities=( + (*migration.missing_resume_mappings, *migration.notes) if migration else () + ), + ), + ) + + +def _status(checkpoint: Mapping[str, Any], position: str) -> ExecutionStatus: + if position in {"complete", "__end__"}: + return ExecutionStatus.COMPLETED + if checkpoint.get("is_blocked"): + return ExecutionStatus.BLOCKED + if checkpoint.get("last_error"): + return ExecutionStatus.FAILED + if checkpoint.get("is_paused"): + return ExecutionStatus.WAITING + return ExecutionStatus.RUNNING + + +def _waiting(checkpoint: Mapping[str, Any], status: ExecutionStatus) -> WaitingView | None: + updated_at = _datetime(checkpoint.get("updated_at")) + if status is ExecutionStatus.BLOCKED: + return WaitingView( + code="blocked", + message=str(checkpoint.get("last_error") or "Workflow requires operator intervention"), + since=updated_at, + recovery="Resolve the blocking condition, then issue retry or cancel.", + ) + if status is ExecutionStatus.FAILED: + return WaitingView( + code="failed", + message=str(checkpoint.get("last_error")), + since=updated_at, + recovery="Inspect the failed station/effect and issue retry or cancel.", + ) + if status is ExecutionStatus.WAITING: + return WaitingView( + code="gate", + message=f"Waiting at {checkpoint.get('current_node') or 'an approval gate'}", + since=updated_at, + recovery="Provide an eligible approval, rejection, question, retry, or cancel command.", + ) + return None + + +def _permitted_commands(status: ExecutionStatus, position: str) -> tuple[str, ...]: + if status is ExecutionStatus.COMPLETED: + return () + if status in {ExecutionStatus.BLOCKED, ExecutionStatus.FAILED}: + return ("retry", "cancel") + if status is ExecutionStatus.WAITING: + commands = ["resume", "retry", "cancel"] + if position.endswith("_gate"): + commands[0:0] = ["approve", "reject"] + return tuple(commands) + return ("synchronize", "cancel") + + +def _observation( + observation: Observation | None, + checkpoint: Mapping[str, Any], + now: datetime, + stale_after: timedelta, +) -> ObservationView: + if observation is None: + return ObservationView( + available=False, + conflicting=bool(checkpoint.get("external_state_conflict")), + ) + observed_at = observation.observed_at + comparable_now = now if now.tzinfo else now.replace(tzinfo=UTC) + comparable_observed = observed_at if observed_at.tzinfo else observed_at.replace(tzinfo=UTC) + return ObservationView( + observation_id=observation.observation_id, + source_system=observation.source_system, + observed_at=observation.observed_at, + stale=comparable_now - comparable_observed > stale_after, + conflicting=bool(checkpoint.get("external_state_conflict")), + available=True, + ) + + +def _station_attempts(checkpoint: Mapping[str, Any]) -> tuple[StationAttemptView, ...]: + return tuple( + StationAttemptView( + station_name=str(item.get("station_name") or "unknown"), + invocation_id=str(item.get("invocation_id") or "unknown"), + attempt=int(item.get("attempt") or 1), + status=str(item.get("status") or "unknown"), + completed_at=_datetime(item.get("completed_at")), + reason=item.get("reason"), + ) + for item in checkpoint.get("station_history") or [] + ) + + +def _effect(record: EffectRecord) -> EffectView: + result = record.result + return EffectView( + effect_id=record.command.effect_id, + operation=record.command.operation, + target=record.command.target.external_id, + status=record.status.value, + attempt=record.attempt, + updated_at=record.updated_at, + provider_reference=result.provider_reference if result else None, + error=result.error_message if result else None, + ) + + +def _datetime(value: Any) -> datetime | None: + if isinstance(value, datetime): + return value + if isinstance(value, str) and value: + return datetime.fromisoformat(value) + return None diff --git a/src/forge/read_models/models.py b/src/forge/read_models/models.py new file mode 100644 index 00000000..d0c5f82d --- /dev/null +++ b/src/forge/read_models/models.py @@ -0,0 +1,85 @@ +"""Versioned, execution-neutral operator read models.""" + +from __future__ import annotations + +from datetime import datetime +from enum import StrEnum + +from forge.domain import JsonValue, VersionedDomainModel + + +class ExecutionStatus(StrEnum): + RUNNING = "running" + WAITING = "waiting" + BLOCKED = "blocked" + FAILED = "failed" + COMPLETED = "completed" + + +class DefinitionView(VersionedDomainModel): + name: str + revision: int + digest: str | None = None + available: bool + manifest: dict[str, JsonValue] | None = None + + +class WaitingView(VersionedDomainModel): + code: str + message: str + since: datetime | None = None + recovery: str | None = None + + +class NextTransitionView(VersionedDomainModel): + outcome: str | None = None + target: str + + +class ObservationView(VersionedDomainModel): + observation_id: str | None = None + source_system: str | None = None + observed_at: datetime | None = None + stale: bool | None = None + conflicting: bool = False + available: bool + + +class StationAttemptView(VersionedDomainModel): + station_name: str + invocation_id: str + attempt: int + status: str + completed_at: datetime | None = None + reason: str | None = None + + +class EffectView(VersionedDomainModel): + effect_id: str + operation: str + target: str + status: str + attempt: int + updated_at: datetime + provider_reference: str | None = None + error: str | None = None + + +class MigrationView(VersionedDomainModel): + eligible: bool | None = None + incompatibilities: tuple[str, ...] = () + + +class ExecutionReadModel(VersionedDomainModel): + run_id: str + ticket_key: str + status: ExecutionStatus + current_position: str + definition: DefinitionView + permitted_commands: tuple[str, ...] + next_transitions: tuple[NextTransitionView, ...] + waiting: WaitingView | None = None + last_observation: ObservationView + station_attempts: tuple[StationAttemptView, ...] = () + effects: tuple[EffectView, ...] = () + migration: MigrationView = MigrationView() diff --git a/tests/unit/api/routes/test_executions.py b/tests/unit/api/routes/test_executions.py new file mode 100644 index 00000000..6f92a5f7 --- /dev/null +++ b/tests/unit/api/routes/test_executions.py @@ -0,0 +1,54 @@ +from unittest.mock import AsyncMock + +import pytest + +from forge.api.routes.executions import load_execution_read_model + + +@pytest.mark.asyncio +async def test_load_execution_read_model_uses_pinned_definition_and_effect_history() -> None: + definition = { + "apiVersion": "forge/v1", + "kind": "Workflow", + "metadata": {"name": "short-feature", "revision": 1}, + "spec": { + "state": "feature", + "entry": "generate_prd", + "steps": {"generate_prd": {"next": "__end__"}}, + }, + } + from forge.workflow.declarative.loader import load_workflow_value + + digest = load_workflow_value(definition).digest + checkpointer = AsyncMock() + checkpointer.aget.return_value = { + "channel_values": { + "thread_id": "FORGE-1", + "ticket_key": "FORGE-1", + "workflow_name": "short-feature", + "workflow_revision": 1, + "workflow_digest": digest, + "workflow_definition": definition, + "current_node": "generate_prd", + } + } + journal = AsyncMock() + journal.list_for_workflow.return_value = [] + + model = await load_execution_read_model( + "FORGE-1", checkpointer=checkpointer, effect_journal=journal + ) + + assert model is not None + assert model.definition.available is True + assert model.definition.manifest is not None + assert model.definition.manifest["digest"] == digest + journal.list_for_workflow.assert_awaited_once_with("FORGE-1") + + +@pytest.mark.asyncio +async def test_load_execution_read_model_returns_none_for_unknown_workflow() -> None: + checkpointer = AsyncMock() + checkpointer.aget.return_value = None + + assert await load_execution_read_model("MISSING-1", checkpointer=checkpointer) is None diff --git a/tests/unit/read_models/__init__.py b/tests/unit/read_models/__init__.py new file mode 100644 index 00000000..427fbb5e --- /dev/null +++ b/tests/unit/read_models/__init__.py @@ -0,0 +1 @@ +"""Tests for operator read models.""" diff --git a/tests/unit/read_models/test_execution.py b/tests/unit/read_models/test_execution.py new file mode 100644 index 00000000..96d325a2 --- /dev/null +++ b/tests/unit/read_models/test_execution.py @@ -0,0 +1,155 @@ +from datetime import UTC, datetime, timedelta + +from forge.domain import ( + EffectCommand, + EffectResult, + EffectResultStatus, + Observation, + ObservationSource, + ResourceIdentity, + WorkflowIdentity, +) +from forge.effects import EffectRecord, EffectRecordStatus +from forge.read_models.execution import project_execution +from forge.read_models.models import ExecutionStatus +from forge.workflow.declarative.loader import load_workflow_value +from forge.workflow.declarative.manifest import build_process_manifest + +NOW = datetime(2026, 8, 27, 12, tzinfo=UTC) + + +def _manifest(): + definition = load_workflow_value( + { + "apiVersion": "forge/v1", + "kind": "Workflow", + "metadata": {"name": "feature-flow", "revision": 2}, + "spec": { + "state": "feature", + "entry": "generate_prd", + "steps": { + "generate_prd": {"next": "prd_approval_gate"}, + "prd_approval_gate": { + "route": "route_prd_approval", + "branches": { + "approved": "__end__", + "revise": "generate_prd", + }, + }, + }, + }, + } + ) + return build_process_manifest(definition) + + +def _effect() -> EffectRecord: + workflow = WorkflowIdentity( + run_id="FORGE-1", workflow_name="feature-flow", definition_revision=2 + ) + command = EffectCommand( + effect_id="effect-1", + idempotency_key="effect-1", + workflow=workflow, + operation="jira.comment.create", + target=ResourceIdentity(resource_type="issue", external_id="FORGE-1"), + ) + result = EffectResult( + effect_id="effect-1", + idempotency_key="effect-1", + status=EffectResultStatus.SUCCEEDED, + completed_at=NOW, + provider_reference="comment-7", + ) + return EffectRecord( + command=command, + status=EffectRecordStatus.SUCCEEDED, + attempt=1, + created_at=NOW, + updated_at=NOW, + next_attempt_at=NOW, + result=result, + ) + + +def test_waiting_instance_explains_position_commands_and_next_transitions() -> None: + checkpoint = { + "thread_id": "FORGE-1", + "ticket_key": "FORGE-1", + "workflow_name": "feature-flow", + "workflow_revision": 2, + "workflow_digest": _manifest().digest, + "current_node": "prd_approval_gate", + "is_paused": True, + "updated_at": NOW.isoformat(), + } + + model = project_execution(checkpoint, effects=[_effect()], manifest=_manifest(), now=NOW) + + assert model.status is ExecutionStatus.WAITING + assert model.waiting is not None + assert model.waiting.code == "gate" + assert model.permitted_commands == ("approve", "reject", "resume", "retry", "cancel") + assert {(item.outcome, item.target) for item in model.next_transitions} == { + ("approved", "__end__"), + ("revise", "generate_prd"), + } + assert model.effects[0].provider_reference == "comment-7" + + +def test_blocked_instance_has_recovery_without_logs() -> None: + model = project_execution( + { + "ticket_key": "FORGE-1", + "current_node": "implement_work", + "is_blocked": True, + "last_error": "Required repository credential is unavailable", + } + ) + + assert model.status is ExecutionStatus.BLOCKED + assert model.waiting is not None + assert model.waiting.message == "Required repository credential is unavailable" + assert model.permitted_commands == ("retry", "cancel") + assert model.definition.available is False + + +def test_observation_staleness_is_explicit() -> None: + observation = Observation( + observation_id="observation-1", + source=ObservationSource.POLLER, + source_system="github", + resource=ResourceIdentity(resource_type="change_request", external_id="repo#1"), + observed_at=NOW - timedelta(hours=2), + received_at=NOW - timedelta(hours=2), + ) + + model = project_execution( + {"ticket_key": "FORGE-1", "current_node": "ci_evaluator", "is_paused": True}, + last_observation=observation, + now=NOW, + ) + + assert model.last_observation.available is True + assert model.last_observation.stale is True + + +def test_station_history_is_projected_without_complete_checkpoint_state() -> None: + model = project_execution( + { + "ticket_key": "FORGE-1", + "current_node": "setup_workspace", + "station_history": [ + { + "station_name": "task-routing", + "invocation_id": "invocation-1", + "attempt": 1, + "status": "succeeded", + "completed_at": NOW.isoformat(), + } + ], + } + ) + + assert model.station_attempts[0].station_name == "task-routing" + assert model.station_attempts[0].status == "succeeded" From b81b005bd0e98034a865dc73dcfe136ce40b1489 Mon Sep 17 00:00:00 2001 From: eshulman2 Date: Thu, 27 Aug 2026 21:02:03 +0300 Subject: [PATCH 2/5] feat: retain station attempt evidence in checkpoints --- src/forge/workflow/base.py | 1 + src/forge/workflow/reducers/common.py | 27 +++++++++++++++++++ .../workflow/reducers/implementation_input.py | 3 ++- src/forge/workflow/reducers/task_routing.py | 5 +++- .../workflow/stations/test_task_routing.py | 8 +++--- .../test_implementation_input_station.py | 1 + 6 files changed, 39 insertions(+), 6 deletions(-) diff --git a/src/forge/workflow/base.py b/src/forge/workflow/base.py index bd12dceb..98b88d44 100644 --- a/src/forge/workflow/base.py +++ b/src/forge/workflow/base.py @@ -141,6 +141,7 @@ class BaseState(TypedDict, total=False): workflow_project_key: str workflow_transition_count: int workflow_node_attempts: dict[str, int] + station_history: list[dict[str, Any]] # Generic node-contract capabilities and durable precondition audit trail. # Missing capability keys preserve legacy inference; explicit booleans are diff --git a/src/forge/workflow/reducers/common.py b/src/forge/workflow/reducers/common.py index b6bfcf93..cfcc722c 100644 --- a/src/forge/workflow/reducers/common.py +++ b/src/forge/workflow/reducers/common.py @@ -32,3 +32,30 @@ def validate_station_outcome( request.contract_version, ): raise ValueError("Station outcome contract does not match its request") + + +def append_station_attempt( + state: Mapping[str, Any], + request: StationRequest[InputT], + outcome: StationOutcome[OutputT], +) -> list[dict[str, Any]]: + """Append compact durable evidence without retaining full station payloads.""" + history = list(state.get("station_history") or []) + record = { + "station_name": request.invocation.station_name, + "invocation_id": request.invocation.invocation_id, + "attempt": request.attempt, + "status": outcome.status.value, + "completed_at": outcome.completed_at.isoformat(), + "reason": outcome.reason, + } + for index, existing in enumerate(history): + if ( + existing.get("invocation_id") == request.invocation.invocation_id + and existing.get("attempt") == request.attempt + ): + history[index] = record + break + else: + history.append(record) + return history diff --git a/src/forge/workflow/reducers/implementation_input.py b/src/forge/workflow/reducers/implementation_input.py index acd2a376..6e2c5c4e 100644 --- a/src/forge/workflow/reducers/implementation_input.py +++ b/src/forge/workflow/reducers/implementation_input.py @@ -4,7 +4,7 @@ from typing import Any from forge.domain import StationOutcome, StationOutcomeStatus, StationRequest -from forge.workflow.reducers.common import validate_station_outcome +from forge.workflow.reducers.common import append_station_attempt, validate_station_outcome from forge.workflow.stations.implementation_input import ImplementationInput, ImplementationOutput @@ -27,6 +27,7 @@ def reduce_implementation_input( previous if previous and previous.get("status") == "completed" else work_unit ) return { + "station_history": append_station_attempt(state, request, outcome), "artifacts": list(artifacts_by_id.values()), "work_units": list(units_by_id.values()), "current_work_unit_id": work_unit["id"], diff --git a/src/forge/workflow/reducers/task_routing.py b/src/forge/workflow/reducers/task_routing.py index 6e2f6c4e..92141371 100644 --- a/src/forge/workflow/reducers/task_routing.py +++ b/src/forge/workflow/reducers/task_routing.py @@ -6,7 +6,7 @@ from typing import Any from forge.domain import StationOutcome, StationOutcomeStatus, StationRequest -from forge.workflow.reducers.common import validate_station_outcome +from forge.workflow.reducers.common import append_station_attempt, validate_station_outcome from forge.workflow.stations.task_routing import ( RepositoryAggregationInput, RepositoryAggregationOutput, @@ -25,12 +25,14 @@ def reduce_task_routing( raise ValueError("Task-routing station returned no output") if outcome.status is StationOutcomeStatus.BLOCKED: return { + "station_history": append_station_attempt(state, request, outcome), "last_error": outcome.reason or "No tasks available for routing", "current_node": "route_tasks", } if outcome.status is not StationOutcomeStatus.SUCCEEDED: raise ValueError(f"Task-routing station did not succeed: {outcome.status}") return { + "station_history": append_station_attempt(state, request, outcome), "repos_to_process": list(outcome.output.repositories), "current_repo": outcome.output.first_repository, "repos_completed": [], @@ -49,6 +51,7 @@ def reduce_repository_aggregation( if outcome.status is not StationOutcomeStatus.SUCCEEDED or outcome.output is None: raise ValueError(f"Repository aggregation did not succeed: {outcome.status}") return { + "station_history": append_station_attempt(state, request, outcome), "pr_urls": list(outcome.output.pull_request_urls), "repos_completed": list(outcome.output.completed_repositories), "implemented_tasks": list(outcome.output.implemented_tasks), diff --git a/tests/unit/workflow/stations/test_task_routing.py b/tests/unit/workflow/stations/test_task_routing.py index 297454d4..c13fd5dc 100644 --- a/tests/unit/workflow/stations/test_task_routing.py +++ b/tests/unit/workflow/stations/test_task_routing.py @@ -58,6 +58,7 @@ def test_reducer_owns_legacy_topology_mapping() -> None: assert update["current_node"] == "setup_workspace" assert update["current_repo"] == "acme/api" assert set(update) == { + "station_history", "repos_to_process", "current_repo", "repos_completed", @@ -77,10 +78,9 @@ def test_empty_mapping_returns_structured_blocked_outcome() -> None: assert outcome.status is StationOutcomeStatus.BLOCKED assert outcome.failure is not None assert outcome.failure.code == "no_tasks" - assert update == { - "last_error": "No tasks available for routing", - "current_node": "route_tasks", - } + assert update["last_error"] == "No tasks available for routing" + assert update["current_node"] == "route_tasks" + assert update["station_history"][0]["status"] == "blocked" def test_stale_outcome_is_rejected() -> None: diff --git a/tests/unit/workflow/test_implementation_input_station.py b/tests/unit/workflow/test_implementation_input_station.py index a88eda70..a068bff2 100644 --- a/tests/unit/workflow/test_implementation_input_station.py +++ b/tests/unit/workflow/test_implementation_input_station.py @@ -63,6 +63,7 @@ def test_reducer_owns_only_documented_checkpoint_fields() -> None: update = reduce_implementation_input({"unrelated": "preserved"}, station_request, outcome) assert set(update) == { + "station_history", "artifacts", "work_units", "current_work_unit_id", From 2ed855902a6ba2ddecda4e657792ee7f59680d9a Mon Sep 17 00:00:00 2001 From: eshulman2 Date: Thu, 27 Aug 2026 21:50:49 +0300 Subject: [PATCH 3/5] feat: expose durable execution timeline --- src/forge/api/routes/executions.py | 39 ++++++++++-- src/forge/config.py | 4 ++ src/forge/read_models/__init__.py | 4 +- src/forge/read_models/execution.py | 72 ++++++++++++++++++++++ src/forge/read_models/models.py | 18 ++++++ src/forge/workflow/base.py | 1 + src/forge/workflow/declarative/compiler.py | 19 ++++++ tests/unit/read_models/test_execution.py | 44 +++++++++++++ 8 files changed, 195 insertions(+), 6 deletions(-) diff --git a/src/forge/api/routes/executions.py b/src/forge/api/routes/executions.py index 6825a509..f3859155 100644 --- a/src/forge/api/routes/executions.py +++ b/src/forge/api/routes/executions.py @@ -2,19 +2,30 @@ from __future__ import annotations -from typing import Any +import secrets +from typing import Annotated, Any -from fastapi import APIRouter, HTTPException +from fastapi import APIRouter, Depends, Header, HTTPException, Query +from forge.config import get_settings from forge.effects import RedisEffectJournal from forge.orchestrator.checkpointer import get_checkpointer -from forge.read_models import ExecutionReadModel, project_execution +from forge.read_models import ExecutionReadModel, TimelinePage, project_execution from forge.workflow.declarative.loader import load_workflow_value from forge.workflow.declarative.manifest import build_process_manifest router = APIRouter(prefix="/api/v1/workflows", tags=["workflows"]) +def require_operator(authorization: Annotated[str | None, Header()] = None) -> None: + configured = get_settings().forge_operator_token.get_secret_value() + if not configured: + raise HTTPException(status_code=503, detail="Operator API token is not configured") + scheme, _, supplied = (authorization or "").partition(" ") + if scheme.lower() != "bearer" or not secrets.compare_digest(supplied, configured): + raise HTTPException(status_code=401, detail="Invalid operator credentials") + + async def load_execution_read_model( ticket_key: str, *, @@ -39,7 +50,9 @@ async def load_execution_read_model( @router.get("/{ticket_key}/execution", response_model=ExecutionReadModel) -async def get_execution(ticket_key: str) -> ExecutionReadModel: +async def get_execution( + ticket_key: str, _authorized: Annotated[None, Depends(require_operator)] +) -> ExecutionReadModel: try: model = await load_execution_read_model(ticket_key) except ValueError as exc: @@ -47,3 +60,21 @@ async def get_execution(ticket_key: str) -> ExecutionReadModel: if model is None: raise HTTPException(status_code=404, detail=f"Workflow {ticket_key} was not found") return model + + +@router.get("/{ticket_key}/execution/timeline", response_model=TimelinePage) +async def get_execution_timeline( + ticket_key: str, + _authorized: Annotated[None, Depends(require_operator)], + cursor: Annotated[int, Query(ge=0)] = 0, + limit: Annotated[int, Query(ge=1, le=200)] = 50, +) -> TimelinePage: + model = await load_execution_read_model(ticket_key) + if model is None: + raise HTTPException(status_code=404, detail=f"Workflow {ticket_key} was not found") + end = min(cursor + limit, len(model.timeline)) + return TimelinePage( + items=model.timeline[cursor:end], + next_cursor=end if end < len(model.timeline) else None, + total=len(model.timeline), + ) diff --git a/src/forge/config.py b/src/forge/config.py index 33519cc3..88a7cc67 100644 --- a/src/forge/config.py +++ b/src/forge/config.py @@ -430,6 +430,10 @@ def model_policy_resolver(self): default=False, description="Disable /docs, /redoc, and /openapi.json endpoints", ) + forge_operator_token: SecretStr = Field( + default=SecretStr(""), + description="Bearer token required by workflow execution/operator APIs", + ) @property def skills_install_dir(self) -> Path: diff --git a/src/forge/read_models/__init__.py b/src/forge/read_models/__init__.py index 2979b97d..eaee0f64 100644 --- a/src/forge/read_models/__init__.py +++ b/src/forge/read_models/__init__.py @@ -1,6 +1,6 @@ """Operator-facing projections over durable workflow records.""" from forge.read_models.execution import project_execution -from forge.read_models.models import ExecutionReadModel +from forge.read_models.models import ExecutionReadModel, TimelinePage -__all__ = ["ExecutionReadModel", "project_execution"] +__all__ = ["ExecutionReadModel", "TimelinePage", "project_execution"] diff --git a/src/forge/read_models/execution.py b/src/forge/read_models/execution.py index b18ef38c..a2b7c1ed 100644 --- a/src/forge/read_models/execution.py +++ b/src/forge/read_models/execution.py @@ -17,6 +17,7 @@ NextTransitionView, ObservationView, StationAttemptView, + TimelineEntry, WaitingView, ) from forge.workflow.declarative.manifest import ProcessChangeImpact, ProcessManifest @@ -71,6 +72,7 @@ def project_execution( (*migration.missing_resume_mappings, *migration.notes) if migration else () ), ), + timeline=_timeline(checkpoint, effects), ) @@ -183,3 +185,73 @@ def _datetime(value: Any) -> datetime | None: if isinstance(value, str) and value: return datetime.fromisoformat(value) return None + + +def _timeline( + checkpoint: Mapping[str, Any], effects: Sequence[EffectRecord] +) -> tuple[TimelineEntry, ...]: + entries: list[TimelineEntry] = [] + for item in checkpoint.get("command_decisions") or []: + entries.append( + TimelineEntry( + event_id=str(item.get("decision_id") or item.get("command_id") or "command"), + kind="command_decision", + occurred_at=_datetime(item.get("decided_at")), + status=item.get("status"), + summary=str(item.get("reason") or "Command evaluated"), + details={ + key: value + for key, value in { + "command_id": item.get("command_id"), + "command_type": item.get("command_type"), + "observation_id": item.get("observation_id"), + }.items() + if value is not None + }, + ) + ) + for item in checkpoint.get("transition_history") or []: + entries.append( + TimelineEntry( + event_id=str(item.get("transition_id") or "transition"), + kind="transition", + occurred_at=_datetime(item.get("occurred_at")), + status="committed", + summary=f"{item.get('source', 'unknown')} → {item.get('target', 'unknown')}", + details={"source": str(item.get("source")), "target": str(item.get("target"))}, + ) + ) + for item in checkpoint.get("station_history") or []: + entries.append( + TimelineEntry( + event_id=str(item.get("invocation_id") or "station"), + kind="station_attempt", + occurred_at=_datetime(item.get("completed_at")), + status=str(item.get("status") or "unknown"), + summary=f"Station {item.get('station_name', 'unknown')} attempt {item.get('attempt', 1)}", + details={"reason": str(item["reason"])} if item.get("reason") else {}, + ) + ) + for record in effects: + entries.append( + TimelineEntry( + event_id=record.command.effect_id, + kind="effect", + occurred_at=record.updated_at, + status=record.status.value, + summary=f"{record.command.operation} on {record.command.target.external_id}", + details={ + "attempt": record.attempt, + "idempotency_key": record.command.idempotency_key, + "replay_count": record.replay_count, + }, + ) + ) + entries.sort( + key=lambda entry: ( + entry.occurred_at or datetime.min.replace(tzinfo=UTC), + entry.kind, + entry.event_id, + ) + ) + return tuple(entries) diff --git a/src/forge/read_models/models.py b/src/forge/read_models/models.py index d0c5f82d..c393139d 100644 --- a/src/forge/read_models/models.py +++ b/src/forge/read_models/models.py @@ -5,6 +5,8 @@ from datetime import datetime from enum import StrEnum +from pydantic import Field + from forge.domain import JsonValue, VersionedDomainModel @@ -70,6 +72,21 @@ class MigrationView(VersionedDomainModel): incompatibilities: tuple[str, ...] = () +class TimelineEntry(VersionedDomainModel): + event_id: str + kind: str + occurred_at: datetime | None = None + status: str | None = None + summary: str + details: dict[str, JsonValue] = Field(default_factory=dict) + + +class TimelinePage(VersionedDomainModel): + items: tuple[TimelineEntry, ...] + next_cursor: int | None = None + total: int = 0 + + class ExecutionReadModel(VersionedDomainModel): run_id: str ticket_key: str @@ -83,3 +100,4 @@ class ExecutionReadModel(VersionedDomainModel): station_attempts: tuple[StationAttemptView, ...] = () effects: tuple[EffectView, ...] = () migration: MigrationView = MigrationView() + timeline: tuple[TimelineEntry, ...] = () diff --git a/src/forge/workflow/base.py b/src/forge/workflow/base.py index 98b88d44..f46259b9 100644 --- a/src/forge/workflow/base.py +++ b/src/forge/workflow/base.py @@ -141,6 +141,7 @@ class BaseState(TypedDict, total=False): workflow_project_key: str workflow_transition_count: int workflow_node_attempts: dict[str, int] + transition_history: list[dict[str, Any]] station_history: list[dict[str, Any]] # Generic node-contract capabilities and durable precondition audit trail. diff --git a/src/forge/workflow/declarative/compiler.py b/src/forge/workflow/declarative/compiler.py index e7cc50e1..877f96fc 100644 --- a/src/forge/workflow/declarative/compiler.py +++ b/src/forge/workflow/declarative/compiler.py @@ -9,6 +9,7 @@ from langgraph.graph import END, StateGraph from langgraph.types import Send +from forge.domain import stable_identity from forge.workflow.declarative.capabilities import ( KNOWN_EFFECT_CAPABILITIES, effect_capability_scope, @@ -336,10 +337,28 @@ async def run(state: dict[str, Any]) -> dict[str, Any]: (result.get("last_error"), result.get("is_paused"), result.get("is_blocked")) ): result = {**result, "current_node": "complete", "is_paused": False} + target = str(result.get("current_node") or node_name) + occurred_at = str(result.get("updated_at") or state.get("updated_at") or "") + transition = { + "transition_id": stable_identity( + "workflow-transition", + { + "run_id": state.get("thread_id") or state.get("ticket_key"), + "count": transitions, + "source": node_name, + "target": target, + }, + ), + "source": node_name, + "target": target, + "occurred_at": occurred_at, + } + history = list(state.get("transition_history") or []) return { **result, "workflow_transition_count": transitions, "workflow_node_attempts": attempts, + "transition_history": [*history, transition], } run.__name__ = f"declarative_{node_name}" diff --git a/tests/unit/read_models/test_execution.py b/tests/unit/read_models/test_execution.py index 96d325a2..470f9391 100644 --- a/tests/unit/read_models/test_execution.py +++ b/tests/unit/read_models/test_execution.py @@ -153,3 +153,47 @@ def test_station_history_is_projected_without_complete_checkpoint_state() -> Non assert model.station_attempts[0].station_name == "task-routing" assert model.station_attempts[0].status == "succeeded" + + +def test_timeline_combines_durable_decisions_transitions_stations_and_effects() -> None: + model = project_execution( + { + "ticket_key": "FORGE-1", + "current_node": "generate_prd", + "command_decisions": [ + { + "decision_id": "decision-1", + "decided_at": (NOW - timedelta(minutes=3)).isoformat(), + "status": "accepted", + "reason": "eligible signal", + "command_type": "start", + } + ], + "transition_history": [ + { + "transition_id": "transition-1", + "source": "entry", + "target": "generate_prd", + "occurred_at": (NOW - timedelta(minutes=2)).isoformat(), + } + ], + "station_history": [ + { + "station_name": "prd-generation", + "invocation_id": "station-1", + "attempt": 1, + "status": "succeeded", + "completed_at": (NOW - timedelta(minutes=1)).isoformat(), + } + ], + }, + effects=[_effect()], + now=NOW, + ) + + assert [entry.kind for entry in model.timeline] == [ + "command_decision", + "transition", + "station_attempt", + "effect", + ] From 8604f1b396c15c558ca0f7356b09afef740cfa72 Mon Sep 17 00:00:00 2001 From: eshulman2 Date: Fri, 28 Aug 2026 00:29:20 +0300 Subject: [PATCH 4/5] Complete execution read models and operations --- docs/architecture/option-b-completion-plan.md | 11 +- docs/architecture/phase-7-read-models-plan.md | 61 +- docs/reference/api.md | 44 ++ src/forge/api/routes/__init__.py | 2 + src/forge/api/routes/effects.py | 50 +- src/forge/api/routes/executions.py | 73 ++- src/forge/api/routes/metrics.py | 110 ++++ src/forge/api/routes/org_pulse.py | 33 ++ src/forge/integrations/org_pulse.py | 67 +++ src/forge/main.py | 2 + src/forge/read_models/__init__.py | 40 +- src/forge/read_models/execution.py | 536 +++++++++++++++++- src/forge/read_models/models.py | 45 ++ src/forge/read_models/timeline.py | 225 ++++++++ src/forge/reconciliation/__init__.py | 2 + src/forge/reconciliation/ledger.py | 41 ++ src/forge/workflow/base.py | 5 + tests/unit/api/routes/test_executions.py | 63 ++ tests/unit/api/routes/test_metrics.py | 58 ++ tests/unit/api/routes/test_org_pulse.py | 27 + tests/unit/api/test_effects.py | 70 ++- .../test_read_model_boundaries.py | 72 +++ tests/unit/integrations/test_org_pulse.py | 30 + tests/unit/read_models/test_execution.py | 80 +++ tests/unit/read_models/test_timeline_store.py | 181 ++++++ tests/unit/reconciliation/test_ledger.py | 23 +- 26 files changed, 1909 insertions(+), 42 deletions(-) create mode 100644 src/forge/api/routes/org_pulse.py create mode 100644 src/forge/integrations/org_pulse.py create mode 100644 src/forge/read_models/timeline.py create mode 100644 tests/unit/api/routes/test_org_pulse.py create mode 100644 tests/unit/architecture/test_read_model_boundaries.py create mode 100644 tests/unit/integrations/test_org_pulse.py create mode 100644 tests/unit/read_models/test_timeline_store.py diff --git a/docs/architecture/option-b-completion-plan.md b/docs/architecture/option-b-completion-plan.md index 06aacba7..c8f4fd9f 100644 --- a/docs/architecture/option-b-completion-plan.md +++ b/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. diff --git a/docs/architecture/phase-7-read-models-plan.md b/docs/architecture/phase-7-read-models-plan.md index 780daf64..8e0d0d3f 100644 --- a/docs/architecture/phase-7-read-models-plan.md +++ b/docs/architecture/phase-7-read-models-plan.md @@ -1,6 +1,6 @@ # Phase 7 implementation plan: process and execution read models -**Status:** In progress +**Status:** Complete. **Depends on:** Versioned process definitions, station outcomes and durable effects @@ -23,9 +23,58 @@ or execute an effect. 5. **Org Pulse and metrics.** Consume the API for dashboards and measure waiting age, retries, blocked causes, stale observations and migration incompatibilities. -## Current slice +## Completion evidence -This PR implements slices 1–3 and the read-side contracts required by slices 4–5. It -uses existing checkpoint and Phase 3 effect records. Legacy checkpoints remain readable; -they explicitly report when their canonical definition or observation history predates -the read model instead of guessing from current Jira state. +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. diff --git a/docs/reference/api.md b/docs/reference/api.md index cd1983ed..0f71d809 100644 --- a/docs/reference/api.md +++ b/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 diff --git a/src/forge/api/routes/__init__.py b/src/forge/api/routes/__init__.py index 4184c292..907a0aae 100644 --- a/src/forge/api/routes/__init__.py +++ b/src/forge/api/routes/__init__.py @@ -5,6 +5,7 @@ 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", @@ -13,5 +14,6 @@ "health_router", "jira_router", "metrics_router", + "org_pulse_router", ] from forge.api.routes.effects import router as effects_router diff --git a/src/forge/api/routes/effects.py b/src/forge/api/routes/effects.py index 61eab64d..392305a6 100644 --- a/src/forge/api/routes/effects.py +++ b/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 diff --git a/src/forge/api/routes/executions.py b/src/forge/api/routes/executions.py index f3859155..9402c2c6 100644 --- a/src/forge/api/routes/executions.py +++ b/src/forge/api/routes/executions.py @@ -3,14 +3,22 @@ from __future__ import annotations import secrets +from time import perf_counter from typing import Annotated, Any from fastapi import APIRouter, Depends, Header, HTTPException, Query +from forge.api.routes.metrics import observe_read_model_latency, record_execution_read_model from forge.config import get_settings from forge.effects import RedisEffectJournal from forge.orchestrator.checkpointer import get_checkpointer -from forge.read_models import ExecutionReadModel, TimelinePage, project_execution +from forge.read_models import ( + ExecutionReadModel, + RedisExecutionTimelineStore, + TimelinePage, + project_execution, +) +from forge.reconciliation import RedisObservationLedger from forge.workflow.declarative.loader import load_workflow_value from forge.workflow.declarative.manifest import build_process_manifest @@ -31,7 +39,13 @@ async def load_execution_read_model( *, checkpointer: Any = None, effect_journal: Any = None, + observation_ledger: Any = None, + timeline_store: Any = None, ) -> ExecutionReadModel | None: + # Injected checkpointers are used by tests and migration tooling. Avoid + # opening external Redis adapters in those callers while the production + # route (which supplies no adapter) gets the durable stores by default. + production_defaults = checkpointer is None saver = checkpointer or await get_checkpointer() raw = await saver.aget({"configurable": {"thread_id": ticket_key}}) if raw is None: @@ -46,19 +60,72 @@ async def load_execution_read_model( manifest = build_process_manifest(definition) journal = effect_journal or RedisEffectJournal() effects = await journal.list_for_workflow(str(checkpoint.get("thread_id") or ticket_key)) - return project_execution(checkpoint, effects=effects, manifest=manifest) + 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) + if history_for_run is not None: + decisions = tuple(await history_for_run(run_id)) + + store = timeline_store + if store is None and production_defaults: + store = RedisExecutionTimelineStore() + persisted_timeline = () + if store is not None: + list_records = getattr(store, "list", None) + if list_records is not None: + persisted_timeline = tuple(await list_records(run_id)) + + return project_execution( + checkpoint, + effects=effects, + manifest=manifest, + observation_decisions=_deduplicate_observation_decisions( + [*checkpoint.get("observation_history", ()), *decisions] + ), + timeline_entries=persisted_timeline, + ) + + +def _deduplicate_observation_decisions(decisions: list[Any]) -> tuple[Any, ...]: + """Merge checkpoint and ledger history without duplicate deliveries.""" + result = [] + seen: set[tuple[str, str | None]] = set() + for item in decisions: + if isinstance(item, dict): + delivery = item.get("delivery_identity") or item.get("observation_id") + disposition = item.get("disposition") or item.get("status") + else: + delivery = getattr(item, "delivery_identity", None) + observation = getattr(item, "observation", None) + delivery = delivery or getattr(observation, "observation_id", None) + disposition = getattr(item, "disposition", None) + disposition = getattr(disposition, "value", disposition) + key = (str(delivery or ""), str(disposition) if disposition is not None else None) + if key in seen: + continue + seen.add(key) + result.append(item) + return tuple(result) @router.get("/{ticket_key}/execution", response_model=ExecutionReadModel) async def get_execution( ticket_key: str, _authorized: Annotated[None, Depends(require_operator)] ) -> ExecutionReadModel: + started = perf_counter() try: model = await load_execution_read_model(ticket_key) except ValueError as exc: + observe_read_model_latency("execution", perf_counter() - started) raise HTTPException(status_code=409, detail=str(exc)) from exc + observe_read_model_latency("execution", perf_counter() - started) if model is None: raise HTTPException(status_code=404, detail=f"Workflow {ticket_key} was not found") + record_execution_read_model(model) return model @@ -69,7 +136,9 @@ async def get_execution_timeline( cursor: Annotated[int, Query(ge=0)] = 0, limit: Annotated[int, Query(ge=1, le=200)] = 50, ) -> TimelinePage: + started = perf_counter() model = await load_execution_read_model(ticket_key) + observe_read_model_latency("timeline", perf_counter() - started) if model is None: raise HTTPException(status_code=404, detail=f"Workflow {ticket_key} was not found") end = min(cursor + limit, len(model.timeline)) diff --git a/src/forge/api/routes/metrics.py b/src/forge/api/routes/metrics.py index 24087eef..03103b47 100644 --- a/src/forge/api/routes/metrics.py +++ b/src/forge/api/routes/metrics.py @@ -166,6 +166,51 @@ ["operation"], ) +# Execution read-model metrics. These deliberately use bounded labels (status, +# drift class, and blocking code) so an issue key or arbitrary provider message +# can never create an unbounded Prometheus time series. +READ_MODEL_LATENCY = Histogram( + "forge_read_model_latency_seconds", + "Latency of authenticated execution read-model requests", + ["operation"], + buckets=[0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5], +) + +EXECUTION_WAITING_AGE = Histogram( + "forge_execution_waiting_age_seconds", + "Age of executions currently waiting for an external or operator action", + ["code"], + buckets=[60, 300, 900, 3600, 21600, 86400, 604800], +) + +EXECUTION_RETRIES = Gauge( + "forge_execution_retry_count", + "Retry count in the most recently sampled execution read model", + ["kind"], +) + +EXECUTION_DRIFT = Gauge( + "forge_execution_drift_state", + "Drift state in the most recently sampled execution read model (0 or 1)", + ["class"], +) + +EXECUTION_BLOCKED = Gauge( + "forge_execution_blocked_state", + "Blocked state in the most recently sampled execution read model (0 or 1)", + ["code"], +) + +EXECUTION_MIGRATION_ELIGIBILITY = Gauge( + "forge_execution_migration_eligibility", + "Current execution migration eligibility (1 eligible, 0 ineligible, -1 unknown)", + ["state"], +) + +_BLOCKING_CODES = ("blocked", "failed", "gate", "unknown") +_DRIFT_CLASSES = ("operator_required", "stale") +_MIGRATION_STATES = ("eligible", "ineligible", "unknown") + @router.get("/metrics") async def metrics() -> Response: @@ -258,6 +303,71 @@ def record_effect_replay(operation: str) -> None: EFFECT_REPLAYS.labels(operation=operation).inc() +def observe_read_model_latency(operation: str, duration: float) -> None: + """Observe one authenticated read-model request latency.""" + READ_MODEL_LATENCY.labels(operation=operation).observe(max(0.0, duration)) + + +def record_execution_read_model(model: object) -> None: + """Record bounded operational signals from an execution projection. + + ``model`` is intentionally accepted as an object rather than importing the + read-model package. This keeps the metrics module usable by projection and + API code without introducing an import cycle. + """ + raw_status = getattr(model, "status", "") + status = str(getattr(raw_status, "value", raw_status)) + for known_code in _BLOCKING_CODES: + EXECUTION_BLOCKED.labels(code=known_code).set(0) + waiting = getattr(model, "waiting", None) + if waiting is not None: + raw_code = str(getattr(waiting, "code", "unknown")) + code = raw_code if raw_code in _BLOCKING_CODES else "unknown" + EXECUTION_BLOCKED.labels(code=code).set(1 if status == "blocked" else 0) + since = getattr(waiting, "since", None) + if since is not None: + from datetime import UTC, datetime + + if since.tzinfo is None: + since = since.replace(tzinfo=UTC) + EXECUTION_WAITING_AGE.labels(code=code).observe( + max(0.0, (datetime.now(UTC) - since).total_seconds()) + ) + + # A projection has attempt numbers for both station and durable-effect + # work. Count only additional attempts (attempt 1 is the initial try). + retry_count = 0 + for attempt in ( + *(getattr(item, "attempt", 1) for item in getattr(model, "station_attempts", ())), + *(getattr(item, "attempt", 1) for item in getattr(model, "effects", ())), + ): + retry_count += max(0, int(attempt) - 1) + EXECUTION_RETRIES.labels(kind="execution").set(retry_count) + + observations = [getattr(model, "last_observation", None)] + observations.extend(getattr(model, "stale_observations", ())) + observations.extend(getattr(model, "conflicting_observations", ())) + drift_counts = {"operator_required": 0, "stale": 0} + for observation in observations: + if observation is not None: + if getattr(observation, "conflicting", False): + drift_counts["operator_required"] += 1 + elif getattr(observation, "stale", False): + drift_counts["stale"] += 1 + for drift_class in _DRIFT_CLASSES: + count = drift_counts[drift_class] + EXECUTION_DRIFT.labels(**{"class": drift_class}).set(1 if count else 0) + + migration = getattr(model, "migration", None) + eligible = getattr(migration, "eligible", None) + migration_state = "eligible" if eligible is True else "ineligible" if eligible is False else "unknown" + for state in _MIGRATION_STATES: + EXECUTION_MIGRATION_ELIGIBILITY.labels(state=state).set(0) + EXECUTION_MIGRATION_ELIGIBILITY.labels(state=migration_state).set( + 1 if eligible is True else 0 if eligible is False else -1 + ) + + def record_proposal_review_decision(artifact_type: str, disposition: str) -> None: """Record one semantic proposal-review thread decision.""" PROPOSAL_REVIEW_DECISIONS.labels( diff --git a/src/forge/api/routes/org_pulse.py b/src/forge/api/routes/org_pulse.py new file mode 100644 index 00000000..dd26f287 --- /dev/null +++ b/src/forge/api/routes/org_pulse.py @@ -0,0 +1,33 @@ +"""Read-only Org Pulse integration endpoint.""" + +from __future__ import annotations + +from typing import Annotated + +from fastapi import APIRouter, Depends, HTTPException + +from forge.api.routes.executions import load_execution_read_model, require_operator +from forge.integrations.org_pulse import OrgPulseExecution + +router = APIRouter(prefix="/api/v1/org-pulse", tags=["org-pulse"]) + + +@router.get( + "/workflows/{ticket_key}", + response_model=OrgPulseExecution, + summary="Get the dashboard-safe execution summary", +) +async def get_pulse_execution( + ticket_key: str, + _authorized: Annotated[None, Depends(require_operator)], +) -> OrgPulseExecution: + """Return the stable summary used by Org Pulse. + + The endpoint is intentionally authenticated and read-only. Org Pulse should + retain the ``schema_version`` field and tolerate additive fields in future + responses. + """ + model = await load_execution_read_model(ticket_key) + if model is None: + raise HTTPException(status_code=404, detail=f"Workflow {ticket_key} was not found") + return OrgPulseExecution.from_execution(model) diff --git a/src/forge/integrations/org_pulse.py b/src/forge/integrations/org_pulse.py new file mode 100644 index 00000000..acd47b8d --- /dev/null +++ b/src/forge/integrations/org_pulse.py @@ -0,0 +1,67 @@ +"""Stable, read-only contract consumed by Org Pulse dashboards. + +Org Pulse must not need to understand checkpoint internals or provider payloads. +This contract is deliberately a compact summary of the execution read model and +contains no commands or mutation affordances. +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +from pydantic import Field + +from forge.domain import VersionedDomainModel +from forge.read_models.models import ExecutionReadModel + + +class OrgPulseExecution(VersionedDomainModel): + """Dashboard-safe execution status, versioned independently of checkpoints.""" + + run_id: str + ticket_key: str + status: str + current_position: str + workflow: str + workflow_revision: int + waiting_code: str | None = None + waiting_since: datetime | None = None + blocking_reason: str | None = None + retry_count: int = Field(ge=0) + observation_available: bool + observation_stale: bool | None = None + observation_conflicting: bool + migration_eligible: bool | None = None + migration_incompatibilities: tuple[str, ...] = () + + @classmethod + def from_execution(cls, execution: ExecutionReadModel) -> OrgPulseExecution: + waiting = execution.waiting + retries = sum(max(0, item.attempt - 1) for item in execution.station_attempts) + retries += sum(max(0, item.attempt - 1) for item in execution.effects) + return cls( + run_id=execution.run_id, + ticket_key=execution.ticket_key, + status=execution.status.value, + current_position=execution.current_position, + workflow=execution.definition.name, + workflow_revision=execution.definition.revision, + waiting_code=waiting.code if waiting else None, + waiting_since=waiting.since if waiting else None, + blocking_reason=(waiting.message if waiting and execution.status.value == "blocked" else None), + retry_count=retries, + observation_available=execution.last_observation.available, + observation_stale=execution.last_observation.stale, + observation_conflicting=execution.last_observation.conflicting, + migration_eligible=execution.migration.eligible, + migration_incompatibilities=execution.migration.incompatibilities, + ) + + +def pulse_timestamp(value: datetime | None) -> str | None: + """Return a normalized timestamp for clients that serialize pulse records.""" + if value is None: + return None + if value.tzinfo is None: + value = value.replace(tzinfo=UTC) + return value.isoformat() diff --git a/src/forge/main.py b/src/forge/main.py index c206d51d..bfc27c42 100644 --- a/src/forge/main.py +++ b/src/forge/main.py @@ -19,6 +19,7 @@ health_router, jira_router, metrics_router, + org_pulse_router, ) from forge.config import get_settings from forge.integrations.source_control.registry import get_registry @@ -149,6 +150,7 @@ def create_app() -> FastAPI: app.include_router(jira_router) app.include_router(github_router) app.include_router(executions_router) + app.include_router(org_pulse_router) return app diff --git a/src/forge/read_models/__init__.py b/src/forge/read_models/__init__.py index eaee0f64..00833b56 100644 --- a/src/forge/read_models/__init__.py +++ b/src/forge/read_models/__init__.py @@ -1,6 +1,40 @@ """Operator-facing projections over durable workflow records.""" -from forge.read_models.execution import project_execution -from forge.read_models.models import ExecutionReadModel, TimelinePage +from forge.read_models.execution import project_execution, rebuild_execution_timeline +from forge.read_models.models import ( + EffectAttemptView, + ExecutionReadModel, + RecoveryOptionView, + RuleClauseView, + RuleExplanationView, + TimelineEntry, + TimelinePage, +) +from forge.read_models.timeline import ( + ExecutionTimelineStore, + InMemoryExecutionTimelineStore, + InMemoryTimelineStore, + RedisExecutionTimelineStore, + RedisTimelineStore, + TimelineStore, + timeline_entry, +) -__all__ = ["ExecutionReadModel", "TimelinePage", "project_execution"] +__all__ = [ + "ExecutionReadModel", + "EffectAttemptView", + "RecoveryOptionView", + "RuleClauseView", + "RuleExplanationView", + "TimelinePage", + "TimelineEntry", + "ExecutionTimelineStore", + "InMemoryExecutionTimelineStore", + "InMemoryTimelineStore", + "RedisExecutionTimelineStore", + "RedisTimelineStore", + "TimelineStore", + "project_execution", + "rebuild_execution_timeline", + "timeline_entry", +] diff --git a/src/forge/read_models/execution.py b/src/forge/read_models/execution.py index a2b7c1ed..8bfe8679 100644 --- a/src/forge/read_models/execution.py +++ b/src/forge/read_models/execution.py @@ -10,17 +10,22 @@ from forge.effects import EffectRecord from forge.read_models.models import ( DefinitionView, + EffectAttemptView, EffectView, ExecutionReadModel, ExecutionStatus, MigrationView, NextTransitionView, ObservationView, + RecoveryOptionView, + RuleClauseView, + RuleExplanationView, StationAttemptView, TimelineEntry, WaitingView, ) from forge.workflow.declarative.manifest import ProcessChangeImpact, ProcessManifest +from forge.workflow.preconditions import has_capability def project_execution( @@ -29,9 +34,13 @@ def project_execution( effects: Sequence[EffectRecord] = (), manifest: ProcessManifest | None = None, last_observation: Observation | None = None, + observation_decisions: Sequence[Any] = (), migration: ProcessChangeImpact | None = None, now: datetime | None = None, stale_after: timedelta = timedelta(hours=1), + migrations: Sequence[Mapping[str, Any]] = (), + operator_actions: Sequence[Mapping[str, Any]] = (), + timeline_entries: Sequence[TimelineEntry] = (), ) -> ExecutionReadModel: """Build a read-only explanation without consulting Jira labels or logs.""" now = now or datetime.now(UTC) @@ -40,7 +49,7 @@ def project_execution( position = str(checkpoint.get("current_node") or "entry") status = _status(checkpoint, position) waiting = _waiting(checkpoint, status) - permitted = _permitted_commands(status, position) + permitted = _permitted_commands(status, position, checkpoint, manifest) transitions = tuple( NextTransitionView(outcome=item.outcome, target=item.target) for item in (manifest.transitions if manifest else ()) @@ -48,12 +57,27 @@ def project_execution( ) definition = DefinitionView( name=str(checkpoint.get("workflow_name") or checkpoint.get("ticket_type") or "legacy"), - revision=int(checkpoint.get("workflow_revision") or 1), - digest=checkpoint.get("workflow_digest"), - available=manifest is not None, - manifest=manifest.model_dump(mode="json") if manifest else None, + revision=_definition_revision(checkpoint), + digest=_definition_digest(checkpoint), + available=manifest is not None or isinstance(checkpoint.get("workflow_definition"), dict), + # A pinned canonical artifact is the authoritative definition. The + # compiled manifest remains useful as a fallback for callers that only + # have inspection data (and for legacy checkpoints). + manifest=( + { + **checkpoint["workflow_definition"], + # Canonical workflow artifacts intentionally do not require a + # derived digest field; retain it in the view for clients + # that consumed the original manifest-shaped response. + **({"digest": manifest.digest} if manifest else {}), + } + if isinstance(checkpoint.get("workflow_definition"), dict) + else manifest.model_dump(mode="json") if manifest else None + ), ) - observation_view = _observation(last_observation, checkpoint, now, stale_after) + decisions = tuple(observation_decisions) or _checkpoint_observation_decisions(checkpoint) + observation_view = _observation(last_observation, checkpoint, now, stale_after, decisions) + stale_inputs, conflicting_inputs = _input_views(decisions, now, stale_after) return ExecutionReadModel( run_id=run_id, ticket_key=ticket_key, @@ -64,18 +88,52 @@ def project_execution( next_transitions=transitions, waiting=waiting, last_observation=observation_view, + stale_observations=stale_inputs, + conflicting_observations=conflicting_inputs, station_attempts=_station_attempts(checkpoint), effects=tuple(_effect(record) for record in effects), + recovery_options=_recovery_options(waiting, permitted), + explanations=_rule_explanations(checkpoint, position, manifest), migration=MigrationView( eligible=migration.compatible_for_in_flight if migration else None, incompatibilities=( (*migration.missing_resume_mappings, *migration.notes) if migration else () ), ), - timeline=_timeline(checkpoint, effects), + timeline=_timeline( + checkpoint, + effects, + decisions, + migrations=migrations, + operator_actions=operator_actions, + timeline_entries=timeline_entries, + ), ) +def _definition_revision(checkpoint: Mapping[str, Any]) -> int: + value = checkpoint.get("workflow_definition_revision", checkpoint.get("workflow_revision")) + if value is None: + definition = checkpoint.get("workflow_definition") + if isinstance(definition, Mapping): + metadata = definition.get("metadata") + if isinstance(metadata, Mapping): + value = metadata.get("revision") + value = value or 1 + try: + return int(value or 1) + except (TypeError, ValueError): + return 1 + + +def _definition_digest(checkpoint: Mapping[str, Any]) -> str | None: + value = checkpoint.get("workflow_definition_digest", checkpoint.get("workflow_digest")) + # Canonical definitions do not carry their digest; callers that have a + # compiled manifest still supply it separately. Never hash a possibly + # non-canonical mapping on the read side. + return str(value) if value else None + + def _status(checkpoint: Mapping[str, Any], position: str) -> ExecutionStatus: if position in {"complete", "__end__"}: return ExecutionStatus.COMPLETED @@ -92,10 +150,17 @@ def _waiting(checkpoint: Mapping[str, Any], status: ExecutionStatus) -> WaitingV updated_at = _datetime(checkpoint.get("updated_at")) if status is ExecutionStatus.BLOCKED: return WaitingView( - code="blocked", - message=str(checkpoint.get("last_error") or "Workflow requires operator intervention"), + code=str(checkpoint.get("wait_code") or checkpoint.get("block_code") or "blocked"), + message=str( + checkpoint.get("blocking_reason") + or checkpoint.get("last_error") + or "Workflow requires operator intervention" + ), since=updated_at, - recovery="Resolve the blocking condition, then issue retry or cancel.", + recovery=str( + checkpoint.get("recovery_reason") + or "Resolve the blocking condition, then issue retry or cancel." + ), ) if status is ExecutionStatus.FAILED: return WaitingView( @@ -106,22 +171,42 @@ def _waiting(checkpoint: Mapping[str, Any], status: ExecutionStatus) -> WaitingV ) if status is ExecutionStatus.WAITING: return WaitingView( - code="gate", - message=f"Waiting at {checkpoint.get('current_node') or 'an approval gate'}", + code=str(checkpoint.get("wait_code") or "gate"), + message=str( + checkpoint.get("waiting_reason") + or checkpoint.get("wait_reason") + or f"Waiting at {checkpoint.get('current_node') or 'an approval gate'}" + ), since=updated_at, - recovery="Provide an eligible approval, rejection, question, retry, or cancel command.", + recovery=str( + checkpoint.get("recovery_reason") + or "Provide an eligible approval, rejection, question, retry, or cancel command." + ), ) return None -def _permitted_commands(status: ExecutionStatus, position: str) -> tuple[str, ...]: +def _permitted_commands( + status: ExecutionStatus, + position: str, + checkpoint: Mapping[str, Any], + manifest: ProcessManifest | None, +) -> tuple[str, ...]: + # A persisted decision is authoritative when available. This keeps this + # read side from silently inventing commands for an unfamiliar workflow. + explicit = checkpoint.get("permitted_commands") + if isinstance(explicit, (list, tuple)): + return tuple(str(command) for command in explicit) if status is ExecutionStatus.COMPLETED: return () if status in {ExecutionStatus.BLOCKED, ExecutionStatus.FAILED}: return ("retry", "cancel") if status is ExecutionStatus.WAITING: commands = ["resume", "retry", "cancel"] - if position.endswith("_gate"): + node = next((item for item in (manifest.nodes if manifest else ()) if item.name == position), None) + # Gate-ness comes from the pinned process manifest, never from a name + # convention such as ``*_gate``. + if node is not None and node.kind.value == "gate": commands[0:0] = ["approve", "reject"] return tuple(commands) return ("synchronize", "cancel") @@ -132,6 +217,7 @@ def _observation( checkpoint: Mapping[str, Any], now: datetime, stale_after: timedelta, + decisions: Sequence[Any] = (), ) -> ObservationView: if observation is None: return ObservationView( @@ -145,10 +231,150 @@ def _observation( observation_id=observation.observation_id, source_system=observation.source_system, observed_at=observation.observed_at, - stale=comparable_now - comparable_observed > stale_after, - conflicting=bool(checkpoint.get("external_state_conflict")), + stale=_observation_is_stale(observation, decisions, comparable_now, comparable_observed, stale_after), + conflicting=bool(checkpoint.get("external_state_conflict")) + or _observation_has_disposition(observation, decisions, "conflict"), available=True, + disposition="accepted", + resource_revision=observation.resource_revision, + revision_order=observation.revision_order, + ) + + +def _decision_disposition(item: Any) -> str | None: + value = ( + item.get("disposition") + if isinstance(item, Mapping) + else getattr(item, "disposition", None) ) + return getattr(value, "value", value) + + +def _checkpoint_observation_decisions(checkpoint: Mapping[str, Any]) -> tuple[Any, ...]: + value = checkpoint.get("observation_history") or checkpoint.get("observations") or () + return tuple(value) if isinstance(value, (list, tuple)) else () + + +def _decision_observation(item: Any) -> Any: + observation = ( + item.get("observation") + if isinstance(item, Mapping) + else getattr(item, "observation", None) + ) + if isinstance(observation, Mapping): + return _MappingObservation(observation) + return observation + + +class _MappingObservation: + """Small adapter for JSON checkpoints containing flattened observations.""" + + def __init__(self, value: Mapping[str, Any]) -> None: + self._value = value + self.observation_id = str(value.get("observation_id") or "observation") + self.source_system = value.get("source_system") + self.source = value.get("source", "unknown") + self.resource_revision = value.get("resource_revision") + self.revision_order = value.get("revision_order") + self.observed_at = _datetime(value.get("observed_at")) or datetime.min.replace(tzinfo=UTC) + + +def _observation_id(observation: Any) -> str | None: + value = getattr(observation, "observation_id", None) + return str(value) if value else None + + +def _decision_observation_id(item: Any) -> str | None: + observation = _decision_observation(item) + identity = _observation_id(observation) + if identity: + return identity + value = item.get("observation_id") if isinstance(item, Mapping) else None + return str(value) if value else None + + +def _decision_delivery_identity(item: Any) -> str | None: + value = item.get("delivery_identity") if isinstance(item, Mapping) else getattr(item, "delivery_identity", None) + return str(value) if value else None + + +def _observation_has_disposition( + observation: Any, decisions: Sequence[Any], disposition: str +) -> bool: + identity = _observation_id(observation) + return any( + _decision_disposition(item) == disposition + and (_decision_observation_id(item) in {None, identity}) + for item in decisions + ) + + +def _observation_is_stale( + observation: Any, + decisions: Sequence[Any], + comparable_now: datetime, + comparable_observed: datetime, + stale_after: timedelta, +) -> bool: + return _observation_has_disposition(observation, decisions, "stale") or ( + comparable_now - comparable_observed > stale_after + ) + + +def _input_view(item: Any, now: datetime, stale_after: timedelta) -> ObservationView: + observation = _decision_observation(item) + disposition = _decision_disposition(item) + reason = item.get("reason") if isinstance(item, Mapping) else getattr(item, "reason", None) + if observation is not None: + observed_at = observation.observed_at + current_now = now if now.tzinfo else now.replace(tzinfo=UTC) + current_observed = observed_at if observed_at.tzinfo else observed_at.replace(tzinfo=UTC) + return ObservationView( + observation_id=observation.observation_id, + source_system=str(observation.source_system) if observation.source_system else None, + observed_at=observed_at, + stale=disposition == "stale" or current_now - current_observed > stale_after, + conflicting=disposition == "conflict", + available=True, + disposition=disposition, + reason=reason, + resource_revision=observation.resource_revision, + revision_order=observation.revision_order, + ) + # Checkpoint JSON may contain a flattened decision record. Keep the + # record visible even when older checkpoints cannot hydrate Observation. + return ObservationView( + observation_id=( + str(item.get("observation_id")) + if isinstance(item, Mapping) and item.get("observation_id") + else None + ), + source_system=( + str(item.get("source_system")) + if isinstance(item, Mapping) and item.get("source_system") + else None + ), + stale=disposition == "stale", + conflicting=disposition == "conflict", + available=False, + disposition=disposition, + reason=reason, + ) + + +def _input_views( + decisions: Sequence[Any], now: datetime, stale_after: timedelta +) -> tuple[tuple[ObservationView, ...], tuple[ObservationView, ...]]: + stale: list[ObservationView] = [] + conflicting: list[ObservationView] = [] + for item in decisions: + disposition = _decision_disposition(item) + view = _input_view(item, now, stale_after) + if disposition == "stale": + stale.append(view) + elif disposition == "conflict": + conflicting.append(view) + return tuple(stale), tuple(conflicting) def _station_attempts(checkpoint: Mapping[str, Any]) -> tuple[StationAttemptView, ...]: @@ -167,6 +393,31 @@ def _station_attempts(checkpoint: Mapping[str, Any]) -> tuple[StationAttemptView def _effect(record: EffectRecord) -> EffectView: result = record.result + attempts = [ + EffectAttemptView( + status=attempt.status.value, + completed_at=attempt.completed_at, + provider_reference=attempt.provider_reference, + error=attempt.error_message, + ) + for attempt in record.attempt_history + ] + # The journal stores prior outcomes in attempt_history and the latest + # outcome separately. Expose both so an operator can account for every + # provider call, including a successful final retry. + if result is not None and ( + not attempts + or attempts[-1].completed_at != result.completed_at + or attempts[-1].status != result.status.value + ): + attempts.append( + EffectAttemptView( + status=result.status.value, + completed_at=result.completed_at, + provider_reference=result.provider_reference, + error=result.error_message, + ) + ) return EffectView( effect_id=record.command.effect_id, operation=record.command.operation, @@ -176,6 +427,104 @@ def _effect(record: EffectRecord) -> EffectView: updated_at=record.updated_at, provider_reference=result.provider_reference if result else None, error=result.error_message if result else None, + attempts=tuple(attempts), + ) + + +def _recovery_options( + waiting: WaitingView | None, + permitted: Sequence[str], +) -> tuple[RecoveryOptionView, ...]: + descriptions = { + "approve": "Provide the approval required by the current gate.", + "reject": "Reject the current gate and follow its configured branch.", + "resume": "Resume execution from the persisted checkpoint.", + "synchronize": "Reconcile the latest external observations.", + "retry": "Retry the failed or blocked operation from its durable boundary.", + "cancel": "Cancel the execution without changing external state.", + } + return tuple( + RecoveryOptionView( + command=command, + description=(waiting.recovery if command == "retry" and waiting and waiting.recovery else descriptions.get(command, "Issue this permitted command.")), + ) + for command in permitted + ) + + +def _rule_explanations( + checkpoint: Mapping[str, Any], position: str, manifest: ProcessManifest | None +) -> tuple[RuleExplanationView, ...]: + """Project evaluated contract clauses, including clauses that are true. + + The durable precondition result/history is retained as evidence, while the + current clause values are evaluated against the checkpoint's explicit + capabilities (or the compatibility predicates for legacy state). + """ + profile_name = checkpoint.get("workflow_state_profile") or ( + manifest.state_profile if manifest else None + ) + contract = None + if profile_name: + try: + from forge.workflow.declarative.catalog import get_state_profile + + contract = get_state_profile(str(profile_name)).contracts.get(position) + except (KeyError, ValueError): + contract = None + persisted = checkpoint.get("precondition_result") + if contract is None and not isinstance(persisted, Mapping): + return () + + clauses: list[RuleClauseView] = [] + if contract is not None: + for requirement in contract.requires: + capability = ( + requirement.capability.value + if hasattr(requirement.capability, "value") + else str(requirement.capability) + ) + clauses.append( + RuleClauseView( + capability=capability, + satisfied=has_capability(checkpoint, requirement.capability), + on_missing=requirement.on_missing.value, + reason=requirement.reason, + ) + ) + # For custom contracts, retain false clauses recorded by the runtime even + # though this process cannot import an arbitrary project predicate. + if not clauses and isinstance(persisted, Mapping): + missing = persisted.get("missing") or () + missing_names = {str(value) for value in missing} + for name in sorted(missing_names): + clauses.append( + RuleClauseView( + capability=name, + satisfied=False, + on_missing=str(persisted.get("action")) if persisted.get("action") else None, + reason=str(persisted.get("reason")) if persisted.get("reason") else None, + ) + ) + action = persisted.get("action") if isinstance(persisted, Mapping) else None + satisfied = all(clause.satisfied for clause in clauses) if clauses else action in {None, "proceed"} + summary = ( + "All required workflow rules are satisfied." + if satisfied + else str(persisted.get("reason")) if isinstance(persisted, Mapping) and persisted.get("reason") + else "One or more required workflow rules are false." + ) + # A checkpoint can have several evaluations over time. The current + # result is the primary explanation; history is represented in timeline. + return ( + RuleExplanationView( + rule="node_preconditions", + node=position, + satisfied=satisfied, + action=str(action) if action else None, + summary=summary, + clauses=tuple(clauses), + ), ) @@ -188,9 +537,16 @@ def _datetime(value: Any) -> datetime | None: def _timeline( - checkpoint: Mapping[str, Any], effects: Sequence[EffectRecord] + checkpoint: Mapping[str, Any], + effects: Sequence[EffectRecord], + decisions: Sequence[Any] = (), + *, + migrations: Sequence[Mapping[str, Any]] = (), + operator_actions: Sequence[Mapping[str, Any]] = (), + timeline_entries: Sequence[TimelineEntry] = (), ) -> tuple[TimelineEntry, ...]: - entries: list[TimelineEntry] = [] + """Aggregate all durable execution records into a stable event stream.""" + entries: list[TimelineEntry] = list(timeline_entries) for item in checkpoint.get("command_decisions") or []: entries.append( TimelineEntry( @@ -232,7 +588,107 @@ def _timeline( details={"reason": str(item["reason"])} if item.get("reason") else {}, ) ) + for item in decisions: + observation = _decision_observation(item) + if observation is not None: + occurred_at = observation.observed_at + event_id = observation.observation_id + details = { + "source": getattr(observation.source, "value", observation.source), + "source_system": observation.source_system, + "resource_revision": observation.resource_revision, + "revision_order": observation.revision_order, + } + else: + occurred_at = _datetime(item.get("decided_at")) if isinstance(item, Mapping) else getattr(item, "decided_at", None) + event_id = str(item.get("observation_id") or "observation") if isinstance(item, Mapping) else "observation" + details = {} + disposition = _decision_disposition(item) + delivery_identity = _decision_delivery_identity(item) + if delivery_identity or disposition: + # One provider revision may legitimately have several durable + # decisions (accepted, duplicate, stale, or conflict). Include + # decision identity so projection does not collapse that audit + # history into one observation event. + event_id = ":".join( + part for part in (event_id, delivery_identity, disposition) if part + ) + reason = item.get("reason") if isinstance(item, Mapping) else getattr(item, "reason", None) + entries.append( + TimelineEntry( + event_id=event_id, + kind="observation", + occurred_at=occurred_at, + status=disposition, + summary=str(reason or "External observation evaluated"), + details={key: value for key, value in details.items() if value is not None}, + ) + ) + for item in checkpoint.get("precondition_history") or []: + entries.append( + TimelineEntry( + event_id=str(item.get("event_id") or item.get("node") or "precondition"), + kind="rule_evaluation", + occurred_at=_datetime(item.get("occurred_at") or item.get("evaluated_at")), + status=str(item.get("action") or "evaluated"), + summary=str(item.get("reason") or "Workflow rule evaluated"), + details={ + key: value + for key, value in item.items() + if key not in {"event_id", "node", "occurred_at", "evaluated_at", "action", "reason"} + }, + ) + ) + for item in [*(checkpoint.get("migration_history") or []), *migrations]: + entries.append( + TimelineEntry( + event_id=str(item.get("migration_id") or item.get("event_id") or "migration"), + kind="migration", + occurred_at=_datetime(item.get("occurred_at") or item.get("updated_at")), + status=str(item.get("status") or item.get("classification") or "recorded"), + summary=str(item.get("reason") or "Workflow definition migration evaluated"), + details={ + key: value + for key, value in item.items() + if key not in {"reason", "occurred_at", "updated_at"} + }, + ) + ) + for item in [ + *(checkpoint.get("operator_actions") or []), + *(checkpoint.get("operator_history") or []), + *operator_actions, + ]: + entries.append( + TimelineEntry( + event_id=str(item.get("action_id") or item.get("event_id") or "operator-action"), + kind="operator_action", + occurred_at=_datetime(item.get("occurred_at") or item.get("acted_at")), + status=str(item.get("status") or "recorded"), + summary=str(item.get("summary") or item.get("action") or "Operator action recorded"), + details={key: value for key, value in item.items() if key not in {"summary", "action", "occurred_at", "acted_at"}}, + ) + ) for record in effects: + # EffectResult.attempt_history is the durable source for retries. The + # summary effect remains for compatibility and represents the current + # journal record; attempt events expose each individual outcome. + for attempt, result in enumerate(record.attempt_history, start=1): + entries.append( + TimelineEntry( + event_id=f"{record.command.effect_id}:attempt:{attempt}", + kind="effect_attempt", + occurred_at=result.completed_at, + status=result.status.value, + summary=f"{record.command.operation} attempt {attempt}", + details={ + "effect_id": record.command.effect_id, + "idempotency_key": record.command.idempotency_key, + **({"provider_reference": result.provider_reference} if result.provider_reference else {}), + **({"error": result.error_message} if result.error_message else {}), + }, + ) + ) entries.append( TimelineEntry( event_id=record.command.effect_id, @@ -247,11 +703,41 @@ def _timeline( }, ) ) - entries.sort( - key=lambda entry: ( - entry.occurred_at or datetime.min.replace(tzinfo=UTC), - entry.kind, - entry.event_id, + # Records can be read from both a checkpoint and an append-only store. + # Identity-based collapse makes a rebuild idempotent. + by_id = {entry.event_id: entry for entry in entries} + return tuple( + sorted( + by_id.values(), + key=lambda entry: ( + entry.occurred_at or datetime.min.replace(tzinfo=UTC), + entry.kind, + entry.event_id, + ), ) ) - return tuple(entries) + + +def rebuild_execution_timeline( + checkpoint: Mapping[str, Any], + *, + effects: Sequence[EffectRecord] = (), + observation_decisions: Sequence[Any] = (), + migrations: Sequence[Mapping[str, Any]] = (), + operator_actions: Sequence[Mapping[str, Any]] = (), + timeline_entries: Sequence[TimelineEntry] = (), +) -> tuple[TimelineEntry, ...]: + """Rebuild the timeline solely from durable records. + + This explicit entry point is useful for audits and deterministic replay; + it intentionally does not consult Jira, provider APIs, or worker logs. + """ + decisions = tuple(observation_decisions) or _checkpoint_observation_decisions(checkpoint) + return _timeline( + checkpoint, + effects, + decisions, + migrations=migrations, + operator_actions=operator_actions, + timeline_entries=timeline_entries, + ) diff --git a/src/forge/read_models/models.py b/src/forge/read_models/models.py index c393139d..bb94f4ca 100644 --- a/src/forge/read_models/models.py +++ b/src/forge/read_models/models.py @@ -45,6 +45,46 @@ class ObservationView(VersionedDomainModel): stale: bool | None = None conflicting: bool = False available: bool + disposition: str | None = None + reason: str | None = None + resource_revision: str | None = None + revision_order: int | None = None + + +class RuleClauseView(VersionedDomainModel): + """The result of evaluating one persisted workflow rule clause. + + Both satisfied and unsatisfied clauses are retained. In particular, an + operator must be able to see which prerequisite was false rather than + reverse-engineering a reason from the current node name. + """ + + capability: str + satisfied: bool + on_missing: str | None = None + reason: str | None = None + + +class RuleExplanationView(VersionedDomainModel): + rule: str + node: str + satisfied: bool + action: str | None = None + summary: str + clauses: tuple[RuleClauseView, ...] = () + + +class RecoveryOptionView(VersionedDomainModel): + command: str + description: str + available: bool = True + + +class EffectAttemptView(VersionedDomainModel): + status: str + completed_at: datetime + provider_reference: str | None = None + error: str | None = None class StationAttemptView(VersionedDomainModel): @@ -65,6 +105,7 @@ class EffectView(VersionedDomainModel): updated_at: datetime provider_reference: str | None = None error: str | None = None + attempts: tuple[EffectAttemptView, ...] = () class MigrationView(VersionedDomainModel): @@ -97,7 +138,11 @@ class ExecutionReadModel(VersionedDomainModel): next_transitions: tuple[NextTransitionView, ...] waiting: WaitingView | None = None last_observation: ObservationView + stale_observations: tuple[ObservationView, ...] = () + conflicting_observations: tuple[ObservationView, ...] = () station_attempts: tuple[StationAttemptView, ...] = () effects: tuple[EffectView, ...] = () + recovery_options: tuple[RecoveryOptionView, ...] = () + explanations: tuple[RuleExplanationView, ...] = () migration: MigrationView = MigrationView() timeline: tuple[TimelineEntry, ...] = () diff --git a/src/forge/read_models/timeline.py b/src/forge/read_models/timeline.py new file mode 100644 index 00000000..6036bc2e --- /dev/null +++ b/src/forge/read_models/timeline.py @@ -0,0 +1,225 @@ +"""Durable execution timeline records and storage adapters. + +The workflow checkpoint is the source of truth for control state, but it is not +an event log. This module provides a small append-only boundary for the +operator timeline. Both adapters use the same idempotency and ordering rules, +which makes rebuilding a projection from a checkpoint and its records +deterministic. +""" + +from __future__ import annotations + +import asyncio +from collections.abc import Mapping, Sequence +from datetime import UTC, datetime +from typing import Any, Protocol + +from forge.domain import JsonValue +from forge.orchestrator.checkpointer import get_redis_client +from forge.read_models.models import TimelineEntry + +_PREFIX = "forge:execution-timeline:" +_EVENT_PREFIX = f"{_PREFIX}event:" + +_APPEND_SCRIPT = """ +-- The marker and list append must share one Redis atomic execution. If a +-- client disappears after SETNX, a later retry must still be able to observe +-- that the complete operation committed (or retry the complete operation if +-- the script did not commit). +if redis.call('SETNX', KEYS[1], ARGV[1]) == 1 then + redis.call('RPUSH', KEYS[2], ARGV[1]) + return 1 +end +return 0 +""" + + +def _sort_key(entry: TimelineEntry) -> tuple[datetime, str, str]: + occurred = entry.occurred_at + if occurred is None: + occurred = datetime.min.replace(tzinfo=UTC) + elif occurred.tzinfo is None: + occurred = occurred.replace(tzinfo=UTC) + return occurred, entry.kind, entry.event_id + + +class ExecutionTimelineStore(Protocol): + """Durable append-only storage for normalized timeline entries.""" + + async def append(self, run_id: str, entry: TimelineEntry) -> bool: ... + + async def append_many(self, run_id: str, entries: Sequence[TimelineEntry]) -> int: ... + + async def list(self, run_id: str) -> Sequence[TimelineEntry]: ... + + async def purge_before(self, cutoff: datetime) -> int: ... + + +class InMemoryExecutionTimelineStore: + """Deterministic adapter used by projection and contract tests.""" + + def __init__(self) -> None: + self._entries: dict[str, dict[str, TimelineEntry]] = {} + self._lock = asyncio.Lock() + + async def append(self, run_id: str, entry: TimelineEntry) -> bool: + async with self._lock: + bucket = self._entries.setdefault(str(run_id), {}) + if entry.event_id in bucket: + return False + bucket[entry.event_id] = entry + return True + + async def append_many(self, run_id: str, entries: Sequence[TimelineEntry]) -> int: + added = 0 + async with self._lock: + bucket = self._entries.setdefault(str(run_id), {}) + for entry in entries: + if entry.event_id in bucket: + continue + bucket[entry.event_id] = entry + added += 1 + return added + + async def list(self, run_id: str) -> Sequence[TimelineEntry]: + async with self._lock: + return tuple(sorted(self._entries.get(str(run_id), {}).values(), key=_sort_key)) + + async def purge_before(self, cutoff: datetime) -> int: + removed = 0 + async with self._lock: + for run_id, bucket in list(self._entries.items()): + stale = [ + event_id + for event_id, entry in bucket.items() + if entry.occurred_at is not None and entry.occurred_at < cutoff + ] + for event_id in stale: + del bucket[event_id] + removed += 1 + if not bucket: + self._entries.pop(run_id, None) + return removed + + +class RedisExecutionTimelineStore: + """Redis adapter with atomic, idempotent appends. + + Entries are kept in a per-run list for inexpensive reads and in an event + key for deduplication. Ordering is applied after decoding, so retries and + out-of-order writers produce the same projection. + """ + + def __init__(self, redis_client: Any = None) -> None: + self._redis = redis_client + + async def _client(self) -> Any: + if self._redis is None: + self._redis = await get_redis_client() + return self._redis + + @staticmethod + def _run_key(run_id: str) -> str: + return f"{_PREFIX}{run_id}" + + @staticmethod + def _event_key(run_id: str, event_id: str) -> str: + return f"{_EVENT_PREFIX}{run_id}:{event_id}" + + async def append(self, run_id: str, entry: TimelineEntry) -> bool: + redis = await self._client() + event_key = self._event_key(run_id, entry.event_id) + encoded = entry.model_dump_json() + # A Lua script makes the idempotency marker and per-run append one + # atomic Redis operation. There is no crash window in which SETNX can + # commit while RPUSH is lost, and concurrent retries return exactly + # one successful append. + created = await redis.eval( + _APPEND_SCRIPT, + 2, + event_key, + self._run_key(run_id), + encoded, + ) + return bool(created) + + async def append_many(self, run_id: str, entries: Sequence[TimelineEntry]) -> int: + added = 0 + for entry in entries: + if await self.append(run_id, entry): + added += 1 + return added + + async def list(self, run_id: str) -> Sequence[TimelineEntry]: + redis = await self._client() + values = await redis.lrange(self._run_key(run_id), 0, -1) + decoded = [] + for value in values: + if isinstance(value, bytes): + value = value.decode() + decoded.append(TimelineEntry.model_validate_json(value)) + # A writer may append a later event first; sorting is the read contract. + return tuple(sorted(decoded, key=_sort_key)) + + async def purge_before(self, cutoff: datetime) -> int: + redis = await self._client() + cursor: int | bytes = 0 + removed = 0 + while True: + cursor, keys = await redis.scan(cursor=cursor, match=f"{_PREFIX}*", count=100) + for raw_key in keys: + key = raw_key.decode() if isinstance(raw_key, bytes) else raw_key + if key.startswith(_EVENT_PREFIX): + continue + run_id = key[len(_PREFIX) :] + entries = await self.list(run_id) + keep = [entry for entry in entries if entry.occurred_at is None or entry.occurred_at >= cutoff] + if len(keep) == len(entries): + continue + await redis.delete(key) + for entry in entries: + await redis.delete(self._event_key(run_id, entry.event_id)) + if keep: + await self.append_many(run_id, keep) + removed += len(entries) - len(keep) + if cursor in {0, b"0", "0"}: + break + return removed + + +# Short names make the adapter easy to discover without breaking the explicit +# class names used in architecture documentation. +InMemoryTimelineStore = InMemoryExecutionTimelineStore +RedisTimelineStore = RedisExecutionTimelineStore +TimelineStore = ExecutionTimelineStore + + +def timeline_entry( + *, + event_id: str, + kind: str, + occurred_at: datetime | None, + summary: str, + status: str | None = None, + details: Mapping[str, JsonValue] | None = None, +) -> TimelineEntry: + """Build a normalized record for producers outside the read projection.""" + return TimelineEntry( + event_id=event_id, + kind=kind, + occurred_at=occurred_at, + status=status, + summary=summary, + details=dict(details or {}), + ) + + +__all__ = [ + "ExecutionTimelineStore", + "InMemoryExecutionTimelineStore", + "RedisExecutionTimelineStore", + "InMemoryTimelineStore", + "RedisTimelineStore", + "TimelineStore", + "timeline_entry", +] diff --git a/src/forge/reconciliation/__init__.py b/src/forge/reconciliation/__init__.py index acbf1620..ad8f4017 100644 --- a/src/forge/reconciliation/__init__.py +++ b/src/forge/reconciliation/__init__.py @@ -5,6 +5,7 @@ ObservationLedger, RedisObservationLedger, classify_observation, + observation_run_id, resource_identity, ) from forge.reconciliation.models import ( @@ -23,5 +24,6 @@ "RedisObservationLedger", "ReconciledResource", "classify_observation", + "observation_run_id", "resource_identity", ] diff --git a/src/forge/reconciliation/ledger.py b/src/forge/reconciliation/ledger.py index c07bb3f7..e5b5d8a0 100644 --- a/src/forge/reconciliation/ledger.py +++ b/src/forge/reconciliation/ledger.py @@ -45,6 +45,7 @@ _RESOURCE_PREFIX = "forge:observations:resource:" _DELIVERY_PREFIX = "forge:observations:delivery:" _HISTORY_PREFIX = "forge:observations:history:" +_RUN_HISTORY_PREFIX = "forge:observations:run:" class ObservationLedger(Protocol): @@ -54,6 +55,22 @@ async def latest(self, observation: Observation) -> ReconciledResource | None: . async def history(self, observation: Observation) -> Sequence[ObservationDecision]: ... + async def history_for_run(self, run_id: str) -> Sequence[ObservationDecision]: ... + + +def observation_run_id(observation: Observation) -> str | None: + """Extract workflow correlation without coupling the ledger to Jira.""" + for value in ( + observation.correlation.get("workflow_ticket_key"), + observation.correlation.get("ticket_key"), + ): + if isinstance(value, str) and value: + return value + issue = observation.facts.get("issue") + if isinstance(issue, dict) and isinstance(issue.get("key"), str): + return issue["key"] + return None + def resource_identity(observation: Observation) -> str: return stable_identity( @@ -74,6 +91,7 @@ def __init__(self) -> None: self._resources: dict[str, ReconciledResource] = {} self._history: dict[str, list[ObservationDecision]] = {} self._deliveries: dict[str, ObservationDecision] = {} + self._run_history: dict[str, list[ObservationDecision]] = {} self._lock = asyncio.Lock() async def record(self, observation: Observation) -> ObservationDecision: @@ -140,6 +158,9 @@ async def record(self, observation: Observation) -> ObservationDecision: def _append(self, observation: Observation, decision: ObservationDecision) -> None: self._history.setdefault(resource_identity(observation), []).append(decision) + run_id = observation_run_id(observation) + if run_id: + self._run_history.setdefault(run_id, []).append(decision) async def latest(self, observation: Observation) -> ReconciledResource | None: return self._resources.get(resource_identity(observation)) @@ -147,6 +168,9 @@ async def latest(self, observation: Observation) -> ReconciledResource | None: async def history(self, observation: Observation) -> Sequence[ObservationDecision]: return tuple(self._history.get(resource_identity(observation), ())) + async def history_for_run(self, run_id: str) -> Sequence[ObservationDecision]: + return tuple(self._run_history.get(run_id, ())) + class RedisObservationLedger: """Production ledger using optimistic transactions for monotonic acceptance.""" @@ -171,6 +195,7 @@ async def record(self, observation: Observation) -> ObservationDecision: await (await self._client()).rpush( self._history_key(observation), decision.model_dump_json() ) + await self._index_run(observation, decision) return decision redis = await self._client() @@ -225,6 +250,11 @@ async def record(self, observation: Observation) -> ObservationDecision: ) pipeline.multi() pipeline.rpush(self._history_key(observation), decision.model_dump_json()) + run_id = observation_run_id(observation) + if run_id: + pipeline.rpush( + f"{_RUN_HISTORY_PREFIX}{run_id}", decision.model_dump_json() + ) if prior is None: pipeline.set(delivery_key, decision.model_dump_json()) if decision.disposition is ObservationDisposition.ACCEPTED: @@ -247,6 +277,17 @@ async def history(self, observation: Observation) -> Sequence[ObservationDecisio values = await (await self._client()).lrange(self._history_key(observation), 0, -1) return tuple(ObservationDecision.model_validate_json(value) for value in values) + async def history_for_run(self, run_id: str) -> Sequence[ObservationDecision]: + values = await (await self._client()).lrange(f"{_RUN_HISTORY_PREFIX}{run_id}", 0, -1) + return tuple(ObservationDecision.model_validate_json(value) for value in values) + + async def _index_run(self, observation: Observation, decision: ObservationDecision) -> None: + run_id = observation_run_id(observation) + if run_id: + await (await self._client()).rpush( + f"{_RUN_HISTORY_PREFIX}{run_id}", decision.model_dump_json() + ) + @staticmethod def _resource_key(observation: Observation) -> str: return f"{_RESOURCE_PREFIX}{resource_identity(observation)}" diff --git a/src/forge/workflow/base.py b/src/forge/workflow/base.py index f46259b9..a5e8c3f7 100644 --- a/src/forge/workflow/base.py +++ b/src/forge/workflow/base.py @@ -126,6 +126,9 @@ class BaseState(TypedDict, total=False): # Durable ingress audit trail. Entries are provider-neutral command decisions, # bounded by the worker to keep checkpoint growth predictable. command_decisions: list[dict[str, Any]] + # Normalized observation decisions are retained for read-model rebuilds; + # unlike provider payloads they contain only contract metadata and reason. + observation_history: list[dict[str, Any]] # Declarative workflow identity. Built-in workflows leave these unset. workflow_name: str @@ -143,6 +146,8 @@ class BaseState(TypedDict, total=False): workflow_node_attempts: dict[str, int] transition_history: list[dict[str, Any]] station_history: list[dict[str, Any]] + migration_history: list[dict[str, Any]] + operator_actions: list[dict[str, Any]] # Generic node-contract capabilities and durable precondition audit trail. # Missing capability keys preserve legacy inference; explicit booleans are diff --git a/tests/unit/api/routes/test_executions.py b/tests/unit/api/routes/test_executions.py index 6f92a5f7..aa12b365 100644 --- a/tests/unit/api/routes/test_executions.py +++ b/tests/unit/api/routes/test_executions.py @@ -1,8 +1,10 @@ +from datetime import UTC, datetime from unittest.mock import AsyncMock import pytest from forge.api.routes.executions import load_execution_read_model +from forge.read_models.models import TimelineEntry @pytest.mark.asyncio @@ -52,3 +54,64 @@ async def test_load_execution_read_model_returns_none_for_unknown_workflow() -> checkpointer.aget.return_value = None assert await load_execution_read_model("MISSING-1", checkpointer=checkpointer) is None + + +@pytest.mark.asyncio +async def test_loader_rehydrates_observation_and_timeline_records_after_restart() -> None: + checkpointer = AsyncMock() + checkpointer.aget.return_value = { + "channel_values": { + "thread_id": "FORGE-2", + "ticket_key": "FORGE-2", + "current_node": "ci_evaluator", + "observation_history": [], + } + } + journal = AsyncMock() + journal.list_for_workflow.return_value = [] + ledger = AsyncMock() + ledger.history_for_run.return_value = [ + { + "observation_id": "observation-1", + "delivery_identity": "delivery-1", + "disposition": "stale", + "decided_at": "2026-08-28T11:58:00+00:00", + "reason": "older provider revision", + } + ] + timeline = AsyncMock() + timeline.list.return_value = [ + TimelineEntry( + event_id="operator-1", + kind="operator_action", + occurred_at=datetime(2026, 8, 28, 11, 59, tzinfo=UTC), + status="accepted", + summary="retry", + ) + ] + + first = await load_execution_read_model( + "FORGE-2", + checkpointer=checkpointer, + effect_journal=journal, + observation_ledger=ledger, + timeline_store=timeline, + ) + # Simulate a process restart: all records are re-read from the durable + # adapters rather than relying on in-process projection state. + second = await load_execution_read_model( + "FORGE-2", + checkpointer=checkpointer, + effect_journal=journal, + observation_ledger=ledger, + timeline_store=timeline, + ) + + assert first is not None and second is not None + assert first.timeline == second.timeline + assert [entry.kind for entry in first.timeline] == [ + "observation", + "operator_action", + ] + ledger.history_for_run.assert_awaited_with("FORGE-2") + timeline.list.assert_awaited_with("FORGE-2") diff --git a/tests/unit/api/routes/test_metrics.py b/tests/unit/api/routes/test_metrics.py index 8a4e8b00..07831749 100644 --- a/tests/unit/api/routes/test_metrics.py +++ b/tests/unit/api/routes/test_metrics.py @@ -6,6 +6,64 @@ from forge.main import app +def test_execution_metrics_record_bounded_operational_signals() -> None: + from datetime import UTC, datetime, timedelta + from types import SimpleNamespace + + from forge.api.routes.metrics import ( + EXECUTION_BLOCKED, + EXECUTION_DRIFT, + EXECUTION_MIGRATION_ELIGIBILITY, + EXECUTION_RETRIES, + EXECUTION_WAITING_AGE, + record_execution_read_model, + ) + + model = SimpleNamespace( + status=SimpleNamespace(value="blocked"), + waiting=SimpleNamespace( + code="credential", since=datetime.now(UTC) - timedelta(seconds=5), message="missing" + ), + station_attempts=(SimpleNamespace(attempt=2),), + effects=(), + last_observation=SimpleNamespace(conflicting=True, stale=False), + migration=SimpleNamespace(eligible=False), + ) + record_execution_read_model(model) + + assert EXECUTION_BLOCKED.labels(code="unknown")._value.get() == 1 + assert EXECUTION_WAITING_AGE.labels(code="unknown")._sum.get() > 0 + assert EXECUTION_RETRIES.labels(kind="execution")._value.get() == 1 + assert EXECUTION_DRIFT.labels(**{"class": "operator_required"})._value.get() == 1 + assert EXECUTION_MIGRATION_ELIGIBILITY.labels(state="ineligible")._value.get() == 0 + + +def test_execution_sampled_metrics_do_not_accumulate_on_repeated_reads() -> None: + from types import SimpleNamespace + + from forge.api.routes.metrics import ( + EXECUTION_BLOCKED, + EXECUTION_DRIFT, + EXECUTION_RETRIES, + record_execution_read_model, + ) + + model = SimpleNamespace( + status="blocked", + waiting=SimpleNamespace(code="blocked", since=None), + station_attempts=(SimpleNamespace(attempt=3),), + effects=(), + last_observation=SimpleNamespace(conflicting=True, stale=False), + migration=SimpleNamespace(eligible=None), + ) + record_execution_read_model(model) + record_execution_read_model(model) + + assert EXECUTION_BLOCKED.labels(code="blocked")._value.get() == 1 + assert EXECUTION_RETRIES.labels(kind="execution")._value.get() == 2 + assert EXECUTION_DRIFT.labels(**{"class": "operator_required"})._value.get() == 1 + + class TestMetricsEndpoint: """Tests for /metrics endpoint.""" diff --git a/tests/unit/api/routes/test_org_pulse.py b/tests/unit/api/routes/test_org_pulse.py new file mode 100644 index 00000000..85442d1b --- /dev/null +++ b/tests/unit/api/routes/test_org_pulse.py @@ -0,0 +1,27 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from httpx import ASGITransport, AsyncClient +from pydantic import SecretStr + +from forge.api.routes import org_pulse as route +from forge.main import app + + +@pytest.mark.asyncio +async def test_org_pulse_endpoint_requires_operator_token(monkeypatch) -> None: + monkeypatch.setattr( + "forge.api.routes.executions.get_settings", + lambda: SimpleNamespace(forge_operator_token=SecretStr("pulse-secret")), + ) + monkeypatch.setattr(route, "load_execution_read_model", AsyncMock(return_value=None)) + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + unauthorized = await client.get("/api/v1/org-pulse/workflows/FORGE-7") + missing = await client.get( + "/api/v1/org-pulse/workflows/FORGE-7", + headers={"Authorization": "Bearer pulse-secret"}, + ) + assert unauthorized.status_code == 401 + assert missing.status_code == 404 diff --git a/tests/unit/api/test_effects.py b/tests/unit/api/test_effects.py index e43f8f76..85a58f6d 100644 --- a/tests/unit/api/test_effects.py +++ b/tests/unit/api/test_effects.py @@ -1,13 +1,21 @@ +from datetime import UTC, datetime from types import SimpleNamespace import pytest from httpx import ASGITransport, AsyncClient from pydantic import SecretStr -from forge.api.routes.effects import get_effect_service -from forge.domain import EffectCommand, ResourceIdentity, WorkflowIdentity +from forge.api.routes.effects import get_effect_service, get_timeline_store +from forge.domain import ( + EffectCommand, + EffectResult, + EffectResultStatus, + ResourceIdentity, + WorkflowIdentity, +) from forge.effects import EffectExecutorRegistry, EffectService, InMemoryEffectJournal from forge.main import app +from forge.read_models import InMemoryExecutionTimelineStore def _command() -> EffectCommand: @@ -53,3 +61,61 @@ async def test_operator_can_inspect_workflow_effect_history(monkeypatch) -> None assert response.json()[0]["command"]["idempotency_key"] == "effect-1" finally: app.dependency_overrides.pop(get_effect_service, None) + + +@pytest.mark.asyncio +async def test_authenticated_effect_replay_is_durable_operator_timeline_evidence(monkeypatch) -> None: + journal = InMemoryEffectJournal() + service = EffectService(journal, EffectExecutorRegistry()) + command = _command() + await service.submit(command) + await journal.complete( + EffectResult( + effect_id=command.effect_id, + idempotency_key=command.idempotency_key, + status=EffectResultStatus.TERMINAL_FAILURE, + completed_at=datetime.now(UTC), + error_message="provider unavailable", + ) + ) + timeline = InMemoryExecutionTimelineStore() + app.dependency_overrides[get_effect_service] = lambda: service + app.dependency_overrides[get_timeline_store] = lambda: timeline + monkeypatch.setattr( + "forge.api.routes.effects.get_settings", + lambda: SimpleNamespace(effect_operator_token=SecretStr("operator-secret")), + ) + try: + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.post( + "/api/v1/effects/effect-1/replay", + headers={"Authorization": "Bearer operator-secret"}, + ) + assert response.status_code == 200 + records = await timeline.list("FORGE-1") + assert len(records) == 1 + assert records[0].kind == "operator_action" + assert records[0].details["target"] == "FORGE-1" + assert records[0].details["result_status"] == "pending" + finally: + app.dependency_overrides.pop(get_effect_service, None) + app.dependency_overrides.pop(get_timeline_store, None) + + +@pytest.mark.asyncio +async def test_unauthenticated_effect_replay_does_not_write_operator_evidence(monkeypatch) -> None: + timeline = InMemoryExecutionTimelineStore() + app.dependency_overrides[get_timeline_store] = lambda: timeline + monkeypatch.setattr( + "forge.api.routes.effects.get_settings", + lambda: SimpleNamespace(effect_operator_token=SecretStr("operator-secret")), + ) + try: + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + response = await client.post("/api/v1/effects/effect-1/replay") + assert response.status_code == 401 + assert await timeline.list("FORGE-1") == () + finally: + app.dependency_overrides.pop(get_timeline_store, None) diff --git a/tests/unit/architecture/test_read_model_boundaries.py b/tests/unit/architecture/test_read_model_boundaries.py new file mode 100644 index 00000000..78a6ebcf --- /dev/null +++ b/tests/unit/architecture/test_read_model_boundaries.py @@ -0,0 +1,72 @@ +"""Keep execution inspection a strictly read-only architectural boundary.""" + +from __future__ import annotations + +import ast +from pathlib import Path + +ROOT = Path(__file__).parents[3] +READ_MODELS = ROOT / "src" / "forge" / "read_models" +OPERATOR_ROUTES = ROOT / "src" / "forge" / "api" / "routes" + +# These methods either mutate a workflow checkpoint or execute/re-schedule an +# external effect. Timeline append/purge are intentionally absent: they write +# the projection's own append-only evidence, not workflow/effect state. +MUTATION_METHODS = frozenset( + { + "ainvoke", + "astream", + "aupdate_state", + "adelete_thread", + "advance", + "claim", + "claim_due", + "complete", + "execute_now", + "execute_required", + "replay", + "retry", + "submit", + } +) + + +def _python_files(directory: Path) -> list[Path]: + return sorted(set(directory.rglob("*.py"))) + + +def _calls_to_mutation_methods(path: Path) -> list[str]: + tree = ast.parse(path.read_text(), filename=str(path)) + return [ + f"{path.relative_to(ROOT)}:{node.lineno}:{node.func.attr}" + for node in ast.walk(tree) + if isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr in MUTATION_METHODS + ] + + +def test_read_models_do_not_mutate_workflows_or_effects() -> None: + violations = [item for path in _python_files(READ_MODELS) for item in _calls_to_mutation_methods(path)] + assert violations == [], f"Read-model code crossed a mutation boundary: {violations}" + + +def test_operator_read_routes_do_not_mutate_workflows_or_effects() -> None: + # Limit this guard to operator read routes. Webhook/effect routes are + # mutation surfaces by design and are covered by their own tests. + operator_files = [OPERATOR_ROUTES / "executions.py", OPERATOR_ROUTES / "org_pulse.py"] + violations = [item for path in operator_files for item in _calls_to_mutation_methods(path)] + assert violations == [], f"Operator read API crossed a mutation boundary: {violations}" + + +def test_read_models_do_not_import_effect_execution_services() -> None: + violations: list[str] = [] + for path in _python_files(READ_MODELS): + tree = ast.parse(path.read_text(), filename=str(path)) + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and node.module in { + "forge.effects.service", + "forge.effects.executors", + }: + violations.append(f"{path.relative_to(ROOT)}:{node.lineno}:{node.module}") + assert violations == [], f"Read-model code imported effect execution services: {violations}" diff --git a/tests/unit/integrations/test_org_pulse.py b/tests/unit/integrations/test_org_pulse.py new file mode 100644 index 00000000..3f5a25c0 --- /dev/null +++ b/tests/unit/integrations/test_org_pulse.py @@ -0,0 +1,30 @@ +from datetime import UTC, datetime + +from forge.integrations.org_pulse import OrgPulseExecution +from forge.read_models.execution import project_execution + + +def test_org_pulse_contract_is_versioned_and_contains_operational_state() -> None: + model = project_execution( + { + "thread_id": "FORGE-7", + "ticket_key": "FORGE-7", + "workflow_name": "feature", + "workflow_revision": 4, + "current_node": "approval_gate", + "is_paused": True, + "updated_at": datetime(2026, 8, 28, tzinfo=UTC).isoformat(), + "station_history": [ + {"station_name": "approval", "invocation_id": "a-1", "attempt": 2} + ], + } + ) + + pulse = OrgPulseExecution.from_execution(model) + + assert pulse.schema_version == "1.0" + assert pulse.ticket_key == "FORGE-7" + assert pulse.status == "waiting" + assert pulse.waiting_code == "gate" + assert pulse.retry_count == 1 + assert pulse.migration_eligible is None diff --git a/tests/unit/read_models/test_execution.py b/tests/unit/read_models/test_execution.py index 470f9391..e93e5b06 100644 --- a/tests/unit/read_models/test_execution.py +++ b/tests/unit/read_models/test_execution.py @@ -197,3 +197,83 @@ def test_timeline_combines_durable_decisions_transitions_stations_and_effects() "station_attempt", "effect", ] + + +def test_projection_retains_stale_and_conflicting_observation_decisions() -> None: + model = project_execution( + { + "ticket_key": "FORGE-1", + "current_node": "ci_evaluator", + "observation_history": [ + { + "observation_id": "observation-old", + "source_system": "github", + "disposition": "stale", + "reason": "older provider revision", + }, + { + "observation_id": "observation-conflict", + "source_system": "github", + "disposition": "conflict", + "reason": "same revision contains different facts", + }, + ], + }, + now=NOW, + ) + + assert [item.observation_id for item in model.stale_observations] == ["observation-old"] + assert model.conflicting_observations[0].reason == "same revision contains different facts" + assert [item.kind for item in model.timeline] == ["observation", "observation"] + + +def test_rule_explanation_includes_true_and_false_contract_clauses() -> None: + model = project_execution( + { + "ticket_key": "FORGE-1", + "workflow_state_profile": "feature", + "current_node": "implement_work", + "capabilities": { + "repositories_resolved": True, + "workspace_ready": False, + "planning_context_available": True, + }, + "precondition_result": { + "action": "block", + "missing": ["workspace_ready"], + "reason": "Workspace must exist before implementation", + }, + "is_blocked": True, + } + ) + + explanation = model.explanations[0] + assert explanation.satisfied is False + assert {clause.capability: clause.satisfied for clause in explanation.clauses} == { + "repositories_resolved": True, + "workspace_ready": False, + "planning_context_available": True, + } + + +def test_pinned_canonical_definition_revision_is_not_replaced_by_legacy_alias() -> None: + model = project_execution( + { + "ticket_key": "FORGE-1", + "workflow_name": "feature-flow", + "workflow_revision": 2, + "workflow_digest": "old-digest", + "workflow_definition_revision": 7, + "workflow_definition_digest": "pinned-digest", + "workflow_definition": { + "apiVersion": "forge/v1", + "kind": "Workflow", + "metadata": {"name": "feature-flow", "revision": 7}, + "spec": {"state": "feature", "entry": "generate_prd", "steps": {}}, + }, + } + ) + + assert model.definition.available is True + assert model.definition.revision == 7 + assert model.definition.digest == "pinned-digest" diff --git a/tests/unit/read_models/test_timeline_store.py b/tests/unit/read_models/test_timeline_store.py new file mode 100644 index 00000000..216c3195 --- /dev/null +++ b/tests/unit/read_models/test_timeline_store.py @@ -0,0 +1,181 @@ +import asyncio +from datetime import UTC, datetime, timedelta + +import pytest + +from forge.read_models import ( + InMemoryExecutionTimelineStore, + RedisExecutionTimelineStore, + project_execution, + rebuild_execution_timeline, + timeline_entry, +) + +NOW = datetime(2026, 8, 28, 12, tzinfo=UTC) + + +class _AtomicFakeRedis: + """Tiny fake that implements the Lua append contract, not Redis itself.""" + + def __init__(self) -> None: + self.values: dict[str, str] = {} + self.lists: dict[str, list[str]] = {} + self.eval_calls = 0 + self.fail_once = False + + async def eval(self, _script: str, _key_count: int, event_key: str, run_key: str, value: str) -> int: + self.eval_calls += 1 + if self.fail_once: + self.fail_once = False + raise RuntimeError("simulated script interruption before commit") + if event_key in self.values: + return 0 + # This method is the atomic boundary in the fake: no await occurs + # between marker creation and list append. + self.values[event_key] = value + self.lists.setdefault(run_key, []).append(value) + return 1 + + async def lrange(self, run_key: str, start: int, end: int) -> list[str]: + values = self.lists.get(run_key, []) + return values[start:] if end == -1 else values[start : end + 1] + + async def rpush(self, *_args: object) -> None: + raise AssertionError("timeline append must use the atomic Lua operation") + + +@pytest.mark.asyncio +async def test_timeline_store_is_idempotent_and_orders_replayed_records() -> None: + store = InMemoryExecutionTimelineStore() + late = timeline_entry( + event_id="transition-1", + kind="transition", + occurred_at=NOW - timedelta(minutes=1), + summary="entry → work", + ) + early = timeline_entry( + event_id="observation-1", + kind="observation", + occurred_at=NOW - timedelta(minutes=2), + summary="accepted", + status="accepted", + ) + + assert await store.append("RUN-1", late) is True + assert await store.append("RUN-1", late) is False + assert await store.append_many("RUN-1", [early, late]) == 1 + assert [item.event_id for item in await store.list("RUN-1")] == [ + "observation-1", + "transition-1", + ] + + +@pytest.mark.asyncio +async def test_redis_timeline_append_is_atomic_under_concurrent_duplicates() -> None: + redis = _AtomicFakeRedis() + store = RedisExecutionTimelineStore(redis) + entry = timeline_entry( + event_id="operator-1", + kind="operator_action", + occurred_at=NOW, + summary="retry", + ) + + outcomes = await asyncio.gather( + *(store.append("RUN-1", entry) for _ in range(8)) + ) + + assert outcomes.count(True) == 1 + assert outcomes.count(False) == 7 + assert len(await store.list("RUN-1")) == 1 + assert redis.eval_calls == 8 + + +@pytest.mark.asyncio +async def test_redis_timeline_retry_after_script_interruption_is_complete() -> None: + redis = _AtomicFakeRedis() + redis.fail_once = True + store = RedisExecutionTimelineStore(redis) + entry = timeline_entry( + event_id="operator-1", + kind="operator_action", + occurred_at=NOW, + summary="retry", + ) + + with pytest.raises(RuntimeError): + await store.append("RUN-1", entry) + assert await store.list("RUN-1") == () + assert await store.append("RUN-1", entry) is True + assert len(await store.list("RUN-1")) == 1 + + +def test_projection_rebuilds_timeline_from_all_durable_record_categories() -> None: + model = project_execution( + { + "ticket_key": "RUN-1", + "current_node": "work", + "observation_history": [ + { + "observation_id": "obs-1", + "disposition": "stale", + "decided_at": (NOW - timedelta(minutes=4)).isoformat(), + "reason": "older provider revision", + } + ], + "command_decisions": [ + { + "decision_id": "command-1", + "decided_at": (NOW - timedelta(minutes=3)).isoformat(), + "status": "ignored", + "reason": "duplicate command", + } + ], + "transition_history": [ + { + "transition_id": "transition-1", + "source": "entry", + "target": "work", + "occurred_at": (NOW - timedelta(minutes=2)).isoformat(), + } + ], + "migration_history": [ + { + "migration_id": "migration-1", + "occurred_at": (NOW - timedelta(minutes=1)).isoformat(), + "status": "blocked", + "reason": "missing resume mapping", + } + ], + "operator_actions": [ + { + "action_id": "operator-1", + "occurred_at": NOW.isoformat(), + "action": "retry", + "actor": "operator@example.test", + } + ], + }, + now=NOW, + ) + + assert [entry.kind for entry in model.timeline] == [ + "observation", + "command_decision", + "transition", + "migration", + "operator_action", + ] + assert model.timeline[0].status == "stale" + assert model.timeline[-1].details["actor"] == "operator@example.test" + + rebuilt = rebuild_execution_timeline( + { + "ticket_key": "RUN-1", + "current_node": "work", + }, + timeline_entries=tuple(reversed(model.timeline)), + ) + assert rebuilt == tuple(sorted(model.timeline, key=lambda item: ( + item.occurred_at or datetime.min.replace(tzinfo=UTC), item.kind, item.event_id + ))) diff --git a/tests/unit/reconciliation/test_ledger.py b/tests/unit/reconciliation/test_ledger.py index 7f71e899..eb3668fb 100644 --- a/tests/unit/reconciliation/test_ledger.py +++ b/tests/unit/reconciliation/test_ledger.py @@ -28,7 +28,7 @@ def observation( revision_order=order, observed_at=now, received_at=now, - facts={"status": status}, + facts={"status": status}, ) @@ -78,6 +78,27 @@ async def test_stale_delivery_cannot_overwrite_latest_projection() -> None: assert (await ledger.latest(stale)).latest.facts == {"status": "merged"} +@pytest.mark.asyncio +async def test_observation_history_can_be_rebuilt_by_workflow_run() -> None: + ledger = InMemoryObservationLedger() + current = observation(ObservationSource.WEBHOOK, 5) + current = current.model_copy( + update={"correlation": {"workflow_ticket_key": "FORGE-17"}} + ) + older = observation(ObservationSource.POLLER, 3).model_copy( + update={"correlation": {"workflow_ticket_key": "FORGE-17"}} + ) + + await ledger.record(current) + await ledger.record(older) + + history = await ledger.history_for_run("FORGE-17") + assert [item.disposition for item in history] == [ + ObservationDisposition.ACCEPTED, + ObservationDisposition.STALE, + ] + + @pytest.mark.asyncio async def test_same_revision_with_different_facts_requires_operator() -> None: ledger = InMemoryObservationLedger() From c84edbe2e666b8ad8e1207e05e9714146cf280ad Mon Sep 17 00:00:00 2001 From: eshulman2 Date: Mon, 31 Aug 2026 14:33:15 +0300 Subject: [PATCH 5/5] style: format execution read models --- src/forge/api/routes/metrics.py | 4 +- src/forge/integrations/org_pulse.py | 4 +- src/forge/read_models/execution.py | 74 ++++++++++++++++++++--------- src/forge/read_models/timeline.py | 6 ++- src/forge/reconciliation/ledger.py | 4 +- 5 files changed, 64 insertions(+), 28 deletions(-) diff --git a/src/forge/api/routes/metrics.py b/src/forge/api/routes/metrics.py index 03103b47..d70a5f11 100644 --- a/src/forge/api/routes/metrics.py +++ b/src/forge/api/routes/metrics.py @@ -360,7 +360,9 @@ def record_execution_read_model(model: object) -> None: migration = getattr(model, "migration", None) eligible = getattr(migration, "eligible", None) - migration_state = "eligible" if eligible is True else "ineligible" if eligible is False else "unknown" + migration_state = ( + "eligible" if eligible is True else "ineligible" if eligible is False else "unknown" + ) for state in _MIGRATION_STATES: EXECUTION_MIGRATION_ELIGIBILITY.labels(state=state).set(0) EXECUTION_MIGRATION_ELIGIBILITY.labels(state=migration_state).set( diff --git a/src/forge/integrations/org_pulse.py b/src/forge/integrations/org_pulse.py index acd47b8d..183eed4c 100644 --- a/src/forge/integrations/org_pulse.py +++ b/src/forge/integrations/org_pulse.py @@ -48,7 +48,9 @@ def from_execution(cls, execution: ExecutionReadModel) -> OrgPulseExecution: workflow_revision=execution.definition.revision, waiting_code=waiting.code if waiting else None, waiting_since=waiting.since if waiting else None, - blocking_reason=(waiting.message if waiting and execution.status.value == "blocked" else None), + blocking_reason=( + waiting.message if waiting and execution.status.value == "blocked" else None + ), retry_count=retries, observation_available=execution.last_observation.available, observation_stale=execution.last_observation.stale, diff --git a/src/forge/read_models/execution.py b/src/forge/read_models/execution.py index 8bfe8679..e5e36cab 100644 --- a/src/forge/read_models/execution.py +++ b/src/forge/read_models/execution.py @@ -72,7 +72,9 @@ def project_execution( **({"digest": manifest.digest} if manifest else {}), } if isinstance(checkpoint.get("workflow_definition"), dict) - else manifest.model_dump(mode="json") if manifest else None + else manifest.model_dump(mode="json") + if manifest + else None ), ) decisions = tuple(observation_decisions) or _checkpoint_observation_decisions(checkpoint) @@ -203,7 +205,9 @@ def _permitted_commands( return ("retry", "cancel") if status is ExecutionStatus.WAITING: commands = ["resume", "retry", "cancel"] - node = next((item for item in (manifest.nodes if manifest else ()) if item.name == position), None) + node = next( + (item for item in (manifest.nodes if manifest else ()) if item.name == position), None + ) # Gate-ness comes from the pinned process manifest, never from a name # convention such as ``*_gate``. if node is not None and node.kind.value == "gate": @@ -231,7 +235,9 @@ def _observation( observation_id=observation.observation_id, source_system=observation.source_system, observed_at=observation.observed_at, - stale=_observation_is_stale(observation, decisions, comparable_now, comparable_observed, stale_after), + stale=_observation_is_stale( + observation, decisions, comparable_now, comparable_observed, stale_after + ), conflicting=bool(checkpoint.get("external_state_conflict")) or _observation_has_disposition(observation, decisions, "conflict"), available=True, @@ -243,9 +249,7 @@ def _observation( def _decision_disposition(item: Any) -> str | None: value = ( - item.get("disposition") - if isinstance(item, Mapping) - else getattr(item, "disposition", None) + item.get("disposition") if isinstance(item, Mapping) else getattr(item, "disposition", None) ) return getattr(value, "value", value) @@ -257,9 +261,7 @@ def _checkpoint_observation_decisions(checkpoint: Mapping[str, Any]) -> tuple[An def _decision_observation(item: Any) -> Any: observation = ( - item.get("observation") - if isinstance(item, Mapping) - else getattr(item, "observation", None) + item.get("observation") if isinstance(item, Mapping) else getattr(item, "observation", None) ) if isinstance(observation, Mapping): return _MappingObservation(observation) @@ -294,7 +296,11 @@ def _decision_observation_id(item: Any) -> str | None: def _decision_delivery_identity(item: Any) -> str | None: - value = item.get("delivery_identity") if isinstance(item, Mapping) else getattr(item, "delivery_identity", None) + value = ( + item.get("delivery_identity") + if isinstance(item, Mapping) + else getattr(item, "delivery_identity", None) + ) return str(value) if value else None @@ -446,7 +452,11 @@ def _recovery_options( return tuple( RecoveryOptionView( command=command, - description=(waiting.recovery if command == "retry" and waiting and waiting.recovery else descriptions.get(command, "Issue this permitted command.")), + description=( + waiting.recovery + if command == "retry" and waiting and waiting.recovery + else descriptions.get(command, "Issue this permitted command.") + ), ) for command in permitted ) @@ -507,11 +517,14 @@ def _rule_explanations( ) ) action = persisted.get("action") if isinstance(persisted, Mapping) else None - satisfied = all(clause.satisfied for clause in clauses) if clauses else action in {None, "proceed"} + satisfied = ( + all(clause.satisfied for clause in clauses) if clauses else action in {None, "proceed"} + ) summary = ( "All required workflow rules are satisfied." if satisfied - else str(persisted.get("reason")) if isinstance(persisted, Mapping) and persisted.get("reason") + else str(persisted.get("reason")) + if isinstance(persisted, Mapping) and persisted.get("reason") else "One or more required workflow rules are false." ) # A checkpoint can have several evaluations over time. The current @@ -600,8 +613,16 @@ def _timeline( "revision_order": observation.revision_order, } else: - occurred_at = _datetime(item.get("decided_at")) if isinstance(item, Mapping) else getattr(item, "decided_at", None) - event_id = str(item.get("observation_id") or "observation") if isinstance(item, Mapping) else "observation" + occurred_at = ( + _datetime(item.get("decided_at")) + if isinstance(item, Mapping) + else getattr(item, "decided_at", None) + ) + event_id = ( + str(item.get("observation_id") or "observation") + if isinstance(item, Mapping) + else "observation" + ) details = {} disposition = _decision_disposition(item) delivery_identity = _decision_delivery_identity(item) @@ -610,9 +631,7 @@ def _timeline( # decisions (accepted, duplicate, stale, or conflict). Include # decision identity so projection does not collapse that audit # history into one observation event. - event_id = ":".join( - part for part in (event_id, delivery_identity, disposition) if part - ) + event_id = ":".join(part for part in (event_id, delivery_identity, disposition) if part) reason = item.get("reason") if isinstance(item, Mapping) else getattr(item, "reason", None) entries.append( TimelineEntry( @@ -635,7 +654,8 @@ def _timeline( details={ key: value for key, value in item.items() - if key not in {"event_id", "node", "occurred_at", "evaluated_at", "action", "reason"} + if key + not in {"event_id", "node", "occurred_at", "evaluated_at", "action", "reason"} }, ) ) @@ -665,8 +685,14 @@ def _timeline( kind="operator_action", occurred_at=_datetime(item.get("occurred_at") or item.get("acted_at")), status=str(item.get("status") or "recorded"), - summary=str(item.get("summary") or item.get("action") or "Operator action recorded"), - details={key: value for key, value in item.items() if key not in {"summary", "action", "occurred_at", "acted_at"}}, + summary=str( + item.get("summary") or item.get("action") or "Operator action recorded" + ), + details={ + key: value + for key, value in item.items() + if key not in {"summary", "action", "occurred_at", "acted_at"} + }, ) ) for record in effects: @@ -684,7 +710,11 @@ def _timeline( details={ "effect_id": record.command.effect_id, "idempotency_key": record.command.idempotency_key, - **({"provider_reference": result.provider_reference} if result.provider_reference else {}), + **( + {"provider_reference": result.provider_reference} + if result.provider_reference + else {} + ), **({"error": result.error_message} if result.error_message else {}), }, ) diff --git a/src/forge/read_models/timeline.py b/src/forge/read_models/timeline.py index 6036bc2e..2465d18b 100644 --- a/src/forge/read_models/timeline.py +++ b/src/forge/read_models/timeline.py @@ -173,7 +173,11 @@ async def purge_before(self, cutoff: datetime) -> int: continue run_id = key[len(_PREFIX) :] entries = await self.list(run_id) - keep = [entry for entry in entries if entry.occurred_at is None or entry.occurred_at >= cutoff] + keep = [ + entry + for entry in entries + if entry.occurred_at is None or entry.occurred_at >= cutoff + ] if len(keep) == len(entries): continue await redis.delete(key) diff --git a/src/forge/reconciliation/ledger.py b/src/forge/reconciliation/ledger.py index e5b5d8a0..7a638c93 100644 --- a/src/forge/reconciliation/ledger.py +++ b/src/forge/reconciliation/ledger.py @@ -252,9 +252,7 @@ async def record(self, observation: Observation) -> ObservationDecision: pipeline.rpush(self._history_key(observation), decision.model_dump_json()) run_id = observation_run_id(observation) if run_id: - pipeline.rpush( - f"{_RUN_HISTORY_PREFIX}{run_id}", decision.model_dump_json() - ) + pipeline.rpush(f"{_RUN_HISTORY_PREFIX}{run_id}", decision.model_dump_json()) if prior is None: pipeline.set(delivery_key, decision.model_dump_json()) if decision.disposition is ObservationDisposition.ACCEPTED: