Status: Planning — not yet implemented
Motivation: Eliminate implicit payload contracts scattered across participants. Messages become self-describing objects; participants no longer know string key names.
The bus provides transport-level loose coupling, but payload structure is an implicit contract encoded in string literals at every publish and consume site:
# Publisher (generator_agent.py)
{"chunk": chunk.message.thinking, "session_id": session_id, "query_id": query_id}
# Consumer (ws_bridge.py — has to know the same keys)
payload.get("chunk", "")
payload.get("session_id")
Rename a key anywhere → silent breakage. Add a field → no way to know all consumers. The message_type string ("thinking", "response", "critique") is also redundant with the subject — set manually at every publish site.
# src/local/protocol/messages.py
from __future__ import annotations
from dataclasses import dataclass, field, asdict
from typing import ClassVar
from local.protocol.subjects import GENERATION_THINKING
@dataclass
class GenerationThinking:
subject: ClassVar[str] = GENERATION_THINKING
message_type: ClassVar[str] = "thinking" # derived from class — no longer passed manually
chunk: str
session_id: str
query_id: str
# --- bus boundary ---
def to_envelope(self, sender_id: str, correlation_id: str) -> MessageEnvelope:
return MessageEnvelope.create(
message_type=self.message_type,
subject=self.subject,
sender_id=sender_id,
payload=asdict(self),
correlation_id=correlation_id,
metadata={"session_id": self.session_id},
)
@classmethod
def from_envelope(cls, envelope: MessageEnvelope) -> "GenerationThinking":
p = envelope.payload
return cls(
chunk=p.get("chunk", ""),
session_id=p.get("session_id", ""),
query_id=p.get("query_id", ""),
)
ZmqPublisher.publish() gains an overload that accepts a message object directly:
def publish(self, item: MessageEnvelope | BusMessage, *, sender_id: str = "", correlation_id: str = "") -> None:
if isinstance(item, MessageEnvelope):
envelope = item
else:
envelope = item.to_envelope(sender_id=sender_id, correlation_id=correlation_id)
# ... existing send logic
Call sites become:
# Before
self._pub.publish(self._make_envelope(
GENERATION_THINKING, "thinking",
{"chunk": chunk.message.thinking, "session_id": session_id, "query_id": query_id},
correlation_id, session_id,
))
# After
self._pub.publish(
GenerationThinking(chunk=chunk.message.thinking, session_id=session_id, query_id=query_id),
sender_id=self.id, correlation_id=correlation_id,
)
# Before
chunk = payload.get("chunk", "")
session_id = payload.get("session_id", "")
# After
msg = GenerationThinking.from_envelope(envelope)
chunk = msg.chunk
session_id = msg.session_id
# Abstract base — all messages
class BusMessage:
subject: ClassVar[str]
message_type: ClassVar[str]
def to_envelope(self, sender_id: str, correlation_id: str) -> MessageEnvelope: ...
@classmethod
def from_envelope(cls, envelope: MessageEnvelope) -> "BusMessage": ...
# Tool call/result pairs share correlation_id + tool name
@dataclass
class ToolCall(BusMessage):
tool: str # tool name — determines subject dynamically
args: dict
correlation_id: str
@property
def subject(self) -> str: # override: dynamic per tool name
return f"tool.call.{self.tool}"
@dataclass
class ToolResult(BusMessage):
tool: str
result: str
correlation_id: str
@property
def subject(self) -> str:
return f"tool.result.{self.tool}"
For tools with fixed names, subclasses can lock the tool name as a class constant. ToolCall and ToolResult use a dynamic subject property because the subject includes the tool name.
message_type eliminationWith message_type as a ClassVar on every message class, the separate MESSAGE_TYPE lookup dict discussed earlier is no longer needed. The class itself is the source of truth.
| Class | Subject constant | message_type | Key fields | Publisher → Consumers |
|---|---|---|---|---|
QueryReceived | QUERY_RECEIVED | query | query, session_id, query_id, attachments | UI/gateway → generator |
GenerationThinking | GENERATION_THINKING | thinking | chunk, session_id, query_id | generator → ws_bridge |
ResponseGeneration | RESPONSE_GENERATION | response | query, answer, thinking, tool_calls, session_id, query_id, prompt_tokens, error | generator → critic, memory, ws_bridge |
AnswerDialog | ANSWER_DIALOG | dialog | query, answer, session_id, query_id | generator → memory |
ToolSchema | TOOL_SCHEMA | tool_schema | schema (dict) | tools → generator |
ToolSchemaRequest | TOOL_SCHEMA_REQUEST | tool_schema_request | sender_id | generator → tools |
ToolCall | tool.call.{name} | tool_call | tool, args, correlation_id | tool_dispatcher → tools |
ToolResult | tool.result.{name} | tool_result | tool, result, correlation_id | tools → tool_dispatcher / ws_bridge |
ToolActivity | tool.activity.{name} | tool_activity | tool, event, args, result, elapsed_ms | tools → tool_window |
CritiqueResult | CRITIQUE | critique | score, feedback, query_id, session_id, query, answer | critic → memory, UI |
UserFeedback | USER_FEEDBACK | user_feedback | query_id, session_id, value (+1/-1) | UI → reward |
RewardEvent | REWARD_EVENT | reward | query_id, session_id, score, sender_id | reward → agents |
AgentTransition | AGENT_TRANSITION | agent_transition | agent_id, from_state, to_state, action | agents → UI |
CompactionRequest | COMPACTION_REQUEST | compaction_request | session_id | UI/auto → generator |
CompactionResult | COMPACTION_RESULT | compaction_result | session_id, tokens_before, tokens_after, error | compaction_service → generator/UI |
GeneratorStatus | GENERATOR_STATUS | generator_status | instance_id, model, state, token_count, tool_names, system_prompt | generator → UI |
ConfigReload | CONFIG_RELOAD | config_reload | target (tool name or *) | UI → tools |
| File | Change |
|---|---|
src/local/protocol/messages.py | NEW All message dataclasses + BusMessage base |
src/local/transport/zmq_publisher.py | MOD publish() accepts BusMessage | MessageEnvelope |
src/local/agents/generator_agent.py | MOD All publish sites → message objects; _make_envelope removed; from_envelope on receive |
src/local/agents/critic_agent.py | MOD from_envelope(ResponseGeneration) on consume; publish CritiqueResult object |
src/local/agents/memory_agent.py | MOD from_envelope on ResponseGeneration + CritiqueResult consume |
src/local/agents/base_agent.py | MOD publish AgentTransition object |
src/local/tools/base_tool.py | MOD publish ToolResult + ToolSchema + ToolActivity objects |
src/local/tools/tool_dispatcher.py | MOD publish ToolCall; from_envelope(ToolResult) on receive |
src/local/api/ws_bridge.py | MOD from_envelope on GenerationThinking, ResponseGeneration, ToolRequest, ToolResult |
src/local/api/gateway.py | MOD publish QueryReceived, CompactionRequest, UserFeedback objects |
src/local/services/reward_service.py | MOD from_envelope(UserFeedback); publish RewardEvent object |
src/local/services/compaction_service.py | MOD publish CompactionResult; _make_envelope dependency removed |
src/local/ui/main_window.py | MOD publish QueryReceived, CompactionRequest, UserFeedback objects |
src/local/ui/monitor_app.py | MOD from_envelope on CritiqueResult, ToolSchema |
src/local/ui/tool_window.py / tool_panel.py | MOD from_envelope on ToolActivity |
src/local/session/local_session.py | MOD publish QueryReceived object |
Two options:
from_envelope uses .get() with defaults — same as today, but centralized. Never raises. Old envelopes from before the refactor still parse.ValueError on missing required fields. Catches wiring bugs earlier but breaks during incremental migration.Recommended: lenient during migration, strict later once all sites are migrated.
Since the subject includes the tool name (tool.call.web_search), it can't be a fixed ClassVar. Options:
@property on the instance (shown above) — ClassVar[str] convention breaksWebSearchCall(ToolCall)) with fixed ClassVar — more classes but cleaner typingToolCall class with a tool field and @property subject — simplest, matches how tool_dispatcher already worksRecommended: single ToolCall / ToolResult with @property subject.
Once all generator publish sites use message objects, _make_envelope can be removed. CompactionService currently receives it as a bound method parameter — that dependency is eliminated when CompactionService publishes a CompactionResult object directly via its own publisher.
The wire format (JSON payload) does not change — to_envelope calls asdict(self) which produces the same dict as today. Migration can be incremental, one participant at a time. If the wire format needs to change for any message, don't preserve backward compatibility — just update all sites together.
Because the wire format is unchanged, the refactor can be done incrementally:
src/local/protocol/messages.py with all dataclasses — no behavior changes yetZmqPublisher.publish() to accept BusMessage objectsfrom_envelope()_make_envelope from GeneratorAgent once all its publish sites are migratedsrc/local/protocol/messages.py NEWsrc/local/transport/zmq_publisher.py MOD