Add structured event logging keyed by per-process identity

[fzyzcjy]

Add the structured event logger that records typed events keyed by per-process
identity, wire it through the logging helper and CLI argument, and start it from
the train entrypoints.

• miles/utils/event_logger/logger.py, logging_utils.py, arguments.py and entrypoint wiring, with tests.

[gemini-code-assist]

Code Review

This pull request introduces a structured EventLogger utility with corresponding unit tests, and refactors logging configuration across various entry points to support process-specific identities. The code review feedback highlights several critical improvements: the event_logger_context decorator needs to support asynchronous methods to prevent context loss; the assertion in the finally block of with_context should be removed to avoid context leaks on failure; configure_logger should be restored in TrainRayActor to prevent unconfigured logging in training processes; and file operations in read_events should be wrapped in error handling to avoid crashes from concurrent file deletions or permission issues.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

The event_logger_context decorator does not support asynchronous methods (coroutines). If a decorated method is asynchronous, calling it returns a coroutine object immediately, which exits the with_context block before the coroutine is actually executed or awaited. This causes the context to be completely lost during execution.

To fix this, use inspect.iscoroutinefunction to handle both synchronous and asynchronous methods appropriately.

    import inspect

    def decorator(method: Callable) -> Callable:
        if inspect.iscoroutinefunction(method):
            @functools.wraps(method)
            async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                if not is_event_logger_initialized():
                    return await method(*args, **kwargs)

                ctx_value = ctx_fn(*args, **kwargs)
                with get_event_logger().with_context(ctx_value):
                    return await method(*args, **kwargs)

            return async_wrapper

        @functools.wraps(method)
        def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
            if not is_event_logger_initialized():
                return method(*args, **kwargs)

            ctx_value = ctx_fn(*args, **kwargs)
            with get_event_logger().with_context(ctx_value):
                return method(*args, **kwargs)

        return sync_wrapper

[Shi-Dong]

Perhaps just add an assertion assert not inspect.iscoroutinefunction(method) for now?

[gemini-code-assist]

The assert self._context_var.get() == merged statement in the finally block is dangerous. If the assertion fails, it raises an AssertionError which prevents the critical cleanup logic self._context_var.reset(token) from executing, leading to a context leak. Additionally, raising an exception in a finally block can mask or suppress any original exception raised inside the try block, making debugging extremely difficult. The assertion should be removed entirely to ensure that reset(token) is always executed.

Suggested change

	finally:
	assert self._context_var.get() == merged
	self._context_var.reset(token)
	finally:
	self._context_var.reset(token)

References

1. To prevent resource leaks (e.g., counters that are not decremented), use constructs like try...finally or a with statement to ensure cleanup logic is always executed, even in the case of exceptions or early returns.

[Shi-Dong]

I think this makes sense and having an assertion in the finally block here is dangerous. Please fix.

[gemini-code-assist]

Removing configure_logger() entirely from TrainRayActor means that the training actor processes will run with unconfigured logging (no custom format, no event logger initialization, etc.). Since args is not available in __init__, configure_logger should be called inside the init method of TrainRayActor once args is available, using a TrainProcessIdentity constructed from the actor's role and rank.

[gemini-code-assist]

open(jsonl_path, encoding="utf-8") is called directly without a try...except block. If any of the matched files are deleted concurrently, or if there are permission issues reading a specific file, the entire read_events function will crash. Wrapping the file opening and reading in a try...except block would make the function much more robust.

Suggested change

	for jsonl_path in jsonl_files:
	with open(jsonl_path, encoding="utf-8") as f:
	for line_num, raw_line in enumerate(f, start=1):
	raw_line = raw_line.strip()
	if not raw_line:
	continue
	try:
	event = _event_adapter.validate_json(raw_line)
	events.append(event)
	except Exception:
	logger.warning(
	"Failed to parse event at %s:%d",
	jsonl_path,
	line_num,
	exc_info=True,
	)
	for jsonl_path in jsonl_files:
	try:
	with open(jsonl_path, encoding="utf-8") as f:
	for line_num, raw_line in enumerate(f, start=1):
	raw_line = raw_line.strip()
	if not raw_line:
	continue
	try:
	event = _event_adapter.validate_json(raw_line)
	events.append(event)
	except Exception:
	logger.warning(
	"Failed to parse event at %s:%d",
	jsonl_path,
	line_num,
	exc_info=True,
	)
	except Exception:
	logger.warning("Failed to open or read event file %s", jsonl_path, exc_info=True)

[Shi-Dong]

Perhaps just add an assertion assert not inspect.iscoroutinefunction(method) for now?

[Shi-Dong]

Let's still move all imports to the top.

[Shi-Dong]

I think this makes sense and having an assertion in the finally block here is dangerous. Please fix.