Event-core correctness fixes + refactoring (isolation, provider DRY, EventCodec, mypy→0)

docs/docs/getting-started/features.md

@@ -39,7 +39,9 @@ Graphite adopts an event-driven architecture where topics function as logical me
 
 Consumption events are only recorded once the entire node processing completes successfully. Until that point, the system treats partial or failed node invokes as if they never happened, preventing duplicated outputs or broken states. Should a node encounter an error (e.g., an LLM connection failure, external API issue, or function exception), Graphite detects the unconsumed events upon restoration and places the associated node(s) back into the invoke queue. This design ensures the node can safely retry from the same input without creating conflicting or duplicated consumption records.
 
-By storing each event exactly once and withholding consumption records until success, Graphite guarantees idempotent behavior. Even if a node issues multiple invocations due to an error, the event logs and consumption rules still reconstruct a single, consistent path from invocation to response. This approach produces correct outcomes on retries while maintaining a complete, conflict-free audit trail.
+By recording each event idempotently and withholding consumption records until success, Graphite makes its **event persistence and replay** idempotent: the event logs and consumption rules reconstruct a single, consistent path from invocation to response, maintaining a complete, conflict-free audit trail.
+
+This is not the same as exactly-once execution of external side effects. Node and tool invocation is **at-least-once**: on retry, a node runs again from the same input, so any external action it performs (an LLM call, an outbound API request, a write) can happen more than once unless the tool itself is idempotent. Make tools with external side effects idempotent (e.g. via an idempotency key) if you need exactly-once *effects*.
 
 ## Auditability
 

docs/docs/user-guide/event-driven-workflow.md

@@ -100,14 +100,14 @@ The asynchronous execution model provides sophisticated event-driven processing
 1. **Workflow Initialization**: Sets up initial state with `init_workflow`
 2. **Concurrent Node Processing**: Spawns individual tasks for each node using `_invoke_node`  
 3. **Output Listening**: Creates listeners for each output topic to capture results
-4. **Event Streaming**: Uses `MergeIdleQueue` to stream events as they become available
+4. **Event Streaming**: Uses `AsyncOutputQueue` to stream output events as they become available
 5. **Proper Termination**: Coordinates workflow completion using `AsyncNodeTracker`
 
 #### Key Components
 
-- **AsyncNodeTracker**: Manages active node state and idle detection
+- **AsyncNodeTracker**: Manages active node state and quiescence detection
 - **Output Listeners**: Monitor output topics for new events
-- **MergeIdleQueue**: Coordinates between event availability and workflow idle state
+- **AsyncOutputQueue**: Streams output-topic events and coordinates with workflow quiescence
 - **Offset Management**: Commits events immediately to prevent duplicates
 
 ### Event-Driven Node Triggering
@@ -185,7 +185,9 @@ async for event in output_queue:
 The async workflow implements proper offset management to prevent duplicate data:
 
 ```python
-async for event in MergeIdleQueue(queue, self._tracker):
+output_queue = AsyncOutputQueue(output_topics, self.name, self._tracker)
+await output_queue.start_listeners()
+async for event in output_queue:
     consumed_output_event = ConsumeFromTopicEvent(...)
 
     # Commit BEFORE yielding to prevent duplicate data

docs/docs/user-guide/tools/function.md

@@ -39,7 +39,7 @@ The tool automatically handles different response types in its `to_messages` met
 - **`BaseModel` instances**: Serialized to JSON using `model_dump_json()`
 - **Lists of `BaseModel` objects**: Converted to JSON arrays using `model_dump()` for each item
 - **String responses**: Used directly as message content
-- **Other types**: Encoded using `jsonpickle` for complex object serialization
+- **Other types**: Encoded with `json.dumps(..., default=str)` (no pickle)
 
 ### Async Support
 

grafi/assistants/assistant.py

@@ -69,6 +69,8 @@ def generate_manifest(self, output_dir: str = ".") -> str:
         with open(output_path, "w") as f:
             f.write(json.dumps(manifest_dict, indent=4))
 
+        return output_path
+
     @classmethod
     async def from_dict(cls, data: dict[str, Any]) -> "Assistant":
         """

grafi/assistants/assistant_base.py

@@ -44,8 +44,12 @@ def model_post_init(self, _context: Any) -> None:
         self._construct_workflow()
 
     def _construct_workflow(self) -> "AssistantBase":
-        """Construct the workflow for the assistant."""
-        pass
+        """Construct the workflow for the assistant.
+
+        Subclasses override this to build ``self.workflow`` and return ``self``.
+        The base is a no-op for a bare assistant with a default workflow.
+        """
+        return self
 
     async def invoke(
         self, input_data: PublishToTopicEvent, is_sequential: bool = False

grafi/common/decorators/llm_function.py

@@ -149,7 +149,7 @@ def _type_to_schema(tp: Any) -> JsonSchema:
             f_schema = _type_to_schema(f_type)
             if f.default is not dataclasses.MISSING:
                 f_schema.setdefault("default", f.default)
-            elif f.default_factory is not dataclasses.MISSING:  # type: ignore[attr-defined]
+            elif f.default_factory is not dataclasses.MISSING:
                 # we can't serialize the factory, just mark as optional
                 pass
             else:
@@ -164,7 +164,7 @@ def _type_to_schema(tp: Any) -> JsonSchema:
     import enum
 
     if isinstance(tp, type) and issubclass(tp, enum.Enum):
-        values = [m.value for m in tp]  # type: ignore[arg-type]
+        values = [m.value for m in tp]
         # derive base type from first value
         base = _type_to_schema(type(values[0])) if values else {}
         base["enum"] = values
@@ -242,11 +242,11 @@ def from_function(
 
         # Unwrap callable classes: use __call__
         if not inspect.isroutine(fn) and hasattr(fn, "__call__"):
-            fn = fn.__call__  # type: ignore[assignment]
+            fn = fn.__call__
 
         # Unwrap staticmethod
         if isinstance(fn, staticmethod):
-            fn = fn.__func__  # type: ignore[assignment]
+            fn = fn.__func__
 
         sig = inspect.signature(fn)
         type_hints = get_type_hints(fn, include_extras=True)
@@ -301,7 +301,7 @@ def from_function(
         return_schema: Optional[JsonSchema] = None
         return_ann = type_hints.get("return", sig.return_annotation)
 
-        if return_ann not in (inspect._empty, None, Any, ...):  # type: ignore[attr-defined]
+        if return_ann not in (inspect._empty, None, Any, ...):
             # Convert return type to schema
             return_schema = _type_to_schema(return_ann)
 

grafi/common/decorators/record_base.py

@@ -9,6 +9,7 @@
 from typing import Callable
 from typing import Dict
 from typing import List
+from typing import Optional
 from typing import Type
 from typing import TypeVar
 from typing import Union
@@ -20,7 +21,6 @@
 from pydantic import ConfigDict
 from pydantic_core import to_jsonable_python
 
-from grafi.assistants.assistant_base import AssistantBase
 from grafi.common.containers.container import container
 from grafi.common.env import env_bool
 from grafi.common.events.component_base import ComponentEvent
@@ -34,9 +34,6 @@
 from grafi.common.models.default_id import default_id
 from grafi.common.models.invoke_context import InvokeContext
 from grafi.common.models.message import Message
-from grafi.nodes.node_base import NodeBase
-from grafi.tools.tool import Tool
-from grafi.workflows.workflow import Workflow
 
 T = TypeVar("T")
 
@@ -63,9 +60,10 @@ class ComponentConfig:
     event_types: Dict[
         str, Type[ComponentEvent]
     ]  # Maps 'invoke', 'respond', 'failed' to event classes
-    extract_metadata: Callable[
-        [Union[AssistantBase, Workflow, NodeBase, Tool]], EventContext
-    ]  # Extracts component-specific metadata
+    # Extracts component-specific metadata. Typed as ``Any`` so this common-layer
+    # module does not import the higher-layer component classes (Assistant /
+    # Workflow / Node / Tool) just for an annotation.
+    extract_metadata: Callable[[Any], EventContext]
     process_async_result: Callable[[List], Any]
     span_name_suffix: str = "invoke"  # Suffix for span name
 
@@ -237,27 +235,33 @@ def create_async_decorator(config: ComponentConfig) -> Callable:
     def decorator(func: Callable) -> Callable:
         @functools.wraps(func)
         async def wrapper(
-            self: Union[AssistantBase, Workflow, NodeBase, Tool],
-            *args,
-            **kwargs,
+            self: Any,
+            *args: Any,
+            **kwargs: Any,
         ) -> AsyncGenerator[Union[PublishToTopicEvent, List[Message]], None]:
             # Extract metadata using component-specific logic
             metadata = config.extract_metadata(self)
 
-            input_data: Union[
-                List[ConsumeFromTopicEvent], List[Message], PublishToTopicEvent
+            input_data: Optional[
+                Union[List[ConsumeFromTopicEvent], List[Message], PublishToTopicEvent]
             ] = None
 
             if isinstance(args[0], InvokeContext):
                 invoke_context: InvokeContext = args[0]
                 input_data = args[1]
             else:
-                # Assistant and workflow
+                # Assistant and workflow: args[0] is the input event (Any), which
+                # carries invoke_context. Read it off args[0] directly so mypy
+                # does not widen to the input_data union (which has no such attr).
                 input_data = args[0]
-                invoke_context = input_data.invoke_context
-
-            # Create invoke event
-            invoke_event = config.event_types["invoke"](
+                invoke_context = args[0].invoke_context
+
+            # Create invoke event. The factory maps each string key to the
+            # matching InvokeEvent/RespondEvent/FailedEvent subclass, so these
+            # per-event kwargs (input_data/output_data/error/...) are correct at
+            # runtime but unprovable to mypy through the Dict value's
+            # ComponentEvent base type -- hence the targeted call-arg ignores.
+            invoke_event = config.event_types["invoke"](  # type: ignore[call-arg]
                 id=metadata.id,
                 name=metadata.name,
                 type=metadata.type,
@@ -268,52 +272,60 @@ async def wrapper(
 
             # Execute with tracing
             output_data = None
+            error_details: Optional[Dict[str, Any]] = None
 
             try:
                 with container.tracer.start_as_current_span(
                     f"{metadata.name}.{config.span_name_suffix}"
                 ) as span:
-                    # Set span attributes
-                    for key, value in metadata.model_dump().items():
-                        if value is not None:
-                            span.set_attribute(key, value)
-
-                    span.set_attributes(invoke_context.model_dump())
-
-                    # Set input (size-bounded; omitted if payloads are disabled)
-                    input_payload = _span_payload(input_data)
-                    if input_payload is not None:
-                        span.set_attribute("input", input_payload)
-
-                    # Handle streaming
-                    result_list: List = []
-
-                    async for result in func(self, *args, **kwargs):
-                        yield result
-                        result_list.append(result)
-
-                    output_data = config.process_async_result(result_list)
-
-                    output_payload = _span_payload(output_data)
-                    if output_payload is not None:
-                        span.set_attribute("output", output_payload)
+                    try:
+                        # Set span attributes
+                        for key, value in metadata.model_dump().items():
+                            if value is not None:
+                                span.set_attribute(key, value)
+
+                        span.set_attributes(invoke_context.model_dump())
+
+                        # Set input (size-bounded; omitted if payloads are disabled)
+                        input_payload = _span_payload(input_data)
+                        if input_payload is not None:
+                            span.set_attribute("input", input_payload)
+
+                        # Handle streaming
+                        result_list: List = []
+
+                        async for result in func(self, *args, **kwargs):
+                            yield result
+                            result_list.append(result)
+
+                        output_data = config.process_async_result(result_list)
+
+                        output_payload = _span_payload(output_data)
+                        if output_payload is not None:
+                            span.set_attribute("output", output_payload)
+                    except Exception as e:
+                        # Build structured details once, while the traceback is
+                        # still attached to the exception, and enrich the span
+                        # WHILE it is still recording. Attributes set after the
+                        # span context manager exits are dropped by the backend.
+                        error_details = _build_error_details(e, metadata)
+                        _record_span_error(span, e, error_details)
+                        raise
 
             except Exception as e:
-                # Build structured details once, while the traceback is still
-                # attached to the exception.
-                error_details = _build_error_details(e, metadata)
-
-                # Enrich the active span with structured error information.
-                if "span" in locals():
-                    _record_span_error(span, e, error_details)
+                # error_details is normally built above, inside the live span.
+                # Rebuild defensively only if the failure happened before the
+                # inner try (e.g. span creation itself).
+                if error_details is None:
+                    error_details = _build_error_details(e, metadata)
 
                 # Log a full traceback once at the layer closest to the failure;
                 # outer layers log a concise summary (see _log_component_exception).
                 _log_component_exception(e, metadata, invoke_context, error_details)
 
                 # Record failed event with both the human-readable string (kept
                 # for backward compatibility) and the structured details.
-                failed_event = config.event_types["failed"](
+                failed_event = config.event_types["failed"](  # type: ignore[call-arg]
                     id=metadata.id,
                     name=metadata.name,
                     type=metadata.type,
@@ -326,7 +338,7 @@ async def wrapper(
                 raise
             else:
                 # Record respond event
-                respond_event = config.event_types["respond"](
+                respond_event = config.event_types["respond"](  # type: ignore[call-arg]
                     id=metadata.id,
                     name=metadata.name,
                     type=metadata.type,