DEV Community: Alain Airom (Ayrom)

Harnessing LMCache: Building a Pluggable KV-Cache Management & Testing Framework

Alain Airom (Ayrom) — Tue, 10 Nov 2026 08:17:51 +0000

LMCache test & implementation.

Introduction

As Large Language Model (LLM) architectures scale and context windows expand to tens of thousands of tokens, prefill latency has become a major bottleneck in AI application performance. Processing long, repeated prompt prefixes—such as system prompts, core reference documents in Retrieval-Augmented Generation (RAG), and extensive agent chat histories—consumes substantial compute cycles and drives up time-to-first-token (TTFT).

What is LMCache?

LMCache is an open-source Key-Value (KV) cache management layer designed to solve this exact bottleneck. Rather than recomputing attention Key and Value tensors for every incoming request, LMCache captures, indexes, and stores these tensors outside the core inference engine, re-injecting them on demand.
Key features of LMCache include:

Chunk-Level Prefix Reuse: Token streams are partitioned into fixed-size chunks (defaulting to 256 tokens). Each chunk is assigned a rolling prefix hash (using blake3), guaranteeing that matching prefixes immediately hit cached states.
Tiered Storage Architecture: Cache snapshots flow seamlessly across device memory (GPU/CPU RAM), local disk (L2), and remote backends (e.g., Redis, S3, Mooncake, NIXL) with Least Recently Used (LRU) eviction policies.
Engine Independence: Cache snapshots survive engine restarts and slot evictions, allowing cross-instance sharing and disaggregated prefill/decode topologies.

The LMCache Ecosystem

The LMCache ecosystem spans mainstream open-source projects, enterprise hardware vendors, and infrastructure orchestrators:

Inference Engines: Native support targets server-grade Linux/CUDA engines such as vLLM, SGLang, and TensorRT-LLM.
Orchestration Frameworks: Integrates with orchestrators like NVIDIA Dynamo, vLLM Production Stack, llm-d, and KServe.
Hardware & Storage Ecosystem: Compatible with storage vendors (Dell, Redis, DDN, Valkey, WEKA, VAST, Samsung, IBM Storage) and accelerator providers (NVIDIA, AMD, Ascend, Intel, Moore Threads, Rebellions, MetaX).
Research Extensions: Incorporates advanced research ideas including TurboQuant, CacheBlend, CacheGen, and Cartridges.

Excerpt from official Github repository of the project:

LMCache is a KV cache management layer for LLM inference. It turns KV cache from a temporary state into reusable AI-native knowledge that can be stored persistently, reused across multiple serving engines, monitored with an observability stack, and transformed for better generation quality. As a result, LMCache reduces TTFT (time-to-first-token) and improves throughput, especially for long-context agentic, multi-turn conversation, and knowledge-augmented workloads (e.g., RAG).

LMCache is vendor-neutral. It can be used as a KV cache layer for a range of mainstream open-source serving engines, inference frameworks, hardware vendors, storage systems, and infrastructure providers. The vendor neutrality allows users to freely switch between serving engines and storage vendors, while reusing the stored KV caches.

Key features

Engine-independent deployment: LMCache, as a standalone daemon process, manages KV cache independently from the inference engine process, so that KV cache will not be lost even if the inference engine crashes (i.e., no fate-sharing with engines).
Persistent, tiered KV cache offloading and reuse: Move KV caches out of GPU memory into a tiered storage hierarchy spanning CPU memory, local storage, and remote backends, enabling reuse across requests, sessions, and engine instances to reduce repeated prefill computation and improve TTFT.
Production-level KV cache observability: LMCache provides a rich set of KV cache observability metrics, including typical Kubernetes metrics (health monitoring, performance diagnostics), KV-cache-specific metrics (request-level and token-level prefix cache hits, lifecycle, request-level KV cache performance), management metrics (user-specific usage), and more.
Pluggable storage and transport backends: Easily integrate remote storage and KV transfer backends through a unified interface, enabling KV cache offloading and sharing across storage providers. Through this interface, LMCache supports storage backends including CPU RAM, local disk (SSD), Redis/Valkey, Mooncake, InfiniStore, S3-compatible object storage, NIXL, and GDS.
Non-prefix KV reuse: Extend KV reuse beyond prefix caching by reusing cached KV blocks at any position in the prompt. This leverages CacheBlend to selectively recompute tokens for quality recovery.
PD disaggregation and KV transfer: Support KV cache transfer from prefill workers to decode workers over NVLink, RDMA, or TCP through transport layers such as NIXL.
Pluggable KV transformation: A simple interface for researchers to write compression, token dropping, and custom serialization through a flexible SERDE interface.

LMCache is becoming an integral layer in the LLM inference ecosystem, with community-driven integration with serving engines, inference frameworks, hardware vendors, storage systems, and infrastructure providers:

Implementation

Disclaimer: Officially, LMCache targets server-side GPU inference engines (vLLM, SGLang, TRT-LLM) on Linux; official PyPI wheels require CUDA, and feature requests for Ollama/LM Studio support (e.g., issue #923) were closed as "not planned". However, for a mock test and rapid local prototyping, an LMCache-style framework named KVBridge *was implemented to bring these concepts to my local test environment.*

Application overview & purpose

KVBridge demonstrates the core value proposition of LMCache — a KV cache management layer for scalable LLM inference — on a developer laptop. LMCache itself targets server-side engines (vLLM, SGLang, TensorRT-LLM) and needs CUDA. KVBridge re-implements the LMCache concepts (chunked, content-addressed KV reuse; tiered storage; engine-independent keys) on top of the engines developers actually run: two local ones (llama.cpp, Ollama) and any OpenAI-compatible cloud service, all behind one pluggable engine registry.

What KVBridge is, and is not

KVBridge is an LMCache-style implementation. It is not the LMCache package. It optionally reuses LMCache's native TokenHasher when the real package can be built (scripts/install_lmcache.sh); otherwise a bit-compatible blake3 rolling hash is used.

System Architecture & Components

KVBridge mirrors LMCache's core mechanisms (rolling prefix hashing, tiered storage, and cache persistence) while adding an extensible engine registry to support local engines (llama.cpp, Ollama) and any OpenAI-compatible cloud provider.

Code Layout

The local application, uses Gradio for the UI implementation and is configurable to connect to multiple backends.

lmcache-test/
├── kvbridge/
│   ├── __init__.py
│   ├── __main__.py        # python -m kvbridge
│   ├── app.py             # Gradio GUI (registry-driven), AppState, main(), SIGTERM handler
│   ├── backends.py        # engine registry, LlamaCppBackend, OllamaBackend, CloudBackend, make_backend
│   ├── engine.py          # KVBridgeEngine: lookup, restore, compute, store
│   ├── store.py           # L1 RAM / L2 disk tiered store
│   ├── hashing.py         # chunking + chained hashes
│   ├── metrics.py         # counters, TTFT, hit rates
│   ├── charts.py          # Plotly figures
│   ├── scenarios.py       # demo scenario registry
│   └── config.py          # Settings (GUI/env/.env), presets, update_env_file
├── tests/                 # unit + live tests
├── scripts/               # setup, start, stop, test, cleanup, install_lmcache
├── Docs/                  # Quickstart, Architecture, Research, this file
├── input/  output/        # tracked via .gitkeep, contents ignored
├── requirements.txt
├── pytest.ini
├── .env.example           # every KVB_* variable, local and cloud examples
└── README.md              # includes the Apache-2.0 licence statement

Evaluation Modes

To evaluate performance gains, the architecture routes requests through three distinct operational modes:

cold: Disables native prompt caching (cache_prompt=false) and wipes existing slot state, forcing a full prefill calculation on every request to establish a baseline.
native: Relies strictly on the engine's built-in, in-slot prefix caching (cache_prompt=true).
lmcache: Drops in-slot state, performs chunk-hash lookups against the L1/L2 store, restores cached tensor state into the engine slot, pre-fills only the un-cached tail suffix, and saves new chunk snapshots upon completion.

Full KV State Export and Import (engine.py): the primary pipeline manages KV tensor persistence using (only for mock test purpose) llama.cpp's /slots/{id}?action=save|restore API:

""The cache-aware inference pipeline.

Request flow (llama.cpp, full integration)::

    prompt --tokenize--> tokens --chunk+hash--> [h0 h1 h2 ...]
                                                   |
                    store.lookup (longest prefix) <-+
                          |hit                |miss
            fetch snapshot (L1 RAM / L2 disk)  |
            restore into engine slot           |
                          +---------+----------+
                                    v
                 engine.generate (prefills only the *missing* suffix)
                                    v
                 save slot -> store.put (only when new chunks were computed)

Three execution modes make the benefit measurable:

``cold``    prefix reuse disabled - every request prefills the whole prompt.
``native``  the engine's own slot/prefix cache only (baseline a user has today).
``lmcache`` native cache is dropped first (modelling eviction by another
            session, a restart, ...) and the KV layer restores state instead.
"""
...

def _infer_full(self, prompt: str, rec: RequestRecord, n_predict: int, temperature: float) -> None:
    """llama.cpp path: real KV tensor save/restore handling."""
    b = self.backend
    tokens = b.tokenize(prompt)

    if rec.mode == "cold":
        b.reset_native_cache()
        self._fill(rec, b.generate(prompt, tokens, n_predict, False, temperature))
        return

    if rec.mode == "native":
        self._fill(rec, b.generate(prompt, tokens, n_predict, True, temperature))
        return

    # ---- LMCache Mode Pipeline
    hashes = self.hasher.chunk_hashes(tokens)
    rec.chunks_total = len(hashes)
    rec.chunk_status = self.store.chunk_status(hashes)

    # Drop native slot state to isolate LMCache retrieval benefit
    b.reset_native_cache()                       
    match = self.store.lookup(hashes)
    rec.lmcache_match_tokens, rec.chunks_hit, rec.tier = match.tokens, match.chunks, match.tier

    # Restore matching KV snapshot from L1 RAM or L2 Disk
    if match.entry is not None:
        try:
            blob = self.store.fetch(match.entry)
            n = b.restore_state(blob)
            rec.restore_ms = (time.perf_counter() - t0) * 1000
        except (BackendError, KeyError, OSError) as exc:
            log.warning("Restore failed (%s); falling back to recompute", exc)
            rec.chunks_hit, rec.lmcache_match_tokens, rec.tier = 0, 0, "none"
            b.reset_native_cache()

    # Compute generation (prefills missing suffix only)
    res = b.generate(prompt, tokens, n_predict, True, temperature)
    self._fill(rec, res)

    # Persist newly computed tail chunks into the Tiered Store
    if len(hashes) > rec.chunks_hit:
        blob, n_saved = b.save_state()
        self.store.put(hashes, len(hashes) * self.s.chunk_size, blob,
                       note=f"{n_saved} tok snapshot")

Extensible Engine Registry (backends.py): to prevent hardcoded conditionals across engine implementations, a centralized registry handles dynamic initialization:

"""Inference-engine adapters.

Three adapters share one small interface (:class:`Backend`) and are looked up
through a registry (:func:`register_engine`, :func:`resolve_engine`), so the
engine is chosen purely by configuration (``KVB_ENGINE`` or the GUI):

``LlamaCppBackend``
    Talks to ``llama-server`` (llama.cpp).  llama.cpp exposes the **slot
    save/restore API** (``POST /slots/{id}?action=save|restore|erase``) which
    serialises the *real KV tensors* of a sequence.  That is the integration
    point that lets an external cache layer (LMCache semantics) take KV out of
    the engine and inject it back -> **full integration**.

``OllamaBackend``
    Talks to the Ollama REST API.  Ollama deliberately hides KV state: there
    is no KV export/import and no tokenizer endpoint.  The adapter therefore
    works in **shadow mode** - the LMCache layer indexes/accounts for chunks
    and predicts reuse, while Ollama's own prefix cache does the real work and
    its ``prompt_eval_cached_count`` is used as ground truth.

``CloudBackend``
    Talks to any **OpenAI-compatible** HTTP API (OpenAI, Groq, OpenRouter,
    Together, Mistral, DeepSeek, vLLM, LM Studio, ...).  Hosted engines never
    expose their KV tensors either, so it also runs in **shadow mode**; the
    provider's own prompt-cache counter (``cached_tokens``) is the ground truth.
    Prompts leave the machine - the GUI says so explicitly.
"""

from __future__ import annotations

import json
import logging
import os
import re
import shutil
import signal
import subprocess
import time
import uuid
import zlib
from abc import ABC, abstractmethod
from dataclasses import dataclass
from pathlib import Path
from typing import Optional, Sequence
from urllib.parse import urlparse

import requests

from .config import CLOUD_API_STYLES, DEFAULT_LLAMA_SEARCH_PATHS, Settings
...


@dataclass(frozen=True)
class EngineSpec:
    key: str                      # Identifier (e.g., KVB_ENGINE)
    label: str                    # GUI Label
    cls: type[Backend]            # Adapter implementation class
    blurb: str = ""
    aliases: tuple[str, ...] = ()

_REGISTRY: dict[str, EngineSpec] = {}

def register_engine(spec: EngineSpec) -> EngineSpec:
    _REGISTRY[_norm(spec.key)] = spec
    return spec

def make_backend(kind: str, settings: Settings) -> Backend:
    return resolve_engine(kind).cls(settings)

# Engine Registration
register_engine(EngineSpec("llamacpp", "llama.cpp (local)", LlamaCppBackend,
                           "Full KV save/restore through llama-server slots.", ("llama",)))
register_engine(EngineSpec("ollama", "Ollama (local)", OllamaBackend,
                           "Shadow mode: Ollama hides its KV cache.", ()))
register_engine(EngineSpec("cloud", "Cloud / OpenAI-compatible API", CloudBackend,
                           "Shadow mode against hosted engines.", ("openai", "api")))

Chained Rolling Prefix Hasher (hashing.py): token streams are packed into big-endian uint32 representations and processed sequentially:

"""Chunk-level rolling prefix hashing, compatible with LMCache.

LMCache splits a token sequence into fixed-size chunks (default 256 tokens)
and identifies each chunk by a *rolling prefix hash*::

    h_0 = blake3(prefix_seed        || pack(chunk_0))
    h_i = blake3(h_{i-1}            || pack(chunk_i))

where ``pack`` is the big-endian ``uint32`` encoding of the token ids
(``lmcache/v1/multiprocess/token_codec.py``).  Only *complete* chunks are
hashed; a trailing partial chunk is discarded - exactly like
``lmcache.v1.multiprocess.token_hasher.TokenHasher``.

This module re-implements that algorithm (so the app runs on macOS where the
full LMCache wheel cannot be installed) and can optionally delegate to the
real ``TokenHasher`` when the ``lmcache`` package is importable.  A unit test
verifies both produce identical digests.
"""

from __future__ import annotations

import hashlib
import logging
import struct
from typing import Protocol, Sequence

log = logging.getLogger(__name__)

try:  # blake3 is the algorithm LMCache uses
    import blake3 as _blake3

    def _digest(data_parts: Sequence[bytes]) -> bytes:
        h = _blake3.blake3()
        for part in data_parts:
            h.update(part)
        return h.digest()

    ALGORITHM = "blake3"
except ImportError:  # pragma: no cover - fallback keeps the app usable
    def _digest(data_parts: Sequence[bytes]) -> bytes:
        h = hashlib.sha256()
        for part in data_parts:
            h.update(part)
        return h.digest()

    ALGORITHM = "sha256 (blake3 missing; NOT LMCache compatible)"


def pack_tokens(tokens: Sequence[int]) -> bytes:
    """Pack token ids as big-endian ``uint32`` (LMCache wire format)."""
    return struct.pack(f">{len(tokens)}I", *tokens)


def namespace_seed(namespace: str) -> bytes:
    """Derive the initial prefix hash from a model/config namespace.

    KV tensors are only valid for the exact model, quantisation and context
    configuration that produced them, so the namespace is mixed into the
    root of every hash chain (LMCache keys also embed the model name).
    """
    return _digest([namespace.encode("utf-8")])


class Hasher(Protocol):
    chunk_size: int
    impl: str

    def chunk_hashes(self, tokens: Sequence[int]) -> list[bytes]: ...


class ChunkHasher:
    """Pure-Python implementation of LMCache's rolling chunk hashing."""

    impl = "kvbridge (LMCache-compatible blake3)"

    def __init__(self, chunk_size: int = 256, namespace: str = "") -> None:
        if chunk_size < 1:
            raise ValueError("chunk_size must be >= 1")
        self.chunk_size = chunk_size
        self.namespace = namespace
        self.seed = namespace_seed(namespace)

    def chunk_hashes(self, tokens: Sequence[int]) -> list[bytes]:
        """Return one rolling hash per *complete* chunk of ``tokens``."""
        n_complete = len(tokens) - len(tokens) % self.chunk_size
        prefix = self.seed
        out: list[bytes] = []
        for i in range(0, n_complete, self.chunk_size):
            prefix = _digest([prefix, pack_tokens(tokens[i : i + self.chunk_size])])
            out.append(prefix)
        return out


class LMCacheHasherAdapter:
    """Delegates to ``lmcache.v1.multiprocess.token_hasher.TokenHasher``."""

    def __init__(self, chunk_size: int = 256, namespace: str = "") -> None:
        from lmcache.v1.multiprocess.token_hasher import TokenHasher  # type: ignore

        self.chunk_size = chunk_size
        self.namespace = namespace
        self.seed = namespace_seed(namespace)
        self._hasher = TokenHasher(chunk_size=chunk_size, hash_algorithm="blake3")
        self.impl = "lmcache.TokenHasher (real LMCache)"

    def chunk_hashes(self, tokens: Sequence[int]) -> list[bytes]:
        return self._hasher.compute_chunk_hashes(list(tokens), prefix_hash=self.seed)


def make_hasher(chunk_size: int, namespace: str, prefer_lmcache: bool = False) -> Hasher:
    """Create a hasher, preferring the real LMCache one when requested."""
    if prefer_lmcache:
        try:
            return LMCacheHasherAdapter(chunk_size, namespace)
        except Exception as exc:  # ImportError or LMCache init problems
            log.warning("LMCache TokenHasher unavailable (%s); using built-in hasher", exc)
    return ChunkHasher(chunk_size, namespace)


def common_prefix_len(a: Sequence[int], b: Sequence[int]) -> int:
    """Length of the longest common prefix of two token sequences."""
    n = min(len(a), len(b))
    i = 0
    while i < n and a[i] == b[i]:
        i += 1
    return i

Configuration examples & usage scenarios

Pick one block for .env (or enter the same values in the GUI and press Connect).

.env — local, full integration (llama.cpp):

KVB_ENGINE=llamacpp
KVB_OLLAMA_MODEL=llama3.2:latest      # used to locate the GGUF blob
KVB_CTX_SIZE=8192
KVB_LLAMA_START_TIMEOUT=300           # slow disk? allow more time

.env — local, shadow mode (Ollama):

KVB_ENGINE=ollama
KVB_OLLAMA_URL=http://127.0.0.1:11434
KVB_OLLAMA_MODEL=llama3.2:latest
KVB_OLLAMA_KEEP_ALIVE=30m

.env — hosted provider (key stays in your shell):

KVB_ENGINE=cloud
KVB_CLOUD_PROVIDER=openai
KVB_CLOUD_MODEL=gpt-4o-mini
KVB_CLOUD_API_KEY_ENV=OPENAI_API_KEY  # export OPENAI_API_KEY=... before start.sh
KVB_CLOUD_TOKEN_BUDGET=200000         # spend guard

.env — self-hosted vLLM / any OpenAI-compatible server:

KVB_ENGINE=cloud
KVB_CLOUD_BASE_URL=https://llm.example.internal/v1
KVB_CLOUD_MODEL=meta-llama/Llama-3.1-8B-Instruct
KVB_CLOUD_API_KEY_ENV=INTERNAL_LLM_KEY
KVB_CLOUD_EXTRA_HEADERS={"X-Team": "search"}

.env — free smoke test of the cloud path (no account):

KVB_ENGINE=cloud
KVB_CLOUD_PROVIDER=ollama-local-openai   # http://127.0.0.1:11434/v1
KVB_CLOUD_MODEL=llama3.2:latest

Scenario	Engine	What to do	What you see
A. Maximum benefit, fully offline	llama.cpp	`KVB_ENGINE=llamacpp`, Connect, run Document Q&A	First question prefills (seconds); later ones restore KV in tens of ms; Engine restart shows an L2 restore
B. Observe an existing Ollama setup	Ollama	`KVB_ENGINE=ollama`, Connect, run Shared system prompt	Predicted reuse next to `prompt_eval_cached_count`; SHADOW label
C. Measure a hosted provider's prefix cache	Cloud	Preset + model + key + token budget, Connect, run Shared system prompt with a long input document	`cached_tokens` reported by the provider per request; `cold` adds a nonce and shows no reuse
D. Try the cloud path for free	Cloud	Preset `ollama-local-openai`, model `llama3.2:latest`	Same GUI and metrics, no key, no cost
E. Switch engines without editing files	any	Change the engine radio, edit fields, Connect, optionally Save settings to .env	Status line states engine, location and FULL/SHADOW mode

Programmatic equivalent:

from kvbridge.config import Settings
from kvbridge.backends import make_backend
from kvbridge.engine import CachedEngine

s = Settings.from_env()                  # honours KVB_ENGINE and .env
assert not (problems := s.validate()), problems
engine = CachedEngine(s, make_backend(s.engine, s))
engine.start()

Final Disclaimer and findings after the tests

🚩 As noted earlier, this application was designed to demystify LMCache functionality and practical usage. Operating as a local, GPU-free implementation, it offers a purely educational pathway to observe firsthand how tiered caching and prefix reuse enhance inference efficiency.

Question	Answer
Does LMCache officially support Ollama / llama.cpp?	No Supported: vLLM, SGLang, TRT-LLM. Issue #923 (Ollama / LM Studio) was closed “not planned”.
Can the PyPI wheel run on macOS?	No Wheels are manylinux-only and require CUDA (`CUDA_HOME not set`).
Can LMCache's hasher be used standalone?	Partly With `NO_GPU_EXT=1` plus extra dependencies the native `TokenHasher` builds.
Can llama.cpp export/import KV state?	Yes `/slots/{id}?action=save
Can a hosted (cloud) API export/import KV state?	No Never. Providers run their own prefix cache and report {% raw %}`cached_tokens`, enabling the same accounting-only shadow mode.
Can Ollama export/import KV state?	No No public KV API. It does report `prompt_eval_cached_count`, enabling an accounting-only shadow mode.

Conclusion

Building a practical testbed around LMCache concepts provides valuable insights into modern prompt caching mechanics.
Key takeaways from the project include:

Measurable TTFT Reduction: In benchmark scenarios (e.g., ~1.9k-token document Q&A), average TTFT dropped from 4,554 ms (cold) to 999 ms (lmcache)—achieving a 4.6x overall speedup across a 12-request workload. On warm hits, TTFT fell as low as 11 ms (L1 RAM) and 13 ms (L2 Disk).
Persistence vs. Native In-Slot Caching: While native engine caching can be faster on a single warm slot (e.g., 64 ms), the primary value of an LMCache-style architecture is capacity and persistence—snapshots survive process restarts, span arbitrary numbers of prefixes, and remain usable across multiple sessions.
Observability & Control: Through an intuitive Gradio GUI, automated test suites (over 180 unit/live tests), and robust SIGTERM lifecycle handling, developers can reliably test, visualize, and optimize prompt reuse strategies on any workstation.

Thanks for reading 🌒

Giving AI Agents a Governed Identity: A Comprehensive Guide to Architecture, Implementation, and Enforcement with CUGA

Alain Airom (Ayrom) — Fri, 06 Nov 2026 07:52:17 +0000

Implementing Agent Identity with CUGA Agent.

Introduction & Background

The Shift Towards Autonomous AI Agents

As enterprise AI transitions from stateless chat completion interfaces to autonomous agents capable of reading context, making decisions, executing multi-step workflows, and calling external APIs, traditional application security models begin to break down.
Historically, software integration relied on rigid permission boundaries: either a service used a single, static API key for system-to-system operations, or an application borrowed a user’s session token to perform actions. In an agentic architecture, however, LLM-powered agents plan their own actions, dynamically select tools, and interact across complex execution chains.

┌───────────────────────────────────────────────────────────────────────────┐
│                        UNGOVERNED AGENT ARCHITECTURE                      │
│                                                                           │
│  User ──────► [ Agent Runtime ] ──────► Shared Service Account / API Key  │
│                   │                                 │                     │
│                   ▼                                 ▼                     │
│            Dynamic Tool Call                Unrestricted Access           │
│         (No granular attribution)           (Cannot isolate failures)     │
└───────────────────────────────────────────────────────────────────────────┘

Without a dedicated, governed identity mechanism, agents present several key security risks:

Identity Spoofing & Blurs: When an agent borrows a shared service account or user password, auditors cannot differentiate between an intentional human action and an agent-generated call.
Over-Privileged Execution (Lack of Least Privilege): Agents often run with broad system access, allowing a single prompt injection or hallucinated tool call to inspect, modify, or delete off-limit data.
Uncontrolled Delegation: When an agent acts on behalf of a user, there is often no cryptographic guarantee that the agent’s permission context remains less than or equal to the authorizing user's entitlements.
Lack of Granular Containment (Kill Switch Gap): Disabling a compromised agent requires revoking broad service accounts or shared credentials, which risks taking down unaffected services or agent runtimes.

What is Governed Agent Identity?

An AI agent acts on its own: it reads data, calls tools and sometimes changes things. Agent Identity gives each agent its own verifiable identity in the same way a company gives each employee a badge. The agent is registered, approved by a different person, and then proves who it is on every call.

Its core purpose is to answer four questions about every action an agent takes:

Agent Identity establishes AI agents as first-class identity entities—similar to issuing an employee badge within an enterprise identity framework. Each agent is registered in a secure registry, reviewed under strict Separation of Duties (where the creator cannot approve their own agent), bound to its own OAuth client, and granted short-lived, cryptographically signed tokens.

┌───────────────────────────────────────────────────────────────────────────┐
│                         GOVERNED AGENT IDENTITY                           │
│                                                                           │
│  Developer ──► Register ──► Approver (Separation of Duties) ──► Active    │
│                                                                  │        │
│  User Token + Agent Credentials ──► RFC 8693 Token Exchange ─────┘        │
│                                           │                               │
│                                           ▼                               │
│  Audit Trail ◄── [ Identity Gateway ] ──► Short-Lived Scoped Token        │
│   (Agent ID)          (Verification)              │                       │
│                                                   ▼                       │
│                                          Protected Business Data          │
└───────────────────────────────────────────────────────────────────────────┘

Governed identity answers four critical operational questions for every execution step:

Security Pillar	Operational Requirement	Governed Agent Identity Realization
Accountability	Who performed this action?	Every call carries an unforgeable Agent ID. All runtime and lifecycle actions log to a single correlation trail.
Least Privilege	What is the maximum allowed scope?	Scope ceilings (allowed_scopes) are defined at onboarding. Issued tokens can never exceed this ceiling, even if a tool is exposed.
Safe Delegation	On whose authority is the agent acting?	Delegated tokens utilize RFC 8693 token exchange (sub = User, act.sub = Agent). Scope is bounded by the intersection of user rights and agent rights.
Immediate Control	Can we isolate a rogue agent instantly?	A centralized kill switch shifts status to SUSPENDED, blocking token issuance and forcing instant rejection at identity gateways.

System Architecture & Lifecycle Governance

Component Architecture & Layered Defense

A robust governed agent architecture splits responsibility into two distinct runtime layers: LLM Policy Framing and Resource-Level Identity Enforcement.

Cuga Policy Framing: Runs before the model or tool invocation. Intent guards filter dangerous prompts (e.g., bulk deletion), while tool approval rules freeze execution until human validation is received.
Resource-Level Identity Enforcement: Runs inside the backend tool execution path. The tool requests token minting via an IdentityGateway, which checks credentials, validates identity backend status, and asserts token signatures and scope ceilings.

Complete Agent Lifecycle State Machine

An agent's operational permissions are governed by an explicit state machine. Creation, approval, suspension, and reactivation are managed via formal administrative operations with embedded audit trails.

DEFAULT: The initial state after registration by a developer. The agent has an assigned Agent ID and scope ceiling, but cannot mint tokens or register an OAuth client.
ACTIVE: Set by an independent approver. Dynamic OAuth client registration is unlocked, and short-lived tokens can be issued for authorization.
SUSPENDED (Kill Switch): Suspends token issuance immediately. When coupled with strict status validation on resource tools, active tokens are rejected.

Access Delegation Mechanics & Scoping Modes

                                  GRANT TYPES & ACCESS MODES
                                              │
         ┌────────────────────────────────────┼────────────────────────────────────┐
         ▼                                    ▼                                    ▼
  Autonomous Mode                      Delegated Mode                      Privileged Mode
(client_credentials)                  (RFC 8693 Exchange)               (authorization_details)
         │                                    │                                    │
  • Agent acts alone                   • Agent acts for user                • Sensitive operations
  • Background analytics               • HR / Course enrollment             • Service restarts
  • Claims: agent_id                   • Claims: sub=user, act=agent        • Short TTL + Step-up human check

Governed identities support four primary access modes depending on operational context:

Autonomous Access (client_credentials): Used for automated, non-interactive processes. The agent authenticates using its OAuth client_id and secret, receiving a token scoped to its defined ceiling.
Delegated Access / On-Behalf-Of (RFC 8693 Token Exchange): Used when an agent executes tasks on behalf of an authenticated end user. The identity gateway exchanges the user's JWT for a delegated token where sub is the user and act.sub is the agent.

Privileged Action (Human-in-the-Loop Scoping): Designed for sensitive operations (e.g., service restarts or financial transactions). Requires interactive human approval alongside short-lived tokens carrying explicit authorization_details.
Agent Chaining (Nested Delegation): When a supervisor agent delegates sub-tasks to a downstream worker agent, tokens nest the actor claim (act: worker -> supervisor), maintaining complete auditability across multi-agent steps.

Implementation with CUGA-Agent Integration

The following reference implementation demonstrates how to build a governed agent using Cuga, an Identity Gateway, an isolated Policy Store, and custom Guarded Tools.

The demonstration use-case

The following is what is implemented (containing simulation scenario)

Cuga Agent is used for real: agent construction, tool execution, intent-guard and tool-approval policies, human-in-the-loop resume, supervisor delegation.
Agent Identity has no public SDK. The showcase therefore ships a local mock of the IBM Verify identity service (MockVerifyBackend: RS256-signed JWTs, agent registry, dynamic client registration, client-credentials and RFC 8693 token exchange). HttpVerifyBackend is a reference implementation of the same interface for a real tenant. It has not been tested against a live tenant; treat it as a starting point.
Cuga Harness Kit is not a test runner. It scaffolds skills for coding assistants (11 skills, installed with init --targets bob). The showcase runs it into a temporary folder to prove it works, and uses its own harness/scenarios.py (10 conformance scenarios) for validation.
The LLM engine is selectable: local Ollama (default, granite4.2:latest), llama.cpp (llama-server) or a cloud OpenAI-compatible API. Pick it in the UI or in .env (see LLM engines). Tests and the offline demo use a deterministic fake model, so no LLM is needed to run them.
Live status of the engines: Ollama detection, test and switching were exercised against a real local daemon. llama.cpp and cloud were tested with mocked HTTP only (no llama-server and no cloud key were available), so treat them as unverified against a real server.

Technical Features of the use-case

Agent lifecycle: create, approve (a different person), register an OAuth client, deploy, suspend (kill switch), reactivate.
Four agents covering different identity patterns:
- Learning Coach: delegated (on-behalf-of) user actions, with human approval and a privileged token for expensive enrolments.
- Revenue Analyst: autonomous agent with a read-only scope ceiling.
- Ops SRE Assistant: privileged actions (service restart) bound to a short-lived token.
- Workplace Supervisor: multi-agent orchestration with a nested actor chain in the token.
Defence in depth: Cuga policies (before the LLM), then per-tool identity checks (after the LLM). The LLM never sees a token.
Interactive documentation: six identity flows rendered as sequence diagrams, a token inspector, an audit trail keyed by Agent ID.
Pluggable LLM engines: Ollama, llama.cpp and cloud behind one interface, with detection, a settings page with Test connection, runtime switching without a restart, a documented fallback order and error messages that say how to fix the problem.
Validation: 118 unit tests, 10 harness conformance scenarios, a CLI demo and a UI smoke test.
Operations: structured logging, typed configuration, scripts for start/stop/cleanup, a multi-stage Containerfile.

Identity Models and Gateway Setup: first, we establish core identity models, custom error types, context management, and token validation mechanisms.

# agent_identity/models.py
"""Domain models shared by the identity layer."""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Any


class AgentStatus(str, Enum):
    """Lifecycle states of an agent in the IBM Agent Registry."""

    DEFAULT = "DEFAULT"  # created by the developer, not yet approved
    ACTIVE = "ACTIVE"  # approved: may register clients and obtain tokens
    SUSPENDED = "SUSPENDED"  # blocks NEW token issuance (issued tokens stay valid!)
    UNDER_REVIEW = "UNDER_REVIEW"


class Entitlement(str, Enum):
    """Management-plane entitlements (names follow the IBM docs)."""

    READ_AGENTS = "readAgents"
    WRITE_AGENTS = "writeAgents"
    MANAGE_AGENT_STATUS = "manageAgentStatus"


@dataclass(frozen=True)
class Principal:
    """A human management user (developer or approver)."""

    user_id: str
    display_name: str
    entitlements: frozenset[Entitlement]

    def has(self, ent: Entitlement) -> bool:
        return ent in self.entitlements


@dataclass
class AgentRecord:
    """An entry of the Agent Registry."""

    agent_id: str
    name: str
    description: str
    owner: str  # user_id of the creating developer
    status: AgentStatus = AgentStatus.DEFAULT
    allowed_scopes: frozenset[str] = frozenset()
    approved_by: str | None = None
    client_id: str | None = None
    created_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat(timespec="seconds"))
    metadata: dict[str, Any] = field(default_factory=dict)


@dataclass(frozen=True)
class ClientCredentials:
    """OAuth client bound to an agent. The secret is returned only once."""

    client_id: str
    client_secret: str
    agent_id: str

Next, we implement context handling to store end-user session state securely without exposing authorization tokens directly to the LLM model space.

# agent_identity/gateway.py
"""Identity gateway used by tools.

The LLM never sees a token. Tools call :meth:`IdentityGateway.authorize`, which
reads the *call context* (who is the end user, which tokens are available) from
a ``ContextVar`` set by the runtime, mints a just-in-time token for the right
grant type, validates it like a resource server would, and returns the
validated :class:`~agent_identity.tokens.TokenContext`.
"""

from __future__ import annotations

import contextvars
from dataclasses import dataclass
from typing import Any

from .audit import AuditLog
from .errors import DelegationError, IdentityError
from .lifecycle import AgentIdentityService
from .logging_setup import get_logger
from .tokens import TokenContext, TokenValidator

log = get_logger("gateway")


@dataclass(frozen=True)
class CallContext:
    """Per-request identity context (propagated to tools via a ContextVar)."""

    user_id: str | None = None
    user_token: str | None = None
    # Set when a supervisor delegates to a sub-agent: the supervisor's own
    # on-behalf-of token becomes the *subject* of the sub-agent's exchange, which
    # nests the actors in the ``act`` claim (the transaction/caller chain).
    supervisor_token: str | None = None


_CALL: contextvars.ContextVar[CallContext | None] = contextvars.ContextVar("agent_identity_call", default=None)


def set_call_context(ctx: CallContext | None) -> contextvars.Token:
    return _CALL.set(ctx)


def reset_call_context(token: contextvars.Token) -> None:
    _CALL.reset(token)


def current_call_context() -> CallContext | None:
    return _CALL.get()


class IdentityGateway:
    """Mints and validates tokens for tool executions."""

    def __init__(self, service: AgentIdentityService, issuer: str, public_key: Any, strict_status_check: bool = True) -> None:
        self._service = service
        self._issuer = issuer
        self._key = public_key
        self._strict = strict_status_check
        self._validators: dict[str, TokenValidator] = {}

    @property
    def audit(self) -> AuditLog:
        return self._service.audit

    def validator(self, audience: str) -> TokenValidator:
        """Return (and cache) the resource-server validator for ``audience``."""
        if audience not in self._validators:
            lookup = (lambda aid: self._service.backend.get_agent(aid).status) if self._strict else None
            self._validators[audience] = TokenValidator(
                self._key, self._issuer, audience, self._service.audit, status_lookup=lookup
            )
        return self._validators[audience]

    def authorize(
        self,
        *,
        agent_name: str,
        mode: str,  # "obo" (on behalf of the user) | "autonomous"
        audience: str,
        scopes: list[str],
        resource: str,
        authorization_details: list[dict[str, Any]] | None = None,
    ) -> TokenContext:
        """Mint + validate a token; raises an :class:`IdentityError` subclass on denial."""
        agent = self._service.backend.find_agent(agent_name)
        try:
            if mode == "autonomous":
                token = self._service.agent_token(agent.agent_id, scopes, audience)
                return self.validator(audience).validate(token, tuple(scopes), resource=resource)

            call = current_call_context()
            subject = (call.supervisor_token or call.user_token) if call else None
            if not subject:
                raise DelegationError("No end-user context: on-behalf-of access needs a logged-in user")
            token = self._service.delegated_token(agent.agent_id, subject, scopes, audience, authorization_details)
            return self.validator(audience).validate(
                token,
                tuple(scopes),
                require_delegation=True,
                require_privileged=bool(authorization_details),
                resource=resource,
            )
        except IdentityError as exc:
            log.warning("authorize failed agent=%s resource=%s: %s", agent_name, resource, exc)
            raise

Defining Guarded Tools: tools assert authorization boundaries via the IdentityGateway prior to interacting with backend resources.

# agent_identity/tools.py

"""Identity-guarded tools for the example agents.

Every tool obtains a token through the :class:`IdentityGateway` *before* touching
data. Denials are returned as ``ACCESS DENIED: ...`` text so the model can relay
them faithfully, and are also written to the audit log.
"""


import json
from langchain_core.tools import tool
from agent_identity.models import IdentityError
from agent_identity.gateway import IdentityGateway

COURSES_DB = {
    "SEC301": {"title": "Advanced Cloud Security", "cost": 1200},
    "PY101": {"title": "Python Basics", "cost": 300}
}

def create_learning_coach_tools(gateway: IdentityGateway, agent_name: str):

    @tool
    def list_courses() -> str:
        """List available training courses, costs, and IDs."""
        try:
            gateway.authorize(
                agent_name=agent_name,
                mode="obo",
                audience="hr-service",
                scopes=["courses:read"]
            )
        except IdentityError as exc:
            return f"ACCESS DENIED: {str(exc)}"

        return json.dumps(COURSES_DB)

    @tool
    def enroll_expensive_course(course_id: str) -> str:
        """Enroll the user in a course costing over 500 USD. Requires approval."""
        course = COURSES_DB.get(course_id.upper())
        if not course:
            return f"Error: Course {course_id} not found."

        details = [{
            "type": "course_enrollment",
            "course_id": course_id.upper(),
            "max_cost": course["cost"]
        }]

        try:
            # Privileged token authorization with specific authorization_details
            gateway.authorize(
                agent_name=agent_name,
                mode="obo",
                audience="hr-service",
                scopes=["courses:enroll"],
                authorization_details=details
            )
        except IdentityError as exc:
            return f"ACCESS DENIED: {str(exc)}"

        return f"Successfully enrolled in {course['title']} ({course_id.upper()})."

    return [list_courses, enroll_expensive_course]

Constructing the Governed Cuga Agent Environment:next, we initialize the agent with isolated policy stores and human-in-the-loop gates.

# agents/learning_coach.py
"""Example agent 1 - Learning Coach.

Use case: an employee asks an agent to browse and enrol in training courses.
Identity pattern: **on-behalf-of (RFC 8693)** - tokens carry ``sub=<employee>`` and
``act.sub=<agent>``. Expensive courses need a human approval (Cuga tool-approval
policy) *and* a short-lived privileged token (``authorization_details``).
"""

from __future__ import annotations

from agent_identity.spec import AgentSpec, ApprovalSpec, IntentGuardSpec
from agent_identity.tools import build_course_tools

SPEC = AgentSpec(
    name="learning-coach",
    title="Learning Coach",
    description="Helps employees discover and enrol in training courses on their behalf.",
    use_case="HR / learning - delegated user actions",
    access_mode="obo",
    allowed_scopes=frozenset({"courses:read", "courses:enroll"}),
    tool_builder=build_course_tools,
    instructions=(
        "You are the Learning Coach. Use the provided tools to list courses and enrol the current user. "
        "Courses above 500 USD must be enrolled with enroll_expensive_course. "
        "If a tool answers 'ACCESS DENIED', report that denial verbatim and do not retry."
    ),
    guards=(
        IntentGuardSpec(
            name="No bulk record deletion",
            keywords=("delete all", "wipe", "erase all"),
            response="Blocked by policy: this agent cannot delete records.",
        ),
    ),
    approvals=(
        ApprovalSpec(
            name="Approve expensive enrolments",
            tools=("enroll_expensive_course",),
            message="This enrolment exceeds 500 USD. A human must approve it.",
        ),
    ),
    sample_prompts=(
        "List the available courses.",
        "Enroll me in PY101.",
        "Enroll me in SEC301.",
        "Delete all employee records.",
    ),
)

Runtime Switchable LLM Engine Infrastructure: to remain resilient, the runtime LLM engine layer supports non-disruptive switches between local (Ollama, llama.cpp) and cloud models without discarding context or invalidating agent tokens.

                          RUNTIME ENGINE SWITCHING
                                     │
    ┌────────────────────────────────┼────────────────────────────────┐
    │                                │                                │
    ▼                                ▼                                ▼
1. Gate New Turns           2. Drain Running Turns          3. Rebuild Graphs
Close `_ready` gate         Wait for active turn completion Reset `_graph` references
Pause new messages          Enforce drain timeout           Keep chat history & tokens

# agent_identity/llm_engines.py

"""Pluggable LLM engines: Ollama, llama.cpp and OpenAI-compatible cloud services.

This module is UI-framework free. It provides:

* :class:`LLMEngine` - the abstraction (detect / validate / test_connection / build_model);
* three concrete engines plus an offline ``fake`` engine used by tests and as last resort;
* :class:`EngineManager` - detection, persisted UI choice, and the **resolution order**:

  1. an explicit selection (the UI override saved locally, then ``AI_LLM_ENGINE``);
  2. engines whose endpoint / credentials are configured or detected, in the order of
     ``AI_LLM_FALLBACK_ENGINES``;
  3. a hard-coded fallback (local Ollama), then the offline fake model as a last resort.

Secrets are :class:`pydantic.SecretStr` and are never logged, rendered or written to disk.
Every failure is a :class:`~agent_identity.errors.LLMEngineError` that carries a remediation hint.
"""

from __future__ import annotations

import itertools
import json
import os
import shutil
import stat
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any
from urllib.parse import urlparse

import httpx
from pydantic import SecretStr

from .config import Settings
from .errors import EngineConnectionError, EngineMisconfigured, EngineUnavailable, LLMEngineError
from .logging_setup import get_logger
from .ui_helpers import probe_ollama
...

import asyncio
from typing import Dict, Any

class EngineManager:
    """Manages active LLM engines, handling fallbacks and hot swaps."""

    def __init__(self, current_config):
        self.current_config = current_config
        self._gate_open = asyncio.Event()
        self._gate_open.set()
        self._active_turns = 0

    async def switch_engine(self, new_config: Dict[str, Any], agent_instances: list):
        """Swaps the active LLM engine using a gate-drain-swap-rebuild workflow."""
        # Step 1: Gate incoming execution turns
        self._gate_open.clear()

        # Step 2: Drain active turns
        drain_timeout = 30.0
        start_time = asyncio.get_event_loop().time()
        while self._active_turns > 0:
            if asyncio.get_event_loop().time() - start_time > drain_timeout:
                self._gate_open.set()
                raise RuntimeError("Engine switch timed out while draining active turns.")
            await asyncio.sleep(0.1)

        # Step 3: Swap configuration and rebuild agent execution graphs
        try:
            self.current_config = new_config
            new_model = self._build_model_instance(new_config)

            for agent in agent_instances:
                # Update underlying model reference and signal graph rebuild
                agent._model = new_model
                agent._graph = None
                agent._compiled_graph = None

        except Exception as exc:
            # Revert to safe state if swap fails
            self._gate_open.set()
            raise RuntimeError(f"Engine switch failed: {exc}")

        # Step 4: Re-open gate for new traffic
        self._gate_open.set()

    def _build_model_instance(self, config):
        # Initializes Ollama, llama.cpp, Cloud OpenAI API, or Fake test model
        pass

Conclusion & Realization via CUGA-Agent

Summary of Architecture & Security Benefits

By combining the Cuga Agent Framework with a governed identity architecture, enterprise deployments achieve end-to-end security, regulatory compliance, and operational observability:

┌────────────────────────────────────────────────────────────────────────────┐
│                      GOVERNED AGENT SECURITY MATRIX                        │
│                                                                            │
│  Layer 1: Intent Policy ──► Pre-LLM prompt filtering                       │
│  Layer 2: Tool Approval ──► Human-in-the-Loop validation                   │
│  Layer 3: OAuth Gateway ──► Scoped token issuance (Intersection principle) │
│  Layer 4: Target Tool   ──► Strict identity signature & status checks      │
│  Layer 5: Audit Log     ──► Single trace correlated by Agent ID            │
└────────────────────────────────────────────────────────────────────────────┘

Deterministic Defense-in-Depth: Policy guards run before the model evaluates prompts, while Identity Gateways enforce scope checks inside the execution tool. Even if an LLM is hijacked via prompt injection, it remains constrained by the token's scope ceiling.
True Separation of Duties: Autonomous agents cannot self-approve or inherit excessive access, eliminating untracked administrative expansion.
Seamless Multi-Agent Delegation: RFC 8693 token exchange preserves the chain of custody across complex multi-agent workflows (User -> Supervisor -> Worker), providing clear attribution for downstream audits.
Resilient Architecture: Runtime LLM engine switching allows teams to hot-swap model backends (local or cloud) without disrupting active user sessions, token stores, or audit histories.

Overall implementation: Governed CUGA Agent Implementation

Below is a detailed structural representation of the complete governed agent architecture:

Implementing CUGA-Agent with Governed Identity transitions autonomous systems from opaque risk drivers into auditable enterprise assets:

No Shared Passwords/Keys: Every agent operates under its own client registration.
Cryptographic Bounds: Scope ceilings ensure agents can never exceed their approved boundaries, even when acting for privileged users.
Instant Containment: The strict-status kill switch invalidates active sessions in real time, preventing rogue executions.
Traceable Attribution: Unified audit logging tags every runtime action and lifecycle change with a consistent Agent ID.

Thanks for reading 🦉

[Boost]

Alain Airom (Ayrom) — Thu, 05 Nov 2026 20:58:23 +0000

KL3FT3Z

Oct 25

When Wording Changes Image Safety: A GigaChat Red-Team Case Study

#ai #webdev #machinelearning #redteam

7 min read

Don't asnwer to fake Dev.to (support) message

Alain Airom (Ayrom) — Thu, 05 Nov 2026 15:16:11 +0000

Hi,
Just a short one, since few days I receive the following email;

This is a fake money stilling message. Don't follow the steps and make a feedback to dev.to support.

Building a GitHub Contribution City: Transforming Developer History into a 3D Skyline

Alain Airom (Ayrom) — Tue, 03 Nov 2026 09:46:03 +0000

Build your contribution city!

Developer profiles are our digital resumes, but let's be honest: the standard 53-week GitHub contribution heatmap grid is starting to feel a bit flat. What if instead of looking at flat green squares, we could walk through a bustling cyberpunk metropolis, gaze up at towering skyscrapers built from our busiest commit weeks, and watch our coding history come alive in 3D?

In this post, we’ll explore GitHub Contribution City, a production-grade application built in Go that transforms any GitHub user's contribution history into a stunning 2D SVG skyline, a 3D isometric SVG, and an interactive WebGL 3D world powered by Three.js.

Before moving further, I want to clarify that the inspiration for this project comes from a blog post by Markus Eisele—a highly esteemed IBM colleague—which you can read here. Reading it gave me the idea to build this using IBM Bob. However, Markus is a true Java enthusiast who isn't particularly fond of languages like Python or JavaScript. Personally, I'm not a fan of Java :) so I asked Bob to adapt the codebase into Go instead. We all have our own brand of madness! 😵‍💫

Introduction & Project Inspiration

The initial spark for this project was inspired by Markus Eisele’s fantastic Quarkus Contribution City concept. While the original implementation was written in Java, we wanted to build a modern, lightning-fast, zero-dependency binary in Go that can run locally as a CLI tool, deploy as a high-performance web server, or run automatically via GitHub Actions.
Whether you want to generate a gorgeous SVG badge for your README.md, view your personal tech metropolis in 3D, or spin up an automated daily action that tracks your team's contributions, GitHub Contribution City turns your code commits into architectural marvels.

Detailed Architecture and Implementation

System Architecture

The application is structured around a clean modular pipeline: input configurations and remote API data feed into our core scene builder, which in turn delegates to specialized renderers.

Configuration Management (`pkg/config`)

The configuration manager handles environment variables, .env configuration files, and sensible defaults (such as defaulting to torvalds if no user is specified).

package config

import (
    "bufio"
    "os"
    "strconv"
    "strings"
)

// Config represents runtime configuration for GitHub Contribution City
type Config struct {
    GitHubToken    string `json:"github_token"`
    GitHubUsername string `json:"github_username"`
    Weeks          int    `json:"weeks"`
    MaxHeight      int    `json:"max_height"`
    Theme          string `json:"theme"`
    Port           int    `json:"port"`
    Host           string `json:"host"`
    OutputDir      string `json:"output_dir"`
    InputDir       string `json:"input_dir"`
}

// DefaultConfig returns default sensible parameters
func DefaultConfig() *Config {
    return &Config{
        GitHubToken:    "",
        GitHubUsername: "torvalds",
        Weeks:          53,
        MaxHeight:      14,
        Theme:          "github-dark",
        Port:           8088,
        Host:           "0.0.0.0",
        OutputDir:      "output",
        InputDir:       "input",
    }
}

// LoadConfig loads configuration from optional .env file and environment variables
func LoadConfig(envFile string) *Config {
    cfg := DefaultConfig()

    if envFile != "" {
        _ = loadDotEnv(envFile)
    } else if _, err := os.Stat(".env"); err == nil {
        _ = loadDotEnv(".env")
    }

    if val := os.Getenv("GITHUB_TOKEN"); val != "" {
        cfg.GitHubToken = strings.TrimSpace(val)
    }
    if val := os.Getenv("GITHUB_USERNAME"); val != "" {
        cfg.GitHubUsername = strings.TrimSpace(val)
    }
    if val := os.Getenv("CITY_WEEKS"); val != "" {
        if w, err := strconv.Atoi(val); err == nil && w >= 1 && w <= 53 {
            cfg.Weeks = w
        }
    }
    if val := os.Getenv("CITY_MAX_HEIGHT"); val != "" {
        if h, err := strconv.Atoi(val); err == nil && h >= 2 && h <= 50 {
            cfg.MaxHeight = h
        }
    }
    if val := os.Getenv("CITY_THEME"); val != "" {
        cfg.Theme = strings.TrimSpace(val)
    }
    if val := os.Getenv("PORT"); val != "" {
        if p, err := strconv.Atoi(val); err == nil && p > 0 {
            cfg.Port = p
        }
    }
    return cfg
}

func loadDotEnv(filepath string) error {
    file, err := os.Open(filepath)
    if err != nil {
        return err
    }
    defer file.Close()

    scanner := bufio.NewScanner(file)
    for scanner.Scan() {
        line := strings.TrimSpace(scanner.Text())
        if line == "" || strings.HasPrefix(line, "#") {
            continue
        }
        parts := strings.SplitN(line, "=", 2)
        if len(parts) == 2 {
            key := strings.TrimSpace(parts[0])
            val := strings.TrimSpace(parts[1])
            val = strings.Trim(val, `"'`)
            if os.Getenv(key) == "" {
                _ = os.Setenv(key, val)
            }
        }
    }
    return scanner.Err()
}

Fetching Contribution Calendars (`pkg/github`)

The GitHub client seamlessly switches between authenticated GraphQL API queries (supporting private and internal commits if a PAT is provided) and a robust HTML scraper fallback for public profiles.

package github

import (
    "bytes"
    "context"
    "encoding/json"
    "fmt"
    "io"
    "net/http"
    "time"
)

// ContributionDay represents one day's contributions
type ContributionDay struct {
    ContributionCount int       `json:"contributionCount"`
    ContributionLevel string    `json:"contributionLevel"`
    Date              string    `json:"date"`
    Weekday           int       `json:"weekday"`
    ParsedDate        time.Time `json:"-"`
}

// ContributionWeek represents a week of days (up to 7 days)
type ContributionWeek struct {
    ContributionDays []ContributionDay `json:"contributionDays"`
    FirstDay         string            `json:"firstDay"`
}

// ContributionCalendar is the complete calendar returned by GitHub
type ContributionCalendar struct {
    TotalContributions int                `json:"totalContributions"`
    Weeks              []ContributionWeek `json:"weeks"`
    Colors             []string           `json:"colors"`
    Username           string             `json:"username"`
}

// Client is a GitHub API client that can fetch contribution calendars
type Client struct {
    token      string
    httpClient *http.Client
    graphQLURL string
}

func NewClient(token string) *Client {
    return &Client{
        token: token,
        httpClient: &http.Client{
            Timeout: 15 * time.Second,
        },
        graphQLURL: "https://api.github.com/graphql",
    }
}

// FetchCalendar fetches contribution data using GraphQL or public scraping fallback
func (c *Client) FetchCalendar(ctx context.Context, username string) (*ContributionCalendar, error) {
    if c.token != "" {
        cal, err := c.fetchGraphQL(ctx, username)
        if err == nil && cal != nil {
            return cal, nil
        }
    }
    return c.fetchPublicCalendar(ctx, username)
}

Logarithmic City Layout Algorithm (`pkg/city`)

To prevent single hyper-active weeks from completely distorting the skyline scale, building heights are calculated using logarithmic scaling:

// Calculate building heights using logarithmic scale as specified in Quarkus Contribution City:
buildingHeights := make([]int, numWeeks)
for i, tot := range weeklyTotals {
    if tot == 0 || maxWeekly == 0 {
        buildingHeights[i] = 1
    } else {
        ratio := math.Log1p(float64(tot)) / math.Log1p(float64(maxWeekly))
        h := 1 + int(math.Round(ratio * float64(opts.MaxHeight-1)))
        if h < 1 { h = 1 }
        if h > opts.MaxHeight { h = opts.MaxHeight }
        buildingHeights[i] = h
    }
}

Main Application Entrypoint (`cmd/city/main.go`)

The main CLI dispatcher supports generating single SVGs, 3D isometric views, batch processing multiple usernames from an input folder, or starting an embedded HTTP server.

package main

import (
    "context"
    "encoding/json"
    "flag"
    "fmt"
    "log"
    "os"
    "path/filepath"
    "strings"
    "time"

    "github.com/alainairom/github-contribution-city/pkg/city"
    "github.com/alainairom/github-contribution-city/pkg/config"
    "github.com/alainairom/github-contribution-city/pkg/github"
    "github.com/alainairom/github-contribution-city/pkg/renderer"
)

func main() {
    var (
        serverMode = flag.Bool("server", false, "Start the web visualization server")
        username   = flag.String("user", "", "GitHub username to visualize")
        token      = flag.String("token", "", "GitHub Personal Access Token (PAT)")
        weeks      = flag.Int("weeks", 0, "Number of weeks to render (4 - 53)")
        maxHeight  = flag.Int("height", 0, "Maximum building height (3 - 30)")
        theme      = flag.String("theme", "", "Visual theme")
        format     = flag.String("format", "svg", "Export format: svg, iso, html, json, all")
        outFile    = flag.String("output", "", "Output filepath or directory")
        port       = flag.Int("port", 0, "HTTP server port")
        envFile    = flag.String("env", ".env", "Path to .env configuration file")
    )
    flag.Parse()

    cfg := config.LoadConfig(*envFile)
    if *username != "" { cfg.GitHubUsername = *username }
    if *token != "" { cfg.GitHubToken = *token }
    if *weeks > 0 { cfg.Weeks = *weeks }
    if *maxHeight > 0 { cfg.MaxHeight = *maxHeight }
    if *theme != "" { cfg.Theme = *theme }
    if *port > 0 { cfg.Port = *port }

    _ = os.MkdirAll(cfg.OutputDir, 0755)

    client := github.NewClient(cfg.GitHubToken)
    builder := city.NewBuilder()
    svgRenderer := renderer.NewSVGRenderer()
    htmlRenderer := renderer.NewHTMLRenderer()

    if *serverMode {
        runServer(cfg, client, builder, svgRenderer, htmlRenderer)
        return
    }

    fmt.Printf("🏙️  Generating Contribution City for '%s'...\n", cfg.GitHubUsername)
    ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
    defer cancel()

    cal, err := client.FetchCalendar(ctx, cfg.GitHubUsername)
    if err != nil {
        log.Fatalf("❌ Error fetching contribution data: %v", err)
    }

    scene := builder.Build(cal, city.Options{
        Weeks:     cfg.Weeks,
        MaxHeight: cfg.MaxHeight,
        Theme:     cfg.Theme,
    })

    outPath := filepath.Join(cfg.OutputDir, fmt.Sprintf("%s_city.svg", cfg.GitHubUsername))
    svgContent := svgRenderer.RenderSkyline(scene, cfg.Theme)
    _ = os.WriteFile(outPath, []byte(svgContent), 0644)
    fmt.Printf("✨ SVG Skyline successfully written to: %s\n", outPath)
}

Automation with GitHub Actions

To keep your contribution city continuously updated without manual intervention, the repository features an automated CI/CD and scheduling suite using GitHub Actions.
The workflow automation includes:

Continuous Integration (ci.yml): Runs linting checks, go vet, vulnerability scans via govulncheck, multi-OS unit tests (ubuntu, macos, windows), and cross-compilation on every push and pull request.

  name: Continuous Integration

  on:
    push:
      branches:
        - main
        - master
      paths-ignore:
        - 'Docs/**'
        - 'README.md'
        - 'output/**'
    pull_request:
      branches:
        - main
        - master

  permissions:
    contents: read

  jobs:
    lint:
      name: Lint & Code Style
      runs-on: ubuntu-latest
      steps:
        - name: Checkout Source Code
          uses: actions/checkout@v4

        - name: Setup Go
          uses: actions/setup-go@v5
          with:
            go-version: '1.24'
            cache: true

        - name: Verify Formatting
          run: |
            UNFORMATTED=$(gofmt -l . | grep -v '^vendor/' || true)
            if [ -n "$UNFORMATTED" ]; then
              echo "❌ Code format issues found in following files:"
              echo "$UNFORMATTED"
              exit 1
            fi
            echo "✅ All Go files formatted properly."

        - name: Run Go Vet
          run: go vet ./...

        - name: Run golangci-lint
          uses: golangci/golangci-lint-action@v6
          with:
            version: latest
            args: --timeout=5m
          continue-on-error: true

    vuln-scan:
      name: Security & Vulnerability Scan
      runs-on: ubuntu-latest
      steps:
        - name: Checkout Source Code
          uses: actions/checkout@v4

        - name: Setup Go
          uses: actions/setup-go@v5
          with:
            go-version: '1.24'
            cache: true

        - name: Run govulncheck
          uses: golang/govulncheck-action@v1
          with:
            go-version-input: '1.24'
            go-package: ./...

    ...

Daily Visualization Updates (update-city.yml): Powered by a scheduled cron trigger (03:17 UTC), this workflow fetches fresh contribution matrices for the repository owner and any tracked usernames listed inside the input/ folder, regenerates the respective skylines, and commits the updated graphics back to the repository automatically ([skip ci]).

  name: Daily Contribution City Update

  on:
    schedule:
      # Runs everyday at 03:17 UTC
      - cron: '17 3 * * *'
    workflow_dispatch:
      inputs:
        custom_user:
          description: 'Optional specific GitHub username to generate for'
          required: false
          default: ''
        theme:
          description: 'Theme to use (github-dark, github-light, cyberpunk, mono, neon, matrix)'
          required: false
          default: 'github-dark'

  permissions:
    contents: write

  jobs:
    update-visualizations:
      name: Regenerate Contribution Cities
      runs-on: ubuntu-latest
      steps:
        - name: Checkout Repository
          uses: actions/checkout@v4
          with:
            fetch-depth: 0

        - name: Set up Go
          uses: actions/setup-go@v5
          with:
            go-version: '1.24'
            cache: true

        - name: Build Contribution City CLI
          run: |
            go build -o bin/github-contribution-city cmd/city/main.go

        - name: Generate Visualizations
          env:
            GITHUB_TOKEN: ${{ secrets.CONTRIBUTION_TOKEN || secrets.GITHUB_TOKEN }}
          run: |
            mkdir -p output
            PRIMARY_USER="${{ inputs.custom_user || github.repository_owner }}"
            SELECTED_THEME="${{ inputs.theme || 'github-dark' }}"

            echo "🌆 Generating contribution city for primary user: ${PRIMARY_USER} (Theme: ${SELECTED_THEME})..."

            # 1. Generate primary user 2D SVG Skyline
            ./bin/github-contribution-city \
              -user "${PRIMARY_USER}" \
              -weeks 53 \
              -height 14 \
              -theme "${SELECTED_THEME}" \
              -format svg \
              -output "output/contribution-city.svg"

            # 2. Generate primary user 3D Isometric SVG
            ./bin/github-contribution-city \
              -user "${PRIMARY_USER}" \
              -weeks 53 \
              -height 14 \
              -theme "cyberpunk" \
              -format iso \
              -output "output/contribution-city-3d.svg"

            # 3. Generate primary user 3D Interactive HTML
            ./bin/github-contribution-city \
              -user "${PRIMARY_USER}" \
              -weeks 53 \
              -height 14 \
              -theme "${SELECTED_THEME}" \
              -format html \
              -output "output/contribution-city.html"

            # 4. If tracked users exist in input/ directory, batch process them
            if [ -d "input" ] && [ "$(ls -A input 2>/dev/null)" ]; then
              echo "📁 Processing batch users from input directory..."
              ./bin/github-contribution-city -batch input/ -output output/
            fi

        - name: Commit and Push Changes
          run: |
            git config --global user.name "github-actions[bot]"
            git config --global user.email "github-actions[bot]@users.noreply.github.com"
            git add output/

            if git diff --staged --quiet; then
              echo "ℹ️  No changes detected in contribution calendars. Skipping commit."
            else
              git commit -m "chore(city): auto-update contribution city visualizations [skip ci]"
              git push
              echo "✅ Successfully committed and pushed updated visualizations."
            fi

        - name: Upload Output Artifacts
          uses: actions/upload-artifact@v4
          with:
            name: contribution-city-visualizations
            path: output/
            retention-days: 30

Container & Release Management (docker.yml & release.yml): Automatically builds multi-arch container images (linux/amd64, linux/arm64) for the GitHub Container Registry (ghcr.io) and packages cross-compiled release binaries whenever version tags are pushed.

Conclusion

Building GitHub Contribution City was a fun journey merging developer metrics with procedural city generation and graphics rendering. By leveraging Go, we achieved a blazing fast binary that can be embedded directly into CI/CD pipelines, containerized for Docker/Podman, or served live via an interactive Three.js 3D browser experience.

Thanks for reading 🌃

Building an End-to-End Multilingual Speech and LLM Pipeline with IBM Granite Speech 5.0-2B and Bob

Alain Airom (Ayrom) — Mon, 02 Nov 2026 10:55:32 +0000

Testing the latest granite 🪨 speech 5 capabilities on Apple Silicon.

Introduction

As speech-to-text workflows transition toward more deeply integrated architectures, developers need models that don't just transcribe audio mechanically, but understand context, intent, and translation across multiple languages natively. Enter IBM Granite Speech 5.0-2B (ibm-granite/granite-speech-5.0-2b)—a high-performance, open-weights model leveraging a two-pass design (a CTC speech encoder coupled directly with a Granite-4.0-1B LLM backbone). Released under the permissive Apache 2.0 licence, it delivers state-of-the-art multilingual Automatic Speech Recognition (ASR) and Speech Translation (AST) across English, French, German, Spanish, Portuguese, Italian, and Japanese.

In this guide, we will walk through how to construct a complete, production-ready end-to-end demonstration application. We'll use Bob—our local developer workspace tool—to orchestrate the project structure, and connect Granite Speech's powerful transcription and translation features directly to downstream LLM backends (such as Ollama, llama.cpp, or OpenAI) via a unified Gradio web UI.

IBM Granite Speech 5

Excerpt from Hugging Face model card

Granite Speech 5.0 2B is a compact and efficient speech-language model, specifically designed for multilingual automatic speech recognition (ASR) and bidirectional automatic speech translation (AST) for English, French, German, Spanish, Portuguese and Japanese.

The model was trained on 390,000 hours of audio from public corpora for ASR and AST as well as synthetic datasets tailored to support Japanese ASR, keyword-biased ASR and speech translation. Granite Speech 5.0 2B was trained by modality aligning an intermediate checkpoint of granite-4.0-1b-base to speech on publicly available open source corpora containing audio inputs and text targets. Compared to its predecessor granite-speech-4.1-2b, this model has the same parameter count and provides additional capabilities and improvements:

Higher transcription accuracy for English short-form ASR and increased robustness to noise, reverberation and distant microphone speech
Supports input language id prompting for multilingual ASR
Improved long-form and conversational speech translation
Increased keyword list size (up to 2000) for multilingual keyword-biased ASR and AST

The mode card page proposes several sample test codes and implementations runnable as is.

Sample code using Transformers;

  # Usage with transformers
  # pip install transformers torchaudio soundfile

  import torch
  import torchaudio
  from huggingface_hub import hf_hub_download
  from transformers import AutoModelForSpeechSeq2Seq, AutoProcessor

  device = "cuda" if torch.cuda.is_available() else "cpu"

  model_name = "ibm-granite/granite-speech-5.0-2b"
  processor = AutoProcessor.from_pretrained(model_name)
  tokenizer = processor.tokenizer
  model = AutoModelForSpeechSeq2Seq.from_pretrained(
      model_name, device_map=device, torch_dtype=torch.bfloat16
  )

  # Load audio
  audio_path = hf_hub_download(repo_id=model_name, filename="multilingual_sample.wav")
  wav, sr = torchaudio.load(audio_path, normalize=True)
  assert wav.shape[0] == 1 and sr == 16000  # mono, 16kHz

  # Create text prompt
  user_prompt = "<|audio|>transcribe the speech with proper punctuation and capitalization."
  chat = [
      {"role": "user", "content": user_prompt},
  ]
  prompt = tokenizer.apply_chat_template(chat, tokenize=False, add_generation_prompt=True)

  # Run the processor + model
  model_inputs = processor(prompt, wav, device=device, return_tensors="pt").to(device)
  model_outputs = model.generate(
      **model_inputs, max_new_tokens=200, do_sample=False, num_beams=1
  )

  # Transformers includes the input IDs in the response
  num_input_tokens = model_inputs["input_ids"].shape[-1]
  new_tokens = model_outputs[0, num_input_tokens:].unsqueeze(0)
  output_text = tokenizer.batch_decode(
      new_tokens, add_special_tokens=False, skip_special_tokens=True
  )
  print(f"STT output = {output_text[0]}")

Sample code using vLLM;

# usage vllm
# pip install vllm

from transformers import AutoTokenizer
from vllm import LLM, SamplingParams
from vllm.assets.audio import AudioAsset

model_id = "ibm-granite/granite-speech-5.0-2b"
tokenizer = AutoTokenizer.from_pretrained(model_id)

def get_prompt(question: str, has_audio: bool):
    """Build the input prompt to send to vLLM."""
    if has_audio:
        question = f"<|audio|>{question}"
    chat = [
        {
            "role": "user",
            "content": question
        }
    ]
    return tokenizer.apply_chat_template(chat, tokenize=False)

model = LLM(
    model=model_id,
    max_model_len=2048, # This may be needed for lower resource devices.
    limit_mm_per_prompt={"audio": 1},
)

question = "can you transcribe the speech into a written format?"
prompt_with_audio = get_prompt(
    question=question,
    has_audio=True,
)
audio = AudioAsset("mary_had_lamb").audio_and_sample_rate

inputs = {
    "prompt": prompt_with_audio,
    "multi_modal_data": {
        "audio": audio,
    }
}

outputs = model.generate(
    inputs,
    sampling_params=SamplingParams(
        temperature=0.0,
        max_tokens=64,
    ),
)
print(f"Audio Example - Question: {question}")
print(f"Generated text: {outputs[0].outputs[0].text}")

Sample Application Features and Functions

The application implements the following features;

Speech Transcription: upload a recorded audio file or speak directly into your microphone. The system converts speech to accurate written text using IBM's Granite Speech 5.0-2B model — trained specifically for high-quality transcription in real-world conditions.

Multilingual Translation: spoken audio in English, French, German, Spanish, Portuguese, or Japanese is automatically translated into any other supported language. Ideal for international teams, multilingual content production, and cross-border customer support.

The example above is a sample translation from English to French. Other languages are available too, such as German for another test;

  Der Bezirksarzt näherte sich mit der Gleichgültigkeit eines Mannes, der es gewohnt ist, die Hälfte seiner Zeit unter den Toten zu verbringen. Er hob dann das über das Gesicht gelegte Blatt und öffnete nur die Lippen.

AI-Assisted Analysis: once a transcript is produced, it is forwarded to an AI language model with your custom instruction — "summarise this meeting," "extract all action items," "identify the customer's main concern." The AI response appears alongside the transcript.

One-Click Full Pipeline: the Full Pipeline tab combines transcription and AI analysis into a single button click. Provide audio and an instruction; the system handles everything end-to-end and saves a structured result file automatically.

Live Configuration: switch AI backends, change models, or adjust processing settings from within the interface — no restart required. Configurations take effect immediately, enabling rapid A/B comparison of different AI backends.
Hardware Intelligence: the system automatically detects the capabilities of the host machine — GPU type, available memory, CPU cores — and self-optimises its processing parameters accordingly, ensuring the best performance on every device.

Architecture

The entire pipeline is structured to ensure modularity, separating hardware capability detection, audio processing, configuration management, and multi-backend LLM interfacing.

Implementation

Project Layout & Core Setup

Project's file and organization:

granite-speech5-test/
├── app.py                  # Gradio demo application (entry point)
├── requirements.txt        # Python dependencies
├── .env.example            # Environment variable template
├── config/
│   └── config.yaml         # Application configuration
├── src/
│   ├── speech_processor.py # Granite Speech 5.0-2B wrapper
│   ├── hardware_detector.py# Hardware capability detection
│   ├── llm_client.py       # Multi-backend LLM abstraction
│   └── config_manager.py   # Configuration management
├── tests/
│   ├── test_speech_processor.py
│   ├── test_hardware_detector.py
│   ├── test_llm_client.py
│   └── test_config_manager.py
├── scripts/
│   ├── setup.sh            # One-time environment setup
│   ├── start.sh            # Launch app in background
│   ├── stop.sh             # Graceful shutdown
│   └── cleanup.sh          # Remove artefacts / venv
├── input/                  # Put audio files here
├── output/                 # JSON results written here
└── Docs/
    ├── Quickstart.md
    └── Architecture.md

Hardware Detection & Optimization Advisor

Before loading models, the application queries host resources to assign the optimal compute device (CUDA GPU, Apple Silicon MPS, or CPU) and precision dtype:

"""
Hardware capability detection and optimisation advisor.

Interrogates the host machine for CPU cores, RAM, GPU VRAM and Apple Silicon
presence, then returns a structured HardwareProfile and suggested model-loading
parameters.

No external heavy dependencies – falls back gracefully when optional packages
(torch, psutil) are unavailable so that the module can be imported during the
install phase.
"""

from __future__ import annotations

import logging
import os
import platform
import subprocess
import sys
from dataclasses import dataclass, field
from typing import Optional

...

# Optional torch import at module level so tests can patch it cleanly.
try:
    import torch  # type: ignore
except ImportError:
    torch = None  # type: ignore

logger = logging.getLogger(__name__)

Core Speech Processing Wrapper

The core speech module (GraniteSpeechProcessor) handles raw audio files, normalizes inputs to 16 kHz mono waveforms, and executes inference using the proper prompt templates specified by the model card:

"""
IBM Granite Speech 5.0-2B — Core speech processing module.

Model reference: https://huggingface.co/ibm-granite/granite-speech-5.0-2b
Architecture: two-pass design (CTC speech encoder → Granite-4.0-1B LLM backbone)
Supported tasks: ASR (transcription) and AST (translation) in 6 languages.

Usage constraints (Apache 2.0 licence, no additional restrictions):
- Audio must be 16 kHz mono; longer recordings should be split into ≤2-minute segments.
- Prompt tokens use the <|audio|> placeholder which the processor replaces with audio embeddings.

Performance guide
-----------------
On CPU (M-series or x86) a 2-second clip takes ~60-120 s with float32 because the
Granite-4.0-1B LLM backbone must run greedy-decode token-by-token.  To reach
real-time speed, one of the following is needed:

  1. CUDA GPU (any card ≥ 4 GB VRAM) → bfloat16 → 1-5× real-time
  2. Apple Silicon with MPS           → float16  → ~5× real-time
  3. INT-8 dynamic quantisation       → ~2× CPU speedup (built-in, no GPU)
  4. Reduce max_new_tokens            → linear speedup (safe for short clips)
  5. ONNX runtime (WebGPU / CoreML)   → best CPU/edge throughput

All optimisation flags are controlled via config/config.yaml [speech.perf].
"""

from __future__ import annotations

import logging
import time
from contextlib import contextmanager
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional

import numpy as np
import torch

from src.speech_processor import GraniteSpeechProcessor, PerfConfig
from src.hardware_detector import detect_hardware

hw = detect_hardware()

# Initialise processor with performance profile
processor = GraniteSpeechProcessor(
    device=hw.recommended_device,
    perf=PerfConfig(int8_quantize=False, profile=True)
)
...
# Transcribe audio file
result = processor.transcribe("input/meeting.wav", language="en", verbatim=False)
print("Transcript:", result.text)
print(f"Real-Time Factor (RTFx): {result.realtime_factor:.1f}×")

# Translate speech (e.g., French audio to English text)
translation = processor.translate("input/french_speech.wav", source_language="fr", target_language="en")
print("Translation:", translation.text)

Downstream Processing with Multi-Backend LLMs

Once a transcript is extracted, you can pipe it directly into local or cloud LLM backends using a uniform interface:

"""
LLM backend abstraction layer.

Provides a unified LLMClient interface over multiple local and cloud backends:

  Local
  -----
  • Ollama   – http://localhost:11434 (default)
  • LM Studio – http://localhost:1234  (OpenAI-compatible endpoint)
  • llama.cpp server – http://localhost:9931  (project rule: llama.cpp port)

  Cloud
  -----
  • OpenAI   – https://api.openai.com/v1
  • Anthropic – Claude via Messages API
  • IBM watsonx.ai – https://us-south.ml.cloud.ibm.com (optional)

All backends expose the same .chat(messages) → str interface so the
application layer never needs to know which backend is active.

Usage
-----
>>> from src.llm_client import build_llm_client
>>> llm = build_llm_client(backend="ollama", model="granite3.3:8b")
>>> llm.chat([{"role": "user", "content": "Summarise this transcript: …"}])
"""

from __future__ import annotations

import json
import logging
import time
import urllib.request  # imported at module level for testability
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Optional

logger = logging.getLogger(__name__)

...

# Build a local Ollama client instance pointing to Granite or Llama models
llm = build_llm_client(backend="ollama", model="granite4.2:latest")

response = llm.chat([
    {
        "role": "user", 
        "content": f"Transcript:\n{result.text}\n\nInstruction: Summarise the key action items concisely."
    }
])

print("LLM Reply:", response.text)
print(f"Latency: {response.latency_s:.2f}s | Tokens used: {response.input_tokens} → {response.output_tokens}")

End-to-End Application Process

The user interface combines transcription, translation, and LLM interaction tabs into a unified local control center (app.py).

"""
IBM Granite Speech 5.0-2B — End-to-End Demo Application (Gradio UI).

Tabs
────
1. Transcribe   – upload or record audio → Granite Speech → text output
2. Translate    – speech translation between supported language pairs
3. LLM Interact – send a transcript to the configured LLM backend
4. Config       – inspect and switch settings at runtime
5. Hardware     – display host hardware profile

Run
───
    python app.py
    # or via scripts/start.sh
"""

from __future__ import annotations

import json
import logging
import os
import time
import urllib.request
import urllib.error
from datetime import datetime
from pathlib import Path
from typing import Optional, Tuple

import gradio as gr

from src.config_manager import get_config, reload_config
from src.hardware_detector import detect_hardware
from src.llm_client import build_llm_client
from src.speech_processor import (
    GraniteSpeechProcessor,
    PerfConfig,
    SpeechResult,
    SUPPORTED_LANGUAGES,
)

logger = logging.getLogger(__name__)

...

# ── Entry point ───────────────────────────────────────────────────────────────

def main() -> None:
    cfg = get_config()

    # Ensure output directory exists
    Path(cfg.app.output_dir).mkdir(parents=True, exist_ok=True)

    # Warm up hardware detection
    logger.info("Detecting hardware …")
    hw = _get_hardware_profile()
    logger.info("\n%s", hw.summary_str())

    demo = build_ui()

    # Build SSL kwargs — when cert/key are present the app serves HTTPS,
    # which is required for the browser microphone API to work on localhost.
    ssl_kwargs: dict = {}
    root = Path(__file__).parent
    certfile = cfg.app.ssl_certfile or str(root / "certs" / "cert.pem")
    keyfile = cfg.app.ssl_keyfile or str(root / "certs" / "key.pem")
    if Path(certfile).exists() and Path(keyfile).exists():
        ssl_kwargs = {
            "ssl_certfile": certfile,
            "ssl_keyfile": keyfile,
            # ssl_verify=False tells Gradio's internal httpx health-check to
            # skip cert validation for its own loopback call — required when
            # using a self-signed certificate, otherwise Gradio crashes on
            # startup with CERTIFICATE_VERIFY_FAILED before the browser even
            # connects.
            "ssl_verify": False,
        }
        scheme = "https"
        logger.info("TLS enabled — serving over HTTPS (microphone will work in browser)")
    else:
        scheme = "http"
        logger.warning(
            "No TLS cert found at %s — microphone requires HTTPS.\n"
            "  Run: openssl req -x509 -newkey rsa:2048 -nodes "
            "-keyout certs/key.pem -out certs/cert.pem -days 3650 "
            '-subj "/CN=localhost" -addext "subjectAltName=DNS:localhost,IP:127.0.0.1"',
            certfile,
        )

    logger.info("App URL: %s://localhost:%d", scheme, cfg.app.server_port)
    demo.launch(
        server_port=cfg.app.server_port,
        share=cfg.app.share,
        server_name="0.0.0.0",
        **ssl_kwargs,
    )


if __name__ == "__main__":
    main()

Launch the app via your terminal and use the full pipeline.

Conclusion

By combining IBM Granite Speech 5.0-2B with a modular application architecture, developers gain an either a cloud based or an air-gappable environment, highly accurate pipeline capable of bridging audio input with advanced reasoning engines[cite: 5]. Whether you're targeting low-latency GPU edge nodes, running offline speech translation on Apple Silicon, or hooking local models into Ollama, this reference pattern gives you total data sovereignty and performance observability out of the box.

Thanks for reading 🎙️

Turning Books into Actionable Agent Skills: A Deep Dive into book-to-skill (and Docling)

Alain Airom (Ayrom) — Thu, 29 Oct 2026 18:44:46 +0000

Implementing and testing book-to-skill with a personal touch!

Introduction

As developers, we frequently buy comprehensive technical books or accumulate dense internal documentation repositories, read them once, and struggle to retain or quickly reference specific frameworks later. Traditional approaches—like searching raw PDFs or dumping entire books into an LLM context window—either fail to yield precise answers or suffer from the "discovery loop tax," wasting tokens and causing hallucinations.

book-to-skill completely transforms this workflow by distilling any technical book, document folder, or collection of sources into a structured, on-demand agent skill (SKILL.md and modular chapter files) compatible with cross-agent standards like GitHub Copilot CLI, Claude Code, OpenCode and last but surely not least, IBM Bob.

What immediately sparked my interest in this project is its intelligent extraction strategy, specifically its integration of docling for in-depth document information extraction. While standard text extractors struggle with technical books filled with complex layouts, docling preserves markdown tables, formulas, and code blocks seamlessly (~1.5s/page), ensuring that code examples and technical structures aren't lost in translation.

`book-to-skill` Core Concepts

Image from github repository book-to-skill

To truly appreciate the bridge between document extraction and agent intelligence, it helps to examine the core philosophy and mechanics outlined in the book-to-skill project specification:

The Problem It Solves: Technical books, architecture decision records, and internal documentation are read once and forgotten. Standard workarounds—such as keyword-searching PDFs, asking an unguided LLM to recall contents, or maintaining static notes—frequently lead to hallucinations, token fatigue, or abandoned documents.
The "Discovery Loop Tax": Traditional PDF-reading agents waste resources by constantly re-navigating table-of-contents and reprocessing entire files on every turn. By paying the structural synthesis cost once at conversion time, book-to-skill achieves 24× to 51× fewer tokens per query compared to full context dumps.
Universal Format and Agent Support: The tool transforms files (PDF, EPUB, DOCX, Markdown, HTML, RTF, MOBI) into standard-compliant agent skills (SKILL.md) natively supported across major development hosts including GitHub Copilot CLI, Claude Code, Amp, Codex, Hermes Agent, OpenCode, and OpenClaw.
Granular Artifact Layout: Generated skills are neatly partitioned into on-demand components:
- SKILL.md: Master index and core frameworks (~4,000 tokens).
- chapters/: Individual chapter files loaded strictly on-demand (~1,000 tokens each).
- glossary.md, patterns.md, and cheatsheet.md: Alphabetical key terms, design patterns, and quick-reference decision heuristics.
Privacy and Local-First Execution: Processing runs entirely on the user's local machine, ensuring full data sovereignty and adherence to personal or enterprise privacy policies.

Implementation

Disclaimer: While book-to-skill is fully functional and ready to use straight out of the box, my primary goal here was to wrap its powerful extraction capabilities into a streamlined, user-friendly interface. This implementation serves as a personal sandbox to bring my own vision of automated document intelligence and local-first agent workflows to life.

To make book-to-skill accessible via a modern web interface, I designed a robust, local-first FastAPI backend that wraps the extraction pipeline and orchestrates an automated two-phase processing pipeline:

Phase 1 (Extraction): Runs the core extraction script asynchronously via a non-blocking subprocess, directing output to an isolated per-job directory (output/<job_id>/).
Phase 2 (Skill Generation): Dispatches LLM-based skill generation to a thread pool (ThreadPoolExecutor), producing core mental models, chapter summaries, glossaries, patterns, and cheatsheets.

Architecture Schema

Two-Phase Processing Pipeline

The processing pipeline functions as the following;

User selects file + options (generate_skill: true/false)
         │
         ▼
POST /api/process { file_path, mode, skill_name, generate_skill }
         │
         ▼
FastAPI creates job_id, stores {
  status: "pending",
  skill_status: null
}
         │
         ▼
─────────────────────────────────────────────
PHASE 1 — EXTRACTION (asyncio subprocess)
─────────────────────────────────────────────
  background task sets BOOK_SKILL_WORKDIR=output/<job_id>/
  asyncio.create_subprocess_exec(extract.py …)
         │
         ├─ stdout → job["stdout"]
         ├─ stderr → job["stderr"]
         └─ on exit (rc=0):
              parse output/<job_id>/metadata.json
              read  output/<job_id>/full_text.txt
              job["metadata"] / ["extracted_text"] populated
              job["status"] = "done"
         │
         ▼
─────────────────────────────────────────────
PHASE 2 — SKILL GENERATION (thread pool)
─────────────────────────────────────────────
  (only if generate_skill=True and extraction succeeded)
  job["skill_status"] = "running"

  loop.run_in_executor(thread_pool, generate_skill, …)
         │
         ├─ probe LLM endpoint (8 s timeout)
         │    reachable  → use _chat() for each artifact
         │    unreachable → use stub generators
         │
         ├─ generate per-chapter summaries  → chapters/ch01-*.md …
         ├─ generate glossary               → glossary.md
         ├─ generate patterns               → patterns.md
         ├─ generate cheatsheet             → cheatsheet.md
         └─ generate master SKILL.md        → SKILL.md
              job["skill_status"] = "done" | "error"
              job["skill_result"] = {
                skill_name, skill_dir, files[], chapters,
                generated_at, llm_available, error
              }
         │
         ▼
Frontend polls GET /api/jobs/{id} every 1.5 s
 status == "done" → populate Results tab, switch tab
 skill_status == "running" → keep polling + refresh skill files card live
 skill_status == "done" → render "Generated Skill Files" card
         │
         ▼
GET /api/jobs/{id}/skill
 ├─ _resolve_skill_dir():
 │    1. skill_result["skill_dir"] if set and exists on disk  (normal path)
 │    2. scan output/<job_id>/skill/*/  for first subdir      (server-restart fallback)
 ├─ auto-heal: skill_status "running" → "done" if .md files found on disk
 ├─ synthesise skill_result from disk when skill_result was null
 │    (llm_available = null  →  UI shows "🔍 recovered from disk")
 └─ return { skill_status, skill_result, files[] }

GET /api/jobs/{id}/skill/download/…  → serve individual .md files
 └─ also uses _resolve_skill_dir() — works without skill_result in memory

Application Architecure and Code Excerpts

Below is an excerpt from the FastAPI backend (backend/main.py) showcasing how extraction is executed asynchronously using asyncio.create_subprocess_exec while maintaining isolated per-job working directories via BOOK_SKILL_WORKDIR:

"""
Book-to-Skill Web Application Backend
FastAPI server that wraps the book-to-skill library's extraction pipeline
and triggers LLM-based skill generation after extraction completes.
"""

import asyncio
import json
import logging
import os
import sys
import uuid
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
from pathlib import Path
from typing import List, Optional

from fastapi import FastAPI, HTTPException, BackgroundTasks, UploadFile, File
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse, StreamingResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel

# ─────────────────────────────── Helpers (time) ──────────────────────────────


def _now_iso() -> str:
    """Return current UTC time as an ISO-8601 string."""
    return datetime.now(timezone.utc).isoformat()


# ─────────────────────────────── Logging ─────────────────────────────────────

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
logger = logging.getLogger("book_to_skill_app")

# ─────────────────────────────── Config ──────────────────────────────────────

BASE_DIR = Path(__file__).resolve().parent.parent
INPUT_DIR = BASE_DIR / "input"
OUTPUT_DIR = BASE_DIR / "output"
SKILLS_DIR   = Path(os.environ.get("SKILLS_DIR", Path.home() / ".agents" / "skills"))
_thread_pool = ThreadPoolExecutor(max_workers=4)
BOOK_TO_SKILL_REPO = BASE_DIR / "book_to_skill_repo"

# File-size cap for uploads — configurable via UPLOAD_MAX_MB in .env (0 = unlimited)
_upload_max_mb = int(os.environ.get("UPLOAD_MAX_MB", "100"))
MAX_UPLOAD_BYTES = _upload_max_mb * 1024 * 1024 if _upload_max_mb > 0 else 0

# Supported upload extensions (mirrors what book-to-skill accepts)
ALLOWED_UPLOAD_EXTENSIONS = {
    ".pdf", ".epub", ".docx", ".html", ".htm", ".rtf",
    ".mobi", ".azw", ".azw3", ".txt", ".md", ".rst", ".adoc",
}

# Ensure required directories exist
INPUT_DIR.mkdir(exist_ok=True)
OUTPUT_DIR.mkdir(exist_ok=True)

# In-memory job store  {job_id: JobResult}
_jobs: dict[str, dict] = {}

# ─────────────────────────────── Models ──────────────────────────────────────

class ProcessRequest(BaseModel):
    file_path: str
    mode: str = "text"                 # "text" | "technical"
    skill_name: Optional[str] = None
    output_dir: Optional[str] = None
    generate_skill: bool = True        # set False to skip LLM skill generation


class DirectoryListRequest(BaseModel):
    directory: str = str(INPUT_DIR)


class JobStatus(BaseModel):
    job_id: str
    status: str                  # "pending" | "running" | "done" | "error"
    file_path: str
    started_at: Optional[str] = None
    finished_at: Optional[str] = None
    stdout: str = ""
    stderr: str = ""
    extracted_text: Optional[str] = None
    metadata: Optional[dict] = None
    error: Optional[str] = None
    skill_status: Optional[str] = None   # "pending" | "running" | "done" | "error" | None
    skill_result: Optional[dict] = None  # SkillGenerationResult.to_dict()


# ─────────────────────────────── App ─────────────────────────────────────────
...

async def _run_extraction(
    job_id: str,
    file_path: Path,
    mode: str,
    skill_name: Optional[str],
    generate_skill: bool = True,
) -> None:
    _jobs[job_id]["status"] = "running"
    _jobs[job_id]["started_at"] = _now_iso()

    try:
        script = _resolve_script()
    except FileNotFoundError as exc:
        _jobs[job_id]["status"] = "error"
        _jobs[job_id]["error"] = str(exc)
        return

    job_workdir = OUTPUT_DIR / job_id
    job_workdir.mkdir(parents=True, exist_ok=True)

    cmd: List[str] = [sys.executable, str(script), str(file_path), "--mode", mode]

    proc = await asyncio.create_subprocess_exec(
        *cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        env={
            **os.environ,
            "PYTHONIOENCODING": "utf-8",
            "BOOK_SKILL_WORKDIR": str(job_workdir),
        },
    )
    stdout_bytes, stderr_bytes = await asyncio.wait_for(proc.communicate(), timeout=300)
    # ... handles completion and triggers Phase 2 skill generation ...

Furthermore, the skill generator (backend/skill_generator.py) includes a graceful fallback mechanism and model probing to ensure that even when an LLM endpoint (such as Ollama or llama.cpp) is offline, the application still outputs functional stub structures:

"""
Skill Generator
===============
Reads extraction output (full_text.txt + metadata.json) produced by the
book-to-skill extractor and calls an OpenAI-compatible LLM to generate a
structured agent skill following the SKILL.md spec:

  output/<job_id>/skill/<skill_name>/
    SKILL.md          – master index + core frameworks (~4 000 tokens)
    glossary.md       – alphabetical key-term definitions
    patterns.md       – techniques & design patterns
    cheatsheet.md     – decision rules & quick-reference tables
    chapters/
      ch01-<slug>.md  – per-chapter summary
      ...

The generator degrades gracefully when no LLM is configured: it produces
stub files so the application always has something to show.

Environment variables (all optional — defaults work with Ollama):
  LLM_BASE_URL  – OpenAI-compatible base URL (default: http://localhost:11434/v1)
  LLM_API_KEY   – API key (default: "ollama" — Ollama ignores it)
  LLM_MODEL     – Model name (default: llama3.2:latest).
                  Must exactly match a model installed in Ollama (use
                  `ollama list` to check). Legacy alias "llama3" was removed
                  from Ollama; the current default is "llama3.2:latest".
  LLM_TIMEOUT   – Per-request timeout in seconds (default: 120)
"""

from __future__ import annotations

import json
import logging
import os
import re
import textwrap
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import List, Optional, Tuple

logger = logging.getLogger("book_to_skill_app.skill_generator")

# ─────────────────────────────── LLM config ───────────────────────────────────
# Values are read lazily on every call via _cfg() so that editing .env while
# the server is running takes effect immediately — no restart required.

from dataclasses import dataclass as _dataclass  # local alias avoids shadowing

@_dataclass(frozen=True)
class _LLMConfig:
    base_url: str
    api_key:  str
    model:    str
    timeout:  int


def _cfg() -> _LLMConfig:
    """Return the current LLM configuration, re-reading .env on every call.

    Uses python-dotenv with override=True so values written to .env after
    server startup are immediately visible to the next request without a
    restart.  The .env file is located relative to this file's project root.
    """
    from dotenv import load_dotenv
    from pathlib import Path as _Path
    _env_path = _Path(__file__).resolve().parent.parent / ".env"
    load_dotenv(_env_path, override=True)
    return _LLMConfig(
        base_url = os.environ.get("LLM_BASE_URL", "http://localhost:11434/v1"),
        api_key  = os.environ.get("LLM_API_KEY",  "ollama"),
        model    = os.environ.get("LLM_MODEL",    "llama3.2:latest"),
        timeout  = int(os.environ.get("LLM_TIMEOUT", "120")),
    )

# ─────────────────────────────── Result dataclass ─────────────────────────────

...
def _llm_available() -> bool:
    """Quick probe: returns True when the LLM endpoint answers a tiny request."""
    try:
        models_req = urllib.request.Request(
            f"{LLM_BASE_URL.rstrip('/')}/models",
            headers={"Authorization": f"Bearer {LLM_API_KEY}"},
            method="GET",
        )
        with urllib.request.urlopen(models_req, timeout=8):
            pass  
        return True
    except Exception as exc:
        logger.debug("LLM server not reachable: %s", exc)
        return False

Application's Configurations

The application can provide thresholds for documents to be processed, as well as using LLMs, through an environment configuration.

# ─── Book-to-Skill Web App — Environment Configuration ───────────────────────
# Copy this file to .env and fill in your values.

# Host and port the FastAPI server listens on
APP_HOST=0.0.0.0
APP_PORT=8080

# Upload size limit in MB (default: 100). Set to 0 for unlimited.
# UPLOAD_MAX_MB=100

# Override where generated skills are stored (default: ~/.agents/skills)
# SKILLS_DIR=/path/to/custom/skills

# Override the path to the book_to_skill_repo clone
# BOOK_TO_SKILL_REPO=/path/to/book-to-skill

# Set to "development" to enable FastAPI auto-reload
ENV=production

# ─── LLM / Skill generation ───────────────────────────────────────────────────
# OpenAI-compatible endpoint for skill generation.
# Default points to local Ollama.  Change LLM_BASE_URL for llama.cpp:
#   LLM_BASE_URL=http://localhost:9931/v1
# Or any OpenAI-compatible service (e.g. OpenAI, Groq, Mistral).
LLM_BASE_URL=http://localhost:11434/v1
LLM_API_KEY=ollama
# LLM_MODEL must exactly match a model installed in Ollama (run `ollama list`).
# "llama3" is a legacy alias removed in newer Ollama releases — use "llama3.2:latest".
LLM_MODEL=llama3.2:latest
# Per-LLM-request timeout in seconds (default: 120)
# LLM_TIMEOUT=120

API Accessibility

The original code could be called through command line, so the provided application provides also APIs to be call for instance throgh curl commands.

Processing and Test Outputs

When the backend processes an input document through its two-phase pipeline—combining Docling or format-specific parsers with LLM-powered generation—it builds a structured output hierarchy. Below is an example of a generated chapter summary illustrating how technical sections and core ideas are captured:

# Chapter 30 — Chapter 15: Direct Connect Public Virtual Interface (p. 303)

> **Note:** This file was originally generated as a stub because the LLM detection
> mechanism returned `False` even though Ollama was running.  The root cause has
> been fixed (see summary below); this file has been updated to reflect that Ollama
> **is** detected and available.

---

## LLM Detection Status

| Field | Value |
|-------|-------|
| `llm_available` | **`true`** |
| LLM endpoint | `http://localhost:11434/v1` |
| Model detected | `llama3.2:latest` |
| Detection method | Two-step probe (GET /models → POST /chat/completions) |

---

## Chapter Summary

**Chapter 15: Direct Connect — Public Virtual Interface**

This chapter covers the AWS Direct Connect Public Virtual Interface (VIF), which
provides dedicated private connectivity from on-premises infrastructure to AWS
public services (S3, DynamoDB, CloudFront, etc.) without traversing the public internet.

### Key Concepts

- **Public VIF** — a logical interface over a Direct Connect link that advertises
  AWS public IP prefixes via BGP; traffic to AWS public services exits through
  the Direct Connect circuit rather than over NAT/internet.
- **BGP peering** — each VIF requires a BGP session; the customer advertises their
  public IP prefixes and AWS advertises all AWS public service CIDRs.
- **Route filtering** — AWS enforces strict prefix filters; only globally routable
  (non-RFC 1918) customer prefixes are accepted.
- **Multi-region reach** — a single Public VIF can reach AWS public endpoints in
  **all AWS regions** (not just the Direct Connect location region).
- **Hosted vs. dedicated** — Public VIFs can be provisioned on both dedicated
  (1 Gbps / 10 Gbps) and hosted (sub-1 Gbps) Direct Connect connections.

### Design Patterns

| Pattern | When to use |
|---------|-------------|
| Single Public VIF | Simplest; adequate for non-critical workloads |
| Redundant Public VIFs (two DX locations) | HA + failover for production |
| Public VIF + Site-to-Site VPN fallback | Cost-effective HA when second DX is too expensive |
| LAG (Link Aggregation Group) | Increase bandwidth + redundancy on a single DX location |

### Common Exam Traps

1. **Public VIF ≠ internet access** — traffic stays on the AWS backbone; it does
   not give clients internet egress.
2. **Public VIF ≠ Private VIF** — Private VIF reaches VPCs; Public VIF reaches
   AWS public service endpoints.
3. **BGP AS path** — Direct Connect uses BGP; you must have a public ASN (or use
   the 64512–65534 private ASN range in specific hosted scenarios).
4. **Prefix advertisement** — customers must advertise at least one public IP block;
   you cannot advertise `0.0.0.0/0` on a Public VIF.

Pipeline Verification and LLM Probing

To ensure robust generation, the system features a two-step probe (_llm_available()) that verifies the endpoint via GET /models and checks the model via POST /chat/completions. For instance, verifying technical documentation like AWS Direct Connect chapters confirms correct model response and artifact generation:

LLM Detection Status: Verified active (true) with llama3.2:latest connected via http://localhost:11434/v1.
Test Suite Results: Execution of the pytest suite confirms pipeline integrity across modular extractors and fallback handlers:

  43 passed, 1 warning in 64.44s[cite: 13]

Conclusion

Through this implementation, we successfully realized a complete, local-first web UI and orchestration layer for book-to-skill. Key achievements include:

Advanced Document Extraction: Leveraging Docling to parse technical PDFs accurately without losing code formatting, tables, or structural hierarchies.
Asynchronous & Non-Blocking Architecture: Utilizing FastAPI, background tasks, and thread pools to handle heavy parsing and LLM inference tasks smoothly.
Resilient State Management: Implementing disk-scan fallbacks and auto-healing job statuses so generated skills remain accessible even across server restarts.
Actionable Agent Integration: Translating static book pages into structured, on-demand agent skills (SKILL.md, glossary.md, patterns.md, cheatsheet.md, and chapter files) that drastically cut down token overhead during development workflows.

Thanks for reading 🤹

Agent Context A/B Testing with Intel Dokimi

Alain Airom (Ayrom) — Tue, 27 Oct 2026 19:45:49 +0000

Hands-on A/B agent context testing using dokimi from Intel.

Introduction

What is Agent Context A/B Testing?

Agent Context A/B Testing is a rigorous evaluation methodology used to measure how different pieces of context (such as system prompts, documentation, user profiles, or specialized agent skills) impact an AI model's performance on specific tasks. Rather than guessing which context configuration works best, this approach treats context as a variable in a controlled experiment, comparing a baseline against one or more enhanced "arms".

Key Components of the Evaluation Flow

The A/B testing pipeline generally follows a structured, multi-step process:

Questions / Golden Sets: The process starts with a standardized set of test prompts, questions, or synthetic personas that represent real-world use cases.
Execution Arms: The same base model answers every question across different context "arms" (e.g., a baseline arm versus an arm enhanced with specific documentation, user profiles, or plugins). Agentic arms also record whether and how tools were utilized.
Automated Judging: A separate, dedicated LLM acts as a judge, scoring all generated answers (typically on a 0–100 scale) across multiple core dimensions such as accuracy, completeness, conciseness, and novelty impact. Cross-checks (such as RAGAS) are often integrated to ensure scoring reliability.
Statistical Significance & Results: The framework computes performance deltas (Δ=B−AΔ=B−A) and runs statistical significance tests (such as calculating p*p*-values) to determine whether an observed performance boost is real or merely statistical noise.
Scorecard & Decision Gates: Results are compiled into a final scorecard reporting the value added by the new context (e.g., worth +X+X on model M*M*), guiding developers on whether to ship it or cut it.

Why Use Context A/B Testing?

Multi-Dimensional Visibility: It provides a comprehensive radar view of model behavior, ensuring that an improvement in accuracy doesn't inadvertently hurt conciseness or introduce hallucinations.
Data-Driven Decisions: It replaces subjective tuning with rigorous statistical validation, helping engineering teams optimize prompts and agent context confidently.

> First things first, what is "dokimi"?

> Excerpts from the project's GitHub and documentation:

Dokimi — A/B testing for agent context

Documentation: https://intel.github.io/dokimi/ — guides, plus a CLI reference and an API reference generated from the code.

dokimi — Greek δοκιμή, "test, proof". The name is the method: an artifact does not get trusted because it was written or because it is popular, it gets trusted because it passed a test that could have failed.

Skills, instructions, docs, MCP servers and repository guidance are cheap to write and easy to over-claim. Dokimi tests each one like an experiment: change one thing, keep everything else fixed, run the same question or task set, and report the delta together with its p-value. An artifact ships with the scorecard that earned it, a deliberate contrast to popularity-ranked marketplaces, which structurally bury niche HPC and numerical-accuracy tooling.

What measuring shows (evidence):

Much of a skill repeats what the model already knows, and the rest is exact detail. For claude-opus-5 under lenient grading, 70% of what a skill says the model already produces unprompted. Under strict grading, which also checks qualifiers and exact values, only 22% is already known and 38% is still missing even with the skill in context. Read the two gradings together.
Models believe their context, even when it is wrong. A plausible wrong value planted in a skill is repeated 94% of the time.
Value depends on the model. What a small model needs, a frontier model may already know, so every verdict is per model.

The idea in three points:

An agent is more than its model. It is model + the context you give it: docs, skills, MCP servers, system prompts (agent profiles), plugins.
All of that is cheap to write — and easy to over-claim. Does it really help, or does it just add tokens?
So we test it like an experiment: change one thing, keep everything else fixed, measure the delta — and check it is statistically significant.

What it measures

1. Does it help? Enabled vs disabled arms

An arm is one context condition. Every arm runs the same question or task set against the same model, and the report gives the per-arm delta against the baseline with paired statistics. Arms are given as specs (dokimi arms run --arms …):


  | Arm                    | What the agent gets                                          |
  | ---------------------- | ------------------------------------------------------------ |
  | `baseline`             | nothing extra                                                |
  | `docs:<source>`        | product documentation (local Markdown, URLs, Context7)       |
  | `mcp:<server>`         | an MCP documentation server it can call                      |
  | `skill:<path>`         | one skill, its instructions in context                       |
  | `skill-agent:<path>`   | one skill offered by name and description; the model must decide to load it |
  | `skill-catalog:<root>` | a whole skill catalog, to measure whether the right skill gets selected |
  | `profile:<path>`       | an agent persona / system prompt                             |
  | `guidance:none         | raw                                                          |
  | `guidance:rev:<rev>`   | the same repository with the guidance files of another revision, such as a PR's `AGENTS.md` (repository mode) |
  | `patch:<file>`         | the repository with a unified diff applied on top of its own guidance: a new script, a Makefile target (repository mode) |

The same arms drive question answering (LLM-as-judge with judge-validity checks), executable Docker coding tasks through Codex, Claude Code, Harbor or BenchFlow harnesses (dokimi tasks run), and file/line localization (dokimi explore run). Results roll up per subject (dokimi subjects run): a skill, MCP server, doc source, profile, bundle or repository, each with a scorecard.

2. What does it add? Novel vs known content

An arm delta says whether an artifact helps; it does not say which parts. dokimi novelty analyze --artifact skill:<path>|text:<path> splits the artifact into knowledge atoms and probes each one against the model with and without the artifact. Each atom comes out added (new to the model), redundant (the model already knew it), degraded (the artifact broke something the model knew) or still missing (stated but not picked up). text: takes any file, so the same analysis runs on a documentation page or a repository's AGENTS.md. The output is per (artifact, model) pair, with shortening candidates (novelty shorten) and a check that the artifact can be selected among competitors at all (novelty routing). See skill-novelty.md.

3. Repositories: scan the guidance, then measure agents working in them

A repository is a subject in its own right, described by a repo-eval.yaml manifest it owns:

dokimi repo scan (L0, no LLM): do the links, paths, identifiers, commands and build labels the guidance names exist in the tree?
dokimi repo budget (no LLM): how many characters of guidance and skill listing every model call pays for at start-up, what loads lazily or on demand, and which AGENTS.md files a CLAUDE.md turns off.
dokimi repo reach (L1): which guidance files does a given harness actually load?
dokimi repo inspect (static, no LLM): which instruction files, checks, capability surfaces and risky hooks or scripts the repository offers an agent, with findings per stage.
dokimi repo run: run an agent per (task, arm, model, repetition) inside the repository's image, grade each run with the repository's own tests, and dokimi repo report the per (task, arm, model) table. With the guidance:* arms this measures what the repository's agent guidance does to the agent's work; dokimi repo compare puts each arm against a baseline arm, paired by task.

> End of Excerpts from the project’s GitHub and documentation

Implementation of the Application

Discovering the 'dokimi' repository naturally piqued my interest. Leveraging Bob, I developed an end-to-end application to evaluate and demonstrate its core capabilities in action.

Overall Architecture of the Implementation

The Dokimi Agent Context A/B Testing Platform is structured around a modular, full-stack architecture that combines a FastAPI backend with a responsive single-page dashboard built using the Carbon Design System theme. At its core, the system incorporates Intel's Dokimi evaluation methodology—enforcing paired on/off arm comparisons, knowledge atom decomposition, and rigorous statistical significance testing via paired Student's t-tests and Holm-Bonferroni corrections. The platform supports dual-tier persistence, keeping built-in demonstration fixtures immutable in memory while storing dynamic ad-hoc use-cases in persistent JSON storage (data/adhoc_usecases.json). Furthermore, evaluation requests are dispatched through a flexible execution layer capable of interfacing with domain-specific mock engines, LiteLLM providers (such as OpenAI and Anthropic), and local LLM runtimes like Ollama and llama.cpp, while real-time progress is streamed to the frontend via Server-Sent Events (SSE).

While the application ships with built-in mock fixtures out of the box, it fully supports dynamic ad-hoc use-case submissions. The platform also offers flexible LLM inference support—allowing you to seamlessly run local models via Ollama or llama.cpp, or connect to cloud-based LLM providers, all configurable directly within the application.

It incorporates Intel's Dokimi evaluation principles:

Paired On/Off Arm Comparisons: Every context arm is evaluated against the exact same question/scenario sets to compute delta scores.
Knowledge Atom Decomposition: Context is evaluated not merely on volume, but whether it adds novel facts, repeats redundant knowledge, or degrades accuracy.
Statistical Significance Testing: Paired Student's t-test with Holm-Bonferroni family-wise error rate corrections.

Technical and Functional Implementation Details

Context Arms Under Test


| Arm ID        | Spec Type         | Injected Payload        | Purpose                                                      |
| ------------- | ----------------- | ----------------------- | ------------------------------------------------------------ |
| `baseline`    | `baseline`        | Empty (Unprompted)      | Establishes model foundational performance baseline.         |
| `arm_docs`    | `docs:local:path` | `standard_runbook.md`   | Measures value of standard static procedural documentation.  |
| `arm_profile` | `profile:path`    | `sre_expert_profile.md` | Measures value of domain persona & behavioral guidelines.    |
| `arm_skill`   | `skill:path`      | `SKILL.md`              | Measures value of precision action protocols and remediation scripts]. |

Dokimi Evaluation Metrics

Mean Score (S∈[0,1]*S*∈[0,1]): Weighted composite of Accuracy (45%), Grounding Completeness (40%), and Conciseness (15%).
Score Delta (ΔΔ): Difference in mean score between treatment arm and baseline.
Paired t*t*-test (p*p*-value): Tests null hypothesis that context addition produces no difference.
Holm-Bonferroni (α=0.05*α*=0.05): Controls family-wise error across multiple simultaneous arms.
Novel Knowledge Atoms: Measures count of verified facts introduced beyond base model knowledge.

dokimi-agent-ab-testing/
├── app/
│   ├── main.py              # Application entry point & FastAPI setup
│   ├── core/
│   │   ├── config.py        # Settings and environment configuration
│   │   └── models.py        # Pydantic domain models & schemas
│   ├── services/
│   │   ├── dokimi_engine.py # Dokimi evaluation & statistical runner
│   │   ├── progress_manager.py # Real-time progress tracking & SSE streamer
│   │   └── llm_service.py   # LLM dispatcher (Mock & real providers)
│   ├── api/
│   │   └── router.py        # API routes for experiments, variants, chat, tasks
│   ├── templates/
│   │   └── index.html       # Single-page dashboard UI template
│   └── static/
│       ├── css/style.css    # IBM Carbon-inspired styles
│       └── js/app.js        # Dynamic UI logic & ECharts integration
├── data/
│   ├── docs/                # Markdown runbooks for doc arms
│   ├── skills/              # Skill definitions for skill arms
│   ├── profiles/            # Persona prompts for profile arms
│   └── eval_sets/           # Standardized evaluation questions
├── scripts/
│   ├── start.sh             # Detached launch script with URL output
│   ├── stop.sh              # Graceful shutdown script
│   ├── cleanup.sh           # Clean venvs, cache, and build files
│   └── clean_containers.sh  # Remove built container images
├── Docs/
│   ├── Architecture.md      # This document
│   ├── Quickstart.md        # Step-by-step setup and execution guide
│   └── AdHoc_UseCases.md    # Guide and API specs for ad-hoc use-case submissions
├── tests/                   # Pytest test suite
├── Dockerfile & Containerfile
├── requirements.txt
└── README.md

What the Application Demonstrates?

Core Concepts

When building AI coding and operations agents, developers often inject large prompt templates, extensive markdown documentation, or complex tool suites. However, without systematic testing, it is impossible to know whether that extra context actually improves performance or merely adds latency and cost.

The platform implements Intel's Dokimi evaluation principles:

Treatment Arms: Different context configurations tested in parallel:
- baseline: Unprompted model baseline.
- arm_docs: Static procedural documentation injection.
- arm_profile: Domain persona and behavioral guidelines.
- arm_skill: Precision action protocols and step-by-step remediation commands.
- arm_mcp: Model Context Protocol server capabilities.
Paired On/Off Arm Testing: Every arm is tested against the exact same scenario queries under identical conditions.
Statistical Significance: Computes paired Student's t-test p-values with Holm-Bonferroni multi-arm corrections (α=0.05*α*=0.05).
Knowledge Atom Decomposition: Distinguishes novel actionable facts from redundant base knowledge.

Once the application is launched, several URLs and links are available to demonstrate Dokimi's functionalities:

=====================================================================
✅ Dokimi Agent A/B Testing Platform is LIVE and HEALTHY!
📌 Process PID:         63870
🌐 Web Dashboard URL:   http://127.0.0.1:8000
📚 API Swagger Docs:    http://127.0.0.1:8000/docs
📖 API Redoc:           http://127.0.0.1:8000/redoc
❤️ Health Check:        http://127.0.0.1:8000/health
🎯 Use-Cases Endpoint:  http://127.0.0.1:8000/api/use-cases
🧬 Variants Endpoint:   http://127.0.0.1:8000/api/variants
📋 Questions Endpoint:  http://127.0.0.1:8000/api/questions
📊 Experiments API:     http://127.0.0.1:8000/api/experiments
📄 Application Logs:    output/app.log
=====================================================================

The platform navigation tabs provide full control over the evaluation lifecycle:

📊 A/B Experiment Dashboard: Run paired evaluations, view top performing arms, delta improvements, statistical significance badges, and interactive ECharts (Bar & Radar).

🎯 Use-Case Library: Browse immutable built-in fixtures (SRE Incident Response, Cloud Security Audit) and submit dynamic ad-hoc use-cases.
💬 Dual Agent Chat Playground: Query two context arms side-by-side in real-time.

🧬 Context Arms & Variants: Inspect and configure custom system prompts, documentation payloads, and skill protocols.

📋 Evaluation Question Set: Inspect golden evaluation questions and ground truth assertion keywords.

You can submit a custom domain use-case directly from the Web UI or via the REST API, all documented.

End-to-End Ad-Hoc Use-Case Submission & Evaluation

"""
End-to-End Dokimi A/B Testing Workflow:
1. Submits an ad-hoc use-case (Kafka consumer group rebalancing)
2. Triggers a full Dokimi A/B evaluation across all context arms
3. Inspects statistical significance (p-values, Holm-Bonferroni) and score deltas
4. Cleans up the ad-hoc use-case
"""

import json
import time
import urllib.request
import urllib.error

BASE_URL = "http://127.0.0.1:8000"


def http_post(endpoint: str, data: dict) -> dict:
    url = f"{BASE_URL}{endpoint}"
    req = urllib.request.Request(
        url,
        data=json.dumps(data).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST"
    )
    with urllib.request.urlopen(req) as response:
        return json.loads(response.read().decode("utf-8"))


def http_get(endpoint: str) -> dict:
    url = f"{BASE_URL}{endpoint}"
    req = urllib.request.Request(url, method="GET")
    with urllib.request.urlopen(req) as response:
        return json.loads(response.read().decode("utf-8"))


def http_delete(endpoint: str) -> dict:
    url = f"{BASE_URL}{endpoint}"
    req = urllib.request.Request(url, method="DELETE")
    with urllib.request.urlopen(req) as response:
        return json.loads(response.read().decode("utf-8"))


def main():
    print("=" * 70)
    print("🧪 Dokimi A/B Testing Platform - Programmatic Evaluation")
    print("=" * 70)

    # 1. Define Ad-Hoc Use-Case
    use_case_payload = {
        "name": "Kafka Consumer Lag & Rebalancing Storm Triage",
        "description": "Evaluate context arms on diagnosing consumer rebalance loops and partition skew.",
        "parameters": {
            "domain": "Event Streaming / Messaging",
            "cluster_type": "Apache Kafka / Strimzi",
            "alpha": 0.05
        },
        "expected_outputs": [
            "Diagnose max.poll.interval.ms timeout causes",
            "Tune session.timeout.ms and heartbeat.interval.ms",
            "Provide cooperative-sticky assignor configuration"
        ],
        "scenarios": [
            {
                "id": "kafka_sc_01",
                "prompt": "Consumer group 'order-processing' is stuck in continuous rebalancing loop. Lag is growing by 5k msg/sec. How to mitigate?",
                "expected_topics": [
                    "Check max.poll.interval.ms vs message processing time",
                    "Switch to CooperativeStickyAssignor to prevent stop-the-world rebalance",
                    "Inspect GC pauses on consumer JVM instances",
                    "Scale out partitions and consumers proportionally"
                ],
                "category": "Consumer Rebalance"
            },
            {
                "id": "kafka_sc_02",
                "prompt": "Two partitions in topic 'payments' have 20x higher lag than all other partitions. How to rebalance data distribution?",
                "expected_topics": [
                    "Inspect partition key hashing skew",
                    "Check for hot keys in upstream producer payload",
                    "Verify consumer group partition assignment with kafka-consumer-groups.sh"
                ],
                "category": "Partition Skew"
            }
        ]
    }

    # 2. Submit Use-Case
    print("\n1. Submitting Ad-Hoc Use-Case...")
    use_case = http_post("/api/use-cases/submit", use_case_payload)
    uc_id = use_case["id"]
    print(f"   ✅ Registered: '{use_case['name']}' (ID: {uc_id})")

    # 3. Trigger Dokimi Experiment Run
    print(f"\n2. Executing Dokimi Paired A/B Experiment against use-case '{uc_id}'...")
    exp_payload = {
        "name": "Kafka Rebalance Context Optimization",
        "use_case_id": uc_id,
        "model": "mock-model",
        "provider": "mock"
    }

    start_time = time.time()
    experiment = http_post("/api/experiments/run", exp_payload)
    elapsed = time.time() - start_time

    print(f"   ✅ Experiment Complete (ID: {experiment['id']}) in {elapsed:.2f}s")
    print(f"   📢 Verdict: {experiment['summary_verdict']}\n")

    # 4. Display Results Table
    print(f"{'Arm Spec ID':<16} | {'Mean Score':<10} | {'Delta vs Base':<14} | {'p-value':<10} | {'Sig (p<0.05)':<12} | {'Novel Atoms':<12}")
    print("-" * 88)

    base_score = experiment["metrics_by_arm"]["baseline"]["mean_score"]
    for arm_id, m in experiment["metrics_by_arm"].items():
        delta = m["mean_score"] - base_score
        delta_str = f"+{delta:.4f}" if delta >= 0 else f"{delta:.4f}"
        pval_str = f"{m['p_value']:.4f}" if m["p_value"] is not None else "— (Base)"
        sig_str = "✅ YES" if m["stat_significant"] else ("Base" if arm_id == "baseline" else "❌ NO")
        print(f"{arm_id:<16} | {m['mean_score']:<10.4f} | {delta_str:<14} | {pval_str:<10} | {sig_str:<12} | +{m['novel_atoms_added']:<11}")

    # 5. Clean Up Use-Case
    print(f"\n3. Cleaning up ad-hoc use-case '{uc_id}'...")
    del_res = http_delete(f"/api/use-cases/{uc_id}")
    print(f"   ✅ Cleaned up: {del_res['status']}")
    print("=" * 70)


if __name__ == "__main__":
    main()

Async Client with httpx

import asyncio
import httpx

BASE_URL = "http://127.0.0.1:8000"


async def run_async_evaluation():
    async with httpx.AsyncClient(base_url=BASE_URL, timeout=30.0) as client:
        # 1. Health check
        health_resp = await client.get("/health")
        assert health_resp.status_code == 200

        # 2. Submit use-case
        payload = {
            "name": "Cloud IAM Least Privilege Enforcement",
            "description": "Evaluate context arms on AWS/GCP wildcard permission pruning.",
            "parameters": {"domain": "Cloud Security", "alpha": 0.05},
            "expected_outputs": ["Remove Action: *", "Enforce resource ARNs"],
            "scenarios": [
                {
                    "id": "iam_01",
                    "prompt": "An IAM role has Action: 's3:*' on Resource: '*'. How to remediate to read-only access for bucket 'app-data'?",
                    "expected_topics": [
                        "Replace 's3:*' with 's3:GetObject' and 's3:ListBucket'",
                        "Scope Resource to 'arn:aws:s3:::app-data' and 'arn:aws:s3:::app-data/*'",
                        "Validate using AWS Access Analyzer"
                    ],
                    "category": "IAM Security"
                }
            ]
        }
        res = await client.post("/api/use-cases/submit", json=payload)
        uc = res.json()
        print(f"Registered use-case: {uc['id']}")

        # 3. Run evaluation
        exp_res = await client.post(
            "/api/experiments/run",
            json={
                "name": "IAM Security Eval",
                "use_case_id": uc["id"],
                "model": "mock-model",
                "provider": "mock"
            }
        )
        exp = exp_res.json()
        print(f"Top arm: {exp['summary_verdict']}")

        # 4. Clean up
        await client.delete(f"/api/use-cases/{uc['id']}")
        print("Cleaned up successfully.")

if __name__ == "__main__":
    asyncio.run(run_async_evaluation())

Fetch & Axios Integration

const BASE_URL = "http://127.0.0.1:8000";

async function runDokimiUseCase() {
    try {
        // 1. Submit Ad-Hoc Use-Case
        const submitResponse = await fetch(`${BASE_URL}/api/use-cases/submit`, {
            method: "POST",
            headers: { "Content-Type": "application/json" },
            body: JSON.stringify({
                name: "Postgres Connection Pool Saturation Drill",
                description: "Assess context arms on PgBouncer connection queuing and backend worker starvation.",
                parameters: { domain: "Database SRE", alpha: 0.05 },
                expected_outputs: ["Configure transaction pooling", "Tune max_connections"],
                scenarios: [
                    {
                        id: "pg_pool_01",
                        prompt: "Postgres server reached max_connections (500). Applications receiving 'FATAL: remaining connection slots are reserved'. What is the immediate fix?",
                        expected_topics: [
                            "Deploy PgBouncer in transaction pooling mode",
                            "Identify idle in transaction connections",
                            "Tune application pool maxActive limits",
                            "Terminate non-essential idle backend connections"
                        ],
                        category: "Postgres Tuning"
                    }
                ]
            })
        });

        const useCase = await submitResponse.json();
        console.log(`✅ Registered use-case: ${useCase.name} (ID: ${useCase.id})`);

        // 2. Run A/B Experiment
        const expResponse = await fetch(`${BASE_URL}/api/experiments/run`, {
            method: "POST",
            headers: { "Content-Type": "application/json" },
            body: JSON.stringify({
                name: "Postgres Pooling Evaluation",
                use_case_id: useCase.id,
                model: "mock-model",
                provider: "mock"
            })
        });

        const experiment = await expResponse.json();
        console.log("📊 Experiment Result:", experiment.summary_verdict);

        // 3. Delete Ad-Hoc Use-Case
        await fetch(`${BASE_URL}/api/use-cases/${encodeURIComponent(useCase.id)}`, {
            method: "DELETE"
        });
        console.log("🧹 Cleaned up temporary use-case.");

    } catch (err) {
        console.error("❌ Error running Dokimi integration:", err);
    }
}

runDokimiUseCase();

Command-Line Workflow

# 1. Health check
curl -s http://127.0.0.1:8000/health | jq .

# 2. Submit new ad-hoc use-case
UC_RESP=$(curl -s -X POST http://127.0.0.1:8000/api/use-cases/submit \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Redis Memory Eviction Storm Triage",
    "description": "Assess agent triage on maxmemory policy OOM errors.",
    "parameters": {"domain": "Cache Infrastructure", "alpha": 0.05},
    "expected_outputs": ["Identify keys without TTL", "Tune volatile-lru"],
    "scenarios": [
      {
        "id": "redis_sc_1",
        "prompt": "Redis cluster returning (error) OOM command not allowed when used memory > maxmemory. How to triage?",
        "expected_topics": [
          "Check INFO memory and maxmemory-policy",
          "Identify big keys with redis-cli --bigkeys",
          "Enable volatile-lru or allkeys-lru eviction",
          "Set TTLs on unbounded cache keys"
        ],
        "category": "Redis Caching"
      }
    ]
  }')

echo "$UC_RESP" | jq .
UC_ID=$(echo "$UC_RESP" | jq -r .id)

# 3. Run Dokimi A/B experiment
curl -s -X POST http://127.0.0.1:8000/api/experiments/run \
  -H "Content-Type: application/json" \
  -d "{
    \"name\": \"Redis Eviction Context Optimization\",
    \"use_case_id\": \"$UC_ID\",
    \"model\": \"mock-model\",
    \"provider\": \"mock\"
  }" | jq '{id: .id, summary: .summary_verdict, metrics: .metrics_by_arm}'

# 4. Clean up use-case
curl -s -X DELETE "http://127.0.0.1:8000/api/use-cases/$UC_ID" | jq .

Use-Case Submission & Clean-up

#!/usr/bin/env bash
set -euo pipefail

BASE_URL="http://127.0.0.1:8000"

echo "=== Submitting Multiple Use-Cases ==="

DOMAINS=("Networking" "Storage" "CI_CD")

for DOMAIN in "${DOMAINS[@]}"; do
  echo "Submitting domain: $DOMAIN..."
  RESP=$(curl -s -X POST "${BASE_URL}/api/use-cases/submit" \
    -H "Content-Type: application/json" \
    -d "{
      \"name\": \"Domain Benchmark: ${DOMAIN}\",
      \"description\": \"Automated batch validation for ${DOMAIN}\",
      \"parameters\": {\"domain\": \"${DOMAIN}\", \"alpha\": 0.05},
      \"expected_outputs\": [\"Accuracy > 80%\", \"Valid triage sequence\"],
      \"scenarios\": [
        {
          \"id\": \"${DOMAIN,,}_sc_1\",
          \"prompt\": \"How to diagnose common ${DOMAIN} failure modes?\",
          \"expected_topics\": [\"metrics\", \"logs\", \"remediation\"],
          \"category\": \"${DOMAIN}\"
        }
      ]
    }")

  UC_ID=$(echo "$RESP" | grep -o '"id":"[^"]*' | cut -d'"' -f4)
  echo "  -> Created: $UC_ID"
done

echo "=== Current Registered Use-Cases ==="
curl -s "${BASE_URL}/api/use-cases" | grep -o '"id":"[^"]*' | cut -d'"' -f4

Conclusion: Bringing Rigor to Agent Engineering

Optimizing AI agents has historically been a game of intuition—tweaking system prompts, appending markdown runbooks, or adding tool definitions based on gut feeling rather than empirical evidence. As demonstrated through this implementation combining Bob and Intel's Dokimi framework, moving from subjective tuning to rigorous, statistically grounded A/B testing is essential for production-grade agent deployment.

By evaluating context variations against standardized question sets, enforcing paired on/off comparisons, and utilizing statistical tools like Student's t*t*-tests and Holm-Bonferroni corrections, engineering teams can definitively answer whether an artifact adds actual value or simply introduces costly noise and redundancy. Whether you are running local models via Ollama and llama.cpp or leveraging cloud-based LLM providers adopting this systematic evaluation lifecycle ensures you only ship context configurations that have proven their worth.

Thanks for reading 📊

Why I Crosspost Across Medium, Dev.to, DZone, and ZyVop (And How to Choose)

Alain Airom (Ayrom) — Fri, 23 Oct 2026 11:28:24 +0000

My take on “tech blogging” sites!

When it comes to sharing technical knowledge, tutorials, and insights, one platform is rarely enough. Every blogging network has its own community, culture, and quirks.

To make sure my content reaches as many developers, architects, and tech enthusiasts as possible, I regularly publish my posts across multiple platforms. Here is a quick breakdown of where I publish, along with the pros and cons of each—including a new favorite of mine!

1. Medium

Medium is one of the most recognized publishing platforms on the internet today.

Pros: Huge global reach, millions of active readers, and a rich diversity of topics ranging from deep technical architecture to soft skills.
Cons: Paywall restrictions. Non-paying users often hit the reading limit, which can prevent them from accessing complete, detailed technical tutorials.

2. Dev.to

Dev.to has built an incredible ecosystem specifically tailored for developers and software engineers.

Pros: Highly engaged developer community, completely open and free with no paywalls or reading limits, and excellent support for code formatting and Markdown.
Cons: Because it caters almost exclusively to software developers, non-engineering or high-level business technology topics don't always get the same traction.

3. DZone

DZone is a legacy favorite for enterprise software development and architectural insights.

Pros: High-quality articles with a strong focus on practical developer tips, enterprise trends, and architectural patterns.
Cons: Articles undergo a strict manual moderation process. While quality control is definitely a positive, the review queue can sometimes be so long that publishing gets delayed significantly.

4. ZyVop (My Latest Favorite! 🚀)

ZyVop is a newer platform I recently discovered, and it has quickly become a delight to use.

Pros: A clean, modern environment with near-daily platform enhancements and feature updates. Best of all, its seamless article import feature makes crossposting from other sites totally effortless. I've quickly grown to love this little gem!
Cons: As a growing platform, its reader base is still building compared to giants like Medium or Dev.to—which is all the more reason to help spread the word!

The Solution: Total Freedom for the Reader

Because I want to spread the word while giving everyone the freedom to read on their platform of choice, I syndicate my content across all four networks.

If you want a central hub where you can browse all my articles, project updates, and tutorials in one place—completely free of paywalls or queue delays—I built my own site:

Visit the site to pick the reading format and platform that works best for you!

Thanks for reading 🥸

Can Bob help in Quantum Application Programming?

Alain Airom (Ayrom) — Fri, 23 Oct 2026 09:44:21 +0000

Using Bob and Qiskit to build an end-to-end demonstration.

TL;DR - Quantum Software Ecosystem

The quantum software landscape is anchored by a few dominant open-source frameworks backed by major industry players, with IBM's Qiskit commanding the largest share of developer mindshare and ecosystem adoption. Supported by millions of cumulative downloads (approaching nearly 30 million total PyPI installations) and hundreds of thousands of active users worldwide, Qiskit leads the market due to its deep integration with IBM’s cloud-accessible superconducting hardware, robust transpiler services, and extensive educational resources. Competing frameworks like Google's Cirq—focused heavily on Noisy Intermediate-Scale Quantum (NISQ) processor architectures—and Xanadu's PennyLane—which has gained substantial traction for quantum machine learning and differentiable quantum programming—hold strong academic and specialized niches. Meanwhile, enterprise-focused platforms like Microsoft’s Azure Quantum (Q#) and Amazon’s Braket act as hardware-agnostic multi-tenant middleware layers, collectively driving a rapidly expanding quantum software market that parallels the broader hardware industry's robust annual growth.

What is Qskit

Qiskit is an open-source software development kit (SDK) designed for working with quantum computers at the level of circuits, operators, and primitives. Developed primarily by IBM, it serves as the industry’s most widely adopted framework for quantum computing, bridging the gap between theoretical quantum physics and practical software engineering. By offering a Python-based programming environment backed by high-performance compilation tools written in Rust, Qiskit enables researchers, developers, and enterprise professionals to design quantum algorithms, optimize circuits for specific hardware topologies, and execute workloads seamlessly across local simulators or cloud-accessible utility-scale quantum processors.

Qiskit

Excerpt from Qiskit GitHub.

Qiskit is an open-source SDK for working with quantum computers at the level of extended quantum circuits, operators, and primitives.

This library is the core component of Qiskit, which contains the building blocks for creating and working with quantum circuits, quantum operators, and primitive functions (Sampler and Estimator). It also contains a transpiler that supports optimizing quantum circuits, and a quantum information toolbox for creating advanced operators.

Qiskit provides two public APIs: a Python API and a C API. The Python API is the primary interface and prior to Qiskit 2.0 was the only public API available for Qiskit. The C API is designed to provide direct access to Qiskit's internal data model (which is written in Rust). The C API can be consumed as either a shared library (libqiskit.so) for standalone use, or from its embedding in the qiskit Python package for writing Python extension modules.

Using IBM Bob to Write Qiskit Enabled Applications

Disclaimer: *Although I am far from a quantum computing expert—with only a foundational grasp of its underlying mechanics, current use cases, and future possibilities—I was eager to explore a domain where I am essentially a novice. I wanted to see how Bob could help me design and construct an engaging, user-friendly, and educational application to demystify these complex concepts. So, I put Bob to the test and asked him to help me build it from the ground up!*

Implementation Architecture & Code

Python Algorithmes

Deep Dive: The Python Algorithm Layer & Architectural Design

At the core of the application lies a meticulously engineered Python algorithm layer that brings foundational quantum principles to life through clean, modular dataclasses. Designed as stateless, self-contained units, the application's algorithms encapsulate circuit construction, local simulation via AerSimulator, and semantic result interpretation into robust, reusable components.

As mapped out in the system's high-level architecture, user interactions originating from the Streamlit GUI (app.py) are routed dynamically through sidebar controls into individual algorithm modules under src/algorithms/. Each algorithm implementation conforms to a uniform interface, making it trivial to extend the suite or inspect its inner workings programmatically.

Entanglement and Foundational Circuits: Bell States: the simplest yet most profound demonstration of quantum mechanics in the suite is the Bell State (EPR Pair) implementation (bell_state.py). It models maximal quantum entanglement between two qubits:

@dataclass
class BellStateDemo:
    shots: int = 1024
    circuit: QuantumCircuit | None = field(default=None, repr=False)
    counts: dict[str, int] = field(default_factory=dict)
    circuit_diagram: str = ""

    def build_circuit(self) -> QuantumCircuit:
        qc = QuantumCircuit(2, 2)
        qc.h(0)        # Hadamard: |0⟩ → |+⟩ = (|0⟩+|1⟩)/√2
        qc.cx(0, 1)    # CNOT: entangles qubits 0 and 1
        qc.measure([0, 1], [0, 1])
        return qc

When simulated, the results consistently yield a roughly 50/50 split between 00 and 11 outcomes, proving that measuring one qubit instantaneously determines the state of the other regardless of classical separation.

Query Complexity Advantage: Bernstein-Vazirani & Deutsch-Jozsa: to illustrate quantum computational speedups, the suite features algorithmic oracles that demonstrate exponential query advantages over classical computing. In the Bernstein-Vazirani Algorithm (bernstein_vazirani.py), a hidden n-bit string s*s* is uncovered using a single quantum query compared to n*n* classical queries:

"""
Bernstein-Vazirani Algorithm
=============================
Finds a hidden bit-string s ∈ {0,1}ⁿ encoded in an oracle f(x) = s·x (mod 2)
with a SINGLE quantum query, compared to n queries classically.

The algorithm demonstrates the power of quantum parallelism:
  • Classical: requires n queries (one per bit)
  • Quantum:   requires 1 query

This is one of the clearest examples of an exponential query advantage.

Circuit:
    |0⟩ ─H─●─H─  (qubit carrying bit 0 of s)
    |0⟩ ─H─●─H─  (qubit carrying bit 1 of s)
    |1⟩ ─H─⊕──   (ancilla in |−⟩)

References:
    • Bernstein, E. & Vazirani, U. (1997). Quantum Complexity Theory.
    • Qiskit documentation: https://github.com/Qiskit/documentation
"""

# Ancilla preparation and Hadamard superposition across query qubits
qc.x(n)
qc.h(n)
qc.barrier()
qc.h(range(n))

# Oracle: CNOT on ancilla for every bit=1 in the secret string
for i, bit in enumerate(reversed(self.secret_string)):
    if bit == "1":
        qc.cx(i, n)

qc.h(range(n))  # Inverse Hadamard to interfere and expose the secret

Similarly, the Deutsch-Jozsa Algorithm (deutsch_jozsa.py) determines whether a Boolean function is constant or balanced with a single query, leveraging quantum parallelism and interference to yield an all-zero measurement for constant functions.

  """
  Deutsch-Jozsa Algorithm
  ========================
  Determines whether a Boolean function f: {0,1}ⁿ → {0,1} is CONSTANT
  (same output for all inputs) or BALANCED (0 for half, 1 for half) using
  a single quantum query.

  Classical algorithm: needs up to 2^(n-1)+1 queries in the worst case.
  Quantum algorithm:   needs exactly 1 query — exponential speedup.

  This was the first demonstration of a quantum algorithm outperforming
  any deterministic classical algorithm.

  References:
      • Deutsch, D. & Jozsa, R. (1992). Rapid Solution of Problems by Quantum Computation.
      • Qiskit documentation: https://github.com/Qiskit/documentation
  """
  def _interpret(self) -> str:
          """Generate a human-readable result interpretation.

          Rule: all-zero measurement → CONSTANT; any non-zero bit → BALANCED.
          """
          total = sum(self.counts.values())
          zero_state = "0" * self.num_qubits
          zero_count = self.counts.get(zero_state, 0)
          measured_constant = zero_count / total > 0.9

          expected_constant = self.oracle_type.startswith("constant")
          verdict = "CONSTANT" if measured_constant else "BALANCED"
          correct = "✓ Correct" if measured_constant == expected_constant else "✗ Unexpected"

          return (
              f"Oracle type: {self.oracle_type}\n"
              f"Measured all-zero ({zero_state}): {zero_count}/{total} times "
              f"({zero_count/total*100:.1f}%)\n"
              f"Verdict: Function is {verdict}  [{correct}]\n\n"
              "Rule: all-zero measurement → CONSTANT; any non-zero → BALANCED\n"
              f"Classical worst case: {2**(self.num_qubits-1)+1} queries needed.\n"
              "Quantum: always 1 query — the first-ever proven quantum advantage."
          )

Quadratic Speedups & Amplitude Amplification: Grover's Search: for unstructured database search, Grover's Search Algorithm (grover.py) provides a quadratic speedup (O(√N) vs. classical O(N)) by utilizing iterative amplitude amplification. The algorithm computes the optimal number of iterations based on the search space size (iterations = ⌊(π/4) × √(2ⁿ)⌋) and alternates between a phase-flip oracle and the Grover diffuser:

def __post_init__(self) -> None:
    n = 2 ** self.num_qubits
    self.iterations = max(1, int(math.pi / 4 * math.sqrt(n)))

def build_circuit(self) -> QuantumCircuit:
    qc = QuantumCircuit(self.num_qubits, self.num_qubits)
    qc.h(range(self.num_qubits))   # Equal superposition across all states
    for _ in range(self.iterations):
        qc.compose(self._oracle(), inplace=True)
        qc.compose(self._diffuser(), inplace=True)
    qc.measure(range(self.num_qubits), range(self.num_qubits))
    return qc

Advanced Protocols: Quantum Teleportation and Feed-Forward Control: advanced protocols such as Quantum Teleportation (teleportation.py) move an arbitrary quantum state between parties using pre-shared entanglement and classical communication channels. Crucially, Qiskit 1.0’s modern control-flow builder interface is utilized to handle post-measurement classical feed-forward operations conditionally:

qc.measure(0, 0)
qc.measure(1, 1)

# Conditional corrections using Qiskit 1.0 if_test context managers
with qc.if_test((qc.clbits[1], 1)):
    qc.x(2)  # Bob applies X if Alice's bit 1 is 1
with qc.if_test((qc.clbits[0], 1)):
    qc.z(2)  # Bob applies Z if Alice's bit 0 is 1

qc.measure(2, 2)

Summary of Python Execution Flow

Every algorithm adheres to a decoupled execution lifecycle:

Instantiation: Parameters (such as shots, target, or secret_string) are injected via standard Python dataclass fields.
Circuit Compilation: .build_circuit() constructs the structural QuantumCircuit using primitive gates.
Simulation: .run() transpiles the circuit and dispatches it to the local qiskit_aer.AerSimulator.
Interpretation: Raw counts are processed into human-readable semantic metrics and visualized via utility modules (such as plot_histogram and automated text/PNG diagram renderers).

Deep Dive: The C Language Demo Layer & Compilation Architecture

To demonstrate the full performance and low-level capabilities of the Qiskit ecosystem, the application includes a dedicated C language demo suite (src/c_demos/) built around the high-performance Qiskit C API (qiskit.h). Rather than limiting quantum programming exclusively to high-level Python wrappers, this layer exposes direct memory management, native circuit allocation, sparse observable construction, and optimized runtime compilation.

The C demonstration layer consists of three main components:

The Source Registry (c_source_registry.py): Stores metadata, build instructions, and expected outputs for standalone C programs like the Bell state (bell_state_c.c), multi-qubit GHZ state generators (ghz_state_c.c), and sparse Pauli-string observable builders (observable_demo_c.c).
The C Demo Manager (CDemoManager): A robust compilation and execution controller that automatically detects available Qiskit C runtime pathways, compiles source files on the fly, and securely executes them.
The C API Integration: Leverages primitives like qk_circuit_new, qk_circuit_gate, and qk_obs_zero to manipulate quantum structures with near-metal performance.

Dual Runtime Compilation Pathways (`CDemoManager`)

Because executing compiled native code within a web framework UI requires careful environment handling, CDemoManager implements a sophisticated runtime detection strategy supporting two distinct pathways alongside a graceful fallback:

PATH A — Standalone Shared Library (libqiskit.{so,dylib}): Triggered when users build Qiskit from source via make c or provide an explicit QISKIT_C_LIB environment variable. The manager compiles standalone C source code via gcc linking directly against -lqiskit and setting runtime path variables (DYLD_LIBRARY_PATH / LD_LIBRARY_PATH).
PATH B — Bundled Python Extension (qiskit.capi): Designed for standard pip installations (pip install qiskit >= 1.0), which ship qiskit.h alongside qiskit._accelerate (the Rust-compiled shared library).

The Thread-Safety Challenge & Subprocess Workaround

A critical engineering hurdle solved by CDemoManager involves macOS dynamic linker (dyld) restrictions. Streamlit executes user code within background worker threads. On macOS, calling dlopen() on a compiled shared extension from a non-main thread triggers a fatal SIGABRT crash because Mach-O segment initializers require the main thread.

To bypass this safely, CDemoManager spawns an isolated child Python process via subprocess.run(). The child compilation workflow performs the following steps securely:

Pre-flight Checks: Verifies that the C compiler (gcc/cc), qiskit.h, Python.h, and -undefined dynamic_lookup linker flags are fully operational.
Dynamic Extension Patching: Wraps the C source with custom Python C-API module bindings (PyMethodDef, PyModuleDef), renaming main() to an internal handler _qk_demo_run() and calling qk_import() inside PyInit_*().
Child Execution: Compiles the extension inside an isolated temporary directory, loads it via importlib.util.exec_module() on the child process's main thread, captures stdout streams via fmemopen, and returns structured JSON telemetry back to the main application without risking server instability.

Core C API Code Patterns

The underlying C implementations mirror high-level Qiskit logic with explicit memory control. For example, constructing a Bell State in bell_state_c.c allocates circuit structures and maps quantum gates directly:

/* ============================================================================
 * bell_state_c.c — Bell State using the Qiskit C API
 * ============================================================================
 * Demonstrates quantum entanglement by creating a Bell state |Φ+⟩:
 *   |Φ+⟩ = (|00⟩ + |11⟩) / √2
 *
 * Build instructions (UNIX/macOS):
 *   1. Build Qiskit C library from source:
 *      git clone https://github.com/Qiskit/qiskit.git
 *      cd qiskit && make c
 *
 *   2. Compile this file:
 *      gcc bell_state_c.c -o bell_state \
 *          -I /path/to/qiskit/dist/c/include \
 *          -L /path/to/qiskit/dist/c/lib \
 *          -lqiskit \
 *          -Wl,-rpath,/path/to/qiskit/dist/c/lib
 *
 *   3. Run:
 *      ./bell_state
 *
 * References:
 *   - Qiskit C API: https://quantum.cloud.ibm.com/docs/en/guides/install-c-api
 *   - Qiskit GitHub: https://github.com/Qiskit/qiskit
 * ============================================================================
 */

#include <stdio.h>
#include <stdint.h>
#include <qiskit.h>

int main(void) {
    printf("=== Bell State Demo (Qiskit C API) ===\\n\\n");

    /* ── 1. Create a 2-qubit circuit with 2 classical bits ──────────── */
    QkCircuit *qc = qk_circuit_new(2, 2);
    if (!qc) {
        fprintf(stderr, "ERROR: Failed to allocate circuit\\n");
        return 1;
    }

    /* ── 2. Apply Hadamard gate to qubit 0 ──────────────────────────── */
    /* H gate: |0⟩ → (|0⟩ + |1⟩) / √2   (creates superposition)       */
    uint32_t q0[1] = {0};
    qk_circuit_gate(qc, QkGate_H, q0, NULL);

    /* ── 3. Apply CNOT (CX) gate: control=0, target=1 ───────────────── */
    /* CX gate: entangles the two qubits                                 */
    uint32_t cx_qubits[2] = {0, 1};
    qk_circuit_gate(qc, QkGate_CX, cx_qubits, NULL);

    /* ── 4. Measure both qubits ─────────────────────────────────────── */
    /* qk_circuit_measure(circuit, qubit_index, clbit_index) — one call per qubit */
    qk_circuit_measure(qc, 0, 0);
    qk_circuit_measure(qc, 1, 1);

    /* ── 5. Print circuit information ──────────────────────────────────*/
    printf("Num instructions: %lu\\n", qk_circuit_num_instructions(qc));
    printf("Num qubits:       %u\\n",  qk_circuit_num_qubits(qc));
    printf("Num clbits:       %u\\n",  qk_circuit_num_clbits(qc));
    char *circ_str = qk_circuit_draw(qc, NULL);
    printf("Circuit string:\\n%s\\n\\n", circ_str);
    qk_str_free(circ_str);

    printf("Quantum state created: |Φ+⟩ = (|00⟩ + |11⟩) / √2\\n");
    printf("Expected measurement:  50%% |00⟩,  50%% |11⟩\\n");
    printf("Entanglement verified: qubits always collapse to the same value.\\n");

    /* ── 6. Free circuit memory ─────────────────────────────────────── */
    qk_circuit_free(qc);

    printf("\\n=== Demo complete ===\\n");
    return 0;
}

Similarly, advanced components like Sparse Observables (observable_demo_c.c) construct complex Pauli-string measurements (O=2⋅(X0⊗Y1⊗Z2)+0.5⋅(Z0⊗Z1)O=2⋅(*X*0⊗*Y*1⊗*Z*2)+0.5⋅(*Z*0⊗*Z*1)) for use with Estimator primitives:

/* ============================================================================
 * observable_demo_c.c — Sparse Observable Construction (Qiskit C API)
 * ============================================================================
 * Demonstrates quantum observable construction using the Qiskit C API.
 * Observables are fundamental to quantum computing: they represent measurable
 * physical quantities (e.g., energy, magnetisation).
 *
 * This example builds the observable:
 *   O = 2·(X₀ ⊗ Y₁ ⊗ Z₂) + 0.5·(Z₀ ⊗ Z₁)
 *
 * Build instructions (UNIX/macOS):
 *   1. Build Qiskit C library:
 *      git clone https://github.com/Qiskit/qiskit.git
 *      cd qiskit && make c
 *
 *   2. Compile:
 *      gcc observable_demo_c.c -o observable_demo \
 *          -I /path/to/qiskit/dist/c/include \
 *          -L /path/to/qiskit/dist/c/lib \
 *          -lqiskit \
 *          -Wl,-rpath,/path/to/qiskit/dist/c/lib
 *
 * References:
 *   - Qiskit C API docs: https://quantum.cloud.ibm.com/docs/en/guides/install-c-api
 *   - Qiskit GitHub:     https://github.com/Qiskit/qiskit
 * ============================================================================
 */

#include <stdio.h>
#include <stdint.h>
#include <qiskit.h>

int main(void) {
    printf("=== Sparse Observable Demo (Qiskit C API) ===\\n\\n");

    uint32_t num_qubits = 10;

    /* ── 1. Allocate a zero observable on 10 qubits ──────────────────── */
    QkObs *obs = qk_obs_zero(num_qubits);
    if (!obs) {
        fprintf(stderr, "ERROR: Failed to allocate observable\\n");
        return 1;
    }

    /* ── 2. Add term:  2·(X₀ Y₁ Z₂) ──────────────────────────────────── */
    {
        QkComplex64 coeff    = {2.0, 0.0};            /* coefficient 2+0i  */
        QkBitTerm   bits[3]  = {QkBitTerm_X,
                                 QkBitTerm_Y,
                                 QkBitTerm_Z};
        uint32_t    idxs[3]  = {0, 1, 2};             /* qubit indices     */
        QkObsTerm   term     = {coeff, 3, bits, idxs, num_qubits};
        qk_obs_add_term(obs, &term);
    }

    /* ── 3. Add term:  0.5·(Z₀ Z₁) ───────────────────────────────────── */
    {
        QkComplex64 coeff   = {0.5, 0.0};
        QkBitTerm   bits[2] = {QkBitTerm_Z, QkBitTerm_Z};
        uint32_t    idxs[2] = {0, 1};
        QkObsTerm   term    = {coeff, 2, bits, idxs, num_qubits};
        qk_obs_add_term(obs, &term);
    }

    /* ── 4. Query properties ─────────────────────────────────────────── */
    printf("Num qubits:  %u\\n",  qk_obs_num_qubits(obs));
    printf("Num terms:   %lu\\n", qk_obs_num_terms(obs));
    char *obs_str = qk_obs_str(obs);
    printf("Observable:\\n  %s\\n\\n", obs_str);
    qk_str_free(obs_str);

    printf("Observable represents: O = 2·(X₀⊗Y₁⊗Z₂) + 0.5·(Z₀⊗Z₁)\\n");
    printf("Use with EstimatorV2 primitive to compute ⟨ψ|O|ψ⟩.\\n");

    /* ── 5. Free memory ──────────────────────────────────────────────── */
    qk_obs_free(obs);

    printf("\\n=== Demo complete ===\\n");
    return 0;
}

Summary: The C Language Demo Layer & Compilation Architecture

The C language demo layer (src/c_demos/) provides high-performance, low-level quantum circuit manipulation and observable construction utilizing the native Qiskit C API (qiskit.h). Designed to bridge high-level Python software with near-metal execution, this layer is organized into three primary modules:

The C Source Registry (c_source_registry.py): Stores metadata, expected outputs, and source code paths for standalone demonstrations including Bell states (bell_state_c.c), multi-qubit GHZ entanglement chains (ghz_state_c.c), and sparse Pauli-string observable builders (observable_demo_c.c).
The C Demo Manager (CDemoManager): A dynamic compilation controller that automatically detects available Qiskit C runtime environments, handles compiler handshakes, and manages execution pathways.
Direct C API Primitives: Employs low-level functions such as qk_circuit_new, qk_circuit_gate, qk_obs_zero, and qk_obs_add_term to construct circuits and observables with precise memory control.

Dual Runtime Compilation Pathways

CDemoManager supports two distinct execution paths alongside a graceful fallback when no C library is present:

PATH A (Standalone Shared Library): Compiles standalone C files directly against a custom build of libqiskit.{so,dylib} generated via git clone and make c, utilizing QISKIT_C_LIB environment variables and linker search paths.
PATH B (Bundled Python Extension): Automatically detects headers from standard pip installations (pip install qiskit >= 1.0) via qiskit.capi and links against the Rust-compiled shared library (qiskit._accelerate). To prevent fatal SIGABRT crashes caused by macOS dynamic linker (dyld) restrictions on background worker threads, compilation and execution are offloaded to an isolated child subprocess where dlopen() safely runs on the main thread.

Conclusion

In conclusion, Bob has successfully delivered a comprehensive, production-ready quantum computing demonstration suite that seamlessly bridges high-level Python application logic with low-level C performance. Through the Streamlit graphical interface and Python algorithm layer, users can interactively explore foundational quantum mechanics—ranging from entangled Bell states and exponential query advantages in the Bernstein-Vazirani and Deutsch-Jozsa algorithms to quadratic speedups in Grover's search, Quantum Fourier Transforms, and quantum teleportation. Furthermore, by engineering the C demo layer and CDemoManager, Bob has provided robust dual-compilation pathways that leverage either standalone shared libraries or bundled pip installations (qiskit.capi). Complete with thread-safe child subprocess execution for macOS stability, automated pre-flight checks, sparse observable construction, and extensive test coverage, Bob’s implementation offers an elegant, highly versatile blueprint for exploring utility-scale quantum software development.

Thanks for reading ⚛️

Links & References (stuff to read :))

Code repository for this post: https://github.com/aairom/qiskit-test-Bob
Qiskit GitHub: https://github.com/Qiskit/qiskit
Qiskit C API Guide: https://quantum.cloud.ibm.com/docs/en/guides/install-c-api
IBM Runtime C Library: https://github.com/Qiskit/qiskit-ibm-runtime-c
Qiskit Documentation: https://github.com/Qiskit/documentation
RustWorkX: https://github.com/Qiskit/rustworkx
IBM Quantum Platform: https://quantum.cloud.ibm.com
Learn Quantum Computing with Python and IBM Quantum: https://www.packtpub.com/en-fr/product/learn-quantum-computing-with-python-and-ibm-quantum-9781803240923
Quantum Computing in Practice with Qiskit® and IBM Quantum Experience®: https://www.packtpub.com/en-fr/product/quantum-computing-in-practice-with-qiskit-and-ibm-quantum-experience-9781838821036
Make Your Org Quantum-Safe Without the Hype - FAST: https://www.packtpub.com/en-fr/product/make-your-org-quantum-safe-without-the-hype-fast-9781808490323
Quantum Computing for AI with Python: https://www.packtpub.com/en-fr/product/quantum-computing-for-ai-with-python-9781808088322
Dancing with Qubits: From qubits to algorithms, embark on the quantum computing journey shaping our future , Second Edition: https://www.packtpub.com/en-fr/product/dancing-with-qubits-9781837634620
Building Quantum Software with Python: https://www.manning.com/books/building-quantum-software-with-python
Quantum Programming in Depth: https://www.manning.com/books/quantum-programming-in-depth
Quantum Computing in Action: https://www.manning.com/books/quantum-computing-in-action
Real-World Cryptography: https://www.manning.com/books/real-world-cryptography

Automating Rules Importation in Bob IDE

Alain Airom (Ayrom) — Mon, 19 Oct 2026 08:25:00 +0000

Automating rules import in Bob IDE.

Introduction

In modern agentic platforms and AI-assisted development environments like IBM Bob, maintaining consistent context and project-specific guidelines ("rules") across multiple repositories is crucial. However, manually copying rule files into every new project creates friction and inconsistency. To streamline this workflow, we can implement a dedicated extension that automatically imports a centralized set of rules into any newly created or opened project workspace.

In the context of agentic AI platforms like IBM Bob or Claude or..., "rules" refer to centralized configuration guidelines, project-specific instructions, and behavioral constraints that define how the assistant should interact, code, or execute tasks within a given workspace. Stored locally within project directories (such as .bob/rules) or managed globally, these instructions ensure that the AI automatically adapts to specific coding standards, architectural patterns, and workflow preferences every time a new project or repository is opened.

This blog post explores how to design and implement a rules synchronization extension for IBM Bob and VS Code, detailing its architecture, core synchronization engine, event wiring, and quickstart setup.

Implementation

The implementation is split into two primary layers: a pure Node.js core module (syncCore.ts) that handles the file system logic safely without any IDE dependencies, and an extension host layer (extension.ts) that manages user settings, workspace events, commands, and visual feedback.

Architecture Overview

The architecture separates core synchronization logic from IDE-specific event handling to ensure clean unit testing and robust path safety.

When a project is opened or a new folder is added to the workspace, the extension triggers the synchronization pipeline:

Core Synchronization Logic (`syncCore.ts`)

The core module operates independently of VS Code/Bob APIs. It enforces strict path-safety models to prevent path traversal and recursive copying issues. For instance, safeChild validates each directory entry name before joining paths:

/**
 * Core sync logic. Intentionally free of any `vscode` import so it can be unit tested with plain Node.
 *
 * Path-safety model:
 *  - The SOURCE directory is chosen by the user in settings (trusted, but must be absolute, NUL-free).
 *  - The TARGET folder is relative and must stay inside the project root (checked by `resolveTargetDir`).
 *  - Every child name read from disk is validated by `safeChild` before being appended to a base path.
 */
import * as fs from 'fs';
import * as os from 'os';
import * as path from 'path';

/* */
/* */
/* */
/* */

export function safeChild(base: string, name: string): string {
  if (
    !name ||
    name === '.' ||
    name === '..' ||
    name.includes('/') ||
    name.includes('\\') ||
    name.includes('\0')
  ) {
    throw new SyncAbortError(`Unsafe file name encountered: ${JSON.stringify(name)}`, 'EUNSAFE_NAME');
  }
  return base.endsWith(path.sep) ? base + name : base + path.sep + name;
}

The primary entry point syncRules verifies the source directory, resolves target folders safely within the project root, checks for recursion risks, and recursively copies files according to the specified conflict strategy (overwrite or skip).

Extension Event Wiring (`extension.ts`)

The extension binds IDE events, user settings, and commands to the synchronization engine. Upon activation, it sets up status bar items, registers commands like bobRulesSync.syncNow, and listens to workspace folder modifications:

/**
 * Bob Rules Sync - VS Code / IBM Bob extension entry point.
 *
 * Triggers a rules sync when:
 *  - the extension activates (a project was opened in Bob), and
 *  - a folder is added to the workspace (a new project was created/added).
 * Users can also run "Bob Rules Sync: Sync Rules Now" at any time.
 */
import * as vscode from 'vscode';
import { ConflictStrategy, SyncAbortError, SyncResult, summarize, syncRules } from './syncCore';

const CONFIG_SECTION = 'bobRulesSync';

let output: vscode.OutputChannel;
let statusBar: vscode.StatusBarItem;
let extensionId = '';

/* */
/* */
/* */
/* */


export function activate(context: vscode.ExtensionContext): void {
  extensionId = context.extension.id;
  output = vscode.window.createOutputChannel('Bob Rules Sync');
  statusBar = vscode.window.createStatusBarItem(vscode.StatusBarAlignment.Left, 0);
  statusBar.command = 'bobRulesSync.syncNow';
  context.subscriptions.push(output, statusBar);

  context.subscriptions.push(
    vscode.commands.registerCommand('bobRulesSync.syncNow', () => syncAll('manual command', true)),
    // New project created / folder added to the workspace.
    vscode.workspace.onDidChangeWorkspaceFolders(async (e) => {
      for (const folder of e.added) {
        await syncFolder(folder, 'folder added', false);
      }
    })
  );

  log('INFO', 'Bob Rules Sync activated.');
  void syncAll('project opened', false);
}

Quickstart Guide

To get started with the rules synchronization extension in your development environment:

Make sure the source folder exists and contains your rules (default /Users/xxxx/Devs/_myRules).
Install the extension using scripts/install.sh or via the built VSIX artifact (.bob/artifacts/bob-rules-sync-1.0.0.vsix) through Extensions > ... > Install from VSIX.
Reload the Bob window and open any project to automatically copy rules into .bob/rules.
Customize target folders or trigger manual syncs via the Bob command palette or status bar item.

Conclusion

By building a dedicated extension for rule importation in Bob IDE, we have successfully automated the distribution of development standards across projects. Decoupling the pure Node.js synchronization core from the IDE extension host ensures high testability, strong path safety against traversal attacks, and a seamless developer experience. This automated approach saves hours of manual configuration time and guarantees that every new workspace starts with the proper context and agent rules ready to go.

Thanks for reading 🤖

Architecting Sovereign AI Agents: The Evolution, Implementation, and Mechanics of the CUGA Agent Harness

Alain Airom (Ayrom) — Sat, 17 Oct 2026 14:47:22 +0000

News on CUGA and CUGA-Agent for Sovereignty.

It has been a while since I’ve written about CUGA, but CUGA has had a lot of changes and advances and could be used in a context of sovereign agent implementation. As enterprises and sovereign technical organizations increasingly demand local control, zero data leakage, and uncompromising guardrails over AI-driven automation, the choice of agent harness becomes paramount. Enter CUGA (Configurable Generalist Agent) — IBM Research’s open-source enterprise agent harness — paired with the comprehensive architecture of the cuga-agent-harness reference implementation.

In this deep-dive blog post, we will explore how CUGA bridges the gap between chaotic LLM behavior and predictable, enterprise-grade, sovereign execution. We will examine its architectural patterns, review production code excerpts from the harness, and discuss how to operationalize it securely.

An Intrduction to CUGA from Official GitHub Repository

In short, CUGA is the enterprise agent harness for sovereign, policy-controlled automation.

CUGA turns business requests into actions across your applications, APIs, and documents. Deploy in your private cloud or prepare an air-gapped environment, enforce fine-grained execution policies, and use open models on infrastructure you control. Connect your tools, configure your workflows, and optimize inference for your workload.

The Moto: Automate enterprise work. Keep control.


| Your priority            | What CUGA gives you                                          |
| ------------------------ | ------------------------------------------------------------ |
| **Sovereignty**          | Run the harness in your environment. Connect privately hosted models and tools. Preload supporting model artifacts for disconnected deployments. |
| **Execution policies**   | Block disallowed intents, require approval for selected tools, guide workflows with playbooks, and configure tool-level guards. |
| **Inference efficiency** | Choose open models, tune reasoning modes, shortlist relevant tools, set tool-call budgets, and inspect token usage with per-run receipts. |
| **Faster delivery**      | Connect MCP servers, OpenAPI services, and Python tools. Configure and test an agent in the UI before publishing a version. |
| **Recurring automation** | Use the separate events service to run agents on schedules, webhooks, and application events, with run history in Events Studio. |
| **Reusable experience**  | Load task-specific skills and use a configured Evolve integration to retrieve relevant guidance and user preferences. |

CUGA in the Context of Sovereign Agent Implementation

The race toward autonomous agents often runs into a brick wall of operational realities: prompt drifting, hallucinations leading to unintended API calls, lack of data provenance, and strict data residency requirements (sovereignty). Traditional wrapper frameworks treat the LLM as the core orchestrator, leaving security and control up to fragile system prompts.

CUGA flips this paradigm. It treats the LLM as a planning and reasoning engine while enclosing it within a strict, deterministically enforced harness. This separation of concerns is what makes CUGA uniquely suited for sovereign agent implementations:

Zero Data Exfiltration: By running the underlying models locally via llama.cpp or Ollama, or securely through enterprise-managed, cloud-hosted LLMs within private virtual networks, every prompt, embedding, and tool execution stays entirely within air-gapped or localized perimeter perimeters.
Harness-Enforced Policies: Guardrails (such as Intent Guards, Tool Approvals, and Playbooks) are evaluated and enforced outside the model's immediate token stream, ensuring compliance even if the model attempts to bypass safety directives.
Auditable Execution Trails: Every run returns structured run receipts and deterministic trajectory logs, making auditing straightforward for compliance-heavy industries

Core Architecture: How the CUGA Showcase & Harness Operate

The CUGA Showcase and agent harness (cuga-agent-harness) are built around a framework-agnostic Python core (cuga_showcase/) paired with an intuitive Streamlit UI or headless execution interface.

System Architecture Flow

The system is designed with graceful degradation in mind. If a local model server becomes unreachable or returns an error, the BackendManager automatically manages fallbacks (e.g., from Live or local inference servers back to deterministic offline simulation) without crashing the user session.

Technical Stack

llama.cpp
Ollama
or with sovereign ‘on-premise cloud’ based LLMs.

Data for the Implementation

The 'CRM' data used for this demo application was generated synthetically.

"""Synthetic CRM data and the in-memory store the demo tools operate on.

All data is fictional and local, so the showcase has no cloud dependency.
"""

from __future__ import annotations

import copy
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

# Account records. ``usage`` is monthly active seats for the last six months.
_ACCOUNTS: Dict[str, Dict[str, Any]] = {
    "A-100": {
        "id": "A-100", "name": "Acme Corp", "plan": "Enterprise", "arr": 240000,
        "renewal_date": "2026-12-01", "nps": 52, "owner": "Dana Whitfield",
        "usage": [820, 845, 890, 910, 960, 1010],
    },
    "A-200": {
        "id": "A-200", "name": "Globex", "plan": "Pro", "arr": 96000,
        "renewal_date": "2026-11-15", "nps": 18, "owner": "Marcus Lee",
        "usage": [700, 660, 610, 540, 500, 455],
    },
    "A-300": {
        "id": "A-300", "name": "Initech", "plan": "Starter", "arr": 18000,
        "renewal_date": "2027-02-01", "nps": 35, "owner": "Priya Nair",
        "usage": [200, 201, 199, 202, 200, 203],
    },
    "A-400": {
        "id": "A-400", "name": "Umbrella", "plan": "Enterprise", "arr": 410000,
        "renewal_date": "2026-12-20", "nps": 41, "owner": "Dana Whitfield",
        "usage": [1500, 1490, 1470, 1460, 1440, 1430],
    },
}

_TICKETS: List[Dict[str, Any]] = [
    {"id": "T-1", "account_id": "A-100", "priority": "P3", "title": "Dashboard colour palette request", "status": "open"},
    {"id": "T-2", "account_id": "A-200", "priority": "P1", "title": "API outage in EU region", "status": "open"},
    {"id": "T-3", "account_id": "A-200", "priority": "P2", "title": "Export job times out", "status": "open"},
    {"id": "T-4", "account_id": "A-200", "priority": "P2", "title": "SSO login loop", "status": "open"},
    {"id": "T-5", "account_id": "A-400", "priority": "P2", "title": "Slow report generation", "status": "open"},
    {"id": "T-6", "account_id": "A-300", "priority": "P3", "title": "Invoice address change", "status": "closed"},
]

# Health score penalties per open ticket priority.
_TICKET_PENALTY = {"P1": 15, "P2": 6, "P3": 2}


def usage_trend_pct(usage: List[int]) -> float:
    """Percent change of the last three months vs the three months before."""
    if len(usage) < 6:
        return 0.0
    prev = sum(usage[-6:-3]) / 3
    last = sum(usage[-3:]) / 3
    return round((last - prev) / prev * 100, 1) if prev else 0.0


def health_label(score: int) -> str:
    """Map a numeric score to a text label (never colour alone)."""
    if score >= 75:
        return "healthy"
    if score >= 50:
        return "needs attention"
    return "at risk"


def compute_health(account: Dict[str, Any], tickets: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Deterministic account health score in the 0-100 range.

    score = 100 - ticket penalties + trend bonus (capped +-15) + NPS adjustment
    """
    penalty = sum(_TICKET_PENALTY.get(t["priority"], 0) for t in tickets if t["status"] == "open")
    trend = usage_trend_pct(account["usage"])
    trend_bonus = max(-15.0, min(15.0, trend * 0.8))
    nps_adj = (account["nps"] - 30) * 0.2
    score = int(round(max(0.0, min(100.0, 100 - penalty + trend_bonus + nps_adj - 10))))
    return {
        "account_id": account["id"],
        "score": score,
        "label": health_label(score),
        "drivers": {
            "ticket_penalty": penalty,
            "usage_trend_pct": trend,
            "nps": account["nps"],
        },
    }


@dataclass
class CrmStore:
    """Mutable in-memory CRM. One instance per browser session."""

    accounts: Dict[str, Dict[str, Any]] = field(default_factory=lambda: copy.deepcopy(_ACCOUNTS))
    tickets: List[Dict[str, Any]] = field(default_factory=lambda: copy.deepcopy(_TICKETS))
    outbox: List[Dict[str, Any]] = field(default_factory=list)
    deleted: List[str] = field(default_factory=list)

    def find_account(self, query: str) -> Optional[Dict[str, Any]]:
        """Find an account by id or (partial, case-insensitive) name."""
        q = (query or "").strip().lower()
        if not q:
            return None
        for acc in self.accounts.values():
            if acc["id"].lower() == q or acc["name"].lower() == q:
                return acc
        for acc in self.accounts.values():
            if q in acc["name"].lower():
                return acc
        return None

    def open_tickets(self, account_id: str) -> List[Dict[str, Any]]:
        return [t for t in self.tickets if t["account_id"] == account_id and t["status"] == "open"]

    def reset(self) -> None:
        """Restore the pristine demo data."""
        self.accounts = copy.deepcopy(_ACCOUNTS)
        self.tickets = copy.deepcopy(_TICKETS)
        self.outbox.clear()
        self.deleted.clear()

Implementation: Architecture, Policies, and Code

To see how this works in practice, let's examine the core components that govern tool execution, policy enforcement, and runtime receipts within the harness.

Tool Envelopes and Safety Envelopes: every action an agent can take is wrapped in a standardized tool specification. To ensure the model can gracefully recover from execution errors (such as missing parameters or unauthorized states), tools return standardized JSON envelopes rather than throwing unhandled exceptions:

{"ok": True,  "data": {...}}
{"ok": False, "code": "not_found", "error": "No account matches 'Foo'."}

The code below shows how tools are registered and categorized into distinct sub-agents under a supervisor framework (cuga_showcase/tools.py pattern):

"""Demo tools shared by the offline simulator and the live ``CugaAgent``.

Each tool is a plain function returning the *envelope* convention used by the
official cuga-apps reference apps::

    {"ok": True,  "data": ...}
    {"ok": False, "code": "not_found", "error": "..."}

``ToolSpec`` carries the metadata (description, sensitivity, owning sub-agent)
so the same registry can be exposed as LangChain tools in Live mode.
"""

from __future__ import annotations

import ast
import operator
from dataclasses import dataclass
from typing import Any, Callable, Dict, List, Optional

from .demo_data import CrmStore, compute_health, usage_trend_pct

Envelope = Dict[str, Any]


def ok(data: Any) -> Envelope:
    """Success envelope."""
    return {"ok": True, "data": data}


def err(code: str, message: str) -> Envelope:
    """Failure envelope."""
    return {"ok": False, "code": code, "error": message}


# --------------------------------------------------------------------------
# Safe calculator (no eval): walks the AST and allows arithmetic only.
# --------------------------------------------------------------------------
_BIN_OPS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
    ast.Pow: operator.pow,
    ast.Mod: operator.mod,
}
_UNARY_OPS = {ast.UAdd: operator.pos, ast.USub: operator.neg}


def _eval_node(node: ast.AST) -> float:
    if isinstance(node, ast.Constant) and isinstance(node.value, (int, float)) and not isinstance(node.value, bool):
        return node.value
    if isinstance(node, ast.BinOp) and type(node.op) in _BIN_OPS:
        left, right = _eval_node(node.left), _eval_node(node.right)
        if isinstance(node.op, ast.Pow) and abs(right) > 12:
            raise ValueError("exponent too large")
        return _BIN_OPS[type(node.op)](left, right)
    if isinstance(node, ast.UnaryOp) and type(node.op) in _UNARY_OPS:
        return _UNARY_OPS[type(node.op)](_eval_node(node.operand))
    raise ValueError("unsupported expression")


def calculator(expression: str) -> Envelope:
    """Evaluate an arithmetic expression such as ``(240000 + 96000) * 0.1``."""
    try:
        tree = ast.parse((expression or "").strip(), mode="eval")
        value = _eval_node(tree.body)
    except ZeroDivisionError:
        return err("math_error", "Division by zero.")
    except (ValueError, SyntaxError):
        return err("bad_expression", "Only plain arithmetic is supported.")
    return ok({"expression": expression, "value": round(value, 6)})


# --------------------------------------------------------------------------
# CRM tools (bound to a CrmStore through ``build_registry``)
# --------------------------------------------------------------------------
@dataclass(frozen=True)
class ToolSpec:
    """Metadata + callable for one tool."""

    name: str
    description: str
    fn: Callable[..., Envelope]
    sensitive: bool = False  # sensitive tools are the usual approval candidates
    agent: str = "cuga"  # owning sub-agent in supervisor mode
    params: Optional[Dict[str, str]] = None  # param name -> short description

    def call(self, **kwargs: Any) -> Envelope:
        """Invoke the tool, converting unexpected exceptions into envelopes."""
        try:
            return self.fn(**kwargs)
        except TypeError as exc:  # wrong/missing arguments
            return err("bad_arguments", str(exc))
        except Exception as exc:  # noqa: BLE001 - tools must never crash a run
            return err("tool_error", str(exc))


def build_registry(
    store: CrmStore, knowledge_search: Optional[Callable[[str], Envelope]] = None
) -> Dict[str, ToolSpec]:
    """Create the tool registry bound to ``store``.

    Args:
        store: The session's CRM store.
        knowledge_search: Optional callable backing ``search_knowledge``.
    """

    def get_account(query: str) -> Envelope:
        acc = store.find_account(query)
        if not acc:
            return err("not_found", f"No account matches '{query}'.")
        return ok({k: v for k, v in acc.items() if k != "usage"})

    def list_open_tickets(account_id: str) -> Envelope:
        acc = store.find_account(account_id)
        if not acc:
            return err("not_found", f"No account matches '{account_id}'.")
        return ok(store.open_tickets(acc["id"]))

    def compute_health_score(account_id: str) -> Envelope:
        acc = store.find_account(account_id)
        if not acc:
            return err("not_found", f"No account matches '{account_id}'.")
        return ok(compute_health(acc, store.open_tickets(acc["id"])))

    def get_usage_trend(account_id: str) -> Envelope:
        acc = store.find_account(account_id)
        if not acc:
            return err("not_found", f"No account matches '{account_id}'.")
        return ok({"account_id": acc["id"], "monthly_seats": acc["usage"], "trend_pct": usage_trend_pct(acc["usage"])})

    def list_accounts() -> Envelope:
        return ok([{"id": a["id"], "name": a["name"], "plan": a["plan"], "arr": a["arr"]} for a in store.accounts.values()])

    def send_email(to: str, subject: str, body: str) -> Envelope:
        if "@" not in (to or ""):
            return err("bad_arguments", "Recipient must be an email address.")
        store.outbox.append({"to": to, "subject": subject, "body": body})
        return ok({"queued": True, "to": to, "subject": subject})

    def delete_account(account_id: str) -> Envelope:
        acc = store.find_account(account_id)
        if not acc:
            return err("not_found", f"No account matches '{account_id}'.")
        store.accounts.pop(acc["id"])
        store.deleted.append(acc["id"])
        return ok({"deleted": acc["id"], "name": acc["name"]})

    def search_knowledge(query: str) -> Envelope:
        if knowledge_search is None:
            return err("unavailable", "Knowledge base is not configured.")
        return knowledge_search(query)

    specs = [
        ToolSpec("list_accounts", "List all CRM accounts.", list_accounts, agent="crm_agent", params={}),
        ToolSpec("get_account", "Look up one account by id or name.", get_account, agent="crm_agent",
                 params={"query": "Account id (A-100) or name (Acme)"}),
        ToolSpec("get_usage_trend", "Monthly active seats and 3-month trend for an account.", get_usage_trend,
                 agent="crm_agent", params={"account_id": "Account id or name"}),
        ToolSpec("list_open_tickets", "Open support tickets for an account.", list_open_tickets, agent="support_agent",
                 params={"account_id": "Account id or name"}),
        ToolSpec("compute_health_score", "Deterministic 0-100 health score with drivers.", compute_health_score,
                 agent="support_agent", params={"account_id": "Account id or name"}),
        ToolSpec("calculator", "Evaluate an arithmetic expression.", calculator, agent="crm_agent",
                 params={"expression": "Arithmetic such as 240000*0.1"}),
        ToolSpec("search_knowledge", "Search the local knowledge base and return cited passages.", search_knowledge,
                 agent="crm_agent", params={"query": "Natural-language question"}),
        ToolSpec("send_email", "Queue an email to a contact.", send_email, sensitive=True, agent="comms_agent",
                 params={"to": "Recipient address", "subject": "Subject line", "body": "Message body"}),
        ToolSpec("delete_account", "Permanently delete an account.", delete_account, sensitive=True,
                 agent="crm_agent", params={"account_id": "Account id or name"}),
    ]
    return {s.name: s for s in specs}


def tool_names(registry: Dict[str, ToolSpec]) -> List[str]:
    """Stable, sorted tool names."""
    return sorted(registry)

Enforcing Harness Policies: policies steer the agent outside the prompt. Five policy types are natively supported in the CUGA harness: **Intent Guard, Playbook, Tool Guide, Tool Approval, and Output Formatter.

When a sensitive tool like delete_account is invoked, a Tool Approval policy intercepts the execution flow, parks the run state (awaiting_approval), and requests human validation before letting the tool execute:

"""Policy engine for the offline simulator.

It reproduces, in plain Python, the *behaviour* of CUGA's five policy types so
users can experiment without an LLM:

* ``intent_guard``     - blocks a request before any planning happens
* ``playbook``         - injects step-by-step guidance when a request matches
* ``tool_guide``       - enriches a tool's description with extra guidance
* ``tool_approval``    - parks a tool call until a human approves it (HITL)
* ``output_formatter`` - reshapes the final answer

In Live mode the very same definitions are registered on the real
``agent.policies`` manager (see ``live.py``).
"""

from __future__ import annotations

import uuid
from dataclasses import dataclass, field
from typing import Dict, List, Optional

POLICY_TYPES = ("intent_guard", "playbook", "tool_guide", "tool_approval", "output_formatter")

TYPE_LABELS = {
    "intent_guard": "Intent Guard",
    "playbook": "Playbook",
    "tool_guide": "Tool Guide",
    "tool_approval": "Tool Approval",
    "output_formatter": "Output Formatter",
}

TYPE_HELP = {
    "intent_guard": "Blocks a request that matches keywords before the agent plans anything.",
    "playbook": "Adds step-by-step guidance when the request matches keywords.",
    "tool_guide": "Appends usage guidance to the descriptions of the target tools.",
    "tool_approval": "Pauses before the listed tools run and asks a human to approve.",
    "output_formatter": "Rewrites the final answer using a format instruction.",
}


@dataclass
class Policy:
    """One policy definition (superset of the fields of the five SDK types)."""

    name: str
    type: str
    keywords: List[str] = field(default_factory=list)
    content: str = ""  # playbook / tool_guide content, formatter instruction
    response: str = ""  # intent_guard response
    tools: List[str] = field(default_factory=list)  # tool_approval / tool_guide targets
    message: str = ""  # tool_approval prompt
    enabled: bool = True
    id: str = field(default_factory=lambda: uuid.uuid4().hex[:8])

    def matches(self, text: str) -> bool:
        """Case-insensitive keyword OR-match (the SDK's default operator)."""
        lowered = (text or "").lower()
        return any(k.strip().lower() in lowered for k in self.keywords if k.strip())


def default_policies() -> List[Policy]:
    """Starter policy set shown in the Policy Lab."""
    return [
        Policy(
            name="Block salary disclosure", type="intent_guard",
            keywords=["salary", "compensation of", "payroll"],
            response="I can't share compensation data. Please contact HR.",
        ),
        Policy(
            name="Churn-risk playbook", type="playbook",
            keywords=["churn", "at risk", "renewal risk"],
            content="1. Fetch the account.\n2. Check open P1/P2 tickets.\n3. Compare the 3-month usage trend.\n4. Recommend one concrete next action.",
        ),
        Policy(
            name="Email etiquette", type="tool_guide", tools=["send_email"],
            content="Keep emails under 120 words. Always include the account owner in the greeting.",
        ),
        Policy(
            name="Approve outbound & destructive actions", type="tool_approval",
            tools=["send_email", "delete_account"],
            message="This action changes data outside the conversation. Approve?",
        ),
        Policy(
            name="Executive summary format", type="output_formatter",
            keywords=[], content="Answer as a bold one-line verdict followed by up to three bullet points.",
            enabled=False,
        ),
    ]


class PolicySet:
    """A mutable collection of policies with the lookups the simulator needs."""

    def __init__(self, policies: Optional[List[Policy]] = None) -> None:
        self._items: Dict[str, Policy] = {}
        for p in policies if policies is not None else default_policies():
            self._items[p.id] = p

    # -- CRUD ---------------------------------------------------------------
    def add(self, policy: Policy) -> Policy:
        if policy.type not in POLICY_TYPES:
            raise ValueError(f"Unknown policy type: {policy.type}")
        if not policy.name.strip():
            raise ValueError("Policy name is required.")
        self._items[policy.id] = policy
        return policy

    def remove(self, policy_id: str) -> bool:
        return self._items.pop(policy_id, None) is not None

    def clear(self) -> None:
        self._items.clear()

    def toggle(self, policy_id: str, enabled: bool) -> None:
        if policy_id in self._items:
            self._items[policy_id].enabled = enabled

    def all(self) -> List[Policy]:
        return list(self._items.values())

    def active(self, ptype: str) -> List[Policy]:
        return [p for p in self._items.values() if p.type == ptype and p.enabled]

    # -- lookups used by the agent ------------------------------------------
    def intent_guard(self, message: str) -> Optional[Policy]:
        return next((p for p in self.active("intent_guard") if p.matches(message)), None)

    def playbooks(self, message: str) -> List[Policy]:
        return [p for p in self.active("playbook") if p.matches(message)]

    def tool_guides(self, tool: str) -> List[Policy]:
        return [p for p in self.active("tool_guide") if tool in p.tools or "*" in p.tools]

    def approval_for(self, tool: str) -> Optional[Policy]:
        return next((p for p in self.active("tool_approval") if tool in p.tools or "*" in p.tools), None)

    def formatter(self) -> Optional[Policy]:
        return next(iter(self.active("output_formatter")), None)

Dynamic Capability Injection via SKILL.md: a standout architectural feature of CUGA is its implementation of skills. Instead of cluttering the system prompt with hundreds of rigid instructions for every conceivable task, skills are modularized into independent directories containing a SKILL.md file featuring YAML front matter (name, description, tags) followed by procedural instructions.

"""Loader for ``SKILL.md`` files (the format used by cuga-skills).

A skill is a Markdown file with YAML-style front matter (``name``,
``description``, ``tags``) followed by instructions. CUGA loads skills from
``.cuga/skills``; here we also match them to requests in the simulator.
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Dict, List, Optional, Tuple

from .knowledge import tokenize


@dataclass(frozen=True)
class Skill:
    name: str
    description: str
    body: str
    tags: Tuple[str, ...] = field(default_factory=tuple)
    path: str = ""

    def score(self, message: str) -> float:
        """Share of the description's keywords found in ``message`` (0-1)."""
        wanted = set(tokenize(self.description))
        if not wanted:
            return 0.0
        return len(wanted & set(tokenize(message))) / len(wanted)


def parse_skill(text: str, path: str = "") -> Optional[Skill]:
    """Parse one SKILL.md; returns None when the front matter is invalid."""
    match = re.match(r"^---\s*\n(.*?)\n---\s*\n?(.*)$", text, re.DOTALL)
    if not match:
        return None
    meta: Dict[str, object] = {}
    current_list: Optional[str] = None
    for line in match.group(1).splitlines():
        if re.match(r"^\s+-\s+", line) and current_list:
            meta.setdefault(current_list, []).append(re.sub(r"^\s+-\s+", "", line).strip())  # type: ignore[union-attr]
            continue
        kv = re.match(r"^([A-Za-z_]+):\s*(.*)$", line)
        if kv:
            key, value = kv.group(1), kv.group(2).strip()
            if value:
                meta[key] = value
                current_list = None
            else:
                meta[key] = []
                current_list = key
    name, description = meta.get("name"), meta.get("description")
    if not isinstance(name, str) or not isinstance(description, str):
        return None
    tags = tuple(meta.get("tags", [])) if isinstance(meta.get("tags"), list) else ()
    return Skill(name=name, description=description, body=match.group(2).strip(), tags=tags, path=path)


def load_skills(folder: Path) -> List[Skill]:
    """Load every ``*/SKILL.md`` below ``folder`` (invalid files are skipped)."""
    skills: List[Skill] = []
    if not folder.is_dir():
        return skills
    for path in sorted(folder.rglob("SKILL.md")):
        try:
            skill = parse_skill(path.read_text(encoding="utf-8"), str(path))
        except (OSError, UnicodeDecodeError):
            continue
        if skill:
            skills.append(skill)
    return skills


def match_skill(skills: List[Skill], message: str, threshold: float = 0.25) -> Optional[Skill]:
    """Best skill for ``message`` when its score reaches ``threshold``."""
    best = max(skills, key=lambda s: s.score(message), default=None)
    return best if best and best.score(message) >= threshold else None

---
name: account-health-review
description: Use when the user asks how healthy, risky or satisfied a customer account is, or whether it may churn.
tags:
  - crm
---

# Account health review

1. Look up the account with `get_account`.
2. Call `compute_health_score` and `list_open_tickets`.
3. Call `get_usage_trend` and report the three-month trend.
4. Finish with one verdict (healthy, needs attention, at risk) and one recommended next action.

The current skills in the catalog:

incident-summary: Automatically invoked when support tickets need triage, grouping by priority (P1 first), and recommending the next remediation steps.
churn-risk-playbook: Evaluates active usage drops, open critical tickets, and ARR value to compile a comprehensive executive risk report.

This modular skill design ensures that sovereign deployments can easily drop in, update, or remove operational workflows without recompiling core agent codebases. The agent reads only the metadata descriptions and dynamically loads the full skill instruction set on demand when a user request requires it

Persona-Based Implementation & Visual Workflows

To make sovereign agent deployments practical, the cuga-agent-harness models real-world enterprise operations through concrete user stories and visual interactive artifacts (.bob/artifacts/user-stories-20261017.md and .bob/artifacts/user-stories-viz-20261017.html).

Persona 1: Sarah (Enterprise Customer Success Manager)

Goal: Quickly assess account health and proactively identify churn risks across enterprise accounts without manual CRM mining.
CUGA Implementation: Sarah interacts with the Supervisor agent, asking: "Which accounts are most at risk?" The CugaSupervisor delegates tasks to the crm_agent and support_agent, compiling usage trends, support ticket counts, and ARR data into a unified, cited report.

Persona 2: Marcus (SecOps & Compliance Officer)

Goal: Ensure all autonomous AI operations comply with strict corporate security policies and data residency mandates.
CUGA Implementation: Marcus configures Tool Approval policies for high-consequence tools
(such as delete_account or send_email). When an agent attempts to purge a record, the harness intercepts the call, transitions the state to awaiting_approval, and halts execution until Marcus explicitly clicks Approve or Deny in the UI.

Persona 3: Elena (Local AI Developer / Infrastructure Engineer)

Goal: Deploy and run advanced agentic workflows locally on air-gapped workstations using local models.
CUGA Implementation: Elena utilizes the llama.cpp, Ollama local backends or cloud based LLMs with automatic health checks and graceful fallbacks, inspecting detailed Run Receipts to verify token consumption, wall-clock timing, and tool execution latencies.

Visualizing the Application and Use-Cases

The interactive visual artifact (https://github.com/aairom/cuga-agent-harness/blob/main/.bob/artifacts/user-stories-viz-20261017.html) provides a rendered, node-based representation of these persona pathways, mapping user intents directly to harness policy validation gates, multi-agent delegations, and tool envelope outcomes.

Inspecting Run Receipts: unlike standard LLM telemetry that tracks monetary costs (which vary wildly and complicate sovereign budgeting), CUGA's run receipt focuses strictly on cryptographic and operational determinism: tokens, LLM call counts, wall-clock timing, and tool execution durations.

Conclusion

The evolution of CUGA as demonstrated in the cuga-agent-harness proves that autonomous AI agents do not need to operate as black boxes. By combining local LLM backends (via Ollama or llama.cpp), or cloud based LLM connections, deterministic harness-level policy enforcement, and structured tool envelopes, organizations can finally deploy sovereign agents that respect data boundaries, require human-in-the-loop approvals for critical actions, and provide rigorous audit trails.

Whether you are modernizing legacy enterprise workflows or building specialized local development assistants, CUGA offers a robust blueprint for trustworthy, sovereign agent architecture.

Thanks for reading🦉

Links and references

Code repository for this post: https://github.com/aairom/cuga-agent-harness
CUGAConfigurable Generalist Agent — Agent Harness for the enterprise: https://cuga.dev/
cuga-project home: https://github.com/cuga-project
cuga agent: https://github.com/cuga-project/cuga-agent
cuga harness kit: https://github.com/cuga-project/cuga-harness-kit
cuga app samples: https://github.com/cuga-project/cuga-apps
cuga eval: https://github.com/cuga-project/cuga-eval
cuga flow: https://github.com/cuga-project/cuga-flo
cuga skills: https://github.com/cuga-project/cuga-skills
cuga viz: https://github.com/cuga-project/cuga-viz
CUGA Explained: IBM's Open-Source Agent Harness for Enterprise AI: https://www.youtube.com/watch?v=64WEB4S78nQ
Open by Design: Generalist and Pre‑Built Agents in the Sovereign Core: https://community.ibm.com/community/user/blogs/shikha-srivastava1/2026/04/30/open-by-design-generalist-and-prebuilt-agents-in-t
Build real agentic apps using CUGA: two dozen working examples on a lightweight harness: https://huggingface.co/blog/ibm-research/cuga-apps

DEV Community: Alain Airom (Ayrom)

Harnessing LMCache: Building a Pluggable KV-Cache Management & Testing Framework

Introduction

What is LMCache?

The LMCache Ecosystem

Key features

Implementation

Application overview & purpose

System Architecture & Components

Code Layout

Evaluation Modes

Configuration examples & usage scenarios

Final Disclaimer and findings after the tests

Conclusion

Links

Giving AI Agents a Governed Identity: A Comprehensive Guide to Architecture, Implementation, and Enforcement with CUGA

Introduction & Background

The Shift Towards Autonomous AI Agents

What is Governed Agent Identity?

System Architecture & Lifecycle Governance

Component Architecture & Layered Defense

Complete Agent Lifecycle State Machine

Access Delegation Mechanics & Scoping Modes

Implementation with CUGA-Agent Integration

The demonstration use-case

Technical Features of the use-case

Conclusion & Realization via CUGA-Agent

Summary of Architecture & Security Benefits

Overall implementation: Governed CUGA Agent Implementation

Links

[Boost]

When Wording Changes Image Safety: A GigaChat Red-Team Case Study

Don't asnwer to fake Dev.to (support) message

Building a GitHub Contribution City: Transforming Developer History into a 3D Skyline

Introduction & Project Inspiration

Detailed Architecture and Implementation

System Architecture

Configuration Management (pkg/config)

Fetching Contribution Calendars (pkg/github)

Logarithmic City Layout Algorithm (pkg/city)

Main Application Entrypoint (cmd/city/main.go)

Automation with GitHub Actions

Conclusion

Links

Building an End-to-End Multilingual Speech and LLM Pipeline with IBM Granite Speech 5.0-2B and Bob

Introduction

IBM Granite Speech 5

Sample Application Features and Functions

Architecture

Implementation

Project Layout & Core Setup

Hardware Detection & Optimization Advisor

Core Speech Processing Wrapper

Downstream Processing with Multi-Backend LLMs

End-to-End Application Process

Conclusion

Links

Turning Books into Actionable Agent Skills: A Deep Dive into book-to-skill (and Docling)

Introduction

book-to-skill Core Concepts

Implementation

Architecture Schema

Two-Phase Processing Pipeline

Application Architecure and Code Excerpts

Application's Configurations

API Accessibility

Processing and Test Outputs

Pipeline Verification and LLM Probing

Conclusion

Links

Agent Context A/B Testing with Intel Dokimi

Introduction

What is Agent Context A/B Testing?

Key Components of the Evaluation Flow

Why Use Context A/B Testing?

Dokimi — A/B testing for agent context

What it measures

1. Does it help? Enabled vs disabled arms

2. What does it add? Novel vs known content

3. Repositories: scan the guidance, then measure agents working in them

Configuration Management (`pkg/config`)

Fetching Contribution Calendars (`pkg/github`)

Logarithmic City Layout Algorithm (`pkg/city`)

Main Application Entrypoint (`cmd/city/main.go`)

`book-to-skill` Core Concepts

Dual Runtime Compilation Pathways (`CDemoManager`)

Core Synchronization Logic (`syncCore.ts`)

Extension Event Wiring (`extension.ts`)