fix(dynamic-blocks): tee stdout/stderr so print() still reaches Docke… (#2258)

* fix(dynamic-blocks): tee stdout/stderr so print() still reaches Docker logs

* fix(workflows): return 400 on Modal DynamicBlockCodeError

Make `inner_error_type` and `inner_error_message` optional on
`WorkflowErrorResponse` so errors raised without an `inner_error`
(e.g. from the Modal/OCI executors) don't fail pydantic validation
and surface as a 500 instead of the intended 400.

* return also std out/err on success calls

* return on error and on ok with stdout/err data

* remove duplicated lgos

* undo unnecessary change for returning

* fix tabs/spaces

---------

Co-authored-by: Paweł Pęczek <146137186+PawelPeczek-Roboflow@users.noreply.github.com>

Author: rafel-roboflow

inference/core/entities/responses/workflows.py

@@ -200,6 +200,6 @@ class WorkflowErrorResponse(BaseModel):
     message: str
     error_type: str
     context: str
-    inner_error_type: str
-    inner_error_message: str
-    blocks_errors: Optional[List[WorkflowBlockError]]
+    inner_error_type: Optional[str] = None
+    inner_error_message: Optional[str] = None
+    blocks_errors: Optional[List[WorkflowBlockError]] = None

inference/core/workflows/execution_engine/v1/dynamic_blocks/block_scaffolding.py

@@ -108,6 +108,8 @@ def run(self, *args, **kwargs) -> BlockResult:
             import_lines_count = len(_get_python_code_imports(python_code).splitlines())
             try:
                 with capture_output() as (stdout_buf, stderr_buf):
+                    # stdout/stderr already reach the process streams in real time via the
+                    # tee in capture_output(); buffers are only used to attach context on error.
                     return run_function(self, *args, **kwargs)
             except Exception as error:
                 raise create_dynamic_block_code_error(

inference/core/workflows/execution_engine/v1/dynamic_blocks/error_utils.py

@@ -14,27 +14,39 @@
 
 
 class _ThreadDispatchStream:
-    """Stream wrapper that dispatches writes to a per-thread StringIO buffer
-    when one is active, otherwise falls through to the original stream.
+    """Stream wrapper that tees writes into a per-thread StringIO buffer
+    (when one is active) while still forwarding them to the original stream.
 
-    This avoids the need for a global lock: each thread captures to its own
-    buffer independently, while threads that are not capturing see normal
-    stdout/stderr behaviour.
+    Threads that are not capturing see normal stdout/stderr behaviour; threads
+    that are capturing get both: the buffer keeps an in-memory copy for error
+    payloads, and the original stream still receives the bytes so ``print()``
+    output continues to reach Docker / the process stdout.
     """
 
     def __init__(self, original, attr_name: str):
         object.__setattr__(self, "_original", original)
         object.__setattr__(self, "_attr_name", attr_name)
 
-    def _get_target(self):
-        buf = getattr(_thread_local, self._attr_name, None)
-        return buf if buf is not None else self._original
+    def _get_buffer(self):
+        return getattr(_thread_local, self._attr_name, None)
 
     def write(self, data):
-        return self._get_target().write(data)
+        buf = self._get_buffer()
+        if buf is not None:
+            try:
+                buf.write(data)
+            except Exception:
+                pass
+        return self._original.write(data)
 
     def flush(self):
-        return self._get_target().flush()
+        buf = self._get_buffer()
+        if buf is not None:
+            try:
+                buf.flush()
+            except Exception:
+                pass
+        return self._original.flush()
 
     def fileno(self):
         return self._original.fileno()
@@ -43,7 +55,7 @@ def isatty(self):
         return self._original.isatty()
 
     def __getattr__(self, name):
-        return getattr(self._get_target(), name)
+        return getattr(self._original, name)
 
 
 def _install_dispatchers() -> None:

inference/core/workflows/execution_engine/v1/dynamic_blocks/modal_executor.py

@@ -8,6 +8,7 @@
 import base64
 import json
 import os
+import sys
 from typing import Any, Dict, Optional
 
 import numpy as np
@@ -352,6 +353,15 @@ def execute_remote(
                     stderr=result.get("stderr"),
                 )
 
+            stdout = result.get("stdout")
+            stderr = result.get("stderr")
+            if stdout:
+                sys.stdout.write(stdout)
+                sys.stdout.flush()
+            if stderr:
+                sys.stderr.write(stderr)
+                sys.stderr.flush()
+
             # Get the result and deserialize from JSON
             json_result = result.get("result", "{}")
             return deserialize_for_modal_remote_execution(json_result)

modal/modal_app.py

@@ -363,8 +363,12 @@ class BlockSelf:
 
                 json_result = serialize_for_modal_remote_execution(result)
 
-                # Return the serialized result with success flag
-                return {"success": True, "result": json_result}
+                return {
+                    "success": True,
+                    "result": json_result,
+                    "stdout": stdout_buf.getvalue() or None,
+                    "stderr": stderr_buf.getvalue() or None,
+                }
             except Exception as e:
                 # On error, capture stdout/stderr and return error details
                 result = {