wf-agent2

Легковесный Linux-агент для сбора событий, нормализации, хранения в памяти, обработки команд и отправки исходящих сообщений.

Проект ориентирован на системный запуск под Linux и использует:

• collectors для источников событий;
• единый канонический формат EventEnvelope на protobuf v1;
• bounded in-memory cache и outbox;
• единый control plane для CLI, scheduler и derivator;
• senders и transports, разделённые по ответственности.

Что умеет агент

• подключается к нескольким источникам событий через отдельные collectors;
• прогоняет каждое событие через prefilter, validation, normalization и marking;
• складывает короткое окно событий в bounded ring buffer;
• строит derived events поверх recent cache windows;
• принимает команды через единый CommandCoordinator;
• маршрутизирует команды в executors;
• формирует command-result events и пропускает их через Director;
• отправляет outbound events через bounded outbox и configurable sender/transport.

Быстрый старт

Зависимости

• Python 3.11+
• Linux
• protobuf
• python-dotenv
• typer не требуется, CLI реализован на argparse

Установка

# Для сборки systemd-python необходимы dev-пакеты Linux
sudo apt-get install -y libsystemd-dev pkg-config python3-dev build-essential
cd /opt/wf-agent2
source .venv/bin/activate
python -m pip install -e .
python -m pip install systemd-python dbus-next

Для чтения системного журнала системному пользователю процесса надо дать права, например: включив его в группу:

sudo usermod -aG systemd-journal wifi-agent

или именно процессу при запуске сервиса:

SupplementaryGroups=systemd-journal

Проверка конфигурации

agent config check --config config/agent.toml --no-env-file

Запуск

agent run --config config/agent.toml --no-env-file

Статус и статистика

agent status --config config/agent.toml --no-env-file
agent stats --config config/agent.toml --no-env-file

Архитектура

Ниже показан основной поток данных внутри приложения.

Collectors
  +- JournalCollector
  +- DbusCollector
  +- WpaCtrlCollector
  +- NavigationCollector
  L- Future collectors

Each collector:
  source connection
  -> prefilter
  -> validation
  -> normalization to EventEnvelope
  -> marking/enrichment
  -> Director

Director
  -> Cache
  -> Outbox
  -> Derivator interest

Derivator
  -> reads cache windows
  -> emits derived EventEnvelope
  -> may emit CommandRequest
  -> CommandCoordinator

Command initiators
  +- CLI
  +- Scheduler
  L- Derivator

CommandCoordinator
  -> policy
  -> queue
  -> executor router
  -> executors
  -> command-result event
  -> Director

Sender loops
  -> read outbox only
  -> batch / drain / retry
  -> transport

Ключевые принципы

• EventEnvelope - единый канонический формат для всех collectors и command-result событий.
• EventRecord - компактная runtime-форма для hot path cache.
• Все очереди и буферы bounded.
• CLI, scheduler и derivator не вызывают executors напрямую.
• Senders читают только outbox.
• Transport отвечает только за доставку, sender - за стратегию draining.

Компоненты и связи

Collectors

Collectors отвечают за источник данных и lifecycle соединения.

Общие обязанности:

• start() / stop();
• health() / stats();
• reconnect и recovery;
• применение pipeline stages;
• отправка нормализованного EventEnvelope в Director.

Реализованные collectors:

• JournalCollector - systemd journal;
• DbusCollector - D-Bus сигналы;
• WpaCtrlCollector - wpa_supplicant ctrl_iface;
• NavigationCollector - адаптер навигации.

Collector pipeline

Pipeline одинаков по смыслу для всех collectors.

1. Prefilter

• ранний отсев шума;
• минимизация нагрузки на CPU и память.

2. Validation

• проверка структуры сырого события;
• reject или downgrade malformed records.

3. Normalization

• перевод в канонический EventEnvelope;
• заполнение hot fields и source-specific payload.

4. Marking

• добавление workflow tags, labels, routing hints, severity overrides.

Director

Director решает, куда попадёт canonical event:

• в cache;
• в outbox;
• в derivator-interest path.

Он опирается на metadata envelope, hot normalized fields и правила routing. Глубоко payload не парсит.

Cache

Cache состоит из:

• bounded ring buffer;
• payload store;
• secondary indexes.

Индексы строятся по:

• interface;
• event type;
• source kind;
• ssid.

Для derivator cache является коротким окном recent events.

Commands

Команды представлены как CommandRequest и проходят одинаковую цепочку:

CLI / Scheduler / Derivator
  -> CommandRequest
  -> CommandCoordinator
  -> CommandPolicy
  -> CommandQueue
  -> ExecutorRouter
  -> Executor
  -> CommandResult event
  -> Director

Это гарантирует одинаковую политику dedup, cooldown, priority и exclusive key.

Executors

Executors выполняют control actions, а не ingestion.

Реализованные executors:

• WpaCtrlExecutor - команды для wpa_supplicant;
• SystemExecutor - системные действия, например system.set_regdomain.

Executors эмитят canonical command-result events через Director.

Scheduler

Scheduler создаёт CommandRequest для interval и one-shot задач. Он не вызывает executors напрямую.

Команды scheduler идут по тому же пути, что и CLI команды.

Derivator

Derivator читает окна cache, строит derived events и при необходимости может инициировать команды через CommandCoordinator.

В проекте есть workflow wifi_scan_summary.

Sender и transport

Sender управляет тем, когда и как outbox draining происходит. Transport отвечает за физическую доставку.

Поддержаны sender modes:

• none
• strict_priority
• buffered

Поддержаны transports:

• drop
• webhook

Структура проекта

src/agent/
  app.py               bootstrap runtime
  cli.py               CLI entrypoint
  settings.py          env + TOML bootstrap
  env.py               .env loader

  collectors/          source adapters and pipeline
  commands/            command model, policy, queue, coordinator
  core/                hot-path runtime types and enums
  cache/               bounded cache, payload store, indexes, snapshots
  director/            routing engine and decisions
  derivator/           cache-window derivation and workflows
  executors/           command executors and router
  scheduler/           periodic and one-shot task generation
  sender/              outbox, retry, loops, transports
  proto_generated/     checked-in protobuf-style bindings
  utils/               TOML/JSON helpers
  observability/       logging/health/stats helpers

proto/
  common.proto
  events.proto

tests/
  unit/
  integration/
  fixtures/

Конфигурация

.env

Пример:

AGENT_ENV=dev
AGENT_LOG_LEVEL=INFO
AGENT_CONFIG_PATH=./config/agent.toml
HOST_ID=robot-17
SITE_ID=warehouse-a
WEBHOOK_URL=https://example.local/webhook
WEBHOOK_TOKEN=secret-token

TOML

Главные секции:

• [agent] - общие параметры;
• [cache] - размер ring buffer и payload arena;
• [outbox] - bounded queue;
• [sender] и [transport] - режим доставки;
• [collectors.*] - collectors и их pipeline rules;
• [executors.*] - runtime executors;
• [derivator] - derivator worker и workflows;
• [tasks] - scheduler jobs.

Пример минимального конфига:

[agent]
name = "wifi-event-agent"
log_level = "INFO"

[cache]
capacity_events = 50000
payload_arena_bytes = 67108864

[outbox]
capacity_items = 5000

[sender]
type = "buffered"

[transport]
type = "webhook"

[derivator]
enabled = true
poll_interval_ms = 500

[[derivator.workflows]]
name = "wifi_scan_summary"
workflow_name = "wifi_scan_summary"
min_wifi_records = 1

Развёртывание в системе

Обычно агент запускают как systemd service.

Рекомендованный сценарий

• создать системного пользователя, например wf-agent;
• разместить конфиг в /etc/wf-agent/agent.toml;
• хранить .env в /etc/wf-agent/.env или задавать env через unit;
• запускать сервис от отдельного пользователя, а не интерактивно;
• выдавать только необходимые права на источники событий и системные действия.

Пример unit-файла

[Unit]
Description=wf-agent2
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=wf-agent
Group=wf-agent
WorkingDirectory=/opt/wf-agent2
Environment=AGENT_CONFIG_PATH=/etc/wf-agent/agent.toml
Environment=AGENT_ENV=prod
Environment=AGENT_LOG_LEVEL=INFO
ExecStart=/opt/wf-agent2/.venv/bin/agent run --no-env-file --config /etc/wf-agent/agent.toml
Restart=always
RestartSec=2

[Install]
WantedBy=multi-user.target

Права доступа

Зависит от того, какие collectors и executors включены.

JournalCollector

Нужен доступ к systemd journal:

• обычно через группу systemd-journal;
• либо запуск от root;
• либо через policy, если используется более строгая модель доступа.

DbusCollector

Нужен доступ к system bus и соответствующим D-Bus policy rules.

WpaCtrlCollector / WpaCtrlExecutor

Нужен доступ к unix socket wpa_supplicant ctrl_iface. Часто это означает:

• принадлежность к netdev или специальной группе;
• корректные права на socket directory;
• иногда запуск от root, если локальная политика ограничивает доступ.

SystemExecutor

Команды вроде system.set_regdomain обычно требуют привилегий уровня root или CAP_NET_ADMIN. Для production лучше использовать либо:

• отдельный privileged service;
• либо минимальный sudo/polkit policy на конкретное действие.

Практическая рекомендация

Если агент должен только читать события, а не менять системные параметры, держите его в непривилегированном аккаунте. Если включены system.* команды, отделяйте их от read-only deployment или жёстко ограничивайте разрешённые действия.

Примеры запуска команд

Базовые команды

agent run --config config/agent.toml --no-env-file
agent config check --config config/agent.toml --no-env-file
agent status --config config/agent.toml --no-env-file
agent stats --config config/agent.toml --no-env-file

Collector inspection

agent collectors list --config config/agent.toml --no-env-file
agent collectors show wpa_ctrl --config config/agent.toml --no-env-file

Cache inspection

agent cache head --limit 10 --config config/agent.toml --no-env-file
agent cache tail --limit 20 --config config/agent.toml --no-env-file
agent cache find --type wifi.raw.wpa.disconnected --iface wlp3s0 --config config/agent.toml --no-env-file

Outbox and sender

agent outbox show --limit 20 --config config/agent.toml --no-env-file
agent outbox stats --config config/agent.toml --no-env-file
agent sender test --config config/agent.toml --no-env-file
agent sender flush --config config/agent.toml --no-env-file

Derivator

agent derivator status --config config/agent.toml --no-env-file
agent derivator run-once --config config/agent.toml --no-env-file

Ad-hoc commands

agent cmd wpa status --iface wlp3s0 --config config/agent.toml --no-env-file
agent cmd wpa signal-poll --iface wlp3s0 --config config/agent.toml --no-env-file
agent cmd wpa scan --iface wlp3s0 --config config/agent.toml --no-env-file
agent cmd wpa deassociate --iface wlp3s0 --config config/agent.toml --no-env-file
agent cmd system set-regdomain --country US --config config/agent.toml --no-env-file

Scheduler tasks

agent tasks list --config config/agent.toml --no-env-file
agent tasks show wifi-active-scan --config config/agent.toml --no-env-file
agent tasks run-once wifi-active-scan --config config/agent.toml --no-env-file
agent tasks enable wpa-status --config config/agent.toml --no-env-file
agent tasks disable wifi-active-scan --config config/agent.toml --no-env-file
agent tasks stats --config config/agent.toml --no-env-file

Разработка

Проверка кода:

python -m pytest -q
python -m compileall src tests

Состояние проекта

Проект уже содержит:

• collectors, pipeline и supervisor;
• cache core и outbox;
• command model, policy, queue и coordinator;
• executors и router;
• scheduler;
• derivator worker и workflow wifi_scan_summary;
• CLI для status, stats, collectors, cache, outbox, sender, derivator, cmd и tasks.

Дальнейшее развитие описано в AGENTS.md.