feat: access workflow internal API endpoints using compression

Alper Akgunrequested to merge
aa2-use-workflows-internal-api-with-compression into main

What does this merge request do and why?

feat: enable checkpoint data compression in workflow internal API endpoints. Workflow compression API reduces checkpoint size during transport

Previous MR created issues in uncompressed state due to not calling checkpoint_decoder, You can repeat them by commenting out object_hook=checkpoint_decoder in gitlab_workflow.py feat: use workflows internal api with compression (!3621 - merged)

How to set up and validate locally

  1. Test duo workflow flows and agentic chats with compression support
  2. Test duo workflow flows and agentic chats without compression support
  • On gitlab rails console (bin/rails c) enable or disable the feature:
Feature.enable :duo_workflow_compress_checkpoint
Feature.enabled? :duo_workflow_compress_checkpoint
Feature.disable :duo_workflow_compress_checkpoint
  • restart the services and vscode (probably not needed, but to be sure)
gdk restart rails-web duo-workflow-service gitlab-ai-gateway
  • Check gitlab api log or duo workflow service log
tail -n 200 log/api_json.log  | grep "/api/:version/ai/duo_workflows/workflows" | grep compress |  jq
gdk tail duo-workflow-service

Dependencies

AttributeError: 'dict' object has no attribute 'type' 
Traceback (most recent call last):
  File "/home/aigateway/app/duo_workflow_service/workflows/abstract_workflow.py", line 282, in _compile_and_run_graph
    async for type, state in compiled_graph.astream(
  File "/home/aigateway/app/venv/ai-gateway-M6hW6iiC-py3.12/lib/python3.12/site-packages/langgraph/pregel/__init__.py", line 2652, in astream
    while loop.tick(input_keys=self.input_channels):
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/aigateway/app/venv/ai-gateway-M6hW6iiC-py3.12/lib/python3.12/site-packages/langgraph/pregel/loop.py", line 473, in tick
    updated_channels = self._first(input_keys=input_keys)
                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/aigateway/app/venv/ai-gateway-M6hW6iiC-py3.12/lib/python3.12/site-packages/langgraph/pregel/loop.py", line 703, in _first
    mv_writes, _ = apply_writes(
                   ^^^^^^^^^^^^^
  File "/home/aigateway/app/venv/ai-gateway-M6hW6iiC-py3.12/lib/python3.12/site-packages/langgraph/pregel/algo.py", line 305, in apply_writes
    if channels[chan].update(vals) and get_next_version is not None:
       ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/aigateway/app/venv/ai-gateway-M6hW6iiC-py3.12/lib/python3.12/site-packages/langgraph/channels/binop.py", line 91, in update
    self.value = self.operator(self.value, value)
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/aigateway/app/duo_workflow_service/entities/state.py", line 244, in _conversation_history_reducer
    current_msg_roles, current_msg_token = get_messages_profile(
                                           ^^^^^^^^^^^^^^^^^^^^^
  File "/home/aigateway/app/duo_workflow_service/entities/state.py", line 218, in get_messages_profile
    roles = [msg.type for msg in messages]
             ^^^^^^^^
AttributeError: 'dict' object has no attribute 'type'

{
  "event": "'dict' object has no attribute 'type'",
  "logger": "exceptions",
  "level": "error",
  "correlation_id": "848f8fec-722a-49ca-90b1-13f1d884d401",
  "gitlab_global_user_id": "7hbE4xmTa2FltcIHykLtoyOYW0zYQ8kNILhSaz2kjlM=",
  "workflow_id": "310",
  "status_code": null,
  "exception_class": "AttributeError",
  "additional_details": {
    "workflow_id": "310",
    "source": "duo_workflow_service.checkpointer.gitlab_workflow"
  },
  "timestamp": "2025-11-05T17:30:36.139022Z",
  "exception": "Traceback (most recent call last):\n  File \"/home/aakgun/aakgun/1/gdk/gitlab-ai-gateway/duo_workflow_service/workflows/abstract_workflow.py\", line 289, in _compile_and_run_graph\n    async for type, state in compiled_graph.astream(\n  File \"/home/aakgun/aakgun/1/gdk/gitlab-ai-gateway/.venv/lib/python3.12/site-packages/langgraph/pregel/__init__.py\", line 2652, in astream\n    while loop.tick(input_keys=self.input_channels):\n          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/home/aakgun/aakgun/1/gdk/gitlab-ai-gateway/.venv/lib/python3.12/site-packages/langgraph/pregel/loop.py\", line 473, in tick\n    updated_channels = self._first(input_keys=input_keys)\n                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/home/aakgun/aakgun/1/gdk/gitlab-ai-gateway/.venv/lib/python3.12/site-packages/langgraph/pregel/loop.py\", line 703, in _first\n    mv_writes, _ = apply_writes(\n                   ^^^^^^^^^^^^^\n  File \"/home/aakgun/aakgun/1/gdk/gitlab-ai-gateway/.venv/lib/python3.12/site-packages/langgraph/pregel/algo.py\", line 305, in apply_writes\n    if channels[chan].update(vals) and get_next_version is not None:\n       ^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/home/aakgun/aakgun/1/gdk/gitlab-ai-gateway/.venv/lib/python3.12/site-packages/langgraph/channels/binop.py\", line 91, in update\n    self.value = self.operator(self.value, value)\n                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^\n  File \"/home/aakgun/aakgun/1/gdk/gitlab-ai-gateway/duo_workflow_service/entities/state.py\", line 244, in _conversation_history_reducer\n    current_msg_roles, current_msg_token = get_messages_profile(\n                                           ^^^^^^^^^^^^^^^^^^^^^\n  File \"/home/aakgun/aakgun/1/gdk/gitlab-ai-gateway/duo_workflow_service/entities/state.py\", line 218, in get_messages_profile\n    roles = [msg.type for msg in messages]\n             ^^^^^^^^\nAttributeError: 'dict' object has no attribute 'type'"
}

Merge request checklist

  • Tests added for new functionality. If not, please raise an issue to follow up.
  • Documentation added/updated, if needed.
  • If this change requires executor implementation: verified that issues/MRs exist for both Go executor and Node executor or confirmed that changes are backward-compatible and don't break existing executor functionality.
Edited by Alper Akgun

Merge request reports