PulseBot: Technical Design Document

A Lightweight, Stream-Native AI Agent Powered by Timeplus

Version: 2.0 Author: Timeplus Engineering

1. Executive Summary

PulseBot is an ultra-lightweight personal AI agent that leverages Timeplus's streaming SQL engine as its backbone for agent communication, memory management, and observability. It delivers equivalent functionality to complex agent frameworks in ~5,000 lines of Python by treating everything as a stream.

Core Design Principles

Stream-Native Architecture: All agent communication, events, and state changes flow through Timeplus streams
SQL-First Configuration: Agents and workflows are defined and queried via streaming SQL
Radical Simplicity: Target ~5,000 lines of Python
Observable by Default: Every LLM call, tool execution, and agent decision is logged to streams
Extensible Plugin System: Channels and skills as pluggable modules

2. Architecture Overview

┌─────────────────────────────────────────────────────────────────────────────┐
│                              Clients                                         │
│  ┌─────────────┐    ┌─────────────┐    ┌─────────────┐                      │
│  │  Web Chat   │    │     CLI     │    │  Telegram   │                      │
│  │  (Browser)  │    │  (Terminal) │    │    (Bot)    │                      │
│  └──────┬──────┘    └──────┬──────┘    └──────┬──────┘                      │
│         │                  │                  │                              │
│         └──────────────────┼──────────────────┘                              │
│                            │                                                 │
│                   HTTP / WebSocket                                           │
│                            │                                                 │
│                            ▼                                                 │
│  ┌─────────────────────────────────────────────────────────────────────┐    │
│  │                      API Server (FastAPI)                            │    │
│  │  • REST endpoints (/chat, /health, /sessions)                        │    │
│  │  • WebSocket real-time chat (/ws/{session_id})                       │    │
│  │  • Serves web UI (static HTML/CSS/JS)                                │    │
│  └───────────────────────────┬─────────────────────────────────────────┘    │
│                              │                                               │
│              ┌───────────────┴───────────────┐                               │
│              │                               │                               │
│           writes                          reads                              │
│       (user_input)              (agent_response, tool_call)                  │
│              │                               │                               │
│              ▼                               ▼                               │
└─────────────────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────────────────┐
│                    Timeplus Proton (Streaming Database)                      │
│  ┌───────────────────────────────────────────────────────────────────────┐  │
│  │                         Streams                                        │  │
│  │  ┌───────────┐  ┌───────────┐  ┌───────────┐  ┌────────┐  ┌────────┐  ┌──────────────┐  │  │
│  │  │ messages  │  │ llm_logs  │  │ tool_logs │  │ memory │  │ events │  │task_triggers │  │  │
│  │  │           │  │           │  │           │  │        │  │        │  │              │  │  │
│  │  │ user_input│  │ tokens    │  │ tool_name │  │ content│  │ type   │  │ trigger_id   │  │  │
│  │  │ agent_resp│  │ latency   │  │ duration  │  │ embed  │  │ payload│  │ task_name    │  │  │
│  │  │ tool_call │  │ cost      │  │ status    │  │ score  │  │ tags   │  │ prompt       │  │  │
│  │  └─────┬─────┘  └─────┬─────┘  └─────┬─────┘  └───┬────┘  └───┬────┘  └──────┬───────┘  │  │
│  │        │              │              │            │           │               │           │  │
│  │   API Server       Agent          Agent        Agent       Agent+Tasks    Timeplus       │  │
│  │   writes/reads     writes         writes       writes      writes         Tasks write    │  │
│  └───────────────────────────────────────────────────────────────────────────────────────────┘  │
└──────────────────────────────────────────────────────────────────────────────┘
                                   │
                    ┌──────────────┴──────────────┐
                    │                             │
                 listens                       writes
              (user_input)       (agent_response, tool_call, llm_logs,
                    │                  tool_logs, memory)
                    ▼                             │
┌─────────────────────────────────────────────────────────────────────────────┐
│                              Agent Core                                      │
│  ┌───────────────────────────────────────────────────────────────────────┐  │
│  │                         Agent Loop                                     │  │
│  │                                                                        │  │
│  │  1. Listen for user_input messages                                     │  │
│  │  2. Build context (history + memories)                                 │  │
│  │  3. Call LLM provider                                                  │  │
│  │  4. Execute tools if requested                                         │  │
│  │  5. Write responses back to messages stream                            │  │
│  │  6. Log to llm_logs and tool_logs                                      │  │
│  └───────────────────────────────────────────────────────────────────────┘  │
│                              │                                               │
│              ┌───────────────┼───────────────┐                               │
│              │               │               │                               │
│              ▼               ▼               ▼                               │
│  ┌───────────────┐  ┌───────────────┐  ┌───────────────┐                    │
│  │ Context       │  │ LLM Providers │  │ Tool Executor │                    │
│  │ Builder       │  │               │  │               │                    │
│  │               │  │ • Anthropic   │  │ • Skills      │                    │
│  │ • History     │  │ • OpenAI      │  │   - shell     │                    │
│  │ • Memory      │  │ • Ollama      │  │   - file_ops  │                    │
│  │ • System      │  │ • OpenRouter  │  │               │                    │
│  │   prompt      │  │ • NVIDIA      │  │               │                    │
│  └───────────────┘  └───────────────┘  └───────────────┘                    │
└─────────────────────────────────────────────────────────────────────────────┘

Data Flow

Client → API Server: User sends message via HTTP/WebSocket
API Server → messages stream: Writes user_input message
Agent ← messages stream: Listens for new user_input messages
Agent → LLM Provider: Builds context and calls LLM
Agent → llm_logs stream: Logs tokens, latency, cost
Agent → Tools: Executes tools if LLM requests them
Agent → tool_logs stream: Logs tool execution details
Agent → memory stream: Stores extracted memories (optional)
Agent → messages stream: Writes agent_response and tool_call messages
API Server ← messages stream: Listens for responses on the session
Client ← API Server: Forwards responses via WebSocket

Scheduled Task Data Flow:

Timeplus Task → Python UDF: Task fires on its schedule
Python UDF → /api/v1/task-trigger: HTTP POST to PulseBot API
API Server → messages stream: Writes scheduled_task message with global session
Agent ← messages stream: Picks up scheduled_task, processes with LLM
Agent → NotificationDispatcher: Sends result text to dispatcher
NotificationDispatcher → events stream: Writes task_notification event
Telegram / WebSocket ← events stream: All channel adapters fan out to their active connections

Stream Ownership

Stream	Writer	Reader
messages	API Server (user_input), Agent (agent_response, tool_call)	Agent, API Server
llm_logs	Agent	Analytics/Dashboards
tool_logs	Agent	Analytics/Dashboards
memory	Agent	Agent (context builder)
events	Agent (NotificationDispatcher)	Telegram, WebSocket (task_notification), Monitoring
task_triggers	Timeplus Tasks (via Python UDF)	Audit/Analytics
kanban	ManagerAgent, SubAgents	SubAgents, ManagerAgent
kanban_projects	ProjectManager, ManagerAgent	ProjectManager
kanban_agents	SubAgents, ManagerAgent	ProjectManager

3. Core Timeplus Streams

All data flows through five core streams. Streams are created automatically on agent startup using CREATE STREAM IF NOT EXISTS.

3.1 Messages Stream

Central communication hub for all agent interactions.

Column	Type	Description
id	string	Unique message ID (UUID)
timestamp	datetime64(3)	Event timestamp
source	string	Origin: 'telegram', 'webchat', 'agent', 'skill', 'system'
target	string	Destination: 'agent', 'channel:telegram', 'broadcast'
session_id	string	Groups related messages
message_type	string	'user_input', 'agent_response', 'tool_call', 'tool_result', 'error'
content	string	JSON payload
user_id	string	User identifier
channel_metadata	string	Channel-specific data (JSON)
priority	int8	Priority level (-1 to 2)

3.2 LLM Logs Stream

Comprehensive observability for all LLM calls.

Column	Type	Description
id	string	Request ID
timestamp	datetime64(3)	Call timestamp
session_id	string	Associated session
model	string	Model name (e.g., 'claude-sonnet-4-20250514')
provider	string	Provider name ('anthropic', 'openai', 'ollama')
input_tokens	int32	Input token count
output_tokens	int32	Output token count
total_tokens	int32	Total token count
estimated_cost_usd	float32	Estimated cost
latency_ms	int32	Total latency
time_to_first_token_ms	int32	Streaming latency
system_prompt_hash	string	SHA256 of system prompt
user_message_preview	string	First 200 chars of user message
assistant_response_preview	string	First 200 chars of response
tools_called	array(string)	List of tool names called
tool_call_count	int8	Number of tool calls
status	string	'success', 'error', 'rate_limited', 'timeout'
error_message	string	Error details if failed

3.3 Tool Logs Stream

Dedicated logging for tool executions with timing and status.

Column	Type	Description
id	string	Log entry ID
timestamp	datetime64(3)	Execution timestamp
session_id	string	Associated session
llm_request_id	string	Links to triggering LLM call
tool_name	string	Name of tool executed
skill_name	string	Parent skill name
arguments	string	JSON of tool arguments
status	string	'started', 'success', 'error'
result_preview	string	First 500 chars of result
error_message	string	Error details if failed
duration_ms	int32	Execution duration

3.4 Memory Stream

Persistent memory with vector search capability using append-only design with soft delete.

Column	Type	Description
id	string	Memory ID
timestamp	datetime64(3)	Creation timestamp
memory_type	string	'fact', 'preference', 'conversation_summary', 'skill_learned'
category	string	'user_info', 'project', 'schedule', 'general'
content	string	The memory content
source_session_id	string	Originating session
embedding	array(float32)	Vector embedding (1536-dim for OpenAI)
importance	float32	0.0 to 1.0, affects retrieval priority
is_deleted	bool	Soft delete flag (append-only pattern)

Memory Features:

Semantic search using cosine similarity
Hybrid scoring: vector similarity weighted by importance
Optional feature - requires OpenAI API key for embeddings
Soft delete pattern for append-only streams

3.5 Events Stream

System events for monitoring, alerting, and cross-channel broadcast.

Column	Type	Description
id	string	Event ID
timestamp	datetime64(3)	Event timestamp
event_type	string	`'heartbeat'`, `'channel_connected'`, `'skill_loaded'`, `'error'`, `'alert'`, `'task_notification'`
source	string	Event source
severity	string	`'debug'`, `'info'`, `'warning'`, `'error'`, `'critical'`
payload	string	JSON event data
tags	array(string)	Filtering tags

The task_notification event type is written by NotificationDispatcher after every scheduled task execution. Its payload JSON contains task_name, text, and session_id.

3.6 Task Triggers Stream

Audit log for every scheduled task invocation — inserted by the Timeplus Python UDFs embedded in each TASK.

Column	Type	Description
trigger_id	string	UUID for this invocation
task_id	string	Sanitised internal task name
task_name	string	Same as task_id
prompt	string	Instruction executed on this run
execution_id	string	UUID returned by `/api/v1/task-trigger`
triggered_at	datetime64(3)	UTC timestamp of invocation

4. Agent Core Components

4.1 Agent Loop

The main processing loop that:

Listens to the messages stream for incoming requests (user_input and scheduled_task)
Builds context from memory and conversation history
Calls LLM for reasoning
Executes tools and writes results back to stream
Broadcasts tool call status to UI/CLI in real-time
Routes scheduled_task messages through NotificationDispatcher instead of a normal channel response

Features:

Maximum 10 iterations per request (prevent infinite loops)
Automatic stream creation on startup
Tool call broadcasting for real-time UI updates
Error handling with client notification

4.2 Context Builder

Assembles complete context for LLM prompts:

System prompt with agent identity and instructions
Conversation history from messages stream
Relevant memories via vector search (if configured)
Available tools from loaded skills

4.3 Tool Executor

Executes tools from loaded skills:

Routes tool calls to appropriate skill
Handles errors gracefully
Returns structured results
Logs execution to tool_logs stream

4.4 Router

Routes messages between components:

Channel → Agent (user input)
Agent → Channel (responses)
Agent → Tools (execution)
Tools → Agent (results)

4.5 NotificationDispatcher

Broadcasts scheduled-task results to all channels via the events stream.

Receives the agent's response text after a scheduled_task execution
Writes a task_notification event to pulsebot.events
Channel adapters (Telegram, WebSocket) each subscribe independently and fan out to all active connections
Decoupled: adding a new channel only requires subscribing to the events stream

5. LLM Providers

Multi-provider support with unified interface:

Provider	Models	Features
Anthropic	Claude 4 Opus, Claude 4.5 Sonnet	Primary provider, tool use
OpenAI	GPT-4o, GPT-4 Turbo	Alternative provider
Ollama	Llama 3, Mistral, etc.	Local development
OpenRouter	Multiple providers	Unified access to many models
NVIDIA	NIM models	Enterprise deployment

6. Channels

Input/output channels for user interaction:

6.1 Web Chat (Built-in)

Single-file HTML/CSS/JS interface
WebSocket for real-time communication
Tool call status indicators with animations
Session-based conversation history
Minimal dependencies

6.2 CLI

Rich terminal UI with progress indicators
Real-time tool call display
Interactive chat mode
Session management

6.3 Telegram

Bot API integration
User allowlist for security
Persistent sessions per user
Subscribes to events stream for task_notification events and broadcasts to all active chat IDs
Restores sessions from message history on restart so task broadcasts survive agent restarts
Falls back to allow_from list when no interactive sessions exist

7. Skills System

Pluggable tool system with base interface:

Built-in Skills

Skill	Tools	Description
shell	`run_command`	Shell execution with security guards
file_ops	`read_file`, `write_file`, `list_directory`	Sandboxed file operations
workspace	`workspace_create_app`, ...	Dynamic artifact and web-app publishing
scheduler	`create_interval_task`, `create_cron_task`, `list_tasks`, `pause_task`, `resume_task`, `delete_task`	LLM-callable task management backed by Timeplus native Tasks

Skill Interface

Skills implement:

get_tools() - Return tool definitions
execute(tool_name, arguments) - Execute a tool
name and description properties

Built-in Skills (complete list)

Skill	Tools	Description
shell	`run_command`	Shell execution with security guards
file_ops	`read_file`, `write_file`, `list_directory`	Sandboxed file operations
workspace	`workspace_create_app`, ...	Dynamic artifact and web-app publishing
scheduler	`create_interval_task`, `create_cron_task`, ...	Timeplus-native recurring tasks
project_manager	`create_project`, `list_projects`, `get_project_status`, `cancel_project`	Multi-agent project orchestration

8. Multi-Agent System

PulseBot supports multi-agent projects where specialized worker agents collaborate via a kanban-style message queue stored in Timeplus streams.

8.1 Architecture

ProjectManager
    │
    ├── spawns ──► ManagerAgent (1 per project)
    │                 │  dispatches tasks via pulsebot.kanban
    │                 ▼
    └── spawns ──► SubAgent A          SubAgent B
                      │  (Researcher)      │  (Analyst)
                      │                   │
                      └── kanban msg ─────►│
                                           │
                                           └── result ──► ManagerAgent
                                                              │
                                                              └── pulsebot.messages ──► User

8.2 Components

ProjectManager (pulsebot/agents/project_manager.py)

Receives create_project tool calls from the main agent
Generates a project_id, stamps all agent specs, resolves human-readable target_agents names to agent IDs
Spawns one ManagerAgent and N SubAgent asyncio tasks
Provides list_projects, get_project_status, cancel_project operations

ManagerAgent (pulsebot/agents/manager_agent.py)

Extends SubAgent with project coordination responsibilities
Dispatches initial task messages to workers via pulsebot.kanban
Listens for result/error/status messages from workers
Delivers the final result to pulsebot.messages (→ user's WebSocket session)
Cancels all workers and marks project complete

SubAgent (pulsebot/agents/sub_agent.py)

Listens for task and control messages in pulsebot.kanban filtered by its agent ID
Runs the full LLM + tool loop for each task
Routes results to target_agents with msg_type="task" (worker targets) or msg_type="result" (manager target)
Persists checkpoints to pulsebot.kanban_agents for resumability

8.3 Kanban Streams

Stream	Purpose
`pulsebot.kanban`	Inter-agent message queue (task, result, error, control)
`pulsebot.kanban_projects`	Project lifecycle metadata
`pulsebot.kanban_agents`	Per-agent state and `_tp_sn` checkpoints

8.4 Message Routing Rules

Worker agents subscribe with msg_type IN ('task', 'control'). The ManagerAgent subscribes with msg_type IN ('result', 'error', 'status').

When a SubAgent routes its output:

To another worker: msg_type = "task" (so the receiving worker can pick it up)
To the manager: msg_type = "result" (so the manager can pick it up)

This enables both sequential pipelines (A → B → C → Manager) and fan-out patterns.

8.5 Session Routing

The session_id from the user's WebSocket connection is threaded through the entire project:

Stored in ProjectState and passed to ManagerAgent
Written to pulsebot.messages when the manager delivers the final result
The API server's WebSocket subscription filters by this session_id to deliver the response to the correct client

9. Real-time Features

9.1 Tool Call Broadcasting

Tool executions are broadcast in real-time:

Started: Tool name and formatted arguments
Completed: Success status with duration
Error: Error message with details

Displayed in:

Web Chat: Animated indicators with shimmer effect
CLI: Compact status lines with timing

9.2 WebSocket Streaming

Real-time message delivery via WebSocket:

Agent responses stream as they're generated
Tool call status updates
Error notifications
Connection state management

10. API Endpoints

REST API

Method	Endpoint	Description
GET	`/`	Web chat interface
GET	`/health`	Health check
POST	`/chat`	Send message (async)
POST	`/api/v1/task-trigger`	Scheduled task callback from Timeplus Python UDF
GET	`/sessions/{id}/history`	Get conversation history

WebSocket

Endpoint	Description
`/ws/{session_id}`	Real-time bidirectional chat

WebSocket message types:

message — User input / agent response
tool_call — Tool execution status
task_notification — Scheduled task broadcast result
error — Error notifications

11. Configuration

YAML-based configuration with environment variable substitution:

agent:
  name: "PulseBot"
  model: "claude-sonnet-4-20250514"
  provider: "anthropic"

providers:
  anthropic:
    api_key: ${ANTHROPIC_API_KEY}
  openai:
    api_key: ${OPENAI_API_KEY}
  ollama:
    enabled: true
    host: "http://localhost:11434"

timeplus:
  host: "localhost"
  port: 8123

channels:
  telegram:
    enabled: true
    token: ${TELEGRAM_BOT_TOKEN}
  webchat:
    enabled: true
    port: 8000

skills:
  builtin:
    - shell
    - file_ops
    - scheduler

12. Deployment

Docker Compose

Services:

Timeplus Proton - Streaming database (ports 8123, 3218, 8463)
PulseBot Agent - Message processing
PulseBot API - REST/WebSocket interface (port 8000)

Environment Variables

Variable	Required	Description
ANTHROPIC_API_KEY	Yes*	Claude API key
OPENAI_API_KEY	No	OpenAI key (also for memory embeddings)
TIMEPLUS_HOST	Yes	Timeplus host
TIMEPLUS_PASSWORD	No	Timeplus password
TELEGRAM_BOT_TOKEN	No	Telegram bot token

*One LLM provider key required

13. CLI Commands

Command	Description
`pulsebot run`	Start the agent loop
`pulsebot serve`	Start FastAPI server
`pulsebot chat`	Interactive CLI chat
`pulsebot init`	Generate config.yaml
`pulsebot task list`	List all scheduled tasks
`pulsebot task create`	Create an interval or cron task
`pulsebot task pause <name>`	Pause a task
`pulsebot task resume <name>`	Resume a paused task
`pulsebot task delete <name>`	Delete a task

14. Design Decisions

Why Append-Only Streams for Memory?

Mutable streams are an enterprise feature. The memory system uses append-only streams with a soft delete pattern:

is_deleted flag marks deleted records
Queries filter WHERE is_deleted = false
Simple and works with open-source Proton

Why Optional Memory?

Memory requires OpenAI API for embeddings. When not configured:

Memory features are gracefully disabled
Agent works without semantic memory
No errors or degraded experience

Why Single-File Web UI?

Zero build step required
Easy to customize
Minimal dependencies
Fast to load and modify

Why Tool Call Broadcasting?

Users need visibility into what the agent is doing:

Shows tool name and summarized arguments
Real-time status updates
Duration tracking
Error visibility

15. Future Enhancements

Planned features:

MCP (Model Context Protocol) server support
Additional channels (Slack, WhatsApp)
Analytics dashboard
Cost tracking views

16. Conclusion

PulseBot delivers a powerful, observable, and extensible AI agent by leveraging Timeplus's streaming SQL engine. The stream-native architecture provides:

Simplicity: ~5K lines of Python
Observability: Every event flows through queryable streams
Extensibility: Plugin-based skills and channels
Real-time: Sub-second message routing and tool status
SQL-First: Query and analyze everything via SQL

FilesExpand file tree

design.md

Latest commit

History