fix: daemon mode wasn't reporting progress

src/openjd/adaptor_runtime/adaptors/_base_adaptor.py

@@ -13,6 +13,7 @@
 from typing import Type
 from typing import TypeVar
 
+from ..process._logging import _ADAPTOR_OUTPUT_LEVEL
 from .._utils._constants import _OPENJD_PROGRESS_STDOUT_PREFIX, _OPENJD_STATUS_STDOUT_PREFIX
 from .configuration import AdaptorConfiguration, ConfigurationManager
 from .configuration._configuration_manager import (
@@ -194,16 +195,20 @@ def update_status(
 
         if progress is not None:
             if math.isfinite(progress):
-                sys.stdout.write(f"{cls._OPENJD_PROGRESS_STDOUT_PREFIX}{progress}{os.linesep}")
-                sys.stdout.flush()
+                _logger.log(
+                    _ADAPTOR_OUTPUT_LEVEL,
+                    f"{cls._OPENJD_PROGRESS_STDOUT_PREFIX}{progress}",
+                )
             else:
                 _logger.warning(
                     f"Attempted to set progress to something non-finite: {progress}. "
                     "Ignoring progress update."
                 )
         if status_message is not None:
-            sys.stdout.write(f"{cls._OPENJD_STATUS_STDOUT_PREFIX}{status_message}{os.linesep}")
-            sys.stdout.flush()
+            _logger.log(
+                _ADAPTOR_OUTPUT_LEVEL,
+                f"{cls._OPENJD_STATUS_STDOUT_PREFIX}{status_message}",
+            )
 
     @property
     def path_mapping_rules(self) -> list[PathMappingRule]:

test/openjd/adaptor_runtime/integ/ProgressReportingAdaptor/ProgressReportingAdaptor.json

@@ -0,0 +1 @@
+{}

test/openjd/adaptor_runtime/integ/ProgressReportingAdaptor/__init__.py

@@ -0,0 +1,5 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+from .adaptor import ProgressReportingAdaptor
+
+__all__ = ["ProgressReportingAdaptor"]

test/openjd/adaptor_runtime/integ/ProgressReportingAdaptor/__main__.py

@@ -0,0 +1,18 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+import sys
+
+from openjd.adaptor_runtime import EntryPoint
+from .adaptor import ProgressReportingAdaptor
+
+
+def main():
+    package_name = vars(sys.modules[__name__])["__package__"]
+    if not package_name:
+        raise RuntimeError(f"Must be run as a module. Do not run {__file__} directly")
+
+    EntryPoint(ProgressReportingAdaptor).start()
+
+
+if __name__ == "__main__":
+    main()

test/openjd/adaptor_runtime/integ/ProgressReportingAdaptor/adaptor.py

@@ -0,0 +1,19 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+from openjd.adaptor_runtime.adaptors import Adaptor, SemanticVersion
+
+
+class ProgressReportingAdaptor(Adaptor):
+    """
+    A minimal adaptor that calls update_status to report progress during on_run.
+    Used to test that progress messages propagate from the background adaptor
+    to the frontend process's stdout.
+    """
+
+    @property
+    def integration_data_interface_version(self) -> SemanticVersion:
+        return SemanticVersion(major=0, minor=1)
+
+    def on_run(self, run_data: dict) -> None:
+        progress = run_data.get("progress", 50.0)
+        self.update_status(progress=progress)

test/openjd/adaptor_runtime/integ/adaptors/test_integration_adaptor.py

@@ -17,7 +17,7 @@ class TestRun:
     _OPENJD_PROGRESS_STDOUT_PREFIX: str = "openjd_progress: "
     _OPENJD_STATUS_STDOUT_PREFIX: str = "openjd_status: "
 
-    def test_run(self, capsys) -> None:
+    def test_run(self, capsys, caplog) -> None:
         first_progress = 0.0
         first_status_message = "Starting the printing of run_data"
         second_progress = 100.0
@@ -46,14 +46,15 @@ def integration_data_interface_version(self) -> SemanticVersion:
         adaptor = PrintAdaptor(init_data)
 
         # WHEN
-        adaptor._run(run_data)
+        with caplog.at_level(0):
+            adaptor._run(run_data)
         result = capsys.readouterr().out.strip()
 
         # THEN
-        assert f"{self._OPENJD_PROGRESS_STDOUT_PREFIX}{first_progress}" in result
-        assert f"{self._OPENJD_STATUS_STDOUT_PREFIX}{first_status_message}" in result
-        assert f"{self._OPENJD_PROGRESS_STDOUT_PREFIX}{second_progress}" in result
-        assert f"{self._OPENJD_STATUS_STDOUT_PREFIX}{second_status_message}" in result
+        assert f"{self._OPENJD_PROGRESS_STDOUT_PREFIX}{first_progress}" in caplog.text
+        assert f"{self._OPENJD_STATUS_STDOUT_PREFIX}{first_status_message}" in caplog.text
+        assert f"{self._OPENJD_PROGRESS_STDOUT_PREFIX}{second_progress}" in caplog.text
+        assert f"{self._OPENJD_STATUS_STDOUT_PREFIX}{second_status_message}" in caplog.text
         assert "run_data:\n\tkey1 = value1\n\tkey2 = value2\n\tkey3 = value3" in result
 
     def test_start_end_cleanup(self, tmpdir, capsys) -> None:

test/openjd/adaptor_runtime/integ/test_progress_in_background_mode.py

@@ -0,0 +1,151 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+"""
+Test to reproduce a bug where progress reporting via update_status() does not
+propagate from the background adaptor to the frontend process's stdout.
+
+The flow is:
+  1. Application outputs progress info
+  2. Background adaptor (backend process) calls update_status(progress=X)
+  3. update_status writes "openjd_progress: X" to sys.stdout
+  4. Frontend process should emit "openjd_progress: X" on its own stdout
+     so the worker agent can detect it
+
+The bug: In daemon/background mode, the backend process replaces its stream
+handler with a LogBufferHandler. But update_status() writes directly to
+sys.stdout (not through the logging framework), so the progress message
+never reaches the log buffer, and thus never reaches the frontend.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import pathlib
+import sys
+from logging import INFO
+from unittest.mock import patch
+from pathlib import Path
+
+import pytest
+
+import openjd.adaptor_runtime._entrypoint as runtime_entrypoint
+from openjd.adaptor_runtime import EntryPoint
+
+mod_path = Path(__file__).parent.resolve()
+sys.path.append(str(mod_path))
+
+if (_pypath := os.environ.get("PYTHONPATH")) is not None:
+    os.environ["PYTHONPATH"] = os.pathsep.join([_pypath, str(mod_path)])
+else:
+    os.environ["PYTHONPATH"] = str(mod_path)
+
+from ProgressReportingAdaptor import ProgressReportingAdaptor  # noqa: E402
+
+
+class TestProgressReportingInDaemonMode:
+    """
+    Tests that progress messages from update_status() in the background adaptor
+    are propagated to the frontend process's stdout so the worker agent can
+    detect them.
+    """
+
+    def test_progress_propagates_to_frontend_stdout(
+        self,
+        capfd: pytest.CaptureFixture,
+        caplog: pytest.LogCaptureFixture,
+        tmp_path: pathlib.Path,
+    ):
+        """
+        Reproduces the bug: when running in daemon mode, calling
+        update_status(progress=50) in the backend adaptor should result in
+        "openjd_progress: 50" appearing on the frontend's stdout.
+
+        The worker agent monitors the frontend process's stdout for these
+        messages. If they don't appear, the worker never knows about progress.
+        """
+        caplog.set_level(INFO)
+        connection_file = tmp_path / "connection.json"
+        entrypoint = EntryPoint(ProgressReportingAdaptor)
+
+        # Start the daemon
+        start_argv = [
+            "program_filename.py",
+            "daemon",
+            "start",
+            "--connection-file",
+            str(connection_file),
+        ]
+        with (
+            patch.object(runtime_entrypoint.sys, "argv", start_argv),
+            patch.object(runtime_entrypoint.logging.Logger, "setLevel"),
+        ):
+            entrypoint.start()
+
+        # Run with progress=50
+        run_argv = [
+            "program_filename.py",
+            "daemon",
+            "run",
+            "--connection-file",
+            str(connection_file),
+            "--run-data",
+            json.dumps({"progress": 50.0}),
+        ]
+        with (
+            patch.object(runtime_entrypoint.sys, "argv", run_argv),
+            patch.object(runtime_entrypoint.logging.Logger, "setLevel"),
+        ):
+            entrypoint.start()
+
+        # Stop the daemon
+        stop_argv = [
+            "program_filename.py",
+            "daemon",
+            "stop",
+            "--connection-file",
+            str(connection_file),
+        ]
+        with (
+            patch.object(runtime_entrypoint.sys, "argv", stop_argv),
+            patch.object(runtime_entrypoint.logging.Logger, "setLevel"),
+        ):
+            entrypoint.start()
+
+        # Verify that "openjd_progress: 50" appeared on stdout.
+        # The worker agent is watching stdout for this pattern.
+        captured = capfd.readouterr()
+        assert "openjd_progress: 50" in captured.out, (
+            f"Expected 'openjd_progress: 50' on stdout but got:\n{captured.out}\n\n"
+            f"Log output:\n{caplog.text}"
+        )
+
+    def test_progress_works_in_run_mode(
+        self,
+        capfd: pytest.CaptureFixture,
+        caplog: pytest.LogCaptureFixture,
+    ):
+        """
+        Sanity check: progress reporting works fine in non-daemon 'run' mode
+        because update_status writes directly to sys.stdout and the worker
+        agent is directly reading from that process's stdout.
+        """
+        caplog.set_level(INFO)
+        entrypoint = EntryPoint(ProgressReportingAdaptor)
+
+        run_argv = [
+            "program_filename.py",
+            "run",
+            "--run-data",
+            json.dumps({"progress": 75.0}),
+        ]
+        with (
+            patch.object(runtime_entrypoint.sys, "argv", run_argv),
+            patch.object(runtime_entrypoint.logging.Logger, "setLevel"),
+        ):
+            entrypoint.start()
+
+        captured = capfd.readouterr()
+        assert (
+            "openjd_progress: 75" in captured.out
+        ), f"Expected 'openjd_progress: 75' on stdout but got:\n{captured.out}"

test/openjd/adaptor_runtime/unit/adaptors/test_base_adaptor.py

@@ -316,16 +316,17 @@ class TestStatusUpdate:
             param(1e-5),
         ],
     )
-    def test_progress_update(self, capsys, progress: float):
+    def test_progress_update(self, caplog, progress: float):
         """Tests just updating the progress"""
         # GIVEN
         expected = f"{self._OPENJD_PROGRESS_STDOUT_PREFIX}{progress}"
 
         # WHEN
-        BaseAdaptor.update_status(progress=progress)
+        with caplog.at_level(0):
+            BaseAdaptor.update_status(progress=progress)
 
         # THEN
-        assert expected in capsys.readouterr().out
+        assert expected in caplog.text
 
     @pytest.mark.parametrize(
         "status_message",
@@ -335,16 +336,17 @@ def test_progress_update(self, capsys, progress: float):
             param(""),
         ],
     )
-    def test_status_message_update(self, capsys, status_message: str):
+    def test_status_message_update(self, caplog, status_message: str):
         """Tests just updating the status message"""
         # GIVEN
         expected = f"{self._OPENJD_STATUS_STDOUT_PREFIX}{status_message}"
 
         # WHEN
-        BaseAdaptor.update_status(status_message=status_message)
+        with caplog.at_level(0):
+            BaseAdaptor.update_status(status_message=status_message)
 
         # THEN
-        assert expected in capsys.readouterr().out
+        assert expected in caplog.text
 
     @pytest.mark.parametrize(
         "progress,status_message",
@@ -360,30 +362,29 @@ def test_status_message_update(self, capsys, status_message: str):
             ),
         ],
     )
-    def test_status_update(self, capsys, progress: float, status_message: str):
+    def test_status_update(self, caplog, progress: float, status_message: str):
         """Tests updating both progress and status messages"""
         # GIVEN
         expected_progress = f"{self._OPENJD_PROGRESS_STDOUT_PREFIX}{progress}"
         expected_status_message = f"{self._OPENJD_STATUS_STDOUT_PREFIX}{status_message}"
 
         # WHEN
-        BaseAdaptor.update_status(progress=progress, status_message=status_message)
+        with caplog.at_level(0):
+            BaseAdaptor.update_status(progress=progress, status_message=status_message)
 
         # THEN
-        result = capsys.readouterr().out
-        assert expected_progress in result
-        assert expected_status_message in result
+        assert expected_progress in caplog.text
+        assert expected_status_message in caplog.text
 
-    def test_ignore_status_update(self, capsys):
+    def test_ignore_status_update(self, caplog):
         """Tests we don't send any message if there's nothing to report"""
-        # GIVEN
-        expected = ""  # nothing was captured in stdout
-
         # WHEN
-        BaseAdaptor.update_status(progress=None, status_message=None)
-        BaseAdaptor.update_status(progress=float("NaN"))
-        BaseAdaptor.update_status(progress=float("inf"))
+        with caplog.at_level(0):
+            BaseAdaptor.update_status(progress=None, status_message=None)
+            BaseAdaptor.update_status(progress=float("NaN"))
+            BaseAdaptor.update_status(progress=float("inf"))
 
         # THEN
-        result = capsys.readouterr().out
-        assert expected == result
+        # Only warning messages should be logged, no progress/status messages
+        assert self._OPENJD_PROGRESS_STDOUT_PREFIX not in caplog.text
+        assert self._OPENJD_STATUS_STDOUT_PREFIX not in caplog.text