From a2fc5639db04ac4452a73e3b445239d3dd92f538 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 8 Sep 2026 07:47:18 +0200 Subject: [PATCH] perf(workflow): batch revision reads and add bounded histories --- .../backend/db/models.py | 16 + .../backend/instance_reads.py | 305 ++++++++++ .../backend/instance_service.py | 88 +-- .../backend/manifest.py | 93 ++++ .../9e6b3f8a2c7d_instance_summary_indexes.py | 39 ++ .../backend/router.py | 112 ++++ .../backend/schemas.py | 37 ++ .../backend/work_items.py | 3 +- tests/test_instance_reads.py | 519 ++++++++++++++++++ tests/test_migrations.py | 23 +- 10 files changed, 1189 insertions(+), 46 deletions(-) create mode 100644 src/govoplan_workflow_engine/backend/instance_reads.py create mode 100644 src/govoplan_workflow_engine/backend/migrations/versions/9e6b3f8a2c7d_instance_summary_indexes.py create mode 100644 tests/test_instance_reads.py diff --git a/src/govoplan_workflow_engine/backend/db/models.py b/src/govoplan_workflow_engine/backend/db/models.py index ac31c7a..f1f2995 100644 --- a/src/govoplan_workflow_engine/backend/db/models.py +++ b/src/govoplan_workflow_engine/backend/db/models.py @@ -298,6 +298,19 @@ class WorkflowInstance(Base, TimestampMixin): "status", "updated_at", ), + Index( + "ix_workflow_instances_tenant_created_id", + "tenant_id", + "created_at", + "id", + ), + Index( + "ix_workflow_instances_tenant_definition_created_id", + "tenant_id", + "definition_id", + "created_at", + "id", + ), ) id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid) @@ -383,6 +396,9 @@ class WorkflowInstance(Base, TimestampMixin): ) definition: Mapped[WorkflowDefinition] = relationship(back_populates="instances") + definition_revision: Mapped[WorkflowDefinitionRevision] = relationship( + viewonly=True, + ) steps: Mapped[list["WorkflowInstanceStep"]] = relationship( back_populates="instance", cascade="all, delete-orphan", diff --git a/src/govoplan_workflow_engine/backend/instance_reads.py b/src/govoplan_workflow_engine/backend/instance_reads.py new file mode 100644 index 0000000..6bc9415 --- /dev/null +++ b/src/govoplan_workflow_engine/backend/instance_reads.py @@ -0,0 +1,305 @@ +"""Authorized, bounded projections for workflow progress and retained history.""" + +from __future__ import annotations + +import base64 +import binascii +from datetime import datetime +import json + +from sqlalchemy import and_, func, or_, select +from sqlalchemy.orm import Session, joinedload, load_only, raiseload + +from govoplan_core.auth import ApiPrincipal, has_scope +from govoplan_workflow_engine.backend.db.models import ( + WorkflowDefinitionRevision, + WorkflowInstance, + WorkflowInstanceEvent, + WorkflowInstanceStep, +) +from govoplan_workflow_engine.backend.governance import ( + definition_decision, + require_definition_action, +) +from govoplan_workflow_engine.backend.instance_service import ( + event_response, + step_response, +) +from govoplan_workflow_engine.backend.schemas import ( + WorkflowInstanceEventPageResponse, + WorkflowInstanceStepPageResponse, + WorkflowInstanceSummaryListResponse, + WorkflowInstanceSummaryResponse, +) +from govoplan_workflow_engine.backend.service import WorkflowNotFoundError + + +MAX_PAGE_SIZE = 200 +MAX_HISTORY_SEQUENCE = 2_147_483_647 +_CANDIDATE_BATCH_SIZE = 250 + + +def _require_read(principal: ApiPrincipal) -> None: + if not any( + has_scope(principal, scope) + for scope in ("workflow:instance:read", "workflow:instance:admin") + ): + raise PermissionError("Workflow instance read permission is required.") + + +def _summary_statement(): + # Raiseload makes accidental history/payload expansion fail visibly. Only the + # pinned revision metadata is needed, never its graph or BPMN document. + return ( + select(WorkflowInstance) + .options( + load_only( + WorkflowInstance.tenant_id, + WorkflowInstance.definition_id, + WorkflowInstance.definition_revision_id, + WorkflowInstance.status, + WorkflowInstance.start_origin, + WorkflowInstance.current_step_id, + WorkflowInstance.started_at, + WorkflowInstance.finished_at, + WorkflowInstance.created_at, + WorkflowInstance.updated_at, + raiseload=True, + ), + joinedload(WorkflowInstance.definition), + joinedload(WorkflowInstance.definition_revision).load_only( + WorkflowDefinitionRevision.tenant_id, + WorkflowDefinitionRevision.definition_id, + WorkflowDefinitionRevision.revision, + WorkflowDefinitionRevision.content_hash, + WorkflowDefinitionRevision.execution_mode, + raiseload=True, + ), + raiseload("*"), + ) + .execution_options(populate_existing=True) + ) + + +def _require_evidence(instance: WorkflowInstance) -> None: + definition = instance.definition + revision = instance.definition_revision + if ( + definition is None + or revision is None + or revision.id != instance.definition_revision_id + or definition.id != instance.definition_id + or definition.tenant_id not in {None, instance.tenant_id} + or revision.definition_id != definition.id + or revision.tenant_id != definition.tenant_id + ): + raise WorkflowNotFoundError( + "Workflow instance definition evidence is incomplete." + ) + + +def _summary(instance: WorkflowInstance) -> WorkflowInstanceSummaryResponse: + _require_evidence(instance) + return WorkflowInstanceSummaryResponse( + id=instance.id, + definition_id=instance.definition_id, + definition_name=instance.definition.name, + definition_revision=instance.definition_revision.revision, + definition_hash=instance.definition_revision.content_hash, + execution_mode=instance.definition_revision.execution_mode, + start_origin=instance.start_origin, + status=instance.status, + current_step_id=instance.current_step_id, + started_at=instance.started_at, + finished_at=instance.finished_at, + created_at=instance.created_at, + updated_at=instance.updated_at, + ) + + +def _cursor(instance: WorkflowInstance) -> str: + return base64.urlsafe_b64encode( + json.dumps([instance.created_at.isoformat(), instance.id]).encode("utf-8") + ).decode("ascii") + + +def _cursor_boundary(cursor: str) -> tuple[datetime, str]: + try: + if len(cursor) > 256: + raise ValueError + payload = json.loads(base64.b64decode(cursor, altchars=b"-_", validate=True)) + if ( + not isinstance(payload, list) + or len(payload) != 2 + or not all(isinstance(value, str) for value in payload) + or not 1 <= len(payload[1]) <= 36 + ): + raise ValueError + return datetime.fromisoformat(payload[0]), payload[1] + except (ValueError, TypeError, UnicodeError, binascii.Error) as exc: + raise ValueError("Invalid workflow instance cursor.") from exc + + +def list_instance_summaries( + session: Session, + *, + principal: ApiPrincipal, + registry: object | None, + definition_id: str | None = None, + limit: int = 100, + cursor: str | None = None, +) -> WorkflowInstanceSummaryListResponse: + _require_read(principal) + if not 1 <= limit <= MAX_PAGE_SIZE: + raise ValueError("Workflow page size must be between 1 and 200.") + boundary = _cursor_boundary(cursor) if cursor is not None else None + statement = _summary_statement().where( + WorkflowInstance.tenant_id == principal.tenant_id, + ) + if definition_id is not None: + statement = statement.where(WorkflowInstance.definition_id == definition_id) + statement = statement.order_by( + WorkflowInstance.created_at.desc(), WorkflowInstance.id.desc() + ) + items: list[WorkflowInstanceSummaryResponse] = [] + last_cursor: str | None = None + # Authorization can depend on a Core policy provider, so a raw SQL count or + # SQL limit before authorization cannot define an authorized page. Scan in + # bounded batches until the page and one authorized lookahead are found. + while True: + decisions: dict[str, bool] = {} + batch_statement = statement + if boundary is not None: + created_at, instance_id = boundary + batch_statement = batch_statement.where( + or_( + WorkflowInstance.created_at < created_at, + and_( + WorkflowInstance.created_at == created_at, + WorkflowInstance.id < instance_id, + ), + ) + ) + candidates = list(session.scalars(batch_statement.limit(_CANDIDATE_BATCH_SIZE))) + for instance in candidates: + _require_evidence(instance) + allowed = decisions.get(instance.definition_id) + if allowed is None: + allowed = definition_decision( + instance.definition, + principal=principal, + registry=registry, + action="view", + ).allowed + decisions[instance.definition_id] = allowed + if not allowed: + continue + if len(items) == limit: + return WorkflowInstanceSummaryListResponse( + instances=items, next_cursor=last_cursor + ) + items.append(_summary(instance)) + last_cursor = _cursor(instance) + if len(candidates) < _CANDIDATE_BATCH_SIZE: + return WorkflowInstanceSummaryListResponse(instances=items) + last = candidates[-1] + boundary = (last.created_at, last.id) + + +def _authorized_instance( + session: Session, + *, + principal: ApiPrincipal, + registry: object | None, + instance_id: str, +) -> WorkflowInstance: + _require_read(principal) + instance = session.scalar( + _summary_statement().where( + WorkflowInstance.tenant_id == principal.tenant_id, + WorkflowInstance.id == instance_id, + ) + ) + if instance is None: + raise WorkflowNotFoundError("Workflow instance not found.") + _require_evidence(instance) + require_definition_action( + instance.definition, principal=principal, registry=registry, action="view" + ) + return instance + + +def get_instance_summary( + session: Session, + *, + principal: ApiPrincipal, + registry: object | None, + instance_id: str, +) -> WorkflowInstanceSummaryResponse: + return _summary( + _authorized_instance( + session, principal=principal, registry=registry, instance_id=instance_id + ) + ) + + +def instance_history_page( + session: Session, + *, + principal: ApiPrincipal, + registry: object | None, + instance_id: str, + kind: str, + limit: int = 100, + after_sequence: int = 0, + through_sequence: int | None = None, +) -> WorkflowInstanceStepPageResponse | WorkflowInstanceEventPageResponse: + if kind not in {"steps", "events"}: + raise ValueError("Workflow history kind must be steps or events.") + if ( + not 1 <= limit <= MAX_PAGE_SIZE + or not 0 <= after_sequence <= MAX_HISTORY_SEQUENCE + or ( + through_sequence is not None + and not 0 <= through_sequence <= MAX_HISTORY_SEQUENCE + ) + ): + raise ValueError("Invalid workflow history page bounds.") + _authorized_instance( + session, principal=principal, registry=registry, instance_id=instance_id + ) + model = WorkflowInstanceStep if kind == "steps" else WorkflowInstanceEvent + predicates = [ + model.tenant_id == principal.tenant_id, + model.instance_id == instance_id, + ] + if through_sequence is not None: + predicates.append(model.sequence <= through_sequence) + total, last_sequence = session.execute( + select(func.count(), func.max(model.sequence)).where(*predicates) + ).one() + upper = int(last_sequence or 0) + rows = list( + session.scalars( + select(model) + .where( + *predicates, model.sequence > after_sequence, model.sequence <= upper + ) + .order_by(model.sequence.asc()) + .limit(limit + 1) + .execution_options(populate_existing=True) + ) + ) + next_after = rows[limit - 1].sequence if len(rows) > limit else None + common = { + "total": int(total), + "through_sequence": upper, + "next_after_sequence": next_after, + } + if kind == "steps": + return WorkflowInstanceStepPageResponse( + steps=[step_response(row) for row in rows[:limit]], **common + ) + return WorkflowInstanceEventPageResponse( + events=[event_response(row) for row in rows[:limit]], **common + ) diff --git a/src/govoplan_workflow_engine/backend/instance_service.py b/src/govoplan_workflow_engine/backend/instance_service.py index c29048e..41d69d6 100644 --- a/src/govoplan_workflow_engine/backend/instance_service.py +++ b/src/govoplan_workflow_engine/backend/instance_service.py @@ -38,7 +38,6 @@ from govoplan_core.core.events import ( ) from govoplan_core.db.base import utcnow from govoplan_workflow_engine.backend.db.models import ( - WorkflowDefinition, WorkflowDefinitionRevision, WorkflowInstance, WorkflowInstanceEvent, @@ -100,6 +99,7 @@ def list_instances( selectinload(WorkflowInstance.steps), selectinload(WorkflowInstance.events), selectinload(WorkflowInstance.definition), + selectinload(WorkflowInstance.definition_revision), ) .order_by( WorkflowInstance.updated_at.desc(), @@ -129,6 +129,7 @@ def get_instance( selectinload(WorkflowInstance.steps), selectinload(WorkflowInstance.events), selectinload(WorkflowInstance.definition), + selectinload(WorkflowInstance.definition_revision), ) ) if for_update: @@ -861,12 +862,17 @@ def instance_response( *, replayed: bool = False, ) -> WorkflowInstanceResponse: - revision = session.get( - WorkflowDefinitionRevision, - instance.definition_revision_id, - ) - definition = session.get(WorkflowDefinition, instance.definition_id) - if revision is None or definition is None: + revision = instance.definition_revision + definition = instance.definition + if ( + revision is None + or definition is None + or revision.id != instance.definition_revision_id + or definition.id != instance.definition_id + or definition.tenant_id not in {None, instance.tenant_id} + or revision.definition_id != definition.id + or revision.tenant_id != definition.tenant_id + ): raise WorkflowNotFoundError( "Workflow instance definition evidence is incomplete." ) @@ -894,43 +900,45 @@ def instance_response( created_by=instance.created_by, created_at=instance.created_at, updated_at=instance.updated_at, - steps=[ - WorkflowInstanceStepResponse( - id=step.id, - sequence=step.sequence, - node_id=step.node_id, - node_type=step.node_type, - status=step.status, # type: ignore[arg-type] - attempt=step.attempt, - input=dict(step.input_), - output=dict(step.output_), - handoff=dict(step.handoff), - external_ref=step.external_ref, - started_at=step.started_at, - finished_at=step.finished_at, - error=step.error, - completed_by=step.completed_by, - created_at=step.created_at, - updated_at=step.updated_at, - ) - for step in instance.steps - ], - events=[ - WorkflowInstanceEventResponse( - id=event.id, - sequence=event.sequence, - step_id=event.step_id, - kind=event.kind, - actor_id=event.actor_id, - payload=dict(event.payload), - created_at=event.created_at, - ) - for event in instance.events - ], + steps=[step_response(step) for step in instance.steps], + events=[event_response(event) for event in instance.events], replayed=replayed, ) +def step_response(step: WorkflowInstanceStep) -> WorkflowInstanceStepResponse: + return WorkflowInstanceStepResponse( + id=step.id, + sequence=step.sequence, + node_id=step.node_id, + node_type=step.node_type, + status=step.status, # type: ignore[arg-type] + attempt=step.attempt, + input=dict(step.input_), + output=dict(step.output_), + handoff=dict(step.handoff), + external_ref=step.external_ref, + started_at=step.started_at, + finished_at=step.finished_at, + error=step.error, + completed_by=step.completed_by, + created_at=step.created_at, + updated_at=step.updated_at, + ) + + +def event_response(event: WorkflowInstanceEvent) -> WorkflowInstanceEventResponse: + return WorkflowInstanceEventResponse( + id=event.id, + sequence=event.sequence, + step_id=event.step_id, + kind=event.kind, + actor_id=event.actor_id, + payload=dict(event.payload), + created_at=event.created_at, + ) + + def _drive_instance( session: Session, *, diff --git a/src/govoplan_workflow_engine/backend/manifest.py b/src/govoplan_workflow_engine/backend/manifest.py index 4364145..95b5efe 100644 --- a/src/govoplan_workflow_engine/backend/manifest.py +++ b/src/govoplan_workflow_engine/backend/manifest.py @@ -389,6 +389,99 @@ manifest = ModuleManifest( ), ), documentation=( + DocumentationTopic( + id="workflow.instance-history", + title="Read workflow progress and paged history", + summary="Inspect pinned execution evidence with explicit summary and history pages.", + body=( + "The Workflow Engine API keeps full steps and events on the existing " + "GET /instances and GET /instances/{id} responses; no history is silently " + "truncated. Integrators handling long-running processes can use " + "GET /instances/summaries and GET /instances/{id}/summary for identity, " + "pinned definition revision and hash, execution mode, status, current step " + "identifier, and timestamps without runtime payloads or history. Summary " + "lists accept definition_id, limit (default 100, maximum 200), and cursor. " + "They order by creation time and identifier, newest first. Continue with " + "next_cursor until it is null, keeping the same filter. Pages include only " + "currently visible definitions and do not expose a global total. Newer " + "instances require a fresh first page; updates do not reorder older pages. " + "GET /instances/{id}/steps and /events accept limit (default 100, maximum " + "200), after_sequence (initially 0), and through_sequence (sequence bounds " + "range from 0 to 2147483647). Continue using " + "next_after_sequence and the returned through_sequence until the next " + "position is null. Rows are ordered by increasing sequence; total counts " + "all retained rows up to that boundary, including earlier pages. Keeping " + "the boundary excludes subsequently appended history; omit it for a fresh " + "pass. Step fields reflect their current state, not a historical database " + "snapshot. Each request requires workflow:instance:read or " + "workflow:instance:admin and current definition view authorization in " + "the active tenant. A cursor and the authorization recorded at start " + "never grant access after permission changes. Administrators should use " + "summary and history pages for large runs: revision metadata is loaded " + "in batches, history rows are bounded, and summary discovery checks " + "candidates in batches of 250 until an authorized page is filled. Highly " + "restricted catalogues can still require scanning many candidates. " + "The module migration adds tenant and definition creation-order indexes; " + "building them on a large installation needs disk capacity and may " + "temporarily block writes during the normal database upgrade. " + "Work inbox totals remain exact after current assignment, governance, " + "and search filters; counting them still streams all matching candidates " + "in batches of 250 and can take longer for large inboxes. These read " + "contracts require no editor, Tasks installation, or new permission." + ), + layer="available", + documentation_types=("admin", "user"), + audience=("operator", "module_admin", "integrator", "auditor"), + order=74, + translations={ + "de": { + "title": "Workflow-Fortschritt und paginierten Verlauf lesen", + "summary": "Revisionsgebundene Ausführungsnachweise mit expliziten Übersichts- und Verlaufsseiten prüfen.", + "body": ( + "Die Workflow-Engine-API liefert bei GET /instances und GET " + "/instances/{id} weiterhin vollständige Schritte und Ereignisse; " + "Verläufe werden nicht stillschweigend gekürzt. Für lange Prozesse " + "liefern GET /instances/summaries und GET /instances/{id}/summary " + "Identität, gebundene Definitionsrevision und Hash, Ausführungsmodus, " + "Status, aktuelle Schrittkennung und Zeitstempel ohne Laufzeitnutzdaten " + "oder Verlauf. Übersichtslisten akzeptieren definition_id, limit " + "(Standard 100, höchstens 200) und cursor. Sie sortieren nach " + "Erstellungszeit und Kennung, neueste zuerst. Mit unverändertem Filter " + "und next_cursor fortfahren, bis dieser null ist. Seiten enthalten nur " + "aktuell sichtbare Definitionen und keine globale Gesamtzahl. Neue " + "Instanzen erfordern eine neue erste Seite; Aktualisierungen verschieben " + "ältere Seiten nicht. GET /instances/{id}/steps und /events akzeptieren " + "limit (Standard 100, höchstens 200), after_sequence (anfangs 0) und " + "through_sequence (Sequenzgrenzen von 0 bis 2147483647). Mit " + "next_after_sequence und der zurückgegebenen " + "through_sequence fortfahren, bis die nächste Position null ist. Zeilen " + "sind nach aufsteigender Sequenz sortiert; total zählt alle erhaltenen " + "Zeilen bis zur Grenze einschließlich vorheriger Seiten. Eine feste " + "Grenze schließt später angehängte Einträge aus; für einen neuen " + "Durchlauf wird sie weggelassen. Schrittfelder zeigen ihren aktuellen " + "Zustand, keinen historischen Datenbank-Snapshot. Jede Anfrage benötigt " + "workflow:instance:read oder workflow:instance:admin und die aktuelle " + "Sichtberechtigung für die Definition im aktiven Mandanten. Weder ein " + "Cursor noch die beim Start gespeicherte Autorisierung gewährt Zugriff " + "nach Berechtigungsänderungen. Administratoren sollten für große Läufe " + "Übersichts- und Verlaufsseiten verwenden: Revisionsmetadaten werden " + "gebündelt geladen, Verlaufszeilen sind begrenzt, und die Übersicht " + "prüft Kandidaten in Gruppen von 250, bis eine berechtigte Seite gefüllt " + "ist. Stark eingeschränkte Kataloge können weiterhin viele " + "Kandidatenprüfungen erfordern. Die Modulmigration ergänzt nach " + "Erstellungszeit geordnete Mandanten- und Definitionsindizes. Deren " + "Aufbau benötigt bei großen Installationen Speicherplatz und kann " + "während der regulären Datenbankaktualisierung Schreibzugriffe " + "vorübergehend blockieren. Gesamtzahlen im Arbeitskorb bleiben " + "nach aktuellen Zuweisungs-, Governance- und Suchfiltern exakt. Dafür " + "werden weiterhin alle passenden Kandidaten in Gruppen von 250 " + "durchlaufen; große Arbeitskörbe können länger dauern. Diese " + "Leseverträge benötigen weder Editor noch Tasks-Installation noch " + "eine neue Berechtigung." + ), + }, + }, + ), DocumentationTopic( id="workflow.external-campaign-handoffs", title="Resume Workflow from accountable Campaign work", diff --git a/src/govoplan_workflow_engine/backend/migrations/versions/9e6b3f8a2c7d_instance_summary_indexes.py b/src/govoplan_workflow_engine/backend/migrations/versions/9e6b3f8a2c7d_instance_summary_indexes.py new file mode 100644 index 0000000..8bee2bc --- /dev/null +++ b/src/govoplan_workflow_engine/backend/migrations/versions/9e6b3f8a2c7d_instance_summary_indexes.py @@ -0,0 +1,39 @@ +"""Index tenant-scoped workflow summary pagination. + +Revision ID: 9e6b3f8a2c7d +Revises: 8d5a2f7c1b4e +""" + +from __future__ import annotations + +from alembic import op + + +revision = "9e6b3f8a2c7d" +down_revision = "8d5a2f7c1b4e" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_index( + "ix_workflow_instances_tenant_created_id", + "workflow_instances", + ["tenant_id", "created_at", "id"], + ) + op.create_index( + "ix_workflow_instances_tenant_definition_created_id", + "workflow_instances", + ["tenant_id", "definition_id", "created_at", "id"], + ) + + +def downgrade() -> None: + op.drop_index( + "ix_workflow_instances_tenant_definition_created_id", + table_name="workflow_instances", + ) + op.drop_index( + "ix_workflow_instances_tenant_created_id", + table_name="workflow_instances", + ) diff --git a/src/govoplan_workflow_engine/backend/router.py b/src/govoplan_workflow_engine/backend/router.py index 3124222..2d7ddd9 100644 --- a/src/govoplan_workflow_engine/backend/router.py +++ b/src/govoplan_workflow_engine/backend/router.py @@ -83,8 +83,12 @@ from govoplan_workflow_engine.backend.schemas import ( WorkflowGraphValidationRequest, WorkflowGraphValidationResponse, WorkflowInstanceListResponse, + WorkflowInstanceEventPageResponse, WorkflowInstanceResponse, WorkflowInstanceStartRequest, + WorkflowInstanceStepPageResponse, + WorkflowInstanceSummaryListResponse, + WorkflowInstanceSummaryResponse, WorkflowNodeLibraryResponse, WorkflowNodeTypeResponse, WorkflowPortResponse, @@ -102,6 +106,12 @@ from govoplan_workflow_engine.backend.instance_service import ( resolve_step, start_instance, ) +from govoplan_workflow_engine.backend.instance_reads import ( + MAX_HISTORY_SEQUENCE, + get_instance_summary, + instance_history_page, + list_instance_summaries, +) from govoplan_workflow_engine.backend.runtime import get_registry from govoplan_workflow_engine.backend.service import ( WorkflowBpmnValidationError, @@ -699,6 +709,31 @@ def api_list_instances( raise _http_error(exc) from exc +@router.get("/instances/summaries", response_model=WorkflowInstanceSummaryListResponse) +def api_list_instance_summaries( + definition_id: str | None = None, + limit: int = Query(default=100, ge=1, le=200), + cursor: str | None = Query(default=None, max_length=256), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowInstanceSummaryListResponse: + try: + return list_instance_summaries( + session, + principal=principal, + registry=get_registry(), + definition_id=definition_id, + limit=limit, + cursor=cursor, + ) + except PermissionError as exc: + raise _governance_http_error(exc) from exc + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + except WorkflowError as exc: + raise _http_error(exc) from exc + + @router.post("/standards/reconcile", response_model=dict[str, object]) def api_reconcile_standards( session: Session = Depends(get_session), @@ -875,6 +910,83 @@ def api_get_instance( raise _http_error(exc) from exc +@router.get( + "/instances/{instance_id}/summary", response_model=WorkflowInstanceSummaryResponse +) +def api_get_instance_summary( + instance_id: str, + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowInstanceSummaryResponse: + try: + return get_instance_summary( + session, + principal=principal, + registry=get_registry(), + instance_id=instance_id, + ) + except PermissionError as exc: + raise _governance_http_error(exc) from exc + except WorkflowError as exc: + raise _http_error(exc) from exc + + +@router.get( + "/instances/{instance_id}/steps", response_model=WorkflowInstanceStepPageResponse +) +def api_list_instance_steps( + instance_id: str, + limit: int = Query(default=100, ge=1, le=200), + after_sequence: int = Query(default=0, ge=0, le=MAX_HISTORY_SEQUENCE), + through_sequence: int | None = Query(default=None, ge=0, le=MAX_HISTORY_SEQUENCE), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowInstanceStepPageResponse: + try: + return instance_history_page( + session, + principal=principal, + registry=get_registry(), + instance_id=instance_id, + kind="steps", + limit=limit, + after_sequence=after_sequence, + through_sequence=through_sequence, + ) + except PermissionError as exc: + raise _governance_http_error(exc) from exc + except WorkflowError as exc: + raise _http_error(exc) from exc + + +@router.get( + "/instances/{instance_id}/events", response_model=WorkflowInstanceEventPageResponse +) +def api_list_instance_events( + instance_id: str, + limit: int = Query(default=100, ge=1, le=200), + after_sequence: int = Query(default=0, ge=0, le=MAX_HISTORY_SEQUENCE), + through_sequence: int | None = Query(default=None, ge=0, le=MAX_HISTORY_SEQUENCE), + session: Session = Depends(get_session), + principal: ApiPrincipal = Depends(get_api_principal), +) -> WorkflowInstanceEventPageResponse: + try: + return instance_history_page( + session, + principal=principal, + registry=get_registry(), + instance_id=instance_id, + kind="events", + limit=limit, + after_sequence=after_sequence, + through_sequence=through_sequence, + ) + except PermissionError as exc: + raise _governance_http_error(exc) from exc + except WorkflowError as exc: + raise _http_error(exc) from exc + + @router.post( "/instances/{instance_id}/reconcile", response_model=WorkflowInstanceResponse, diff --git a/src/govoplan_workflow_engine/backend/schemas.py b/src/govoplan_workflow_engine/backend/schemas.py index b90db85..9e7a831 100644 --- a/src/govoplan_workflow_engine/backend/schemas.py +++ b/src/govoplan_workflow_engine/backend/schemas.py @@ -628,3 +628,40 @@ class WorkflowInstanceResponse(BaseModel): class WorkflowInstanceListResponse(BaseModel): instances: list[WorkflowInstanceResponse] + + +class WorkflowInstanceSummaryResponse(BaseModel): + """Instance identity and progress without runtime payloads or history.""" + + id: str + definition_id: str + definition_name: str + definition_revision: int + definition_hash: str + execution_mode: WorkflowExecutionMode + start_origin: WorkflowStartOrigin + status: WorkflowInstanceStatus + current_step_id: str | None + started_at: datetime + finished_at: datetime | None + created_at: datetime + updated_at: datetime + + +class WorkflowInstanceSummaryListResponse(BaseModel): + instances: list[WorkflowInstanceSummaryResponse] + next_cursor: str | None = None + + +class WorkflowInstanceStepPageResponse(BaseModel): + steps: list[WorkflowInstanceStepResponse] + total: int + through_sequence: int + next_after_sequence: int | None = None + + +class WorkflowInstanceEventPageResponse(BaseModel): + events: list[WorkflowInstanceEventResponse] + total: int + through_sequence: int + next_after_sequence: int | None = None diff --git a/src/govoplan_workflow_engine/backend/work_items.py b/src/govoplan_workflow_engine/backend/work_items.py index e20c31f..6e677b2 100644 --- a/src/govoplan_workflow_engine/backend/work_items.py +++ b/src/govoplan_workflow_engine/backend/work_items.py @@ -87,8 +87,8 @@ class WorkflowWorkItemProvider: WorkflowInstanceStep.id.desc(), ) ) + targets = self._targets(principal, query.tenant_id) if not administrative: - targets = self._targets(principal, query.tenant_id) conditions = [ and_( WorkflowInstanceStep.work_assignment_kind == kind, @@ -108,7 +108,6 @@ class WorkflowWorkItemProvider: items: list[WorkItem] = [] total = 0 decisions: dict[str, bool] = {} - targets = self._targets(principal, query.tenant_id) for instance, step, definition in session.execute(statement).yield_per(250): assignment = _step_assignment(instance, step) if not administrative and not _assignment_matches(assignment, targets): diff --git a/tests/test_instance_reads.py b/tests/test_instance_reads.py new file mode 100644 index 0000000..ebddf13 --- /dev/null +++ b/tests/test_instance_reads.py @@ -0,0 +1,519 @@ +from __future__ import annotations + +from contextlib import contextmanager +from dataclasses import replace +from datetime import UTC, datetime, timedelta +import unittest +from unittest.mock import patch + +from fastapi import FastAPI +from fastapi.testclient import TestClient +from sqlalchemy import create_engine, event +from sqlalchemy.orm import Session +from sqlalchemy.pool import StaticPool + +from govoplan_core.auth import ApiPrincipal, get_api_principal +from govoplan_core.core.access import PrincipalRef +from govoplan_core.core.tasks import WorkItemQuery +from govoplan_core.db.base import Base +from govoplan_core.db.session import get_session +from govoplan_workflow_engine.backend.db.models import ( + WorkflowDefinition, + WorkflowDefinitionRevision, + WorkflowInstance, + WorkflowInstanceEvent, + WorkflowInstanceStep, +) +from govoplan_workflow_engine.backend.instance_reads import ( + get_instance_summary, + instance_history_page, + list_instance_summaries, +) +from govoplan_workflow_engine.backend.instance_service import ( + instance_response, + list_instances, +) +from govoplan_workflow_engine.backend.router import router +from govoplan_workflow_engine.backend.service import WorkflowNotFoundError +from govoplan_workflow_engine.backend.work_items import WorkflowWorkItemProvider + + +def principal() -> ApiPrincipal: + return ApiPrincipal( + principal=PrincipalRef( + account_id="account-1", + membership_id="membership-1", + tenant_id="tenant-1", + scopes=frozenset({"workflow:instance:read"}), + ), + account=object(), + user=object(), + ) + + +class WorkflowInstanceReadTests(unittest.TestCase): + def setUp(self) -> None: + self.engine = create_engine( + "sqlite://", + poolclass=StaticPool, + connect_args={"check_same_thread": False}, + ) + Base.metadata.create_all( + self.engine, + tables=[ + WorkflowDefinition.__table__, + WorkflowDefinitionRevision.__table__, + WorkflowInstance.__table__, + WorkflowInstanceStep.__table__, + WorkflowInstanceEvent.__table__, + ], + ) + self.session = Session(self.engine) + self.actor = principal() + + def tearDown(self) -> None: + self.session.close() + self.engine.dispose() + + @contextmanager + def statements(self): + statements = [] + + def capture(_connection, _cursor, statement, _parameters, _context, _many): + statements.append(statement) + + event.listen(self.engine, "before_cursor_execute", capture) + try: + yield statements + finally: + event.remove(self.engine, "before_cursor_execute", capture) + + def seed(self, index: int, *, tenant_id="tenant-1", scope_type="tenant", rows=3): + now = datetime(2026, 1, 1, tzinfo=UTC) + definition = WorkflowDefinition( + id=f"definition-{index}", + tenant_id=tenant_id, + scope_type=scope_type, + scope_id=tenant_id if scope_type == "tenant" else "group-other", + scope_key=f"{scope_type}:{tenant_id}:{index}", + definition_key=f"workflow-{index}", + name=f"Workflow {index}", + status="active", + current_revision=index + 1, + active_revision=index + 1, + ) + revision = WorkflowDefinitionRevision( + id=f"revision-{index}", + tenant_id=tenant_id, + definition=definition, + revision=index + 1, + graph={"sensitive_graph": "not needed for summary"}, + bpmn_xml="sensitive BPMN", + content_hash=f"{index:064x}", + library_id="workflow", + library_version="1", + ) + instance = WorkflowInstance( + id=f"instance-{index:05d}", + tenant_id=tenant_id, + definition=definition, + definition_revision_id=revision.id, + status="waiting", + idempotency_key=f"key-{index}", + authorization_={"account_id": "account-1"}, + input_={"sensitive": "input"}, + context_={"sensitive": "context"}, + output_={"sensitive": "output"}, + started_at=now, + created_at=now, + updated_at=now, + ) + self.session.add_all([definition, revision, instance]) + for sequence in range(1, rows + 1): + step = WorkflowInstanceStep( + id=f"step-{index}-{sequence}", + tenant_id=tenant_id, + instance=instance, + sequence=sequence, + node_id="activity", + node_type="workflow.activity", + status="waiting", + idempotency_key=f"step-key-{index}-{sequence}", + handoff={"kind": "activity", "title": "Review application"}, + work_assignment_kind="account", + work_assignment_id="account-1", + ) + self.session.add(step) + self.session.add( + WorkflowInstanceEvent( + id=f"event-{index}-{sequence}", + tenant_id=tenant_id, + instance=instance, + step_id=step.id, + sequence=sequence, + kind="workflow.step.started", + payload={"sequence": sequence}, + created_at=now, + ) + ) + instance.current_step_id = step.id + return instance + + def summaries(self, **kwargs): + return list_instance_summaries( + self.session, principal=self.actor, registry=None, **kwargs + ) + + def history(self, instance_id="instance-00001", **kwargs): + return instance_history_page( + self.session, + principal=self.actor, + registry=None, + instance_id=instance_id, + **kwargs, + ) + + def test_cold_40_distinct_revisions_have_constant_query_count_and_full_history( + self, + ): + for index in range(40): + self.seed(index) + self.seed(100, tenant_id="other-tenant") + self.session.commit() + self.session.expunge_all() + with self.statements() as queries: + instances = list_instances(self.session, tenant_id="tenant-1", limit=40) + responses = [instance_response(self.session, item) for item in instances] + self.assertEqual(5, len(queries)) + self.assertEqual(40, len(responses)) + self.assertEqual( + set(range(1, 41)), {item.definition_revision for item in responses} + ) + for item in responses: + self.assertEqual( + f"{item.definition_revision - 1:064x}", item.definition_hash + ) + self.assertEqual([1, 2, 3], [step.sequence for step in item.steps]) + self.assertEqual([1, 2, 3], [entry.sequence for entry in item.events]) + + def test_summary_query_omits_histories_graphs_and_runtime_payloads(self): + for index in range(40): + self.seed(index) + self.session.commit() + self.session.expunge_all() + with self.statements() as queries: + page = self.summaries(limit=40) + self.assertEqual(1, len(queries)) + self.assertEqual(40, len(page.instances)) + self.assertIsNone(page.next_cursor) + sql = queries[0] + for forbidden in ( + "workflow_instance_steps", + "workflow_instance_events", + ".graph", + ".bpmn_xml", + "workflow_instances.input", + "workflow_instances.context", + "workflow_instances.output", + "workflow_instances.authorization", + ): + self.assertNotIn(forbidden, sql) + self.assertNotIn("steps", page.instances[0].model_dump()) + self.assertNotIn("events", page.instances[0].model_dump()) + self.assertNotIn("total", page.model_dump()) + + def test_summary_queries_use_ordered_tenant_indexes(self): + self.seed(1, rows=0) + self.session.commit() + for definition_id, index_name in ( + (None, "ix_workflow_instances_tenant_created_id"), + ("definition-1", "ix_workflow_instances_tenant_definition_created_id"), + ): + captured = [] + + def capture(_connection, _cursor, sql, parameters, _context, _many): + captured.append((sql, parameters)) + + event.listen(self.engine, "before_cursor_execute", capture) + try: + self.summaries(definition_id=definition_id) + finally: + event.remove(self.engine, "before_cursor_execute", capture) + sql, parameters = captured[0] + plan = " ".join( + row[3] + for row in self.session.connection().exec_driver_sql( + "EXPLAIN QUERY PLAN " + sql, parameters + ) + ) + self.assertIn(index_name, plan) + self.assertNotIn("TEMP B-TREE", plan) + + def test_summary_pages_fill_after_current_governance_and_preserve_ties(self): + for index in range(5): + self.seed(index, rows=0) + # More inaccessible candidates than one SQL batch precede the visible rows. + for index in range(10, 270): + self.seed(index, scope_type="group", rows=0) + self.seed(300, tenant_id="other-tenant", rows=0) + self.session.commit() + self.session.expunge_all() + with self.statements() as queries: + first = self.summaries(limit=2) + self.assertEqual(2, len(queries)) + self.assertEqual( + ["instance-00004", "instance-00003"], [i.id for i in first.instances] + ) + self.assertIsNotNone(first.next_cursor) + # Updating progress must not move an item across the creation-order cursor. + self.session.get(WorkflowInstance, "instance-00002").updated_at += timedelta( + days=2 + ) + second = self.summaries(limit=2, cursor=first.next_cursor) + third = self.summaries(limit=2, cursor=second.next_cursor) + self.assertEqual( + ["instance-00002", "instance-00001"], [i.id for i in second.instances] + ) + self.assertEqual(["instance-00000"], [i.id for i in third.instances]) + self.assertIsNone(third.next_cursor) + filtered = self.summaries(definition_id="definition-2", limit=1) + self.assertEqual(["instance-00002"], [i.id for i in filtered.instances]) + self.assertIsNone(filtered.next_cursor) + + def test_summary_and_history_recheck_governance_in_the_same_session(self): + self.seed(1) + self.session.commit() + first = self.summaries(limit=1) + self.assertEqual(1, len(first.instances)) + initial = self.history(kind="events", limit=1) + # Simulate governance changing after a prior request, including a warm + # identity map. Retained instance authorization is not a read grant. + with Session(self.engine) as writer: + definition = writer.get(WorkflowDefinition, "definition-1") + definition.scope_type = "group" + definition.scope_id = "group-other" + writer.commit() + self.assertEqual([], self.summaries(limit=1).instances) + with self.assertRaises(PermissionError): + self.history(kind="events", after_sequence=initial.next_after_sequence) + with self.assertRaises(PermissionError): + get_instance_summary( + self.session, + principal=self.actor, + registry=None, + instance_id="instance-00001", + ) + + def test_history_has_exact_total_and_fixed_boundary_without_silent_truncation(self): + self.seed(1, rows=405) + self.seed(2, tenant_id="other-tenant") + self.session.commit() + self.session.expunge_all() + for kind in ("steps", "events"): + with self.subTest(kind=kind): + model = ( + WorkflowInstanceStep if kind == "steps" else WorkflowInstanceEvent + ) + loaded = [] + + def record_load(row, _context): + loaded.append(row.id) + + event.listen(model, "load", record_load) + try: + with self.statements() as queries: + first = self.history(kind=kind, limit=200) + finally: + event.remove(model, "load", record_load) + self.assertEqual(3, len(queries)) + self.assertEqual(201, len(loaded)) + self.assertEqual(405, first.total) + self.assertEqual(405, first.through_sequence) + self.assertEqual(200, first.next_after_sequence) + self.assertEqual(200, len(getattr(first, kind))) + second = self.history( + kind=kind, + limit=200, + after_sequence=first.next_after_sequence, + through_sequence=first.through_sequence, + ) + third = self.history( + kind=kind, + limit=200, + after_sequence=second.next_after_sequence, + through_sequence=first.through_sequence, + ) + self.assertEqual(405, third.total) + self.assertIsNone(third.next_after_sequence) + sequences = [ + row.sequence + for page in (first, second, third) + for row in getattr(page, kind) + ] + self.assertEqual(list(range(1, 406)), sequences) + empty = self.history(kind=kind, after_sequence=1000) + self.assertEqual([], getattr(empty, kind)) + self.assertEqual(405, empty.total) + self.session.add( + WorkflowInstanceEvent( + id="later", + tenant_id="tenant-1", + instance_id="instance-00001", + sequence=406, + kind="workflow.step.completed", + payload={}, + created_at=datetime.now(UTC), + ) + ) + self.session.commit() + fixed = self.history(kind="events", after_sequence=400, through_sequence=405) + fresh = self.history(kind="events", after_sequence=400) + self.assertEqual(405, fixed.total) + self.assertEqual(406, fresh.total) + self.assertEqual([401, 402, 403, 404, 405], [i.sequence for i in fixed.events]) + + def test_pages_enforce_tenant_scope_and_read_permission(self): + self.seed(1) + self.seed(2, tenant_id="other-tenant") + self.session.commit() + with self.assertRaises(WorkflowNotFoundError): + self.history(instance_id="instance-00002", kind="events") + with self.assertRaises(WorkflowNotFoundError): + get_instance_summary( + self.session, + principal=self.actor, + registry=None, + instance_id="instance-00002", + ) + self.actor = replace( + self.actor, principal=replace(self.actor.principal, scopes=frozenset()) + ) + with self.assertRaises(PermissionError): + self.summaries() + with self.assertRaises(PermissionError): + self.history(kind="steps") + + def test_history_filters_child_tenant_and_empty_history(self): + self.seed(1, rows=0) + self.session.add( + WorkflowInstanceEvent( + tenant_id="other-tenant", + instance_id="instance-00001", + sequence=1, + kind="should-not-be-disclosed", + payload={}, + created_at=datetime.now(UTC), + ) + ) + self.session.commit() + empty = self.history(kind="events") + self.assertEqual( + (0, 0, None, []), + ( + empty.total, + empty.through_sequence, + empty.next_after_sequence, + empty.events, + ), + ) + + def test_inconsistent_revision_evidence_fails_closed(self): + instance = self.seed(1) + self.seed(2, tenant_id="other-tenant") + instance.definition_revision_id = "revision-2" + self.session.commit() + self.session.expunge_all() + with self.assertRaises(WorkflowNotFoundError): + self.summaries() + with self.assertRaises(WorkflowNotFoundError): + self.history(kind="events") + loaded = list_instances(self.session, tenant_id="tenant-1") + with self.assertRaises(WorkflowNotFoundError): + instance_response(self.session, loaded[0]) + + def test_work_item_total_remains_exact_after_authorization_and_filters(self): + for index in range(40): + self.seed(index, rows=1) + self.seed(50, scope_type="group", rows=1) + self.seed(51, tenant_id="other-tenant", rows=1) + self.seed(52, rows=1) + self.session.flush() + self.session.get( + WorkflowInstanceStep, "step-52-1" + ).work_assignment_id = "someone-else" + self.session.commit() + self.session.expunge_all() + provider = WorkflowWorkItemProvider() + with ( + self.statements() as queries, + patch.object(provider, "_targets", wraps=provider._targets) as targets, + ): + page = provider.list_items( + self.session, + self.actor, + query=WorkItemQuery(tenant_id="tenant-1", limit=2), + ) + self.assertEqual(1, targets.call_count) + self.assertEqual(1, len(queries)) + self.assertEqual((40, 2, True), (page.total, len(page.items), page.truncated)) + filtered = provider.list_items( + self.session, + self.actor, + query=WorkItemQuery(tenant_id="tenant-1", text="not present", limit=2), + ) + self.assertEqual((0, False), (filtered.total, filtered.truncated)) + + def test_http_routes_validate_bounds_and_keep_legacy_contract(self): + self.seed(1) + self.session.commit() + app = FastAPI() + app.include_router(router) + app.dependency_overrides[get_session] = lambda: self.session + app.dependency_overrides[get_api_principal] = lambda: self.actor + with ( + TestClient(app) as client, + patch( + "govoplan_workflow_engine.backend.router.get_registry", + return_value=None, + ), + ): + prefix = router.prefix + summary = client.get(f"{prefix}/instances/summaries") + self.assertEqual(200, summary.status_code, summary.text) + self.assertNotIn("steps", summary.json()["instances"][0]) + detail = client.get(f"{prefix}/instances/instance-00001/summary") + self.assertEqual(200, detail.status_code, detail.text) + legacy = client.get(f"{prefix}/instances/instance-00001") + self.assertEqual(200, legacy.status_code, legacy.text) + self.assertEqual(3, len(legacy.json()["events"])) + for kind in ("steps", "events"): + page = client.get(f"{prefix}/instances/instance-00001/{kind}?limit=2") + self.assertEqual(200, page.status_code, page.text) + self.assertEqual(3, page.json()["total"]) + self.assertEqual(2, page.json()["next_after_sequence"]) + self.assertEqual( + 422, + client.get( + f"{prefix}/instances/instance-00001/{kind}?limit=201" + ).status_code, + ) + self.assertEqual( + 422, + client.get( + f"{prefix}/instances/instance-00001/{kind}?after_sequence=-1" + ).status_code, + ) + self.assertEqual( + 422, + client.get( + f"{prefix}/instances/instance-00001/{kind}?through_sequence=999999999999999999999" + ).status_code, + ) + self.assertEqual( + 400, + client.get(f"{prefix}/instances/summaries?cursor=invalid").status_code, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_migrations.py b/tests/test_migrations.py index fa9c289..c4729a5 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -29,7 +29,7 @@ class WorkflowMigrationTests(unittest.TestCase): try: with engine.connect() as connection: self.assertIn( - "8d5a2f7c1b4e", + "9e6b3f8a2c7d", set(MigrationContext.configure(connection).get_current_heads()), ) self.assertEqual( @@ -75,6 +75,18 @@ class WorkflowMigrationTests(unittest.TestCase): ) } self.assertIn("start_origin", instance_columns) + indexes = { + item["name"]: item["column_names"] + for item in inspect(connection).get_indexes("workflow_instances") + } + self.assertEqual( + ["tenant_id", "created_at", "id"], + indexes["ix_workflow_instances_tenant_created_id"], + ) + self.assertEqual( + ["tenant_id", "definition_id", "created_at", "id"], + indexes["ix_workflow_instances_tenant_definition_created_id"], + ) definition_columns = { item["name"] for item in inspect(connection).get_columns( @@ -107,6 +119,7 @@ class WorkflowMigrationTests(unittest.TestCase): "b2e4f6a8c0d1_", "e4a1f8c2d7b6_", "8d5a2f7c1b4e_", + "9e6b3f8a2c7d_", ) ): continue @@ -151,7 +164,7 @@ class WorkflowMigrationTests(unittest.TestCase): manifest_factories=(get_manifest,), ) - self.assertIn("8d5a2f7c1b4e", result.current_revision or "") + self.assertIn("9e6b3f8a2c7d", result.current_revision or "") engine = create_engine(url) try: upgraded_tables = set(inspect(engine).get_table_names()) @@ -187,6 +200,8 @@ class WorkflowMigrationTests(unittest.TestCase): try: with engine.begin() as connection: for index_name in ( + "ix_workflow_instances_tenant_created_id", + "ix_workflow_instances_tenant_definition_created_id", "ix_workflow_instance_steps_work_assignment", "ix_workflow_instance_steps_work_due_at", "ix_workflow_instance_steps_work_assignment_id", @@ -220,7 +235,7 @@ class WorkflowMigrationTests(unittest.TestCase): connection.execute( text( "UPDATE alembic_version SET version_num = " - "'b2e4f6a8c0d1' WHERE version_num = '8d5a2f7c1b4e'" + "'b2e4f6a8c0d1' WHERE version_num = '9e6b3f8a2c7d'" ) ) finally: @@ -258,7 +273,7 @@ class WorkflowMigrationTests(unittest.TestCase): ) with engine.connect() as connection: self.assertIn( - "8d5a2f7c1b4e", + "9e6b3f8a2c7d", set(MigrationContext.configure(connection).get_current_heads()), ) finally: