Repository navigation
feat(OMN-1610): Implement emit daemon for persistent Kafka connections - #207
Merged
Merged
Conversation
Add Unix socket daemon to maintain persistent Kafka connections for hooks, reducing emit latency from ~2-3s to ~15-40ms. Implementation: - config.py: ModelEmitDaemonConfig with Pydantic validation - event_registry.py: Event type → topic mapping with metadata injection - queue.py: BoundedEventQueue with disk spool and drop-oldest policy - daemon.py: Unix socket server with asyncio, circuit breaker integration - client.py: EmitClient with async/sync APIs and fallback support - cli.py: CLI commands (start/stop/status/emit/config) Design decisions: - Fire-and-forget semantics: "queued" = success from caller perspective - Daemon owns delivery: bounded queue (100 msgs) + disk spool (1000 msgs/10MB) - Typed event interface: daemon owns serialization, topic composition, metadata Test coverage: - 244 unit tests (config, registry, queue, client) - 17 integration tests (lifecycle, health check, event flow)
📝 WalkthroughWalkthroughAdds a new emit-daemon subsystem: CLI entrypoint, Unix-domain socket daemon that validates/enriches and queues events for Kafka, async/sync client, Pydantic config and protocol models, event registry, bounded in-memory queue with disk spool, validation exemptions/constants bump, plus extensive unit and integration tests. Changes
sequenceDiagram
participant User as User
participant CLI as CLI (emit-daemon)
participant Daemon as EmitDaemon (Unix Socket)
participant Registry as EventRegistry
participant Queue as BoundedEventQueue
participant Kafka as Kafka
User->>CLI: emit --event-type "session.started" --payload {...}
CLI->>Daemon: {"event_type":"session.started","payload":{...}}
Daemon->>Registry: resolve_topic("session.started")
Registry-->>Daemon: "omni.{env}.session.started"
Daemon->>Registry: inject_metadata(payload)
Registry-->>Daemon: enriched_payload
Daemon->>Queue: enqueue(ModelQueuedEvent(enriched_payload))
Queue-->>Daemon: queued(event_id)
Daemon-->>CLI: {"status":"queued","event_id":event_id}
loop Background publisher
Daemon->>Queue: dequeue()
Queue-->>Daemon: ModelQueuedEvent
Daemon->>Kafka: publish(topic, key, value, headers)
Kafka-->>Daemon: ack
end
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 9
🤖 Fix all issues with AI agents
In `@src/omnibase_infra/runtime/emit_daemon/cli.py`:
- Around line 416-421: The shutdown warning prints the literal "{pid}" instead
of the actual process id; update the second print call (the one referencing pid
and wait_time in src/omnibase_infra/runtime/emit_daemon/cli.py) to interpolate
the pid variable (e.g., use an f-string or .format) so it prints the real
command (e.g., "You may need to force kill with: kill -9 <pid>") and keep the
surrounding stderr/file=sys.stderr usage and subsequent sys.exit(1) unchanged.
- Around line 159-217: The _send_socket_message function must catch OSError (and
its subclasses like PermissionError/ConnectionRefusedError) around
socket.connect/sendall/recv and normalize them to your OnexError-based CLI error
type (use the project's OnexError subclass) with exception chaining (raise
YourOnexCliError("descriptive message", ...) from e); also replace the existing
raises of built-in ConnectionError for missing socket and empty response with
the same OnexError subclass so cmd_* handlers receive the normalized error;
ensure json.JSONDecodeError still maps to ValueError (or convert to OnexError if
project policy requires) and include the original exception message when
re-raising.
In `@src/omnibase_infra/runtime/emit_daemon/client.py`:
- Around line 59-260: Change EmitClientError to inherit from OnexError (or an
appropriate OnexError subclass) instead of Exception, and update all places that
raise EmitClientError in EmitClient._connect to use explicit exception chaining
(e.g., raise EmitClientError(..., reason=...) from e) so the original
OSError/TimeoutError/etc. are preserved; also find the _run_async function
(referenced in the review) and ensure it raises an OnexError-compatible
exception (wrap or rethrow RuntimeError as an EmitClientError or another
OnexError subclass) using "from" to chain the original exception. Ensure
constructor signatures still accept message and reason and maintain existing
logging/behavior.
In `@src/omnibase_infra/runtime/emit_daemon/config.py`:
- Around line 150-288: Replace all direct ValueError raises in the validators
with OnexError using the project's error pattern: import OnexError from
omnibase_core.errors and use EnumCoreErrorCode.VALIDATION_ERROR for the
error_code; for places where an exception is caught (e.g., the except ValueError
as e in validate_bootstrap_servers_format) raise OnexError(message="...",
error_code=EnumCoreErrorCode.VALIDATION_ERROR) from e, and for direct validation
failures (e.g., in validate_file_path_parent_exists_or_creatable,
validate_spool_dir_creatable, validate_bootstrap_servers_format,
validate_spool_limits_consistency) replace raise ValueError(...) with raise
OnexError(message="...", error_code=EnumCoreErrorCode.VALIDATION_ERROR) carrying
the same descriptive message so all validators consistently emit OnexError.
In `@src/omnibase_infra/runtime/emit_daemon/daemon.py`:
- Around line 136-137: The EventRegistry is instantiated with a hardcoded
environment="dev"; make this configurable by adding an environment: str field
(default "dev") to ModelEmitDaemonConfig and thread that value into the daemon
constructor (or accept an environment param) so the daemon uses
ModelEmitDaemonConfig.environment when creating EventRegistry(self._registry =
EventRegistry(environment=self.config.environment)); likewise add an environment
field to ModelKafkaEventBusConfig and use it where that config constructs or
configures EventRegistry/topic resolution to avoid hardcoded "dev".
- Around line 651-657: The header currently creates a new correlation_id
(uuid4()) which breaks tracing; instead extract and reuse the payload's
correlation id that was injected by inject_metadata() and set that on
ModelEventHeaders. Locate where ModelEventHeaders is constructed (around the
creation of headers in daemon.py) and replace the uuid4() assignment with the
correlation id from the event payload (e.g., event.payload.correlation_id or
payload.correlation_id depending on the local variable name) so the published
header preserves the original distributed-trace id.
In `@src/omnibase_infra/runtime/emit_daemon/event_registry.py`:
- Around line 243-249: Replace the ValueError used for unknown event types with
OnexError (or a suitable OnexError subclass): import OnexError into
event_registry.py, then change the raise in the method that looks up
self._registrations (the block referencing registration =
self._registrations.get(event_type),
registration.topic_template.format(env=self._environment)) to raise
OnexError(f"Unknown event type: '{event_type}'. Registered types:
{registered}"). Also scan the rest of this module for any other ValueError
usages and convert them to OnexError or a domain-specific OnexError subclass to
comply with the runtime error policy.
In `@src/omnibase_infra/runtime/emit_daemon/queue.py`:
- Around line 223-283: When memory is full, enqueue() must respect
spooling-disabled config: if either self._max_spool_messages == 0 or
self._max_spool_bytes == 0 (meaning spooling is disabled) do not call
_spool_event; instead log/record the drop and return False (or drop the event)
to avoid disk writes. Add this guard in enqueue() right before the call to await
self._spool_event(event); also add a defensive check at the start of
_spool_event(event) to return False immediately if spooling is disabled
(checking the same self._max_spool_messages/self._max_spool_bytes) so the
behavior is consistent even if _spool_event is called elsewhere.
In `@tests/integration/runtime/emit_daemon/test_emit_daemon_integration.py`:
- Around line 170-194: The fixture emit_daemon_config currently declares its
return type as AsyncGenerator but returns a ModelEmitDaemonConfig instance
directly; change the type hint to ModelEmitDaemonConfig (or the concrete type
from omnibase_infra.runtime.emit_daemon.config) so the signature matches the
actual return, e.g., replace "-> AsyncGenerator" with "->
ModelEmitDaemonConfig"; alternatively, if you intend to yield for teardown,
convert the function to use "yield config" and keep AsyncGenerator—update the
fixture accordingly in the emit_daemon_config definition.
🧹 Nitpick comments (12)
tests/unit/validation/test_validator_defaults.py (1)
87-89: Update the threshold history in the docstring to include the new entry.The assertion value is correctly updated to 112, but the threshold history documented in the docstring (lines 57-84) ends at
105 (2026-01-21)and doesn't include the new112 (2026-01-27): OMN-1610entry. This creates a documentation inconsistency within the same test method.📝 Suggested docstring addition (after line 84)
- 105 (2026-01-21): Contract-driven handler config loading (+4 unions) - ModelHandlerContract transport config fields and lifecycle types + - 112 (2026-01-27): OMN-1610 emit daemon for persistent Kafka connections (+7 unions) + - BoundedEventQueue, EmitClient, EventRegistry return types Target: Keep below 150 - if this grows, consider typed patterns from omnibase_core.tests/integration/runtime/emit_daemon/conftest.py (1)
3-13: Scaffold file with documented but unimplemented fixtures.The docstring describes fixtures for "Kafka topic management, daemon lifecycle, and test isolation" but no fixtures are currently implemented. If this is intentional scaffolding for future work, consider adding a
TODOcomment to clarify this is pending implementation.tests/unit/runtime/emit_daemon/test_event_registry.py (2)
19-19: Unused import:timezone.The
timezoneimport is not used in this file; onlyUTCis used.-from datetime import UTC, datetime, timezone +from datetime import UTC, datetime
67-83: Consider catchingpydantic.ValidationErrorexplicitly.Using
pytest.raises(Exception)is overly broad and could mask unrelated errors. SinceModelEventRegistrationis a Pydantic model withfrozen=Trueandextra="forbid", these tests should catchpydantic.ValidationErrorfor better specificity.♻️ Suggested improvement
+from pydantic import ValidationError + # In test_registration_is_frozen: -with pytest.raises(Exception): # ValidationError for frozen model +with pytest.raises(ValidationError): reg.event_type = "modified.event" # type: ignore[misc] # In test_registration_forbids_extra_fields: -with pytest.raises(Exception): # ValidationError for extra fields +with pytest.raises(ValidationError): ModelEventRegistration(tests/integration/runtime/emit_daemon/test_emit_daemon_integration.py (2)
5-7: Remove environment-specific IP from docstring.The hardcoded IP
192.168.86.200:29092in the docstring is environment-specific and may confuse future readers. Consider using a generic placeholder.-These tests validate the full emit daemon lifecycle with Redpanda/Kafka -at 192.168.86.200:29092. They verify that events flow through the daemon -to Kafka and can be consumed correctly. +These tests validate the full emit daemon lifecycle with Redpanda/Kafka. +They verify that events flow through the daemon to Kafka and can be consumed correctly.
943-967: Potential flaky test with 1ms timeout.Using a 1ms timeout (
timeout=0.001) is extremely short and may cause flaky test results depending on system load. The test acknowledges this with "will likely timeout", but this approach may not reliably test the timeout handling code path.Consider using a slightly longer timeout (e.g., 10-50ms) that's still short enough to test timeout behavior but less prone to system-dependent failures.
tests/unit/runtime/emit_daemon/test_client.py (1)
41-41: Unused import:TYPE_CHECKING.The
TYPE_CHECKINGimport is not used in this file.-from typing import TYPE_CHECKINGsrc/omnibase_infra/runtime/emit_daemon/daemon.py (1)
232-233: Consider formalizing EventBusKafka interface with a Protocol.The
hasattrchecks forstart()andclose()methods suggest the interface isn't well-defined. Consider defining a Protocol for the event bus to improve type safety and make the expected interface explicit.from typing import Protocol class EventBusProtocol(Protocol): async def start(self) -> None: ... async def close(self) -> None: ... async def publish(self, topic: str, key: bytes | None, value: bytes, headers: ModelEventHeaders) -> None: ...This would allow removing the
hasattrchecks and improve IDE support.Also applies to: 325-326
src/omnibase_infra/runtime/emit_daemon/cli.py (1)
159-163: Use JsonType for socket payload/response types.
Align the message/response/payload annotations withJsonTypeto enforce JSON-compatibility and project standards. As per coding guidelines, useJsonTypefor JSON-compatible values.♻️ Suggested typing update
from pathlib import Path import yaml +from omnibase_core.types import JsonType @@ def _send_socket_message( socket_path: Path, - message: dict[str, object], + message: dict[str, JsonType], timeout: float = DEFAULT_SOCKET_TIMEOUT, -) -> dict[str, object]: +) -> dict[str, JsonType]: @@ - message: dict[str, object] = { + message: dict[str, JsonType] = { "event_type": args.event_type, "payload": payload, }Also applies to: 500-504
src/omnibase_infra/runtime/emit_daemon/queue.py (1)
93-100: Use JsonType for queued event payload.
payloadshould bedict[str, JsonType]to match JSON compatibility across the system. As per coding guidelines, useJsonTypefor JSON-compatible values.♻️ Suggested typing update
from pydantic import BaseModel, ConfigDict, Field, field_validator +from omnibase_core.types import JsonType @@ - payload: dict[str, object] = Field( + payload: dict[str, JsonType] = Field( ..., description="Event payload data.", )src/omnibase_infra/runtime/emit_daemon/client.py (1)
279-477: Use JsonType for payload/request/response types.
Update payload and request/response annotations toJsonTypeacrossemit,ping, and module-level helpers to enforce JSON-compatibility. As per coding guidelines, useJsonTypefor JSON-compatible values.♻️ Suggested typing update (apply similarly across helpers)
from collections.abc import Awaitable, Callable, Coroutine from pathlib import Path from types import TracebackType from typing import TypeVar +from omnibase_core.types import JsonType @@ - async def _send_request(self, request: dict[str, object]) -> dict[str, object]: + async def _send_request(self, request: dict[str, JsonType]) -> dict[str, JsonType]: @@ - async def emit(self, event_type: str, payload: dict[str, object]) -> str: + async def emit(self, event_type: str, payload: dict[str, JsonType]) -> str: @@ - async def ping(self) -> dict[str, object]: + async def ping(self) -> dict[str, JsonType]: @@ -async def emit_event(event_type: str, payload: dict[str, object], ... ) -> str: +async def emit_event(event_type: str, payload: dict[str, JsonType], ... ) -> str:Also applies to: 630-671
tests/unit/runtime/emit_daemon/test_queue.py (1)
82-97: Use JsonType in test payload annotations.
Even in tests, useJsonTypefor JSON-compatible payloads to match project standards. As per coding guidelines, useJsonTypefor JSON-compatible values.♻️ Suggested typing update
from uuid import uuid4 import pytest +from omnibase_core.types import JsonType @@ - payload: dict | None = None, + payload: dict[str, JsonType] | None = None,
Comment on lines
+159
to
+217
| def _send_socket_message( | ||
| socket_path: Path, | ||
| message: dict[str, object], | ||
| timeout: float = DEFAULT_SOCKET_TIMEOUT, | ||
| ) -> dict[str, object]: | ||
| """Send a message to the daemon via Unix socket. | ||
|
|
||
| Args: | ||
| socket_path: Path to the Unix domain socket. | ||
| message: JSON-serializable message to send. | ||
| timeout: Socket timeout in seconds. | ||
|
|
||
| Returns: | ||
| Parsed JSON response from daemon. | ||
|
|
||
| Raises: | ||
| ConnectionError: If cannot connect to daemon. | ||
| TimeoutError: If operation times out. | ||
| ValueError: If response is invalid JSON. | ||
| """ | ||
| if not socket_path.exists(): | ||
| raise ConnectionError(f"Socket not found: {socket_path}") | ||
|
|
||
| sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) | ||
| sock.settimeout(timeout) | ||
|
|
||
| try: | ||
| sock.connect(str(socket_path)) | ||
|
|
||
| # Send message (newline-delimited JSON) | ||
| request = json.dumps(message) + "\n" | ||
| sock.sendall(request.encode("utf-8")) | ||
|
|
||
| # Read response | ||
| response_data = b"" | ||
| while True: | ||
| chunk = sock.recv(4096) | ||
| if not chunk: | ||
| break | ||
| response_data += chunk | ||
| if b"\n" in response_data: | ||
| break | ||
|
|
||
| if not response_data: | ||
| raise ConnectionError("Empty response from daemon") | ||
|
|
||
| # Parse response | ||
| response_str = response_data.decode("utf-8").strip() | ||
| result = json.loads(response_str) | ||
| if not isinstance(result, dict): | ||
| raise ValueError("Response must be a JSON object") | ||
| return result | ||
|
|
||
| except TimeoutError as e: | ||
| raise TimeoutError(f"Socket operation timed out: {e}") from e | ||
| except json.JSONDecodeError as e: | ||
| raise ValueError(f"Invalid JSON response: {e}") from e | ||
| finally: | ||
| sock.close() |
There was a problem hiding this comment.
Handle socket OSErrors and normalize to project error type.
socket.connect/send/recv can raise PermissionError/ConnectionRefusedError/OSError that currently bubble and bypass the cmd_* handlers (they only catch ConnectionError/TimeoutError/ValueError), leading to tracebacks. Catch OSError and re-raise with chaining using your OnexError-based CLI error type. As per coding guidelines, replace built-in raises with OnexError subclasses and from e chaining.
🔧 Suggested guard (swap ConnectionError for your OnexError subclass)
try:
sock.connect(str(socket_path))
@@
- except TimeoutError as e:
- raise TimeoutError(f"Socket operation timed out: {e}") from e
+ except TimeoutError as e:
+ raise TimeoutError(f"Socket operation timed out: {e}") from e
+ except OSError as e:
+ raise ConnectionError(f"Socket error: {e}") from e
except json.JSONDecodeError as e:
raise ValueError(f"Invalid JSON response: {e}") from e🤖 Prompt for AI Agents
In `@src/omnibase_infra/runtime/emit_daemon/cli.py` around lines 159 - 217, The
_send_socket_message function must catch OSError (and its subclasses like
PermissionError/ConnectionRefusedError) around socket.connect/sendall/recv and
normalize them to your OnexError-based CLI error type (use the project's
OnexError subclass) with exception chaining (raise YourOnexCliError("descriptive
message", ...) from e); also replace the existing raises of built-in
ConnectionError for missing socket and empty response with the same OnexError
subclass so cmd_* handlers receive the normalized error; ensure
json.JSONDecodeError still maps to ValueError (or convert to OnexError if
project policy requires) and include the original exception message when
re-raising.
- Add configurable `environment` field to ModelEmitDaemonConfig - Propagate correlation_id from payload instead of generating new one - Add Windows platform guard for --daemonize option - Add --version flag to CLI - Display log file path before daemonization
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Fix all issues with AI agents
In `@src/omnibase_infra/runtime/emit_daemon/cli.py`:
- Around line 324-339: Before calling _daemonize() and logging.basicConfig,
ensure the directory for config.spool_dir.parent exists so opening the logfile
won't fail; call config.spool_dir.parent.mkdir(parents=True, exist_ok=True) (or
equivalent) prior to printing/daemonizing and before passing filename to
logging.basicConfig in the args.daemonize branch (keep this immediately before
_daemonize and the logging setup), and optionally catch and surface any OSError
so startup errors are visible before stdio is redirected.
In `@src/omnibase_infra/runtime/emit_daemon/daemon.py`:
- Around line 592-623: After the backoff sleep, check the boolean result of
await self._queue.enqueue(event) and handle a False return instead of silently
dropping: capture the result into a variable (e.g., success_enqueue = await
self._queue.enqueue(event)), and if not success_enqueue log an explicit error
(including event.event_id, event.event_type, event.topic and current
event.retry_count), increment a drop counter or move the event to a
dead-letter/spool mechanism, and ensure you do not re-enter infinite retries;
update the retry/drop logic around event.retry_count, self.MAX_RETRY_ATTEMPTS,
self.BACKOFF_BASE_SECONDS and the call to self._queue.enqueue to reflect this
explicit failure handling.
- Around line 204-212: Replace the RuntimeError raised in the daemon startup
block with the project error type: import and raise OnexError (or a
daemon-specific subclass like DaemonAlreadyRunningError) where the current raise
occurs in the same block that calls _check_stale_socket() and _cleanup_stale();
keep the existing message f"Another emit daemon is already running with PID
{pid}" and ensure the import for OnexError (or the subclass) is added at the top
of daemon.py so callers receive the consistent project error type.
- Add configurable retry parameters (max_retry_attempts, backoff_base_seconds) to ModelEmitDaemonConfig with env var support - Add thread-safe async size methods (memory_size_locked, spool_size_locked, total_size_locked) to BoundedEventQueue - Fix CLI logging race by clearing handlers before reconfiguration after daemonize - Move 'import time' to module level in cli.py - Enhance causation_id documentation in event_registry.py
Error handling improvements: - Replace ValueError with OnexError across event_registry.py, config.py - EmitClientError now inherits from OnexError with proper error codes - Use OnexError for "daemon already running" error in daemon.py - Handle socket OSErrors in cli.py with proper exception chaining - Handle failed re-enqueue after backoff in daemon.py Configuration fixes: - Make environment configurable instead of hardcoded "dev" in daemon.py - Respect spooling-disabled configuration in queue.py - Create log directory before daemonizing in cli.py Type improvements: - Use JsonType for payload types in client.py and queue.py - Fix AsyncGenerator type hint mismatch in test fixtures Minor fixes: - Fix literal PID placeholder in shutdown warning (cli.py) - Increase flaky test timeout from 1ms to 10ms - Remove environment-specific IPs from docstrings
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Fix all issues with AI agents
In `@src/omnibase_infra/runtime/emit_daemon/cli.py`:
- Around line 101-111: The YAML-to-dict conversion is treating booleans as
"other" and stringifying them; update the isinstance check to include bool so
True/False are preserved (modify the branch that builds result from data.items
to check isinstance(value, (str, int, float, bool))). Also update the function's
return type annotation (the function that constructs result from data) to
include bool in dict[str, str | int | float | bool] so type hints match runtime
behavior.
- Around line 364-368: The except block in cli.py should catch OnexError from
omnibase_core.errors instead of RuntimeError because daemon.start() raises
OnexError; update the import to include OnexError and change the except clause
to "except OnexError as e:" (preserving the existing print/logging/sys.exit
behavior) so the daemon startup conflict is handled correctly when
daemon.start() raises OnexError.
In `@src/omnibase_infra/runtime/emit_daemon/daemon.py`:
- Around line 506-515: The handlers currently catch ValueError for EventRegistry
operations but EventRegistry.resolve_topic and EventRegistry.validate_payload
now raise OnexError; update the exception handlers to catch OnexError instead of
ValueError so errors are converted to JSON responses. Locate the try/except
blocks around self._registry.resolve_topic(event_type) and
self._registry.validate_payload(event_type, payload) in daemon.py and replace
the except ValueError clauses with except OnexError (import OnexError if
missing) so resolution/validation failures return json.dumps({"status": "error",
"reason": str(e)}).
🧹 Nitpick comments (2)
src/omnibase_infra/runtime/emit_daemon/cli.py (2)
294-304: Consider using a more restrictive umask.
os.umask(0)allows files created by the daemon (socket, PID file, logs, spool) to be world-readable/writable depending on the mode argument. This is overly permissive for a daemon handling event data.🔒 Suggested fix
# Reset file creation mask - os.umask(0) + os.umask(0o022) # rw-r--r-- for files, rwxr-xr-x for directoriesOr use
0o077for more restrictive permissions (owner-only access).
326-351: Move directory creation before_daemonize()for error visibility.The
mkdircall at line 344 happens after_daemonize()(line 334), which redirects stdio to/dev/null. If directory creation fails (e.g., permission denied), the error is silently lost. Move the directory creation before daemonizing so users see startup errors.🔧 Suggested fix
if args.daemonize: - log_file = config.spool_dir.parent / "emit-daemon.log" + log_dir = config.spool_dir.parent + log_file = log_dir / "emit-daemon.log" + # Create log directory before daemonizing so errors are visible + try: + log_dir.mkdir(parents=True, exist_ok=True) + except OSError as e: + print(f"Error: Cannot create log directory: {e}", file=sys.stderr) + sys.exit(1) print("Starting emit-daemon in background...") print(f" Socket: {config.socket_path}") print(f" PID file: {config.pid_path}") print(f" Kafka: {config.kafka_bootstrap_servers}") print(f" Log file: {log_file}") _daemonize() # NOTE: After daemonize, stdio is redirected to /dev/null. - # Startup errors will appear in the log file above. # Clear any existing handlers to ensure reconfiguration works - # (logging.basicConfig() is a no-op if handlers already exist) root_logger = logging.getLogger() for handler in root_logger.handlers[:]: root_logger.removeHandler(handler) - # Ensure log directory exists before setting up file logging - log_dir = config.spool_dir.parent - log_dir.mkdir(parents=True, exist_ok=True) # Set up file-based logging instead logging.basicConfig(
- Update tests to expect OnexError instead of ValueError after code refactoring to use OnexError for validation errors - Fix security issue: change os.umask(0) to os.umask(0o022) for secure file permissions (rw-r--r-- for files) - Fix directory creation timing: move mkdir before _daemonize() so errors are visible to user before stdio is redirected - Remove unused imports (timezone, TYPE_CHECKING)
- Add PUBLISHER_POLL_INTERVAL_SECONDS constant to replace magic number - Document hasattr checks for event_bus.start() and close() methods (required for test mocks that may not implement all methods) - Add debug logging for disconnect cleanup errors in EmitClient
- Add max_backoff_seconds config field (default 60s) to cap exponential backoff and prevent excessive delays (was up to 512s) - Add socket_permissions config field (default 0o660) for deployment flexibility instead of hardcoded permissions - Add validator for socket permissions with clear error messages - Support EMIT_DAEMON_MAX_BACKOFF_SECONDS and EMIT_DAEMON_SOCKET_PERMISSIONS environment variables with octal string parsing - Document acceptable race condition in publisher loop exit condition (intentional trade-off for fire-and-forget semantics) - Bump INFRA_MAX_UNIONS threshold 112 -> 113 for env_mappings type
- Create ModelEmitDaemonConfigInput with all 17 fields as optional - Replace dict[str, Path | str | int | float] union in with_env_overrides() - Early validation via ConfigDict(extra="forbid") catches typos at parse time - Decrement INFRA_MAX_UNIONS from 113 to 111 (eliminated 2 unions) - Add architecture exemption for tightly coupled input model
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@src/omnibase_infra/runtime/emit_daemon/config.py`:
- Around line 498-516: The ValueError in the env override loop is being
swallowed; instead catch ValueError around the conversions in the loop over
env_mappings and raise an OnexError chained from the original exception (use
raise OnexError(...) from e) including the env_var, env_value and target
field_name (and field_type like OCTAL_INT/Path) so misconfigurations fail fast;
update the block that populates input_fields (the loop referencing env_mappings,
input_fields, OCTAL_INT, and Path) to raise OnexError with a clear message from
the caught ValueError rather than silently passing.
🧹 Nitpick comments (1)
src/omnibase_infra/runtime/emit_daemon/daemon.py (1)
452-479: Use JsonType for JSON request typing.
Align request/payload signatures with the canonical JSON type alias.♻️ Suggested update
-from omnibase_core.errors import OnexError +from omnibase_core.errors import OnexError +from omnibase_core.types import JsonType @@ - async def _handle_command(self, request: dict[str, object]) -> str: + async def _handle_command(self, request: dict[str, JsonType]) -> str: @@ - async def _handle_event(self, request: dict[str, object]) -> str: + async def _handle_event(self, request: dict[str, JsonType]) -> str:As per coding guidelines: Use
JsonType(fromomnibase_core.types) as canonical type alias for JSON-compatible values.
Comment on lines
+498
to
+516
| for env_var, (field_name, field_type) in env_mappings.items(): | ||
| env_value = os.environ.get(env_var) | ||
| if env_value is not None: | ||
| try: | ||
| if field_type is Path: | ||
| input_fields[field_name] = Path(env_value) | ||
| elif field_type == OCTAL_INT: | ||
| # Parse as octal string (e.g., "660" -> 0o660 = 432) | ||
| # Handles both "660" and "0o660" formats | ||
| input_fields[field_name] = int(env_value, 8) | ||
| elif field_type is int: | ||
| input_fields[field_name] = int(env_value) | ||
| elif field_type is float: | ||
| input_fields[field_name] = float(env_value) | ||
| else: | ||
| input_fields[field_name] = env_value | ||
| except ValueError: | ||
| # Skip invalid env values, let Pydantic validation handle | ||
| pass |
There was a problem hiding this comment.
Don’t silently ignore invalid env overrides.
Swallowing ValueError hides misconfiguration and can lead to unexpected defaults. Fail fast with a clear OnexError (chained).
🔧 Suggested fix
- except ValueError:
- # Skip invalid env values, let Pydantic validation handle
- pass
+ except ValueError as e:
+ raise OnexError(
+ f"Invalid value for {env_var}: {env_value}"
+ ) from eAs per coding guidelines: Only raise OnexError or its subclasses — always use raise OnexError(...) from e for error chaining.
🤖 Prompt for AI Agents
In `@src/omnibase_infra/runtime/emit_daemon/config.py` around lines 498 - 516, The
ValueError in the env override loop is being swallowed; instead catch ValueError
around the conversions in the loop over env_mappings and raise an OnexError
chained from the original exception (use raise OnexError(...) from e) including
the env_var, env_value and target field_name (and field_type like
OCTAL_INT/Path) so misconfigurations fail fast; update the block that populates
input_fields (the loop referencing env_mappings, input_fields, OCTAL_INT, and
Path) to raise OnexError with a clear message from the caught ValueError rather
than silently passing.
- Combine OMN-1518 dispatch engine simplification with OMN-1610 emit daemon - INFRA_MAX_UNIONS: 106 (main) + 6 (emit daemon) = 112 - Update test expectation for merged union count
CRITICAL fixes: - cli.py: Replace ValueError with OnexError throughout - cli.py: Catch OnexError instead of RuntimeError in exception handlers MAJOR fixes: - daemon.py: Fix exception type mismatch (catch OnexError from EventRegistry) - queue.py: Add defensive guard in _spool_event() for disabled spooling - cli.py: Use dict[str, object] for YAML config to avoid union complexity MINOR fixes: - Remove unused timezone import from model_handler_source_config.py - Replace environment-specific IPs (192.168.x.x) with localhost in docstrings - Add TODO note about Protocol formalization for EventBusKafka All 244 unit tests pass.
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/omnibase_infra/models/handlers/model_handler_source_config.py (1)
171-179: ReplaceValueErrorwithOnexErrorin the validator.
The coding guidelines require raising onlyOnexErroror its subclasses. Use the simpler pattern already established in the codebase (seeemit_daemon/config.pyvalidators): raiseOnexErrordirectly with the message, no need to wrapValueError.🔧 Proposed fix
-from datetime import UTC, datetime +from datetime import UTC, datetime + +from omnibase_core.errors import OnexError @@ if value.tzinfo is None: - raise ValueError( + raise OnexError( "bootstrap_expires_at must be timezone-aware (UTC recommended). " "Use datetime.now(timezone.utc) or datetime(..., tzinfo=timezone.utc)." )
🤖 Fix all issues with AI agents
In `@src/omnibase_infra/validation/validation_exemptions.yaml`:
- Around line 1828-1839: The violation_pattern strings for the two UUID
exemptions use the longer wording "Field 'X' should use UUID type instead of
str" which doesn't match the existing exemptions; update the two entries whose
file_pattern values are 'emit_daemon/queue\.py' (field event_id) and
'emit_daemon/config\.py' (field kafka_client_id) so their violation_pattern
values match the existing format "Field 'X' should use UUID" (e.g., "Field
'event_id' should use UUID" and "Field 'kafka_client_id' should use UUID"); keep
the rest of each exemption intact and verify whether these exemptions are still
needed given the ONEX_EXCLUDE: string_id comments in queue.py and config.py.
🧹 Nitpick comments (1)
src/omnibase_infra/runtime/emit_daemon/cli.py (1)
158-218: Socket error handling is improved, but exception types are mixed.The OSError handling (lines 215-216) addresses the past review comment about catching socket errors. However, the function mixes built-in exceptions (
ConnectionError,TimeoutError) with project exceptions (OnexError). Per coding guidelines, consider normalizing all toOnexErrorsubclasses for consistency.Current exception types:
ConnectionError(built-in) for socket/connection issuesTimeoutError(built-in) for timeoutsOnexErrorfor JSON parsing issuesThe callers (
cmd_status,cmd_emit) catch all three types appropriately, so this works functionally.
Comment on lines
+1828
to
+1839
| - file_pattern: 'emit_daemon/queue\.py' | ||
| violation_pattern: "Field 'event_id' should use UUID type instead of str" | ||
| reason: event_id is a string representation of a UUID for JSON serialization compatibility. | ||
| documentation: | ||
| - src/omnibase_infra/runtime/emit_daemon/queue.py | ||
| ticket: OMN-1610 | ||
| - file_pattern: 'emit_daemon/config\.py' | ||
| violation_pattern: "Field 'kafka_client_id' should use UUID type instead of str" | ||
| reason: kafka_client_id is a Kafka client identifier string, not a UUID. | ||
| documentation: | ||
| - Kafka client.id documentation | ||
| ticket: OMN-1610 |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check existing UUID exemption patterns in validation_exemptions.yaml
echo "=== Existing UUID exemption patterns ==="
rg -n "should use UUID" src/omnibase_infra/validation/validation_exemptions.yaml | head -30
# Check for any validator code that generates UUID field validation messages
echo ""
echo "=== Searching for UUID field validation in validator code ==="
rg -n "should use UUID" --type py -A 2 -B 2
# Also search for field type validation messages in general
echo ""
echo "=== General field type validation patterns ==="
rg -n "Field.*should use" --type py -A 1 | head -40Repository: OmniNode-ai/omnibase_infra
Length of output: 1961
🏁 Script executed:
#!/bin/bash
# Search for the validator that generates these messages
echo "=== Searching for UUID field validation implementation ==="
rg -n "Field.*should use" --type py | head -20
# Also search in validation directory
echo ""
echo "=== Checking validation directory structure ==="
fd -t f ".*validator.*" src/omnibase_infra/validation/ | head -20
# Search for any file mentioning UUID type validation
echo ""
echo "=== Searching for UUID validation logic ==="
rg -n "UUID.*type" --type py -B 2 -A 2 | head -50
# Check all Python files in validation module
echo ""
echo "=== Python files in validation module ==="
fd -t f "\.py$" src/omnibase_infra/validation/ | head -30Repository: OmniNode-ai/omnibase_infra
Length of output: 6444
🏁 Script executed:
#!/bin/bash
# Read the validator_any_type.py file to find the UUID validation message
echo "=== Reading validator_any_type.py ==="
wc -l src/omnibase_infra/validation/validator_any_type.py
# Get the file size, then read it
echo ""
head -300 src/omnibase_infra/validation/validator_any_type.py | tail -100Repository: OmniNode-ai/omnibase_infra
Length of output: 4145
🏁 Script executed:
#!/bin/bash
# Search for where the actual violation message is generated
echo "=== Searching for 'should use UUID' in all files ==="
rg -n "should use UUID" --type py --type yaml
# Also search more broadly for field validation messages
echo ""
echo "=== Searching for violation message patterns ==="
rg -n "Field.*should" --type py | head -30Repository: OmniNode-ai/omnibase_infra
Length of output: 5102
🏁 Script executed:
#!/bin/bash
# Check how violation patterns are used and matched
echo "=== Searching for violation_pattern matching logic ==="
rg -n "violation_pattern" --type py -B 2 -A 2 | head -80
# Also check linter_contract.py for where messages might be generated
echo ""
echo "=== Reading linter_contract.py (first 200 lines) ==="
head -200 src/omnibase_infra/validation/linter_contract.py | tail -100Repository: OmniNode-ai/omnibase_infra
Length of output: 10320
🏁 Script executed:
#!/bin/bash
# Search for where field validation violations are created
echo "=== Searching for field type validation violations ==="
rg -n "Field.*should" --type py -B 3 -A 3 | head -100
# Also check if there's any code that generates messages about field types
echo ""
echo "=== Searching for message generation about field types ==="
rg -n "uuid\|UUID" --type py -i -l | head -20
# Check the emit_daemon files mentioned in the review
echo ""
echo "=== Checking emit_daemon files ==="
ls -la src/omnibase_infra/runtime/emit_daemon/ 2>/dev/null || echo "Directory not found"
fd -t f emit_daemon | head -20Repository: OmniNode-ai/omnibase_infra
Length of output: 9083
🏁 Script executed:
#!/bin/bash
# Check the queue.py file for event_id field
echo "=== Checking queue.py for event_id field ==="
rg -n "event_id" src/omnibase_infra/runtime/emit_daemon/queue.py -B 2 -A 2
# Check the config.py file for kafka_client_id field
echo ""
echo "=== Checking config.py for kafka_client_id field ==="
rg -n "kafka_client_id" src/omnibase_infra/runtime/emit_daemon/config.py -B 2 -A 2
# Also look for any UUID type imports in these files
echo ""
echo "=== Checking for UUID imports in emit_daemon ==="
rg -n "UUID\|uuid" src/omnibase_infra/runtime/emit_daemon/queue.py src/omnibase_infra/runtime/emit_daemon/config.py -B 1 -A 1Repository: OmniNode-ai/omnibase_infra
Length of output: 6020
🏁 Script executed:
#!/bin/bash
# Search for how ONEX_EXCLUDE is handled
echo "=== Searching for ONEX_EXCLUDE handling ==="
rg -n "ONEX_EXCLUDE" --type py -B 2 -A 2 | head -100
# Also search for how violations matching works
echo ""
echo "=== Searching for pattern matching in exemption filtering ==="
rg -n "filter.*exempted\|violation.*match" --type py | head -30
# Check if there's a UUID field validator
echo ""
echo "=== Searching for UUID field type checking ==="
rg -n "uuid.*type\|UUID.*field" --type py -i | head -40Repository: OmniNode-ai/omnibase_infra
Length of output: 7721
🏁 Script executed:
#!/bin/bash
# Search for validators that check field types
echo "=== Searching for field type validation ==="
find src/omnibase_infra/validation -name "*.py" -type f | xargs grep -l "field.*type\|type.*field" -i | head -10
# Check runtime_shape validator since it might validate field types
echo ""
echo "=== Checking validator_runtime_shape.py for field validation ==="
rg -n "Field\|field.*type\|should use" src/omnibase_infra/validation/validator_runtime_shape.py | head -40
# Look for any validator that generates "should use" messages
echo ""
echo "=== Searching for 'should use' message generation ==="
rg -n "should use" --type py
# Check the infra_validators.py file
echo ""
echo "=== Checking infra_validators.py ==="
wc -l src/omnibase_infra/validation/infra_validators.pyRepository: OmniNode-ai/omnibase_infra
Length of output: 17147
🏁 Script executed:
#!/bin/bash
# Search for UUID or field type validation in infra_validators.py
echo "=== Searching infra_validators.py for UUID/field type checks ==="
rg -n "uuid\|UUID\|field.*type\|type.*field" src/omnibase_infra/validation/infra_validators.py -i | head -50
# Also search for any validation function names
echo ""
echo "=== Function names in infra_validators.py ==="
rg -n "^def " src/omnibase_infra/validation/infra_validators.py | head -30
# Check if there's any UUID validation logic at all
echo ""
echo "=== Checking for UUID field type validation anywhere ==="
rg -n "UUID.*str\|str.*UUID" --type py -C 2 | head -60Repository: OmniNode-ai/omnibase_infra
Length of output: 1381
🏁 Script executed:
#!/bin/bash
# Find where violation_pattern matching actually happens
echo "=== Searching for where exemption patterns are matched ==="
rg -n "_filter_exempted_errors\|violation_pattern" src/omnibase_infra/validation/infra_validators.py -A 5 -B 2 | head -100
# Also check if pattern matching is regex-based
echo ""
echo "=== Checking pattern matching logic ==="
rg -n "re\.search\|re\.match\|re\.fullmatch" src/omnibase_infra/validation/ --type py | head -30
# Let's see the actual filter_exempted_errors function
echo ""
echo "=== Reading _filter_exempted_errors function ==="
sed -n '566,650p' src/omnibase_infra/validation/infra_validators.pyRepository: OmniNode-ai/omnibase_infra
Length of output: 3483
🏁 Script executed:
#!/bin/bash
# Search for where error messages with "Field" and UUID are generated
echo "=== Searching for error message generation with Field and UUID ==="
rg -n "Field.*UUID\|violation.*message" --type py -B 2 -A 2 | head -100
# Check test files to see what error messages are expected
echo ""
echo "=== Checking test files for UUID field validation messages ==="
rg -n "should use UUID" tests/ --type py -B 2 -A 2 | head -50
# Search for Pydantic validation or field validators
echo ""
echo "=== Searching for Pydantic validators or validators.py ==="
fd -t f "validator" src/omnibase_infra/ | head -20
# Check if there's a dedicated UUID field validator
echo ""
echo "=== Searching all validators for UUID field checking ==="
find src/omnibase_infra/validation -name "*.py" -exec grep -l "UUID" {} \;Repository: OmniNode-ai/omnibase_infra
Length of output: 1939
🏁 Script executed:
#!/bin/bash
# Read validator_runtime_shape.py to find UUID field validation
echo "=== Checking validator_runtime_shape.py ==="
wc -l src/omnibase_infra/validation/validator_runtime_shape.py
# Search for the actual code that generates field validation messages
rg -n "Field\|UUID\|should" src/omnibase_infra/validation/validator_runtime_shape.py | head -40
# Check if there's a contract violation or field validator
echo ""
echo "=== Searching for field type violations ==="
rg -n "field.*type\|type.*field" src/omnibase_infra/validation/ --type py -i | head -30
# Check the actual validation code that might be generating these messages
echo ""
echo "=== Looking for where violations are reported ==="
rg -n "violation\|error.*message" src/omnibase_infra/validation/validator_runtime_shape.py | head -20Repository: OmniNode-ai/omnibase_infra
Length of output: 270
🏁 Script executed:
#!/bin/bash
# Search the entire codebase for any reference to "type instead of str"
echo "=== Searching for 'type instead of str' anywhere ==="
rg -n "type instead of str" --type py --type yaml
# Check if there's any validation of field types at all
echo ""
echo "=== Checking for field type validation in contract linter ==="
head -400 src/omnibase_infra/validation/linter_contract.py | tail -200
# Also check what validators exist and what they validate
echo ""
echo "=== List of validation files and their purpose ==="
for file in src/omnibase_infra/validation/validator_*.py; do
echo "=== $(basename $file) ==="
head -30 "$file" | grep -A 3 "^\"\"\"" | head -4
doneRepository: OmniNode-ai/omnibase_infra
Length of output: 10447
Correct the violation_pattern format to match all existing UUID exemptions.
The new patterns use "Field 'X' should use UUID type instead of str", but all 26 existing UUID exemptions in this file use "Field 'X' should use UUID". Since exemption matching uses regex via re.search(), if the validator generates the shorter message format, these exemptions won't match and won't take effect.
Also note: Both queue.py (line 79) and config.py (line 186) already include ONEX_EXCLUDE: string_id comments, making these exemptions potentially redundant unless the validator explicitly ignores inline comments.
🤖 Prompt for AI Agents
In `@src/omnibase_infra/validation/validation_exemptions.yaml` around lines 1828 -
1839, The violation_pattern strings for the two UUID exemptions use the longer
wording "Field 'X' should use UUID type instead of str" which doesn't match the
existing exemptions; update the two entries whose file_pattern values are
'emit_daemon/queue\.py' (field event_id) and 'emit_daemon/config\.py' (field
kafka_client_id) so their violation_pattern values match the existing format
"Field 'X' should use UUID" (e.g., "Field 'event_id' should use UUID" and "Field
'kafka_client_id' should use UUID"); keep the rest of each exemption intact and
verify whether these exemptions are still needed given the ONEX_EXCLUDE:
string_id comments in queue.py and config.py.
jonahgabriel
commented
Jan 29, 2026
| return event | ||
| except OSError: | ||
| logger.exception("Failed to read spool file %s", filepath) | ||
| return None |
Collaborator
Author
There was a problem hiding this comment.
Bug: Event loss when unlink() fails after successful read
If filepath.unlink() on line 392 raises OSError after the event was successfully read and parsed, the exception is caught by this handler and None is returned, causing the successfully parsed event to be silently lost.
Problem flow:
filepathis popped from_spool_files(line 383)read_text()succeeds (line 386)model_validate_json()succeeds - event is valid (line 387)_spool_bytesis decremented (line 391)unlink()fails →OSErrorcaught here →return None- Event is lost
Additionally, the log message "Failed to read spool file" is misleading when the actual failure is unlink().
Recommended fix: Separate the unlink() into its own try/except block so the event is returned even if file deletion fails:
Fix src/omnibase_infra/runtime/emit_daemon/queue.py:392-401: Separate filepath.unlink() into its own try/except after the main try block, so that if unlink fails, the successfully parsed event is still returned and only a warning is logged about the orphan file.
jonahgabriel
commented
Jan 29, 2026
| response_line = await asyncio.wait_for( | ||
| self._reader.readline(), | ||
| timeout=self._timeout, | ||
| ) |
Collaborator
Author
There was a problem hiding this comment.
Bug: Concurrent emit() calls can mix up responses
This method does not acquire self._lock to protect the write-then-read sequence. While _connect() and _disconnect() use the lock, the actual request/response exchange (lines 322-330) is unprotected.
Concurrent call scenario:
async with EmitClient() as client:
# This can cause response mixing:
await asyncio.gather(
client.emit("event.one", data1), # Coroutine A
client.emit("event.two", data2), # Coroutine B
)Interleaving:
- Coroutine A writes request 1
- Coroutine B writes request 2
- Coroutine A reads (might get response 2)
- Coroutine B reads (might get response 1)
Task A could receive an event_id that belongs to Task B's event.
Note: The module-level emit_event() helper creates a new client per call and is safe. This only affects concurrent usage of a shared EmitClient instance.
Recommended fix: Either:
- Acquire
self._lockat the start of_send_requestto protect the entire write/read sequence, OR - Document that the client is NOT safe for concurrent coroutine use on a single instance
Fix src/omnibase_infra/runtime/emit_daemon/client.py:296-347: Add "async with self._lock:" at the start of _send_request method (after line 309) to protect the write-then-read sequence from concurrent access. You will need to refactor _connect() to have an unlocked variant since it is called within _send_request.
- queue.py: Separate filepath.unlink() into its own try/except block so that if deletion fails after successful read, the event is still returned instead of being silently lost - client.py: Add lock protection to _send_request() to prevent concurrent emit() calls from mixing up responses. Refactored _connect/_disconnect into locked and unlocked variants to avoid deadlock when called from within _send_request. Fixes issues identified in PR #207 code review.
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@src/omnibase_infra/runtime/emit_daemon/client.py`:
- Around line 656-666: The code in the sync wrapper that checks for an active
event loop currently raises a built-in RuntimeError; change this to raise the
project-specific EmitClientError (an OnexError subclass) with the same
descriptive message to comply with the "only raise OnexError subclasses"
guideline; update the raise site inside the function that performs the loop
check (the block that now raises RuntimeError with the text "Cannot use sync
methods from an async context. Use the async emit() or ping() methods instead.")
to instead construct and raise EmitClientError (ensure EmitClientError is
imported or referenced correctly).
Comment on lines
+656
to
+666
| if loop is not None: | ||
| # We're in an async context - use run_until_complete is not allowed | ||
| # Create a new event loop in a thread would be complex | ||
| # Instead, raise an error suggesting async usage | ||
| raise RuntimeError( | ||
| "Cannot use sync methods from an async context. " | ||
| "Use the async emit() or ping() methods instead." | ||
| ) | ||
|
|
||
| # No running loop - safe to use asyncio.run() | ||
| return asyncio.run(coro) |
There was a problem hiding this comment.
Replace RuntimeError with EmitClientError.
Line 660 raises RuntimeError, which violates the coding guideline requiring only OnexError subclasses. As per coding guidelines, "Only raise OnexError or its subclasses."
🔧 Proposed fix
if loop is not None:
# We're in an async context - use run_until_complete is not allowed
# Create a new event loop in a thread would be complex
# Instead, raise an error suggesting async usage
- raise RuntimeError(
+ raise EmitClientError(
"Cannot use sync methods from an async context. "
- "Use the async emit() or ping() methods instead."
+ "Use the async emit() or ping() methods instead.",
+ reason="async_context_error",
+ error_code=EnumCoreErrorCode.OPERATION_FAILED,
)🤖 Prompt for AI Agents
In `@src/omnibase_infra/runtime/emit_daemon/client.py` around lines 656 - 666, The
code in the sync wrapper that checks for an active event loop currently raises a
built-in RuntimeError; change this to raise the project-specific EmitClientError
(an OnexError subclass) with the same descriptive message to comply with the
"only raise OnexError subclasses" guideline; update the raise site inside the
function that performs the loop check (the block that now raises RuntimeError
with the text "Cannot use sync methods from an async context. Use the async
emit() or ping() methods instead.") to instead construct and raise
EmitClientError (ensure EmitClientError is imported or referenced correctly).
…protocol models Add strongly-typed Pydantic models for daemon request/response protocol: - ModelDaemonPingRequest, ModelDaemonEmitRequest for requests - ModelDaemonPingResponse, ModelDaemonQueuedResponse, ModelDaemonErrorResponse for responses - parse_daemon_request() and parse_daemon_response() for type-safe parsing Benefits: - Compile-time type checking instead of runtime isinstance() checks - IDE autocomplete for response fields (response.status vs response["status"]) - Pydantic validation catches invalid responses at parse time - Discriminated unions enable proper type narrowing Updated daemon.py to use _handle_ping() and _handle_emit() with typed models. Updated client.py ping() to return ModelDaemonPingResponse instead of dict. Updated tests to use typed model attributes.
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Fix all issues with AI agents
In `@src/omnibase_infra/runtime/emit_daemon/daemon.py`:
- Around line 206-209: The Raises section of the daemon method docstring still
lists RuntimeError but the code now raises OnexError; update the docstring to
replace "RuntimeError: If another daemon is already running" with "OnexError: If
another daemon is already running" (locate the docstring in the daemon method in
src/omnibase_infra/runtime/emit_daemon/daemon.py that currently lists OSError
and RuntimeError) so the documentation matches the actual exception type.
In `@src/omnibase_infra/runtime/emit_daemon/model_daemon_request.py`:
- Around line 106-131: Replace the ValueError in parse_daemon_request with the
project-specific OnexError and update the docstring to list OnexError instead of
ValueError; specifically, in parse_daemon_request (which returns
ModelDaemonPingRequest or ModelDaemonEmitRequest via
ModelDaemonPingRequest.model_validate / ModelDaemonEmitRequest.model_validate)
raise OnexError("Invalid request: must contain either 'command' or 'event_type'
field") using the error chaining pattern (raise OnexError(...) from e) when
converting or validating input fails, and ensure callers expecting ValueError
are adjusted to handle OnexError.
In `@src/omnibase_infra/runtime/emit_daemon/model_daemon_response.py`:
- Around line 154-182: The function parse_daemon_response currently raises
ValueError for unknown status; replace that with raising the project-standard
OnexError and use exception chaining (i.e., raise OnexError(f"Unknown response
status: {status}") from None or from the original exception if one exists).
Update imports to include OnexError, change the final else branch to raise
OnexError(...) instead of ValueError, and ensure the error message remains
descriptive so the protocol layer consistently emits OnexError instances.
In `@src/omnibase_infra/validation/infra_validators.py`:
- Around line 438-446: Update the unit test assertion that checks
INFRA_MAX_UNIONS: locate the assertion that expects 112 and change the expected
literal to 113 so it matches the current constant INFRA_MAX_UNIONS; ensure the
assertion message (if present) reflects the new expected value and leave
INFRA_MAX_UNIONS itself unchanged.
🧹 Nitpick comments (1)
tests/unit/runtime/emit_daemon/test_client.py (1)
606-615: Test correctly reflects current implementation, but note the pending refactor.This test expects
RuntimeErrorwhen sync methods are called from an async context. This aligns with the current implementation inclient.pyline 692. However, based on the past review comment, there's a pending change to replaceRuntimeErrorwithEmitClientErrorto comply with coding guidelines. When that refactor is applied, this test will need to be updated to expectEmitClientErrorwith reason"async_context_error".
Comment on lines
+206
to
+209
| Raises: | ||
| OSError: If socket creation fails | ||
| RuntimeError: If another daemon is already running | ||
| """ |
There was a problem hiding this comment.
Update the Raises docstring to match OnexError.
The method now raises OnexError, but the docstring still says RuntimeError.
📝 Suggested fix
- RuntimeError: If another daemon is already running
+ OnexError: If another daemon is already running🤖 Prompt for AI Agents
In `@src/omnibase_infra/runtime/emit_daemon/daemon.py` around lines 206 - 209, The
Raises section of the daemon method docstring still lists RuntimeError but the
code now raises OnexError; update the docstring to replace "RuntimeError: If
another daemon is already running" with "OnexError: If another daemon is already
running" (locate the docstring in the daemon method in
src/omnibase_infra/runtime/emit_daemon/daemon.py that currently lists OSError
and RuntimeError) so the documentation matches the actual exception type.
Comment on lines
+106
to
+131
| def parse_daemon_request( | ||
| data: dict[str, object], | ||
| ) -> ModelDaemonPingRequest | ModelDaemonEmitRequest: | ||
| """Parse raw dict into typed request model. | ||
|
|
||
| Discriminates between request types based on field presence: | ||
| - "command" field present -> ModelDaemonPingRequest | ||
| - "event_type" field present -> ModelDaemonEmitRequest | ||
|
|
||
| Args: | ||
| data: Raw request dict from JSON parsing | ||
|
|
||
| Returns: | ||
| Typed request model | ||
|
|
||
| Raises: | ||
| ValueError: If request format is invalid | ||
| """ | ||
| if "command" in data: | ||
| return ModelDaemonPingRequest.model_validate(data) | ||
| elif "event_type" in data: | ||
| return ModelDaemonEmitRequest.model_validate(data) | ||
| else: | ||
| raise ValueError( | ||
| "Invalid request: must contain either 'command' or 'event_type' field" | ||
| ) |
There was a problem hiding this comment.
Use OnexError instead of ValueError for invalid request format.
This module should adhere to the project error contract; swap the local ValueError for OnexError (and update the docstring/callers accordingly). As per coding guidelines: Only raise OnexError or its subclasses - always use raise OnexError(...) from e pattern for error chaining.
🔧 Suggested fix
-from typing import Annotated, Literal
+from typing import Annotated, Literal
+
+from omnibase_core.errors import OnexError
@@
- Raises:
- ValueError: If request format is invalid
+ Raises:
+ OnexError: If request format is invalid
@@
- raise ValueError(
- "Invalid request: must contain either 'command' or 'event_type' field"
- )
+ raise OnexError(
+ "Invalid request: must contain either 'command' or 'event_type' field"
+ )🤖 Prompt for AI Agents
In `@src/omnibase_infra/runtime/emit_daemon/model_daemon_request.py` around lines
106 - 131, Replace the ValueError in parse_daemon_request with the
project-specific OnexError and update the docstring to list OnexError instead of
ValueError; specifically, in parse_daemon_request (which returns
ModelDaemonPingRequest or ModelDaemonEmitRequest via
ModelDaemonPingRequest.model_validate / ModelDaemonEmitRequest.model_validate)
raise OnexError("Invalid request: must contain either 'command' or 'event_type'
field") using the error chaining pattern (raise OnexError(...) from e) when
converting or validating input fails, and ensure callers expecting ValueError
are adjusted to handle OnexError.
Comment on lines
+154
to
+182
| def parse_daemon_response( | ||
| data: dict[str, object], | ||
| ) -> ModelDaemonPingResponse | ModelDaemonQueuedResponse | ModelDaemonErrorResponse: | ||
| """Parse raw dict into typed response model. | ||
|
|
||
| Discriminates between response types based on "status" field: | ||
| - "ok" -> ModelDaemonPingResponse | ||
| - "queued" -> ModelDaemonQueuedResponse | ||
| - "error" -> ModelDaemonErrorResponse | ||
|
|
||
| Args: | ||
| data: Raw response dict from JSON parsing | ||
|
|
||
| Returns: | ||
| Typed response model | ||
|
|
||
| Raises: | ||
| ValueError: If response format is invalid or status unknown | ||
| """ | ||
| status = data.get("status") | ||
|
|
||
| if status == "ok": | ||
| return ModelDaemonPingResponse.model_validate(data) | ||
| elif status == "queued": | ||
| return ModelDaemonQueuedResponse.model_validate(data) | ||
| elif status == "error": | ||
| return ModelDaemonErrorResponse.model_validate(data) | ||
| else: | ||
| raise ValueError(f"Unknown response status: {status}") |
There was a problem hiding this comment.
Raise OnexError for unknown response status.
Swap the local ValueError for OnexError so the protocol layer consistently emits project-standard errors. As per coding guidelines: Only raise OnexError or its subclasses - always use raise OnexError(...) from e pattern for error chaining.
🔧 Suggested fix
-from typing import Annotated, Literal
+from typing import Annotated, Literal
+
+from omnibase_core.errors import OnexError
@@
- Raises:
- ValueError: If response format is invalid or status unknown
+ Raises:
+ OnexError: If response format is invalid or status unknown
@@
- else:
- raise ValueError(f"Unknown response status: {status}")
+ else:
+ raise OnexError(f"Unknown response status: {status}")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| def parse_daemon_response( | |
| data: dict[str, object], | |
| ) -> ModelDaemonPingResponse | ModelDaemonQueuedResponse | ModelDaemonErrorResponse: | |
| """Parse raw dict into typed response model. | |
| Discriminates between response types based on "status" field: | |
| - "ok" -> ModelDaemonPingResponse | |
| - "queued" -> ModelDaemonQueuedResponse | |
| - "error" -> ModelDaemonErrorResponse | |
| Args: | |
| data: Raw response dict from JSON parsing | |
| Returns: | |
| Typed response model | |
| Raises: | |
| ValueError: If response format is invalid or status unknown | |
| """ | |
| status = data.get("status") | |
| if status == "ok": | |
| return ModelDaemonPingResponse.model_validate(data) | |
| elif status == "queued": | |
| return ModelDaemonQueuedResponse.model_validate(data) | |
| elif status == "error": | |
| return ModelDaemonErrorResponse.model_validate(data) | |
| else: | |
| raise ValueError(f"Unknown response status: {status}") | |
| def parse_daemon_response( | |
| data: dict[str, object], | |
| ) -> ModelDaemonPingResponse | ModelDaemonQueuedResponse | ModelDaemonErrorResponse: | |
| """Parse raw dict into typed response model. | |
| Discriminates between response types based on "status" field: | |
| - "ok" -> ModelDaemonPingResponse | |
| - "queued" -> ModelDaemonQueuedResponse | |
| - "error" -> ModelDaemonErrorResponse | |
| Args: | |
| data: Raw response dict from JSON parsing | |
| Returns: | |
| Typed response model | |
| Raises: | |
| OnexError: If response format is invalid or status unknown | |
| """ | |
| status = data.get("status") | |
| if status == "ok": | |
| return ModelDaemonPingResponse.model_validate(data) | |
| elif status == "queued": | |
| return ModelDaemonQueuedResponse.model_validate(data) | |
| elif status == "error": | |
| return ModelDaemonErrorResponse.model_validate(data) | |
| else: | |
| raise OnexError(f"Unknown response status: {status}") |
🤖 Prompt for AI Agents
In `@src/omnibase_infra/runtime/emit_daemon/model_daemon_response.py` around lines
154 - 182, The function parse_daemon_response currently raises ValueError for
unknown status; replace that with raising the project-standard OnexError and use
exception chaining (i.e., raise OnexError(f"Unknown response status: {status}")
from None or from the original exception if one exists). Update imports to
include OnexError, change the final else branch to raise OnexError(...) instead
of ValueError, and ensure the error message remains descriptive so the protocol
layer consistently emits OnexError instances.
…tBusKafka type Follow ONEX protocol-first design by typing _event_bus against the ProtocolEventBusLike protocol rather than the concrete EventBusKafka class. This enables proper test doubles and aligns with duck-typing patterns. - Add ProtocolEventBusLike import from omnibase_infra.protocols - Change event_bus parameter and attribute type to ProtocolEventBusLike | None - Add type: ignore[call-arg] for Kafka-specific headers parameter - Preserve hasattr checks for lifecycle methods (start/close) not in protocol
Collaborator
Author
Code ReviewNo issues found. Checked for bugs and CLAUDE.md compliance. Review SummaryCLAUDE.md Compliance: ✅ Passed
Bug Scan: ✅ No issues found
Reviewed by Claude Code |
- Update test_store.py to expect 'localhost' as default postgres_host (matches config_store.py change to use localhost) - Update test_validator_defaults.py to expect 113 unions (reflects +8 unions from emit daemon typed protocol models)
There was a problem hiding this comment.
Actionable comments posted: 0
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
tests/services/session/test_store.py (1)
28-28: ReplaceAnywithobjectto satisfy typing policy.
create_test_snapshot_datausesAnyin both the return type and local dict typing; this violates the repo typing rule and may fail validation. Useobject(values includeUUID/datetime, soJsonTypeisn’t accurate here).🔧 Proposed fix
-from typing import Any @@ -) -> dict[str, Any]: +) -> dict[str, object]: @@ - data: dict[str, Any] = { + data: dict[str, object] = {As per coding guidelines,
**/*.py: Never useAnytype - useobjectfor generic payloads; only allowAnyin PydanticField()withNOTE:comment when technically required (e.g., discriminated unions); enforce viascripts/validate.py any_types.Also applies to: 112-118
The test_config_has_default_values test was failing in CI because ConfigSessionStorage (a pydantic-settings model) reads from environment variables. CI has POSTGRES_HOST=192.168.86.200 set, overriding the hardcoded default of 'localhost'. Fix: Access model_fields directly to test Field() defaults, bypassing pydantic-settings environment variable loading. This is the proper pattern for testing Pydantic settings defaults in environments where env vars may be set.
This was referenced Feb 1, 2026
This was referenced Feb 11, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Implementation
config.pyModelEmitDaemonConfig- Pydantic configuration with validationevent_registry.pyqueue.pyBoundedEventQueue- In-memory (100 msgs) + disk spool (1000 msgs/10MB)daemon.pyclient.pyEmitClient- Async/sync APIs with fallback supportcli.pyArchitecture
Design Decisions
Test Plan
emit-daemon start/emit/status/stopRelated Tickets
Summary by CodeRabbit
New Features
Tests
Chores
✏️ Tip: You can customize this high-level summary in your review settings.