Grafi 48/simplify events (#71)

* consistant and simplify

* simplify event and record

* lint and unit tests

* update unit tests

* update code

* lint

* update tracer

* fix lint

* update version, minor changes

Author: GuanyiLi-Craig

docs/docs/guide/configuring-event-store.md

@@ -247,7 +247,7 @@ print(f"Events: {events} ")
 2025-07-12 17:38:55.895 | DEBUG    | grafi.common.instrumentations.tracing:is_local_endpoint_available:30 - Endpoint check failed: [Errno 111] Connection refused
 2025-07-12 17:38:55.895 | DEBUG    | grafi.common.instrumentations.tracing:setup_tracing:117 - OTLP endpoint is not available. Using InMemorySpanExporter.
 2025-07-12 17:38:55.905 | INFO     | grafi.common.topics.topic:publish_data:64 - [agent_input_topic] Message published with event_id: 1da4f45008264dc98fd63dc154dcaa6e
-2025-07-12 17:38:55.912 | DEBUG    | grafi.nodes.node:invoke:49 - Executing Node with inputs: [ConsumeFromTopicEvent(event_id='8dd13f159fbc40b29ca030501f4d2859', event_version='1.0', invoke_context=InvokeContext(conversation_id='9051491ca7a84b71a3f3e9d790b79e4f', invoke_id='6e56da25ca824bf4a8d454a731b6f336', assistant_request_id='0e0317d25c9b40e4a76a130bbbb7bc43', user_id='9e396be8f8c442bc9269a31a9b3f69a2'), event_type=<EventType.CONSUME_FROM_TOPIC: 'ConsumeFromTopic'>, timestamp=datetime.datetime(2025, 7, 12, 16, 38, 55, 909090, tzinfo=datetime.timezone.utc), topic_name='agent_input_topic', offset=0, data=[Message(name=None, message_id='ae2638fa5f3047fbbfb43dbd1545e6d2', timestamp=1752338335885634346, content='What is the capital of the United Kingdom', refusal=None, annotations=None, audio=None, role='user', tool_call_id=None, tools=None, function_call=None, tool_calls=None, is_streaming=False)], consumer_name='OpenAIInputNode', consumer_type='OpenAIInputNode')]
+2025-07-12 17:38:55.912 | DEBUG    | grafi.nodes.node:invoke:49 - Executing Node with inputs: [ConsumeFromTopicEvent(event_id='8dd13f159fbc40b29ca030501f4d2859', event_version='1.0', invoke_context=InvokeContext(conversation_id='9051491ca7a84b71a3f3e9d790b79e4f', invoke_id='6e56da25ca824bf4a8d454a731b6f336', assistant_request_id='0e0317d25c9b40e4a76a130bbbb7bc43', user_id='9e396be8f8c442bc9269a31a9b3f69a2'), event_type=<EventType.CONSUME_FROM_TOPIC: 'ConsumeFromTopic'>, timestamp=datetime.datetime(2025, 7, 12, 16, 38, 55, 909090, tzinfo=datetime.timezone.utc), name='agent_input_topic', offset=0, data=[Message(name=None, message_id='ae2638fa5f3047fbbfb43dbd1545e6d2', timestamp=1752338335885634346, content='What is the capital of the United Kingdom', refusal=None, annotations=None, audio=None, role='user', tool_call_id=None, tools=None, function_call=None, tool_calls=None, is_streaming=False)], consumer_name='OpenAIInputNode', consumer_type='OpenAIInputNode')]
 2025-07-12 17:38:57.400 | INFO     | grafi.common.topics.topic:publish_data:69 - [function_call_topic] Message NOT published (condition not met)
 2025-07-12 17:38:57.400 | INFO     | grafi.common.topics.output_topic:publish_data:91 - [agent_output_topic] Message published with event_id: d83d3339fbd94d6e983fc51965177891
 <span style="color:#FF4689">Output from React Agent:</span>  The capital of the United Kingdom is London.

docs/docs/user-guide/assistant.md

@@ -26,11 +26,11 @@ class MyAssistant(AssistantBase):
         # Implementation required
         pass
 
-    def invoke(self, input_event: PublishToTopicEvent) -> List[ConsumeFromTopicEvent]:
+    def invoke(self, input_data: PublishToTopicEvent) -> List[ConsumeFromTopicEvent]:
         # Implementation required
         pass
 
-    async def a_invoke(self, input_event: PublishToTopicEvent) -> AsyncGenerator[ConsumeFromTopicEvent, None]:
+    async def a_invoke(self, input_data: PublishToTopicEvent) -> AsyncGenerator[ConsumeFromTopicEvent, None]:
         # Implementation required
         pass
 ```
@@ -87,14 +87,14 @@ Synchronously processes input events through the configured workflow.
 
 ```python
 @record_assistant_invoke
-def invoke(self, input_event: PublishToTopicEvent) -> List[ConsumeFromTopicEvent]:
-    events = self.workflow.invoke(input_event)
+def invoke(self, input_data: PublishToTopicEvent) -> List[ConsumeFromTopicEvent]:
+    events = self.workflow.invoke(input_data)
     return events
 ```
 
 **Parameters**:
 
-- `input_event`: A `PublishToTopicEvent` containing the data to be processed and context information
+- `input_data`: A `PublishToTopicEvent` containing the data to be processed and context information
 
 **Returns**: List of `ConsumeFromTopicEvent` objects representing the workflow output
 
@@ -106,14 +106,14 @@ Asynchronously processes input events with support for streaming responses.
 
 ```python
 @record_assistant_a_invoke
-async def a_invoke(self, input_event: PublishToTopicEvent) -> AsyncGenerator[ConsumeFromTopicEvent, None]:
-    async for output in self.workflow.a_invoke(input_event):
+async def a_invoke(self, input_data: PublishToTopicEvent) -> AsyncGenerator[ConsumeFromTopicEvent, None]:
+    async for output in self.workflow.a_invoke(input_data):
         yield output
 ```
 
 **Parameters**:
 
-- `input_event`: A `PublishToTopicEvent` containing the data to be processed and context information
+- `input_data`: A `PublishToTopicEvent` containing the data to be processed and context information
 
 **Returns**: Async generator yielding `ConsumeFromTopicEvent` objects
 

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

@@ -234,8 +234,8 @@ async def _invoke_node(self, invoke_context: InvokeContext, node: Node):
                     published_events = await a_publish_events(...)
                     # Notify downstream nodes immediately
                     for event in published_events:
-                        if event.topic_name in self._topic_nodes:
-                            topic = self._topics[event.topic_name]
+                        if event.name in self._topic_nodes:
+                            topic = self._topics[event.name]
                             async with topic.event_cache._cond:
                                 topic.event_cache._cond.notify_all()
 

docs/docs/user-guide/event-interfaces.md

@@ -59,7 +59,8 @@ Created when data is consumed from a topic:
 from grafi.common.events.topic_events.consume_from_topic_event import ConsumeFromTopicEvent
 
 event = ConsumeFromTopicEvent(
-    topic_name="output_topic",
+    name="output_topic",
+    type=TopicType.DEFAULT_TOPIC_TYPE,
     offset=42,
     publisher_name="ProcessorNode",  # Original publisher
     publisher_type="node",
@@ -80,18 +81,18 @@ from grafi.common.events.topic_events.consume_from_topic_event import ConsumeFro
 from typing import List, AsyncGenerator
 
 class MyAssistant(Assistant):
-    def invoke(self, input_event: PublishToTopicEvent) -> List[ConsumeFromTopicEvent]:
+    def invoke(self, input_data: PublishToTopicEvent) -> List[ConsumeFromTopicEvent]:
         """Synchronous processing of events."""
         # Delegate to workflow
-        events = self.workflow.invoke(input_event)
+        events = self.workflow.invoke(input_data)
         return events
 
     async def a_invoke(
         self,
-        input_event: PublishToTopicEvent
+        input_data: PublishToTopicEvent
     ) -> AsyncGenerator[ConsumeFromTopicEvent, None]:
         """Asynchronous streaming of events."""
-        async for output in self.workflow.a_invoke(input_event):
+        async for output in self.workflow.a_invoke(input_data):
             yield output
 ```
 
@@ -132,10 +133,10 @@ from grafi.common.events.topic_events.publish_to_topic_event import PublishToTop
 from grafi.common.events.topic_events.consume_from_topic_event import ConsumeFromTopicEvent
 
 class MyWorkflow(Workflow):
-    def invoke(self, input_event: PublishToTopicEvent) -> List[ConsumeFromTopicEvent]:
+    def invoke(self, input_data: PublishToTopicEvent) -> List[ConsumeFromTopicEvent]:
         """Execute workflow synchronously."""
         # Initialize workflow with input event
-        self.initial_workflow(input_event)
+        self.initial_workflow(input_data)
 
         # Process nodes until completion
         while not self._invoke_queue.empty():
@@ -185,8 +186,8 @@ def invoke(self, invoke_context: InvokeContext, input_data: Messages) -> Message
 
 **New Pattern:**
 ```python
-def invoke(self, input_event: PublishToTopicEvent) -> List[ConsumeFromTopicEvent]:
-    return self.workflow.invoke(input_event)
+def invoke(self, input_data: PublishToTopicEvent) -> List[ConsumeFromTopicEvent]:
+    return self.workflow.invoke(input_data)
 ```
 
 Key changes:

docs/docs/user-guide/events/event_graph.md

@@ -77,7 +77,7 @@ graph TD
 ```python
 # Create mapping for efficient lookup
 topic_offset_to_publish = {
-    f"{event.topic_name}::{event.offset}": event
+    f"{event.name}::{event.offset}": event
     for event in topic_events.values()
     if isinstance(event, PublishToTopicEvent)
 }
@@ -94,7 +94,7 @@ def build_node_relations(consume_event: ConsumeFromTopicEvent) -> None:
     current_node = self._add_event(consume_event)
 
     # Find corresponding publish event
-    publish_key = f"{consume_event.topic_name}::{consume_event.offset}"
+    publish_key = f"{consume_event.name}::{consume_event.offset}"
     publish_event = topic_offset_to_publish.get(publish_key)
 
     if publish_event:
@@ -278,7 +278,7 @@ def find_root_causes(graph: EventGraph):
         {
             "event_id": node.event_id,
             "timestamp": node.event.timestamp,
-            "topic_name": node.event.topic_name,
+            "name": node.event.name,
             "downstream_count": len(node.downstream_events)
         }
         for node in root_nodes
@@ -401,7 +401,7 @@ def save_graph_to_event_store(graph: EventGraph, event_store: EventStore):
     from grafi.common.events.topic_events.topic_event import TopicEvent
 
     graph_event = TopicEvent(
-        topic_name="event_graph_analysis",
+        name="event_graph_analysis",
         offset=0,
         data=graph.to_dict()
     )

docs/docs/user-guide/events/events.md

@@ -273,8 +273,8 @@ Basic event for topic-related activities.
 
 | Field        | Type           | Description                                        |
 |--------------|----------------|----------------------------------------------------|
-| `topic_name` | `str`          | Name of the topic.                                |
-| `topic_type` | `TopicType`    | Type of the topic.                                |
+| `name`       | `str`          | Name of the topic.                                |
+| `type`       | `TopicType`    | Type of the topic.                                |
 | `offset`     | `int`          | Position/offset in the topic stream.              |
 | `data`       | `List[Message]`| Data associated with the topic event.             |
 
@@ -370,7 +370,8 @@ Each event type includes structured context information:
     "consumed_event_ids": ["event_1", "event_2"],
     "publisher_name": "Data Processor",
     "publisher_type": "ProcessingNode",
-    "topic_name": "output_topic",
+    "name": "output_topic",
+    "type": "topic",
     "offset": 42,
     "invoke_context": {...}
 }

docs/docs/user-guide/invoke-decorators.md

@@ -90,9 +90,9 @@ from grafi.common.events.topic_events.publish_to_topic_event import PublishToTop
 from grafi.common.events.topic_events.consume_from_topic_event import ConsumeFromTopicEvent
 
 @record_assistant_invoke
-def invoke(self, input_event: PublishToTopicEvent) -> List[ConsumeFromTopicEvent]:
+def invoke(self, input_data: PublishToTopicEvent) -> List[ConsumeFromTopicEvent]:
     # Assistant implementation
-    return self.workflow.invoke(input_event)
+    return self.workflow.invoke(input_data)
 ```
 
 #### @record_assistant_a_invoke
@@ -117,9 +117,9 @@ from grafi.common.events.topic_events.consume_from_topic_event import ConsumeFro
 from typing import AsyncGenerator
 
 @record_assistant_a_invoke
-async def a_invoke(self, input_event: PublishToTopicEvent) -> AsyncGenerator[ConsumeFromTopicEvent, None]:
+async def a_invoke(self, input_data: PublishToTopicEvent) -> AsyncGenerator[ConsumeFromTopicEvent, None]:
     # Async assistant implementation
-    async for output in self.workflow.a_invoke(input_event):
+    async for output in self.workflow.a_invoke(input_data):
         yield output
 ```
 
@@ -201,7 +201,7 @@ async def a_invoke(self, invoke_context: InvokeContext,
 
 #### @record_tool_invoke
 
-**Location**: `grafi.common.decorators.record_tool_invoke`
+**Location**: `grafi.common.decorators.record_decorators`
 
 Records synchronous tool invocations with event logging and tracing.
 
@@ -215,7 +215,7 @@ Records synchronous tool invocations with event logging and tracing.
 **Usage**:
 
 ```python
-from grafi.common.decorators.record_tool_invoke import record_tool_invoke
+from grafi.common.decorators.record_decorators import record_tool_invoke
 
 @record_tool_invoke
 def invoke(self, invoke_context: InvokeContext, input_data: Messages) -> Messages:
@@ -225,7 +225,7 @@ def invoke(self, invoke_context: InvokeContext, input_data: Messages) -> Message
 
 #### @record_tool_a_invoke
 
-**Location**: `grafi.common.decorators.record_tool_a_invoke`
+**Location**: `grafi.common.decorators.record_decorators`
 
 Records asynchronous tool invocations that return async generators.
 
@@ -239,7 +239,7 @@ Records asynchronous tool invocations that return async generators.
 **Usage**:
 
 ```python
-from grafi.common.decorators.record_tool_a_invoke import record_tool_a_invoke
+from grafi.common.decorators.record_decorators import record_tool_a_invoke
 
 @record_tool_a_invoke
 async def a_invoke(self, invoke_context: InvokeContext, input_data: Messages) -> MsgsAGen:
@@ -271,7 +271,7 @@ from grafi.common.events.topic_events.consume_from_topic_event import ConsumeFro
 from typing import List
 
 @record_workflow_invoke
-def invoke(self, input_event: PublishToTopicEvent) -> List[ConsumeFromTopicEvent]:
+def invoke(self, input_data: PublishToTopicEvent) -> List[ConsumeFromTopicEvent]:
     # Workflow orchestration logic
     # Initialize workflow, execute nodes, return consumed events
     return output_events
@@ -299,10 +299,10 @@ from grafi.common.events.topic_events.consume_from_topic_event import ConsumeFro
 from typing import AsyncGenerator
 
 @record_workflow_a_invoke
-async def a_invoke(self, input_event: PublishToTopicEvent) -> AsyncGenerator[ConsumeFromTopicEvent, None]:
+async def a_invoke(self, input_data: PublishToTopicEvent) -> AsyncGenerator[ConsumeFromTopicEvent, None]:
     # Async workflow orchestration
     # Initialize workflow, execute nodes asynchronously, yield consumed events
-    async for output_event in self._execute_workflow(input_event):
+    async for output_event in self._execute_workflow(input_data):
         yield output_event
 ```
 

docs/docs/user-guide/models.md

@@ -104,7 +104,7 @@ from grafi.tools.tool import Tool
 command = Command.for_tool(my_tool)
 
 # Invoke the command
-response = command.invoke(invoke_context, input_events)
+response = command.invoke(invoke_context, input_datas)
 ```
 
 ## FunctionSpec

docs/docs/user-guide/node.md

@@ -100,11 +100,11 @@ from grafi.common.events.topic_events.publish_to_topic_event import PublishToTop
 
 # Synchronous invocation
 context = InvokeContext(session_id="session-123")
-input_events = [ConsumeFromTopicEvent(...)]  # List of consumed events
-output_event: PublishToTopicEvent = node.invoke(context, input_events)
+input_datas = [ConsumeFromTopicEvent(...)]  # List of consumed events
+output_event: PublishToTopicEvent = node.invoke(context, input_datas)
 
 # Asynchronous invocation
-async for output_event in node.a_invoke(context, input_events):
+async for output_event in node.a_invoke(context, input_datas):
     # Process each PublishToTopicEvent as it's generated
     pass
 ```

docs/docs/user-guide/topics/input_topics.md

@@ -174,7 +174,7 @@ output_event = workflow_output_topic.publish_data(
 
 # When user responds, publish to paired input topic
 user_response = [Message(role="user", content="Approved")]
-input_event = workflow_input_topic.publish_input_data(
+input_data = workflow_input_topic.publish_input_data(
     upstream_event=output_event,
     data=user_response
 )
@@ -213,12 +213,12 @@ async def human_approval_workflow():
 
     # Process user response
     user_feedback = [Message(role="user", content="Looks good, approved!")]
-    input_event = await input_topic.a_publish_input_data(
+    input_data = await input_topic.a_publish_input_data(
         upstream_event=output_event,
         data=user_feedback
     )
 
-    return input_event
+    return input_data
 ```
 
 ## Best Practices
@@ -325,14 +325,14 @@ async def test_input_topics():
 
     # Simulate user response
     user_response = [Message(role="user", content="Here's my input")]
-    input_event = await workflow_input_topic.a_publish_input_data(
+    input_data = await workflow_input_topic.a_publish_input_data(
         upstream_event=output_event,
         data=user_response
     )
 
-    assert input_event is not None
-    assert input_event.data == user_response
-    assert input_event.consumed_events == [output_event]
+    assert input_data is not None
+    assert input_data.data == user_response
+    assert input_data.consumed_events == [output_event]
 ```
 
 ## Topic Type Constants