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 new file mode 100644 index 00000000..8e0d0d3f --- /dev/null +++ b/docs/architecture/phase-7-read-models-plan.md @@ -0,0 +1,80 @@ +# Phase 7 implementation plan: process and execution read models + +**Status:** Complete. + +**Depends on:** Versioned process definitions, station outcomes and durable effects + +**Goal:** Answer operator questions from durable execution records rather than Jira +labels or worker logs. Read models are projections only: they cannot advance a workflow +or execute an effect. + +## Delivery slices + +1. **Execution projection.** Combine checkpoint position, pinned definition, permitted + commands, waiting reason, station history, external observation metadata and effects + into one versioned response. +2. **Pinned-definition visibility.** Retain the canonical declarative definition with + checkpoints so inspection never substitutes a newer Jira project property for the + revision an instance actually runs. +3. **Operator API.** Expose the projection by workflow/ticket identity with explicit + unavailable fields for legacy checkpoints. +4. **Durable decision and observation history.** Persist command decisions and normalized + observations, including ignored/stale reasons, then include them in the timeline. +5. **Org Pulse and metrics.** Consume the API for dashboards and measure waiting age, + retries, blocked causes, stale observations and migration incompatibilities. + +## Completion evidence + +The Phase 7 work items are implemented in the following boundaries: + +1. `src/forge/read_models/timeline.py` provides idempotent in-memory and Redis + timeline stores. `project_execution` rebuilds observations, command decisions, + transitions, station attempts, effect attempts/results, migrations, and operator + actions into a deterministic timeline. Coverage is in + `tests/unit/read_models/test_timeline_store.py` and the read-model tests. +2. `project_execution` exposes the pinned definition, position, permitted commands, + waits/blocks, stale/conflicting observations, effects, recovery options, and + evaluated rule explanations. Legacy checkpoints expose unavailable fields rather + than consulting Jira. +3. `GET /api/v1/workflows/{ticket_key}/execution` and its authenticated timeline + endpoint are the stable operator surface. Timeline pagination is bounded to 200 + entries and returns a deterministic cursor. Authentication and contract behavior + are covered by `tests/unit/api/routes/test_executions.py`. +4. The Org Pulse contract is `GET /api/v1/org-pulse/workflows/{ticket_key}` and the + versioned `OrgPulseExecution` model in `src/forge/integrations/org_pulse.py`. + Contract and authentication coverage is in + `tests/unit/integrations/test_org_pulse.py` and + `tests/unit/api/routes/test_org_pulse.py`. +5. Read-model latency and waiting-age histograms plus bounded-label gauges for + sampled retry count, drift, blocking, and migration eligibility are defined in + `src/forge/api/routes/metrics.py`; recording is covered by + `tests/unit/api/routes/test_metrics.py`. Sampled-state gauges are deliberately + not counters, so repeated Org Pulse GETs do not inflate event totals. Event + counters remain owned by their actual decision/transition writers. +6. `rebuild_execution_timeline` and restart-style loader coverage prove deterministic + reconstruction from durable checkpoint, ledger, timeline, and effect records. + +## Operations, retention, and rollback + +Read models and operator routes are inspection-only: they do not advance checkpoints, +execute effects, or issue provider mutations. The architecture guard in +`tests/unit/architecture/test_read_model_boundaries.py` prevents mutation calls and +effect-execution imports from returning to those boundaries. + +Timeline retention is exposed as the explicit `purge_before` operation on timeline +stores; terminal effect retention remains the explicit +`EffectService.purge_terminal_before` operation. Pending and running effects are not +eligible for terminal retention. Retention is therefore an operator/deployment +operation, not an implicit action during reads, and its deletion is irreversible +without a backup. + +The API and Org Pulse payloads carry `schema_version` (`1.0`). Consumers must tolerate +additive fields and treat absent/`null` legacy fields as unavailable. A read-model +rollback deploys the prior application version; it does not rewrite checkpoints or +effects. If a persisted timeline format changes, take a backup and use an explicit +rebuild/migration before re-enabling the new reader. + +The full local stack suite, integration suite, focused Ruff checks, and targeted mypy +checks pass. The documentation build remains unverified because the local Zensical file +watcher hit the environment's `Too many open files` (`EMFILE`) limit; this is recorded +as an environment limitation, not evidence that the documentation is invalid. 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 2f606fa9..907a0aae 100644 --- a/src/forge/api/routes/__init__.py +++ b/src/forge/api/routes/__init__.py @@ -1,15 +1,19 @@ """API route modules.""" +from forge.api.routes.executions import router as executions_router from forge.api.routes.github import router as github_router from forge.api.routes.health import router as health_router from forge.api.routes.jira import router as jira_router from forge.api.routes.metrics import router as metrics_router +from forge.api.routes.org_pulse import router as org_pulse_router __all__ = [ + "executions_router", "github_router", "effects_router", "health_router", "jira_router", "metrics_router", + "org_pulse_router", ] from forge.api.routes.effects import router as effects_router 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 new file mode 100644 index 00000000..9402c2c6 --- /dev/null +++ b/src/forge/api/routes/executions.py @@ -0,0 +1,149 @@ +"""Operator read API for durable workflow execution state.""" + +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, + 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 + +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, + *, + 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: + 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)) + 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 + + +@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: + 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)) + 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/api/routes/metrics.py b/src/forge/api/routes/metrics.py index 24087eef..d70a5f11 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,73 @@ 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/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/integrations/org_pulse.py b/src/forge/integrations/org_pulse.py new file mode 100644 index 00000000..183eed4c --- /dev/null +++ b/src/forge/integrations/org_pulse.py @@ -0,0 +1,69 @@ +"""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 75c7d0fc..bfc27c42 100644 --- a/src/forge/main.py +++ b/src/forge/main.py @@ -14,10 +14,12 @@ from forge.api.middleware.correlation import CorrelationIdMiddleware from forge.api.routes import ( effects_router, + executions_router, github_router, health_router, jira_router, metrics_router, + org_pulse_router, ) from forge.config import get_settings from forge.integrations.source_control.registry import get_registry @@ -147,6 +149,8 @@ def create_app() -> FastAPI: app.include_router(effects_router) 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 new file mode 100644 index 00000000..00833b56 --- /dev/null +++ b/src/forge/read_models/__init__.py @@ -0,0 +1,40 @@ +"""Operator-facing projections over durable workflow records.""" + +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", + "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 new file mode 100644 index 00000000..e5e36cab --- /dev/null +++ b/src/forge/read_models/execution.py @@ -0,0 +1,773 @@ +"""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, + 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( + checkpoint: Mapping[str, Any], + *, + 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) + 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, checkpoint, manifest) + 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=_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 + ), + ) + 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, + status=status, + current_position=position, + definition=definition, + permitted_commands=permitted, + 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, + 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 + 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=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=str( + checkpoint.get("recovery_reason") + or "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=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=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, + 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"] + 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") + + +def _observation( + observation: Observation | None, + checkpoint: Mapping[str, Any], + now: datetime, + stale_after: timedelta, + decisions: Sequence[Any] = (), +) -> 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=_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, ...]: + 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 + 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, + 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, + 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), + ), + ) + + +def _datetime(value: Any) -> datetime | None: + if isinstance(value, datetime): + return value + if isinstance(value, str) and value: + return datetime.fromisoformat(value) + return None + + +def _timeline( + 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, ...]: + """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( + 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 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, + 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, + }, + ) + ) + # 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, + ), + ) + ) + + +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 new file mode 100644 index 00000000..bb94f4ca --- /dev/null +++ b/src/forge/read_models/models.py @@ -0,0 +1,148 @@ +"""Versioned, execution-neutral operator read models.""" + +from __future__ import annotations + +from datetime import datetime +from enum import StrEnum + +from pydantic import Field + +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 + 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): + 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 + attempts: tuple[EffectAttemptView, ...] = () + + +class MigrationView(VersionedDomainModel): + eligible: bool | None = None + 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 + status: ExecutionStatus + current_position: str + definition: DefinitionView + permitted_commands: tuple[str, ...] + 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..2465d18b --- /dev/null +++ b/src/forge/read_models/timeline.py @@ -0,0 +1,229 @@ +"""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..7a638c93 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,9 @@ 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 +275,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 bd12dceb..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 @@ -141,6 +144,10 @@ 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]] + 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/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/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/api/routes/test_executions.py b/tests/unit/api/routes/test_executions.py new file mode 100644 index 00000000..aa12b365 --- /dev/null +++ b/tests/unit/api/routes/test_executions.py @@ -0,0 +1,117 @@ +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 +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 + + +@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/__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..e93e5b06 --- /dev/null +++ b/tests/unit/read_models/test_execution.py @@ -0,0 +1,279 @@ +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" + + +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", + ] + + +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() 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",