本页暂时只有英文版

Events

SSE Event Bus (app/events/)

Real-time event streaming to connected clients using Server-Sent Events (SSE).

Architecture

AI Agent ──emit()──→ EventBus ──queue──→ SSE Endpoint (/api/sse) ──→ Browser (EventSource)
BrowserManager ─────→ (broadcast)──→ Multiple clients
TaskRouter ─────────→

bus.py — EventBus

Simple asyncio broadcast pattern:

class EventBus:
    _subscribers: list[asyncio.Queue]

    subscribe() -> asyncio.Queue    # Client connects, gets a queue
    unsubscribe(queue)              # Client disconnects
    emit(event_type, data)          # Broadcast to all subscribers

Each subscriber gets its own asyncio.Queue. Events are JSON-serialized with type field merged into the data payload.

SSE Event Types

Event	Fields	Emitted By
task_created	task_id, store_id, task (full TaskResponse)	tasks.py
task_update	task_id, status, error?	tasks.py, claude_backend.py
task_steps_created	task_id, steps[]	runner.py
step_update	task_id, step_id, step_index, status, screenshot_id?, screenshot_b64?, error?	runner.py
task_log	task_id, log_type, content, timestamp_ms	runner.py
task_message	task_id, role, content	claude_backend.py
task_todos	task_id, todos[]	claude_backend.py
task_questions	task_id, request_id, questions[]	claude_backend.py
task_question_interrupted	task_id, request_id	claude_backend_manager.py
image_request	task_id, request_id, prompt, model, models[], reference_images[], output_name, kind	routers/vision.py
image_generating	task_id, request_id, model, kind	routers/vision.py
image_generated	task_id, request_id, path, url, prompt, model, kind	routers/vision.py
image_request_expired	task_id, request_id	routers/vision.py
image_request_interrupted	task_id, request_id	vision.py
agent_done	task_id, return_code, mode?	claude_backend.py
event_created	event_id, title, status	events.py, channels.py
event_updated	event_id, status, old_status?	events.py
event_activity	event_id, activity_id, actor_type, action, content	events.py
schedule_triggered	schedule_id, task_id, title	cron.py, fanout.py, schedules.py
fanout_triggered	schedule_id, batch_id, store_count, title	fanout.py
schedule_skipped	schedule_id, reason, plan_status	cron.py, fanout.py
schedule_waiting_cancelled	schedule_id, task_id, store_id	fanout.py
schedule_plan_ready	schedule_id, plan_version	claude_backend_hooks.py
schedule_plan_timeout	schedule_id, task_id	plan_reaper.py
ws_assistant_message	role, content	workspace_assistant.py
ws_assistant_done	(empty)	workspace_assistant.py
ping	(empty)	SSE keepalive

task_created — live task list

Emitted from POST /api/tasks after the DB commit and before the task is dispatched. The event carries the full TaskResponse so any subscribed tab can render the row without an extra GET, plus a top-level store_id for cheap filtering against the active view.

The dispatch order matters: task_created must precede any task_update for the same task, otherwise a status patch from the background runner hits a list that doesn't yet contain the row and gets silently dropped (see tests/workflow/test_wf_task_created_event.py and frontend/src/__tests__/sseCreateTaskRace.test.tsx).

The originating tab already adds the task from the POST response; the SSE handler dedupes by id so it doesn't double-insert.

Schedule plan lifecycle events

Emitted in addition to the standard task_update stream:

• schedule_skipped — a cron fire was refused by the plan-mode fire-gate. reason='plan_not_ready' (plan-mode schedules require plan_status='ready'). Frontend can surface a toast for unexpected skips.
• schedule_waiting_cancelled — a prior (schedule, store) task in waiting was cancelled so the next fire could create a fresh run. task_id is the cancelled task; store_id is null for single-mode schedules.
• schedule_plan_ready — user approved a plan-only Task; plan is persisted on the Schedule. plan_version matches the new bumped counter.
• schedule_plan_timeout — plan_reaper.py failed a schedule stuck in plan_status='planning' whose planner Task was idle past the threshold.

All four are emitted after the DB commit so consumers never observe an event for state that isn't yet persisted.

Image confirm-gate lifecycle

The vibe_seller_generate_image MCP tool is confirm-gated: every call emits image_request and blocks until the user acts, so the model can never generate without confirmation. One task shows at most one live card at a time.

• image_request — a card is proposed; the tool is parked on a per-request future waiting for the user to confirm/edit/cancel.
• image_generating → image_generated — the user confirmed; the kie.ai call runs (can take a minute) then the saved image lands.
• image_request_expired — a newer image request superseded this one (agent retry / task retry). The old card is retired; the newer request is the live one.
• image_request_interrupted — the user sent a chat message instead of confirming. The parked tool returns at once telling the agent to read that message rather than generate the un-confirmed proposal. This is distinct from expired: nothing replaced the card — the user redirected. task_question_interrupted is the exact analogue for a pending AskUserQuestion retired by a chat follow-up.

Question Format

The task_questions event supports structured questions:

{
  header?: string;        // Optional category/tag
  question: string;       // The question text
  options?: Array<{       // Available choices
    label: string;
    description?: string;
  }>;
  allow_custom?: boolean; // Allow free-text input (default: true)
}

Design Notes

• The bus is a singleton (event_bus), shared across the entire application
• All subscribers receive all events (no filtering). Filtering is done client-side
• Events are fire-and-forget: if a subscriber's queue is full, the emitter still continues
• Keepalive pings every 30 seconds to prevent connection timeouts

Business Event System (app/events_system/)

Extracts business events (deadlines, campaigns, cases) from channel messages via LLM, manages their lifecycle, and syncs to external backends.

Architecture

Channel Poll → extractor.py → Event model → Activity Timeline
                                    ↓
                              syncer.py → backends/ → Dida365, Google Calendar

Components

• extractor.py: Uses Claude Haiku to parse unstructured text (emails, WeChat messages) into structured events with title, description, event_date, deadline, platform
• syncer.py: Abstract backend pattern — EventBackend ABC with sync_event(), update_event(), delete_event(). Uses @register_backend(name) decorator + EVENT_BACKEND_REGISTRY dict
• backends/dida365.py: Dida365 (TickTick China) task sync via REST API
• backends/google_calendar.py: Google Calendar event sync via API

Event Lifecycle

Statuses: draft → open → in_progress → waiting → resolved → closed (+ dismissed)

Each status change recorded as an EventActivity entry in the activity timeline.

Activity Timeline

Action	Actor	Description
created	user/channel/system	Event creation
status_changed	user	Status transitions
note_added	user/ai	Manual notes
synced	system	Sync to external backend