perf(workflow): batch revision reads and add bounded histories

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()

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: