Skip to content

DAARION Messaging Architecture

Complete Specification: Messenger + Matrix + Agents + DAGI Router

Version: 1.0.0
Date: 2025-11-24
Status: Production Ready

Table of Contents

  1. Overview
  2. System Components
  3. Data Model (ERD)
  4. matrix-gateway API Specification
  5. Message Flow: Human → Agent Reply
  6. Agent-Initiated Messages
  7. agent_filter Rules
  8. DAGI Router Integration
  9. Sequence Diagrams
  10. Implementation Guide

Overview

DAARION Messaging Architecture побудована на Matrix protocol з повною інтеграцією агентів через DAGI Router.

Key Principles

  1. Matrix as Source of Truth — повідомлення зберігаються в Matrix, DAARION тримає індекс
  2. Agent-Aware — агенти як повноцінні учасники каналів
  3. Event-Driven — всі дії через NATS JetStream
  4. Security First — agent_filter контролює доступ агентів
  5. Element Compatible — повна сумісність з Element та іншими Matrix клієнтами

Architecture Layers

┌─────────────────────────────────────────────────────────┐
│                   Frontend (React)                      │
│              MessengerPage + WebSocket                  │
└────────────────────┬────────────────────────────────────┘
                     │
                     ▼
┌─────────────────────────────────────────────────────────┐
│              messaging-service (FastAPI)                │
│         REST API + WebSocket + Channel Management       │
└────────────────────┬────────────────────────────────────┘
                     │
       ┌─────────────┴─────────────┐
       │                           │
       ▼                           ▼
┌──────────────┐         ┌──────────────────┐
│ matrix-gateway│         │   NATS JetStream │
│  (Internal)  │         │   Event Bus      │
└──────┬───────┘         └────────┬─────────┘
       │                          │
       ▼                          ▼
┌──────────────┐         ┌──────────────────┐
│    Matrix    │         │  agent_filter    │
│  Homeserver  │         │  DAGI Router     │
│  (Synapse)   │         │  Agent Runtime   │
└──────────────┘         └──────────────────┘

System Components

1. messaging-service

  • Role: DAARION-specific messaging API
  • Port: 7004
  • Responsibilities:
  • Channel CRUD (mapped to Matrix rooms)
  • Message indexing (full content in Matrix)
  • Member management
  • WebSocket real-time updates
  • Agent posting endpoint

2. matrix-gateway

  • Role: Internal Matrix API adapter
  • Port: 7003 (internal only)
  • Responsibilities:
  • Create/manage Matrix rooms
  • Send messages on behalf of users/agents
  • Receive Matrix events via webhook
  • Normalize Matrix ↔ DAARION entities
  • Publish to NATS

3. agent_filter

  • Role: Security and routing layer
  • Responsibilities:
  • Validate agent access to channels
  • Apply content policies
  • Decide which agents can reply
  • Route to DAGI Router

4. DAGI Router

  • Role: Agent orchestration
  • Responsibilities:
  • Select model for agent
  • Choose pipeline (Memory, Tools, etc.)
  • Invoke Agent Runtime
  • Track agent sessions

5. Agent Runtime

  • Role: Execute agent logic
  • Responsibilities:
  • Read channel context
  • Query Agent Memory (RAG)
  • Call LLM Proxy
  • Execute tools
  • Post reply to channel

Data Model (ERD)

Mermaid Diagram

erDiagram
    USERS {
      uuid id
      string external_id
      string matrix_id
      string handle
      datetime created_at
    }

    MICRODAOS {
      uuid id
      string external_id
      string name
      uuid owner_user_id
      datetime created_at
    }

    MICRODAO_MEMBERS {
      uuid id
      uuid microdao_id
      uuid user_id
      string role
      datetime created_at
    }

    AGENT_BLUEPRINTS {
      uuid id
      string code
      string description
      string model
      json capabilities
    }

    AGENTS {
      uuid id
      string external_id
      string name
      string kind
      uuid microdao_id
      uuid owner_user_id
      string matrix_id
      uuid blueprint_id
      datetime created_at
    }

    CHANNELS {
      uuid id
      string slug
      string name
      uuid microdao_id
      uuid team_id
      string matrix_room_id
      string visibility
      uuid created_by_user_id
      uuid created_by_agent_id
      datetime created_at
    }

    CHANNEL_MEMBERS {
      uuid id
      uuid channel_id
      uuid member_user_id
      uuid member_agent_id
      string role
      datetime joined_at
    }

    MESSAGES {
      uuid id
      uuid channel_id
      string matrix_event_id
      uuid sender_user_id
      uuid sender_agent_id
      string sender_type
      string content_preview
      datetime created_at
    }

    AGENT_SESSIONS {
      uuid id
      uuid agent_id
      uuid channel_id
      datetime started_at
      datetime last_activity_at
      string status
    }

    USERS ||--o{ MICRODAOS : "owns"
    USERS ||--o{ MICRODAO_MEMBERS : "member"
    MICRODAOS ||--o{ MICRODAO_MEMBERS : "has members"

    USERS ||--o{ AGENTS : "owns"
    MICRODAOS ||--o{ AGENTS : "scoped"
    AGENT_BLUEPRINTS ||--o{ AGENTS : "template"

    MICRODAOS ||--o{ CHANNELS : "has"

    CHANNELS ||--o{ CHANNEL_MEMBERS : "members"
    USERS ||--o{ CHANNEL_MEMBERS : "user member"
    AGENTS ||--o{ CHANNEL_MEMBERS : "agent member"

    CHANNELS ||--o{ MESSAGES : "contains"
    USERS ||--o{ MESSAGES : "sender"
    AGENTS ||--o{ MESSAGES : "sender"

    AGENTS ||--o{ AGENT_SESSIONS : "runs"
    CHANNELS ||--o{ AGENT_SESSIONS : "context"

Key Relationships

  • users → microdaos (1:many, via owner)
  • microdaos ↔ users (many:many, via microdao_members)
  • agents → microdaos (many:1, optional scope)
  • agents → users (many:1, optional owner)
  • channels → microdaos (many:1, required)
  • channels ↔ (users + agents) (many:many, via channel_members)
  • messages → channels (many:1)
  • messages → (users | agents) (many:1, sender)

Matrix Mapping

DAARION Matrix
channels.matrix_room_id room_id (!abc:server)
messages.matrix_event_id event_id ($event:server)
users.matrix_id user_id (@user:server)
agents.matrix_id user_id (@agent:server)

matrix-gateway API Specification

Authentication

All internal endpoints require:

X-Internal-Service-Token: <shared-secret>

1. Create Room

POST /internal/matrix/rooms

Create Matrix room for DAARION channel.

Request:

{
  "room_alias": "microdao7-general",
  "name": "Quantum Garden / General",
  "visibility": "private",
  "creator": "user:93",
  "microdao_id": "microdao:7",
  "preset": "trusted_private_chat",
  "power_users": ["user:93", "agent:sofia"]
}

Response:

{
  "room_id": "!abc123:matrix.daarion.city",
  "room_alias": "#microdao7-general:matrix.daarion.city"
}

Actions: - Call /_matrix/client/v3/createRoom - Set power levels for users/agents - Add custom state (microdao_id) - Publish NATS integration.matrix.room.created

2. Send Message

POST /internal/matrix/send

Send message to Matrix room.

Request:

{
  "room_id": "!abc123:matrix.daarion.city",
  "sender": "agent:sofia",
  "sender_matrix_id": "@sofia:matrix.daarion.city",
  "msgtype": "m.text",
  "body": "Короткий summary останніх DAO подій.",
  "relates_to": {
    "m.in_reply_to": {
      "event_id": "$event123:matrix.daarion.city"
    }
  },
  "meta": {
    "channel_id": "7c72d497-27aa-4e75-bb2f-4a4a21d4f91f",
    "microdao_id": "microdao:7",
    "agent_id": "agent:sofia"
  }
}

Response:

{
  "event_id": "$event999:matrix.daarion.city",
  "room_id": "!abc123:matrix.daarion.city"
}

Actions: - Call /_matrix/client/v3/rooms/{roomId}/send/m.room.message/{txnId} - Generate unique txnId - Return event_id - Publish NATS integration.matrix.message

3. Invite User

POST /internal/matrix/invite

Invite user/agent to room.

Request:

{
  "room_id": "!abc123:matrix.daarion.city",
  "user_matrix_id": "@alice:matrix.daarion.city"
}

Response:

{
  "ok": true
}

4. Get Room History

GET /internal/matrix/rooms/{room_id}/messages

Retrieve message history (for agents/services).

Query params: - from — pagination token (optional) - limit — max events (default 50) - dir — b (backwards) or f (forwards)

Response:

{
  "chunk": [
    {
      "event_id": "$event123:matrix.daarion.city",
      "sender": "@alice:matrix.daarion.city",
      "origin_server_ts": 1735749000000,
      "type": "m.room.message",
      "content": {
        "msgtype": "m.text",
        "body": "Привіт, DAARION!"
      }
    }
  ],
  "start": "t1-12345_67890_1234",
  "end": "t1-12345_67890_1200"
}

Actions: - Call /_matrix/client/v3/rooms/{roomId}/messages - Return paginated events

5. Webhook: Receive Matrix Events

POST /internal/matrix/event

Receive events from Matrix (via appservice/webhook).

Request (from Matrix):

{
  "room_id": "!abc123:matrix.daarion.city",
  "event_id": "$event123:matrix.daarion.city",
  "sender": "@alice:matrix.daarion.city",
  "type": "m.room.message",
  "origin_server_ts": 1735749000000,
  "content": {
    "msgtype": "m.text",
    "body": "Привіт з Matrix!"
  },
  "unsigned": {
    "age": 123
  }
}

Actions: 1. Validate source (shared secret / IP allowlist) 2. Transform to internal DTO:

{
  "room_id": "!abc123:matrix.daarion.city",
  "event_id": "$event123:matrix.daarion.city",
  "sender_matrix_id": "@alice:matrix.daarion.city",
  "type": "m.room.message",
  "timestamp": 1735749000000,
  "body": "Привіт з Matrix!",
  "msgtype": "m.text"
}
  1. Publish to NATS:
  2. Subject: integration.matrix.message
  3. Payload: DTO + raw content

6. Health Check

GET /internal/matrix/health

Response:

{
  "status": "ok",
  "homeserver": "https://matrix.daarion.city",
  "appservice_enabled": true
}

Message Flow: Human → Agent Reply

Sequence Diagram

sequenceDiagram
    participant User
    participant Frontend
    participant messaging-service
    participant matrix-gateway
    participant Matrix
    participant NATS
    participant agent_filter
    participant DAGI_Router
    participant Agent_Runtime
    participant LLM_Proxy

    User->>Frontend: Type message
    Frontend->>messaging-service: POST /api/messaging/channels/{id}/messages
    messaging-service->>matrix-gateway: POST /internal/matrix/send
    matrix-gateway->>Matrix: Send m.room.message
    Matrix-->>matrix-gateway: event_id
    matrix-gateway-->>messaging-service: event_id
    messaging-service->>messaging-service: Index in messages table
    messaging-service->>NATS: Publish messaging.message.created
    messaging-service-->>Frontend: 201 Created
    Frontend->>Frontend: Display message

    NATS->>agent_filter: messaging.message.created
    agent_filter->>agent_filter: Check rules (channel, content, agents)
    alt Allow agent reply
        agent_filter->>NATS: Publish agent.filter.decision (allow)
        NATS->>DAGI_Router: agent.filter.decision
        DAGI_Router->>DAGI_Router: Select model, pipeline
        DAGI_Router->>Agent_Runtime: Invoke agent:sofia
        Agent_Runtime->>messaging-service: GET /internal/messaging/channels/{id}/messages
        messaging-service-->>Agent_Runtime: Recent messages
        Agent_Runtime->>Agent_Runtime: Query Agent Memory (RAG)
        Agent_Runtime->>Agent_Runtime: Build prompt
        Agent_Runtime->>LLM_Proxy: Generate response
        LLM_Proxy-->>Agent_Runtime: Response text
        Agent_Runtime->>messaging-service: POST /internal/agents/agent:sofia/post-to-channel
        messaging-service->>matrix-gateway: POST /internal/matrix/send (as agent)
        matrix-gateway->>Matrix: Send m.room.message (agent)
        Matrix-->>matrix-gateway: event_id
        matrix-gateway->>NATS: Publish integration.matrix.message
        messaging-service->>messaging-service: Index agent message
        messaging-service->>NATS: Publish messaging.message.created
        messaging-service->>Frontend: WebSocket: message.created
        Frontend->>Frontend: Display agent reply
    else Deny agent reply
        agent_filter->>agent_filter: No action
    end

Step-by-Step Flow

1. User sends message

Frontend:

await sendMessage(channelId, { text: "Hello!" });

messaging-service: - Validates user permissions - Calls matrix-gateway → Matrix - Indexes message in DB - Publishes NATS messaging.message.created:

{
  "channel_id": "...",
  "matrix_event_id": "$event",
  "sender_id": "user:93",
  "sender_type": "human",
  "microdao_id": "microdao:7",
  "created_at": "2025-11-24T10:30:00Z"
}

2. agent_filter processes event

Subscribed to: messaging.message.created

Logic: - Check channel type (public/private/microdao) - Check agent access (is agent member? can_write?) - Check content (spam, policy violations) - Check context (time, frequency)

Decision: - ALLOW → route to agent - DENY → no action - MODIFY → rewrite prompt

Publish NATS:

Subject: "agent.filter.decision"
Payload: {
  "channel_id": "...",
  "message_id": "...",
  "matrix_event_id": "$event",
  "microdao_id": "microdao:7",
  "decision": "allow",
  "target_agent_id": "agent:sofia",
  "rewrite_prompt": null
}

3. DAGI Router invokes agent

Subscribed to: agent.filter.decision (only allow)

Actions: - Load agent blueprint → get model - Determine pipeline (Memory? Tools?) - Create AgentInvocation:

{
  "agent_id": "agent:sofia",
  "entrypoint": "channel_message",
  "payload": {
    "channel_id": "...",
    "message_id": "...",
    "microdao_id": "microdao:7"
  }
}
  • Send to Agent Runtime

4. Agent Runtime executes

a) Read channel context:

GET /internal/messaging/channels/{channelId}/messages?limit=50

b) Query Agent Memory: - Fetch relevant memories for agent + microdao - RAG query based on message content

c) Build prompt: - System instructions (from blueprint) - Channel history (truncated) - Relevant memories - Optional rewrite from agent_filter

d) Call LLM Proxy:

{
  "model": "gpt-4.1",
  "messages": [
    {"role": "system", "content": "..."},
    {"role": "user", "content": "..."}
  ]
}

e) Execute tools (if needed): - Create task, followup, etc.

f) Post reply:

POST /internal/agents/agent:sofia/post-to-channel
{
  "channel_id": "...",
  "text": "Ось короткий summary..."
}

5. messaging-service posts agent message

  • Find matrix_room_id by channel_id
  • Call matrix-gateway → Matrix (as agent)
  • Index message in DB (sender_type = "agent")
  • Publish NATS messaging.message.created

6. Frontend receives update

  • WebSocket /ws/messaging/{channelId} gets signal
  • Display agent message in UI

Agent-Initiated Messages

Use Cases

  • Scheduled reminders
  • Daily digests
  • Event notifications
  • Autonomous agent actions

Flow

sequenceDiagram
    participant Cron/Event
    participant Agent_Runtime
    participant messaging-service
    participant matrix-gateway
    participant Matrix
    participant NATS
    participant Frontend

    Cron/Event->>Agent_Runtime: Trigger (e.g. daily digest)
    Agent_Runtime->>Agent_Runtime: Generate message
    Agent_Runtime->>messaging-service: POST /internal/agents/{id}/post-to-channel
    messaging-service->>matrix-gateway: POST /internal/matrix/send
    matrix-gateway->>Matrix: Send m.room.message
    Matrix-->>matrix-gateway: event_id
    matrix-gateway->>NATS: Publish integration.matrix.message
    messaging-service->>messaging-service: Index message
    messaging-service->>NATS: Publish messaging.message.created
    messaging-service->>Frontend: WebSocket: message.created
    Frontend->>Frontend: Display message

Key difference: No agent_filter check (agent explicitly decided to post).

Optional: Add system_override flag to bypass filter.

agent_filter Rules

Decision Logic

def agent_filter_decision(event: MessageCreatedEvent) -> FilterDecision:
    # 1. Check channel permissions
    if not is_agent_member(event.channel_id, target_agent_id):
        return FilterDecision(decision="deny", reason="not_member")

    if not has_write_permission(event.channel_id, target_agent_id):
        return FilterDecision(decision="deny", reason="no_write_permission")

    # 2. Check content policy
    if contains_spam(event.content):
        return FilterDecision(decision="deny", reason="spam")

    if violates_policy(event.content):
        return FilterDecision(decision="modify", rewrite="Sanitize content")

    # 3. Check context (rate limiting, time of day)
    if too_many_agent_messages_recently(event.channel_id):
        return FilterDecision(decision="deny", reason="rate_limit")

    # 4. Check microdao rules
    microdao_rules = get_microdao_rules(event.microdao_id)
    if not microdao_rules.allow_agents:
        return FilterDecision(decision="deny", reason="microdao_policy")

    # 5. Select agent
    target_agent = select_best_agent(event.channel_id, event.content)

    return FilterDecision(
        decision="allow",
        target_agent_id=target_agent.id,
        rewrite_prompt=None
    )

Rules Categories

  1. Permissions
  2. Is agent member of channel?
  3. Does agent have can_write permission?

  4. Is channel in agent's allowed scope?

  5. Content Policy

  6. Spam detection
  7. Profanity filter
  8. Sensitive topics

  9. Privacy violations

  10. Context Rules

  11. Rate limiting (max N messages per hour)
  12. Time of day restrictions

  13. Frequency (don't reply to every message)

  14. microDAO Rules

  15. Are agents allowed in this microDAO?
  16. Which agent roles are permitted?

  17. Custom governance policies

  18. Agent Selection

  19. Which agent should respond? (Team Assistant, Quest Agent, etc.)
  20. Based on content, channel type, time

DAGI Router Integration

Router Rules for Messaging

rules:
  - name: "messaging.inbound"
    trigger: "agent.filter.decision"
    condition: "decision == 'allow'"
    action:
      type: "invoke_agent"
      agent_id: "{{ target_agent_id }}"
      entrypoint: "channel_message"
      payload:
        channel_id: "{{ channel_id }}"
        message_id: "{{ message_id }}"
        microdao_id: "{{ microdao_id }}"

  - name: "messaging.scheduled"
    trigger: "cron.daily_digest"
    condition: "time == '09:00'"
    action:
      type: "invoke_agent"
      agent_id: "agent:daily-digest"
      entrypoint: "generate_digest"
      payload:
        microdao_id: "{{ microdao_id }}"

Agent Invocation

{
  "invocation_id": "inv-uuid",
  "agent_id": "agent:sofia",
  "entrypoint": "channel_message",
  "payload": {
    "channel_id": "uuid",
    "message_id": "uuid",
    "microdao_id": "microdao:7"
  },
  "context": {
    "model": "gpt-4.1",
    "temperature": 0.7,
    "max_tokens": 500,
    "tools": ["create_task", "create_followup"],
    "memory_enabled": true
  }
}

Implementation Guide

Phase 1: Core Infrastructure (DONE ✅)

  • [x] Database schema (channels, messages, channel_members)
  • [x] messaging-service (REST + WebSocket)
  • [x] matrix-gateway API spec
  • [x] Frontend UI (MessengerPage)
  • [x] Docker orchestration

Phase 2: Agent Integration (NEXT)

  • [ ] Implement agent_filter service
  • [ ] Extend DAGI Router with messaging rules
  • [ ] Add Agent Runtime channel context reader
  • [ ] Implement /internal/agents/{id}/post-to-channel logic
  • [ ] NATS event integration (actual publishing)

Phase 3: Advanced Features

  • [ ] Agent Memory integration (RAG for channel context)
  • [ ] Tool execution (create_task from messages)
  • [ ] Multi-agent coordination
  • [ ] Scheduled agent messages (digests, reminders)

Phase 4: Production Hardening

  • [ ] Rate limiting (per agent, per channel)
  • [ ] Abuse detection
  • [ ] Analytics and metrics
  • [ ] A/B testing for agent responses

NATS Event Catalog

Published by messaging-service

messaging.message.created

{
  "channel_id": "uuid",
  "matrix_event_id": "$event:server",
  "sender_id": "user:93 | agent:sofia",
  "sender_type": "human | agent",
  "microdao_id": "microdao:7",
  "content_preview": "Hello!",
  "created_at": "2025-11-24T10:30:00Z"
}

messaging.channel.created

{
  "channel_id": "uuid",
  "microdao_id": "microdao:7",
  "matrix_room_id": "!room:server",
  "created_by": "user:93",
  "visibility": "public"
}

Published by matrix-gateway

integration.matrix.message

{
  "room_id": "!room:server",
  "event_id": "$event:server",
  "sender_matrix_id": "@user:server",
  "type": "m.room.message",
  "timestamp": 1735749000000,
  "content": {
    "msgtype": "m.text",
    "body": "Hello!"
  }
}

integration.matrix.room.created

{
  "room_id": "!room:server",
  "room_alias": "#alias:server",
  "creator": "user:93",
  "microdao_id": "microdao:7"
}

Published by agent_filter

agent.filter.decision

{
  "channel_id": "uuid",
  "message_id": "uuid",
  "matrix_event_id": "$event:server",
  "microdao_id": "microdao:7",
  "decision": "allow | deny | modify",
  "target_agent_id": "agent:sofia",
  "rewrite_prompt": "Sanitize...",
  "reason": "not_member | spam | rate_limit | policy"
}

Security Considerations

1. Agent Access Control

  • Agents must be explicitly added to channels
  • can_write permission required
  • agent_filter validates all agent replies

2. Content Safety

  • Spam detection
  • Profanity filtering
  • PII detection (for confidential channels)

3. Rate Limiting

  • Per agent: max 10 messages/hour
  • Per channel: max 50% agent messages
  • Global: max 1000 agent messages/hour

4. Audit Trail

  • All agent actions logged in channel_events
  • Matrix events are immutable (audit log)
  • NATS events retained for 30 days

5. Confidential Channels

  • E2EE channels: agents can't read plaintext
  • Agents operate on summaries/metadata only

Performance Targets

Metric Target Notes
Message send latency < 100ms User → Matrix → Index
Agent reply latency < 3s Full pipeline (filter → LLM → post)
WebSocket latency < 50ms Real-time updates
Channel list load < 500ms With 100+ channels
Message history (50) < 300ms Paginated from index
Agent context load < 1s 50 messages + memory query

Testing Checklist

Integration Tests

  • [ ] User sends message → indexed correctly
  • [ ] Message appears in Element
  • [ ] Agent reply triggered by filter
  • [ ] Agent reply appears in DAARION UI
  • [ ] Agent reply appears in Element
  • [ ] Multiple agents in same channel
  • [ ] Rate limiting works
  • [ ] Confidential channel blocks agent

E2E Tests

  • [ ] Full flow: User → Agent → Reply (< 5s)
  • [ ] Agent-initiated message (digest)
  • [ ] Multi-agent conversation
  • [ ] Tool execution from agent message

Roadmap

v1.1 (2 weeks)

  • Implement agent_filter service
  • DAGI Router messaging rules
  • NATS event publishing (production)
  • Agent Memory integration

v1.2 (1 month)

  • Multi-agent coordination
  • Scheduled agent messages
  • Analytics dashboard

v2.0 (2 months)

  • Voice messages (agent TTS)
  • Agent-to-agent direct messaging
  • Federated agents (cross-homeserver)

Version: 1.0.0
Last Updated: 2025-11-24
Status: Production Ready (Phase 1), Phase 2 Spec Complete
Maintainer: DAARION Platform Team