Skip to content

Event Backbone Architecture

CONCEPT:AU-KG.query.vendor-agnostic-traversal — Vendor-Agnostic Event Backbone

Overview

The event backbone provides protocol-based pub/sub event streaming for graph mutations, task queues, telemetry, and evolution triggers. It follows the same pattern as GraphBackend — abstract protocol with an in-memory default and optional distributed backends.

Architecture

graph TD
    subgraph Producers
        IGE["IntelligenceGraphEngine"]
        TM["TaskManager"]
        AHE["AHE Harness"]
    end

    subgraph EventBackend["Event Backbone (Protocol)"]
        MEB["MemoryEventBackend<br/>(in-process, zero-dep)"]
        KEB["RedpandaEventBackend<br/>(distributed, persistent, confluent-kafka)"]
    end

    subgraph Topics
        T1["kg.mutations"]
        T2["kg.tasks"]
        T3["kg.staging"]
        T4["kg.telemetry"]
        T5["kg.evolution"]
    end

    subgraph Consumers
        OBS["Post-commit mutation observers"]
        TELE["Telemetry Pipeline"]
        EVOL["Evolution Engine"]
    end

    IGE -->|publish| T1
    TM -->|publish| T2
    AHE -->|publish| T5
    T1 -->|subscribe| OBS
    T4 -->|subscribe| TELE
    T5 -->|subscribe| EVOL

    style MEB fill:#d5e8d4,stroke:#82b366
    style KEB fill:#dae8fe,stroke:#6c8ebf

Topic Taxonomy

Topic Purpose Retention Cleanup Policy
kg.mutations Graph CRUD events (add/update/remove node/edge) 7 days compact + remove
kg.tasks WorkItem notification and completion events 3 days remove
kg.staging Staged graph payloads awaiting write 1 day remove
kg.telemetry Agent traces, latency, error rates 1 day remove
kg.evolution Self-improvement triggers, AHE cycle events 7 days compact + remove

Event Schema

All events are JSON-serialized with the following base structure:

{
    "action": "add_node",
    "data": {
        "id": "agent-001",
        "properties": {"type": "Agent", "name": "Research Agent"}
    },
    "timestamp": 1716912000.0,
    "source": "IntelligenceGraphEngine"
}

Mutation Actions

Action Data Fields Description
add_node id, properties Add or update a node
add_edge source, target, properties Add or update an edge
remove_node id Remove a node and its edges
remove_edge source, target Remove a specific edge

Backend Selection

from agent_utilities.knowledge_graph.core.event_backend import create_event_backend

# In-memory (default, zero config, single-process)
backend = create_event_backend("memory")

# Redpanda (production, multi-process)
backend = create_event_backend("redpanda", bootstrap_servers="redpanda:9092")

Environment Variables

Variable Default Description
EVENT_BACKEND redpanda Backend type (memory, redpanda) — falls back to memory unless KAFKA_ENABLED=true
KAFKA_ENABLED false Gate that enables the distributed backend; otherwise MemoryEventBackend is used
REDPANDA_BROKERS localhost:9092 Broker addresses (Kafka/Redpanda)
REDPANDA_CONSUMER_GROUP agent-utilities Default consumer group
REDPANDA_SECURITY_PROTOCOL PLAINTEXT Security protocol

Authority and mirror consistency

The Rust engine is the sole graph authority. Writes commit as governed ChangeEnvelope operations; optional mirrors receive the same acknowledged mutations through the durable per-mirror FanOutBackend outbox. Pub/sub events are post-commit notifications and never form an alternate replication path.

sequenceDiagram
    participant ENG as Engine authority (Rust)
    participant FOB as FanOutBackend
    participant OB as Durable mirror outbox
    participant MIR as Mirror (persistent backend)

    FOB->>ENG: ApplyChangeEnvelope
    ENG-->>FOB: durable acknowledgement
    FOB->>OB: append mutation per mirror
    OB->>MIR: ordered replay
    MIR-->>OB: advance durable cursor
    Note over FOB,MIR: Periodic reconcile repairs drift from authority

Failure Modes

Failure Mode Mitigation
Redpanda unavailable Auto-fallback to MemoryEventBackend for notifications
Mirror unavailable Durable outbox retains its unapplied ordered tail
Authority↔mirror drift Explicit authority-to-mirror reconciliation repairs drift
Duplicate delivery Governed mutation identity and mirror cursors keep replay idempotent

Docker Deployment

# Start Kafka (KRaft mode, no Zookeeper) — also runs Redpanda-compatible brokers
docker compose -f docker/kafka-kraft.compose.yml up -d

# Verify topics
docker exec agent-utilities-kafka \
    /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092

Dependencies

# pyproject.toml
[project.optional-dependencies]
event-kafka = ["confluent-kafka>=2.0"]

Ingest Task Queue Scale-Out (CONCEPT:AU-KG.backend.selectable-queue-backend / KG-2.56 / AU-KG.ingest.decoupled-kg-ingest-consumer)

Separate from the pub/sub event backbone above, the durable ingest task queue (the queue submit_task writes and the ingest workers drain) is selectable, fail-loud, and — with Kafka — horizontally scalable.

Three queue modes

Mode Selected by Scope Workers
sqlite default (nothing set) one host (per-host kg_task_queue.db) in-process daemon threads on the flock host
postgres auto when STATE_DB_URI is set, or TASK_QUEUE_BACKEND=postgres fleet (one shared queue, FOR UPDATE SKIP LOCKED claims — AU-KG.ingest.cross-host-safe-kg) in-process threads on every participating host
kafka TASK_QUEUE_BACKEND=kafka fleet (keyed kg_tasks topic) the kg-ingest consumer group: the host engine's pool plus any number of decoupled kg-ingest-worker processes

Selection contract (KG-2.55):

  • TASK_QUEUE_BACKEND unset → auto: postgres when STATE_DB_URI is set, else sqlite. Auto stays graceful — an unreachable Postgres degrades to the per-host SQLite queue with a warning.
  • TASK_QUEUE_BACKEND=kafka|postgres set explicitly → fail-loud: an unreachable broker/state store raises TaskQueueUnavailable at startup with the endpoint and fall-back instructions. Never a silent degrade.

Partition-key hierarchy (KG-2.56)

Producers key every kg_tasks message; Kafka guarantees ordering per key without serializing unrelated work. Ingestion producers derive an opaque partition_ref from the privacy-normalized target and tenant; the notification does not contain the target definition. For generic queue envelopes, first match wins:

  1. tenant:<id> — ambient ActorContext.tenant_id (multi-tenant ordering);
  2. work:<opaque-ref> — durable ingestion partition reference;
  3. corpus:<repo> — portable corpus identifiers on non-ingestion envelopes;
  4. type:<task_type> — coarsest bucket for everything else.

kg_tasks is ensured idempotently at startup with KG_TASKS_PARTITIONS partitions (default 6). Grow-only: raising the flag adds partitions; an existing topic is never shrunk. Partition count bounds the consumer group's maximum parallelism.

Ordering & idempotency guarantees (AU-KG.ingest.decoupled-kg-ingest-consumer)

  • At-least-once delivery. Offsets are committed only after a task completes (or is durably marked failed); worker crashes redeliver.
  • Idempotent claims. job_id addresses a deterministic ingestion WorkItem. Every consumer must win native ClaimWorkItem; lease epoch and fencing token reject duplicate or stale execution. There is no graph-field claim fallback or advisory-lock authority.
  • Per-key ordering only. There is no global order and no cross-partition priority lane (the graph-polling modes' priority=high fast path does not apply in Kafka mode).
  • Lease recovery. Uncommitted offsets redeliver automatically; native ClaimWorkItem atomically reclaims expired leases when the item is eligible.

Worker deployment shape

# Host engine (unchanged): in Kafka mode its worker pool simply joins kg-ingest.
TASK_QUEUE_BACKEND=kafka KAFKA_BOOTSTRAP_SERVERS=kafka:9092 graph-os-daemon

# Scale out: N decoupled workers, any host, NO KG host role required.
# Each is an engine *client* (existing Rust daemon over TLS + OS-5.14 HMAC secret).
GRAPH_SERVICE_ENDPOINTS=tls://engine.example.test:9100 \
ENGINE_TLS_PROFILE_REF=vault://platform/engine/tls-profile \
GRAPH_SERVICE_AUTH_SECRET="${GRAPH_SERVICE_AUTH_SECRET_FROM_SUPERVISOR}" \
KAFKA_BOOTSTRAP_SERVERS=kafka:9092 \
kg-ingest-worker            # or: python -m agent_utilities.knowledge_graph.ingest_worker

Per-host concurrency autosizes with the shared CPU/memory sizer (compute_ingest_worker_count: ~36% of cores, ~3 GB RAM per worker, floor 2); override with --workers / KG_INGESTION_WORKERS.

Backpressure & lag visibility

The leader host's maintenance scheduler samples the queue every pass and publishes to the AU-OS.observability.no-op-without-metrics gateway Prometheus registry:

  • agent_utilities_kg_ingest_queue_depth{backend} — pending tasks in the selected backend (uniform across sqlite/postgres/kafka);
  • agent_utilities_kg_ingest_consumer_lag{topic,group} — total kg-ingest group lag on kg_tasks (Kafka mode).

The batch orchestrator's deferral and the maintenance bulk-defer gate read the same uniform number via engine.ingest_queue_depth(), which counts each non-terminal ingestion WorkItem exactly once, so backpressure does not double count a Kafka notification and its authoritative work record.

Part of the Knuckles-Team agent ecosystem. Source repository · Documentation home