diff --git a/tests/tools/test_completed_process_results.py b/tests/tools/test_completed_process_results.py index c76a57b3f2..958b6a3380 100644 --- a/tests/tools/test_completed_process_results.py +++ b/tests/tools/test_completed_process_results.py @@ -136,6 +136,7 @@ def test_headless_terminal_result_survives_cli_exit(tmp_path): return json.loads(result.stdout) receipt = json.loads((home / "logs" / "process-results" / f"{process_id}.json").read_text(encoding="utf-8")) + assert receipt["parent_session_id"] # CLI owner must be stamped before its reader starts. env["HERMES_SESSION_ID"] = receipt["parent_session_id"] recovered = read_result(home) assert recovered["result"]["status"] == "exited", recovered @@ -205,3 +206,20 @@ def test_receipts_are_bounded_redacted_and_session_scoped(tmp_path, monkeypatch) expired = time.time() - receipts.RESULT_RETENTION_SECONDS - 1 os.utime(path, (expired, expired)) assert fresh.get(recovered.id) is None + + # Multiplex readers must keep the producer's profile on native threads. + from hermes_constants import set_hermes_home_override, reset_hermes_home_override + profile = tmp_path / "thread-profile" + token = set_hermes_home_override(profile) + try: + with scoped_current_session_id("thread-owner"): + child = registry.spawn_local( + shlex.join([Path(sys.executable).as_posix(), "-c", "print('SCOPED_RESULT')"]), + cwd=str(tmp_path), task_id="thread-task") + child._reader_thread.join(timeout=20) + assert not child._reader_thread.is_alive() + assert (profile / "logs" / "process-results" / f"{child.id}.json").exists() + assert "SCOPED_RESULT" in ProcessRegistry().read_log(child.id)["output"] + finally: + reset_hermes_home_override(token) + assert not (get_hermes_home() / "logs" / "process-results" / f"{child.id}.json").exists() diff --git a/tools/process_registry.py b/tools/process_registry.py index 6539921b6f..f63353fdfa 100644 --- a/tools/process_registry.py +++ b/tools/process_registry.py @@ -789,7 +789,11 @@ class ProcessRegistry(ProcessCheckpointMixin): def _track_started(self, session: ProcessSession, reader_target, reader_name: str, extra_args=()) -> None: """Register before the reader can publish completion, even for an exited child.""" - reader = threading.Thread(target=reader_target, args=(session, *extra_args), daemon=True, name=reader_name) + from contextvars import copy_context + + # Reader completion must retain the producer's multiplex profile scope. + reader = threading.Thread(target=copy_context().run, args=(reader_target, session, *extra_args), + daemon=True, name=reader_name) session._reader_thread = reader with self._lock: self._prune_if_needed()