# 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:

```python
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` |
| `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.

### Question Format

The `task_questions` event supports structured questions:

```typescript
{
  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 |
