DEV Community: Alain Airom (Ayrom)

Building Document Intelligence with docling-context

Alain Airom (Ayrom) — Tue, 13 Oct 2026 07:45:56 +0000

Implementing docling-context!

Introduction

Large Language Models excel at reasoning, but enterprise retrieval-augmented generation (RAG) and document processing workflows often flounder when dealing with complex, multi-modal documents. Unstructured files (such as PDFs, Markdown, and Office formats) frequently suffer from structural loss during chunking, leading to imprecise context window allocation, ambiguous citations, and broken document hierarchies.

The docling-context sample application (dcuc) addresses these challenges by implementing an offline-first, local-only architecture centered around immutable DocLang (.dclx) packages. By maintaining a unified local context tree, dcuc transforms plain text and unstructured sources into structured, citable nodes. It automates multi-tiered indexing (summary, outline, and full text), provides hybrid retrieval, and exposes both a lightweight CLI and a Gradio-based control plane

TL;DR-What is Docling-Context?

Excerpt from Github repository.

Docling-context is a local context tree that stores documents as immutable DocLang .dclx packages.

See the capability guide, retrieval guide, session guide, memory guide, agent harness guide, CLI reference, and single PDF and folder examples.

Ingest native .dclg, .dclg.xml, and .dclx sources with Ingestor.add_resource. Pass bytes with a filename, or a local Path under an explicit allowed_roots entry. Each revision stores its full document, a short DocLang summary, and a DocLang TOC in one DCLX package. Collection summaries are rebuilt by CollectionWorker; CollectionWorker.status reports whether a summary is stale.

Document conversion is local and optional: install the conversion extra to enable LocalDoclingConverter for Docling's supported formats, including PDF, Office documents, and images. DoclingServeConverter requires an explicit HTTPS endpoint or loopback HTTP endpoint. SQLite tracks logical URIs, revisions, tenant ownership, and an append-only change outbox. Package bytes live under SHA-256 paths. The API is designed so another storage backend can implement the same contracts later.

Docling-context could be used simply by installing the requirements and the implementation is quite easy;

from doclang import DocLangXDocument
from docling_context import LocalContextStore, NodeAddress, Principal

document = DocLangXDocument()
assert document.read_xml('<doclang version="0.7"><text>Hello</text></doclang>')
package = document.write_bytes()
principal = Principal(tenant_id="acme", user_id="alice")

with LocalContextStore("./context-data") as store:
    record = store.put(principal, "docling://resources/manuals/hello", package)
    citation = NodeAddress(record.document_id, record.revision_id, "/doclang[1]/text[1]")
    print(store.read_node(principal, citation).text)

... coupled with native, out-of-the-box ingestion.

from pathlib import Path
from docling_context import CollectionWorker, Ingestor, LocalContextStore, Principal

principal = Principal("acme", "alice")
source_dir = Path("./documents").resolve()
with LocalContextStore("./context-data") as store:
    record = Ingestor(store, allowed_roots=(source_dir,)).add_resource(
        principal, "docling://resources/manuals/guide", source_dir / "guide.dclx"
    )
    worker = CollectionWorker(store)
    while worker.run_once() is not None:
        pass
    print(worker.status(principal, "docling://resources/manuals"))

2nd TL;DR-What is Docklang?

Excerpt from Github repository.

DocLang is the AI-native markup format for unstructured content — documents, images, audio, and video alike. It is engineered from the ground up around a single principle: mapping cleanly to LLM tokens. That makes DocLang the most efficient way to read, write, and reason about real-world content with modern AI.

The world's knowledge lives in formats — PDF, HTML, Word, LaTeX, MP3, MP4 — that were designed for rendering or playback, not for understanding. The moment you hand them to an AI system, you lose structure, semantics, layout, timing, or all of them at once. And the formats AI does speak fluently fall short too: Markdown can't describe complex content, HTML wastes tokens, LaTeX is ambiguous, and none of them speak audio or video at all.

Implementation

Flow Architecture

The application's structural design is built upon a strict separation of concerns, ensuring modularity, maintainability, and clean abstraction boundaries throughout the codebase. Rather than allowing client-facing layers to interact directly with underlying storage, parsing, or inference primitives, external entry points—including the command-line interface (cli.py), utility automation scripts, and the interactive web interface built with Gradio (app.py)—delegate all operational workflows to a unified facade (ContextPipeline).

This high-level facade acts as the single point of orchestration, coordinating ingestion, collection indexing, search retrieval, and session logging. Furthermore, because web applications often process user interactions across concurrent worker threads, access to the underlying SQLite database and context store is strictly synchronized using an explicit, reentrant thread lock (RLock), guaranteeing thread-safe execution and data integrity under concurrent loads.

Core Implementation & Code Excerpts

Gradio Web User Interface (app.py): the presentation tier provides interactive tabs for monitoring collection status, running batch ingestions, performing targeted searches, and exploring context trees.

"""Gradio web UI for the docling-context sample.

Run with ``uv run python app.py`` (or ``scripts/start.sh`` for detached mode).

Tabs: Status | Ingest | Search | Ask (chat with citations) | Explore (tree + document).

A single :class:`~dcuc.pipeline.ContextPipeline` is shared by all requests; it
serialises store access internally. SIGTERM/SIGINT close the store cleanly so
``scripts/stop.sh`` performs a graceful shutdown.
"""

from __future__ import annotations

import json
import logging
import signal
import sys
import threading
from pathlib import Path

import gradio as gr

from dcuc.config import Settings
from dcuc.logging_setup import configure_logging
from dcuc.pipeline import TIER_NAMES, ContextPipeline
from dcuc.reporting import answer_markdown, ingest_markdown, write_report

log = logging.getLogger("dcuc.app")
"""

def build_ui(pipeline: ContextPipeline) -> gr.Blocks:
    settings = pipeline.settings
    # ---- layout ------------------------------------------------------------ #
    with gr.Blocks(title="docling-context sample") as ui:
        gr.Markdown(
            "# docling-context sample\n"
            f"Collection `docling://resources/{settings.collection}` - tenant "
            f"`{pipeline.principal.tenant_id}` - LLM **{'on' if pipeline.llm else 'off'}** - "
            f"vectors **{'on' if pipeline.vectors_enabled else 'off'}**"
        )
        with gr.Tab("Status"):
            status_box = gr.Code(language="json", value=do_status())
            gr.Button("Refresh").click(do_status, outputs=status_box)
        ...
    return ui

The Application Facade (pipeline.py): the ContextPipeline manages store interactions, coordinating ingestion, collection jobs, retrieval, and session recording under an explicit threading lock.

"""``ContextPipeline`` - the single facade the CLI and the web UI talk to.

It demonstrates the main docling-context building blocks end to end:

====================  =======================================================
docling-context API   Used here for
====================  =======================================================
``Principal``         Tenant/user identity; *every* store call is scoped by it
``LocalContextStore`` SQLite metadata + immutable SHA-256 DCLX packages
``Ingestor``          Convert + summarise + TOC + index one source revision
``CollectionWorker``  Durable jobs that rebuild per-collection summaries
``Retriever``         Scoped lexical / vector / hybrid search with citations
``SessionStore``      Revisioned, append-only conversation transcripts
``NodeAddress``       Stable citation (document, revision, xpath)
====================  =======================================================

Resource management: the store owns an SQLite connection, so the pipeline is a
context manager and must be closed. A single lock serialises store access,
because the web UI runs request handlers on several worker threads while
SQLite connections are not safe for concurrent use.
"""

from __future__ import annotations

import logging
import re
import threading
import time
import uuid
from collections.abc import Callable, Iterator
from dataclasses import dataclass, field
from pathlib import Path

from docling_context import (
    CollectionWorker,
    ContextResult,
    Converter,
    Ingestor,
    LocalContextStore,
    NodeAddress,
    PackageError,
    Principal,
    RecordMissing,
    Retriever,
    SearchScope,
    SessionStore,
)

from .config import Settings
from .converters import MarkdownConverter
from .llm import LLMClient, LLMError
from .summaries import LLMSummaryProvider
from .uris import classify, uri_for, validate_collection

"""
class ContextPipeline:
    """High-level operations over one docling-context store for one principal."""

    def __init__(self, settings: Settings, *, llm_client: LLMClient | None = None):
        self.settings = settings
        self.collection = validate_collection(settings.collection)
        # Principal validation (no '/', '%', '..', control chars) happens in the library.
        self.principal = Principal(settings.tenant_id, settings.user_id)
        self._lock = threading.RLock()

        # Opening the store migrates the schema and reconciles packages: any blob
        # that is missing or fails its SHA-256 check is marked unavailable.
        self.store = LocalContextStore(settings.store_dir)
        try:
            self.llm = llm_client if llm_client is not None else self._make_llm()
            self.retriever = Retriever(self.store, embedder=self._make_embedder())
            self.sessions = SessionStore(self.store)
            self.worker = CollectionWorker(self.store)
        except Exception:
            self.store.close()  # never leak the SQLite connection on a failed start
            raise

Canonical URI Generation & Slugs (uris.py): to prevent ambiguities and ensure safe paths, directory and file names are converted into valid URI segments using custom slugification rules and hash suffixes.

  """Map files in ``input/`` to canonical ``docling://`` URIs.

  docling-context rejects ambiguous URIs (encoded separators, ``..``, control
  characters, ...). Instead of letting arbitrary filenames reach the store, we
  derive a *slug* for each path segment and append a short hash whenever the
  slug differs from the original, so two different filenames can never collapse
  into the same URI. This mirrors the strategy used by the upstream ``dc`` CLI.
  """

  from __future__ import annotations

  import hashlib
  import os
  import re
  from pathlib import Path

  from docling_context import parse_uri

  # Native DocLang sources are always ingestible; others need Docling conversion.
  NATIVE_SUFFIXES = (".dclg.xml", ".dclg", ".dclx")
  # Converted by the lightweight in-repo converter (no heavy dependencies).
  TEXT_SUFFIXES = (".md", ".markdown", ".txt")
  # Converted by Docling when the `conversion` extra is installed.
  DOCLING_SUFFIXES = (
      ".pdf", ".docx", ".pptx", ".xlsx", ".html", ".htm",
      ".png", ".jpg", ".jpeg", ".tif", ".tiff", ".bmp", ".webp", ".csv", ".asciidoc", ".adoc",
  )

  _COLLECTION_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")


  def slug(value: str) -> str:
      """Return a URI-safe segment; append a hash suffix when the value had to change."""
      result = re.sub(r"[^\w.-]+", "-", value, flags=re.UNICODE).strip(".-")
      if not result:
          raise ValueError(f"cannot derive a URI segment from {value!r}")
      if result != value:
          result += "-" + hashlib.sha256(value.encode("utf-8")).hexdigest()[:8]
      return result


  def validate_collection(name: str) -> str:
      """Collections are the first URI segment; keep them simple and lowercase.

      The ``_context`` name is reserved by docling-context for collection summaries.
      """
      if not _COLLECTION_RE.fullmatch(name):
          raise ValueError(
              "collection must be 1-64 chars of lowercase letters, digits, '-' or '_'"
          )
      return name


  def strip_suffix(name: str) -> str:
      """Remove the (possibly double) extension, e.g. ``a.dclg.xml`` -> ``a``."""
      lower = name.lower()
      for suffix in (*NATIVE_SUFFIXES, *TEXT_SUFFIXES, *DOCLING_SUFFIXES):
          if lower.endswith(suffix) and len(name) > len(suffix):
              return name[: -len(suffix)]
      return Path(name).stem or name


  def classify(path: Path) -> str | None:
      """Return ``native``, ``text``, ``docling`` or ``None`` (unsupported)."""
      lower = path.name.lower()
      if lower.endswith(NATIVE_SUFFIXES):
          return "native"
      if lower.endswith(TEXT_SUFFIXES):
          return "text"
      if lower.endswith(DOCLING_SUFFIXES):
          return "docling"
      return None


  def uri_for(path: Path, base: Path, collection: str) -> str:
      """Build ``docling://resources/<collection>/<sub/dirs>/<stem>`` for ``path``.

      Sub-folders of ``input/`` become URI path segments so the context tree mirrors
      the folder layout. The result is validated with the library's own parser.
      """
      validate_collection(collection)
      # Symlinks are deliberately NOT resolved here: the URI follows the name the
      # user sees, and the Ingestor's ``allowed_roots`` check (which does resolve
      # them) remains the single security boundary for reading files.
      relative = Path(os.path.abspath(path)).relative_to(os.path.abspath(base))
      segments = [slug(part) for part in relative.parts[:-1]]
      segments.append(slug(strip_suffix(relative.parts[-1])))
      uri = f"docling://resources/{collection}/{'/'.join(segments)}"
      return parse_uri(uri).value  # raises InvalidURI if anything slipped through

python

## Supplementary Implementation Modules

To complete the architectural overview of the sample application, three additional core modules manage lightweight format conversion, LLM-backed document summarization, and atomic reporting.

Lightweight Markdown Converter (dcuc/converters.py): while heavy ML-based converters like Docling handle complex PDF and office layouts when opted-in, text-first knowledge bases rely on a streamlined converter. Implemented to satisfy the library's Converter protocol, MarkdownConverter parses Markdown and plain-text files directly into structured DocLang XML elements (<heading> and <text>) without heavy dependencies.

class MarkdownConverter:
    """``docling_context.Converter`` implementation for ``.md`` and ``.txt``."""

    def __init__(self, *, max_file_bytes: int = 5_000_000):
        self.max_file_bytes = max_file_bytes

    def fingerprint_options(self) -> dict[str, object]:
        # The Ingestor hashes these options together with the source bytes. If the
        # options or converter version change, files are re-ingested; otherwise an
        # unchanged file is skipped (idempotent ingestion).
        return {"converter_version": CONVERTER_VERSION, "max_file_bytes": self.max_file_bytes}

    def convert(self, source: bytes, filename: str) -> ConversionResult:
        if len(source) > self.max_file_bytes:
            raise ValueError(f"{filename}: exceeds {self.max_file_bytes} bytes")
        try:
            text = source.decode("utf-8-sig")
        except UnicodeDecodeError as exc:
            raise ValueError(f"{filename}: not valid UTF-8 text") from exc

        title = filename.rsplit("/", 1)[-1].rsplit(".", 1)[0]
        xml = blocks_to_doclang(parse_markdown(text), title=title)
        document = DocLangXDocument()
        if not document.read_xml(xml):
            raise PackageError(document.last_error() or "generated DocLang is invalid")
        return ConversionResult(
            package=document.write_bytes(),
            converter="dcuc-markdown",
            version=CONVERTER_VERSION,
            options=self.fingerprint_options(),
        )

LLM-Backed Summary Provider (dcuc/summaries.py): the ingestion pipeline can automatically generate concise structural summaries for document revisions using a local LLM. Implementing the library's SummaryProvider protocol, LLMSummaryProvider extracts bounded document text, queries the local model for a factual summary, and formats the output into safe DocLang XML. If any error occurs during generation, the provider raises an exception, allowing the ingestor to gracefully fall back to an extractive summary.

class LLMSummaryProvider:
    """Summarise each ingested revision with a local LLM."""

    def __init__(self, client: LLMClient):
        self.client = client

    def summarize(self, document: DocLangXDocument, source_revision: str) -> str:
        content = document_text(document)
        if not content:
            raise ValueError("document has no text to summarise")
        summary = self.client.chat(SYSTEM_PROMPT, content, max_tokens=300)
        summary = " ".join(summary.split())[:MAX_SUMMARY_CHARS]
        log.debug("LLM summary for revision %s: %d chars", source_revision, len(summary))
        # The summary sidecar must itself be DocLang; escaping makes it safe.
        return f"<doclang><text>{escape(summary)}</text></doclang>"

Atomic Timestamped Reporting (dcuc/reporting.py): every runtime operation—such as ingestion, searching, and question-answering—generates dual machine-readable JSON and human-readable Markdown reports saved to the output/ directory. To prevent file corruption or half-written artifacts in the event of an interruption, _atomic_write writes data to a temporary file in the target directory before swapping it atomically using os.replace.

"""Timestamped reports written to ``output/``.

Every run produces a pair of files that share one timestamp prefix:
``<YYYYmmdd-HHMMSS>_<name>.json`` (machine-readable) and ``.md`` (human-readable).
Writes go to a temporary file first and are renamed, so a crash never leaves a
half-written report behind.
"""

from __future__ import annotations

import json
import os
import tempfile
from dataclasses import asdict, is_dataclass
from datetime import datetime
from pathlib import Path
from typing import Any

from .pipeline import Answer, IngestReport

"""
def _atomic_write(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=".tmp-", suffix=path.suffix)
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            handle.write(content)
        os.replace(tmp, path)
    except BaseException:
        Path(tmp).unlink(missing_ok=True)
        raise

Tests and samples: to demonstrate the practical output generated by the application, execution artifacts are automatically serialized into dual formats—a machine-readable JSON log and a structured Markdown report (output/). For instance, when querying the system regarding the enterprise refund guidelines (What is the refund policy?), the pipeline performs a tiered lexical-vector search across document nodes and constructs a verifiable response complete with precise URI paths, tier levels, XPath selectors, and relevance scoring. As illustrated in the resulting session transcript, if a local LLM invocation is bypassed or unavailable, the framework seamlessly switches to an extractive fallback method, quoting the exact passages verbatim while preserving exact provenance metadata.

{
  "question": "What is the refund policy?",
  "answer": "Most relevant passages:\n[1] Refund Policy Customers may request a full refund within 30 days of purchase for annual plans. Monthly plans Monthly plans are not refundable, but cancellation stops the next renewal. How to request a refund Open a ticket with the billing team and include the invoice number. Refunds are issued to the original payment method within 5 business days.\n[2] Refund Policy\n[3] Refund Policy",
  "method": "extractive",
  "citations": [
    {
      "number": 1,
      "uri": "docling://resources/knowledge/policies/refund-policy",
      "xpath": "@summary:/doclang[1]/text[1]",
      "tier": 0,
      "score": 1.06,
      "text": "Refund Policy Customers may request a full refund within 30 days of purchase for annual plans. Monthly plans Monthly plans are not refundable, but cancellation stops the next renewal. How to request a refund Open a ticket with the billing team and include the invoice number. Refunds are issued to the original payment method within 5 business days.",
      "document_id": "a67aee64fbb8429baa9afe36f9448c00",
      "revision_id": "448341261cf843b39dc5dd17cae59f69"
    },
    {
      "number": 2,
      "uri": "docling://resources/knowledge/policies/refund-policy",
      "xpath": "/doclang[1]/heading[1]",
      "tier": 1,
      "score": 0.53,
      "text": "Refund Policy",
      "document_id": "a67aee64fbb8429baa9afe36f9448c00",
      "revision_id": "448341261cf843b39dc5dd17cae59f69"
    },
    {
      "number": 3,
      "uri": "docling://resources/knowledge/policies/refund-policy",
      "xpath": "@toc:/doclang[1]/toc[1]/entry[1]/description[1]",
      "tier": 1,
      "score": 0.53,
      "text": "Refund Policy",
      "document_id": "a67aee64fbb8429baa9afe36f9448c00",
      "revision_id": "448341261cf843b39dc5dd17cae59f69"
    }
  ],
  "session_id": "ade35b2fa47b433495cc9ef5ce3f9fa4"
}

Conclusion

By seamlessly fusing lightweight local SQLite metadata indexing with immutable, content-addressed DocLang .dclx package storage, docling-context empowers developers and architects to construct highly resilient, offline-first document intelligence systems. This robust architectural foundation eliminates external cloud dependencies while guaranteeing exact, stable node citations across document revisions. Furthermore, by incorporating multi-tiered search strategies—such as prioritized body-first retrieval alongside robust fallback mechanisms for local LLM inference—the framework ensures continuous, dependable operational performance even when auxiliary services encounter network disruptions or runtime failures.

Thanks for reading 📃

Extending AI Capabilities: Looking into Implementing Lifecycle Hooks with IBM Bob

Alain Airom (Ayrom) — Thu, 08 Oct 2026 08:51:48 +0000

An introdcution to “Hooks” in IBM Bob.

Introduction

As AI-assisted software development rapidly evolves, AI coding assistants are no longer isolated prompt-and-response interfaces. They are becoming deeply integrated members of our development workflows. IBM Bob provides a robust extension mechanism through Lifecycle Hooks.

Hooks allow developers and team leads to inject real-time context, enforce strict safety policies, auto-format or lint modified files, and stream live session telemetry into external security and logging dashboards.

Unlike probabilistic Large Language Model responses, hooks act as deterministic rules executed by the system. Because LLMs can occasionally hallucinate or bypass soft instructions, hooks provide hard boundaries and guaranteed execution logic for critical tasks. Whether enforcing zero-trust compliance, blocking dangerous commands, running exact test suites, or formatting code according to team standards, hooks guarantee that specified automation and checks execute every single time without relying on model discretion.

In this blog post, we walk through to discover what lifecycle hooks are in IBM Bob, explore the architectural model behind them, and look at actual code implementations that demonstrate how hooks can bring governance and intelligence to your AI session.

What Are Lifecycle Hooks in IBM Bob?

In IBM Bob, Hooks are executable scripts or HTTP endpoints that run automatically in response to specific lifecycle events during an agent session.

Instead of letting an LLM agent operate as a black box, hooks give developers full control over the inputs and outputs at critical checkpoints:

Context Injection: Auto-inject workspace details, Node/Git metadata, and architectural rules directly into the LLM context.
Policy Guardrails: Intercept user prompts or tool executions to reject prohibited commands (e.g., destructive SQL queries or broad file deletion) before they hit the agent or workspace.
Automated Feedback Loops: Automatically lint modified files (such as running ESLint after write_file) and supply issues back to Bob to self-correct.
Auditability & Observability: Persist logs and stream event webhooks to central telemetry services.

Overall System Architecture

The mermaid chart below gives the big picture of the overall system architecture.

Lifecycle Execution State Diagram

The life cycle of an IBM Bob task moves systematically through pre-defined state checkpoints:

Hooks Decision Tree

The flow below, gives an overall view of hooks decision tree.

Practical Hook Implementations

This project shows how to use every Bob lifecycle hook in a practical, production-ready way. It is structured to be progressive — start with the simple SessionStart example and work toward advanced patterns like blocking prompts, gating tool calls, and chaining shell + HTTP hooks together.

| Hook               | Script                                                       | Purpose                                                      |
| ------------------ | ------------------------------------------------------------ | ------------------------------------------------------------ |
| `SessionStart`     | [`.bob/hooks/session-start.mjs`](.bob/hooks/session-start.mjs) | Inject git/env context into the model (startup, resume, compact) |
| `UserPromptSubmit` | [`.bob/hooks/prompt-guard.mjs`](.bob/hooks/prompt-guard.mjs) | Block dangerous prompts (structured deny) or enrich with context |
| `PreCompact`       | [`.bob/hooks/pre-compact.mjs`](.bob/hooks/pre-compact.mjs)   | Gate compaction; block manual if uncommitted changes exist   |
| `PostCompact`      | [`.bob/hooks/post-compact.mjs`](.bob/hooks/post-compact.mjs) | Save compact_summary to `output/`; notify HTTP listener      |
| `PreToolUse`       | [`.bob/hooks/tool-gate.mjs`](.bob/hooks/tool-gate.mjs)       | Block dangerous commands (structured deny) or rewrite tool input |
| `PostToolUse`      | [`.bob/hooks/post-tool-observer.mjs`](.bob/hooks/post-tool-observer.mjs) | Run ESLint after writes; feed findings back to model         |
| `Stop`             | [`.bob/hooks/session-stop.mjs`](.bob/hooks/session-stop.mjs) | Generate session summary report; optional git auto-commit    |
| HTTP listener      | [`server.mjs`](server.mjs)                                   | Receive forwarded events; serve live dashboard + report viewer |
| Automation runner  | [`tests/run-all-hooks.mjs`](tests/run-all-hooks.mjs)         | Run all 26 hook scenarios; write timestamped Markdown report |
| Shell entry-point  | [`scripts/run-hooks.sh`](scripts/run-hooks.sh)               | CLI wrapper: `--open`, `--browser`, `--session-id` flags     |
| Report router      | [`routes/reports.mjs`](routes/reports.mjs)                   | Secure Markdown→HTML renderer; `/report/latest`, `/report/list` |

The structure of the files provided;

bob-hook-implementation/
├── .bob/
│   ├── hooks/                      # ← All hook scripts live here
│   │   ├── session-start.mjs       #   SessionStart hook (startup / resume / compact)
│   │   ├── prompt-guard.mjs        #   UserPromptSubmit hook (structured deny + context)
│   │   ├── pre-compact.mjs         #   PreCompact hook (compaction gate)
│   │   ├── post-compact.mjs        #   PostCompact hook (save compact_summary)
│   │   ├── tool-gate.mjs           #   PreToolUse hook (structured deny + input rewrite)
│   │   ├── post-tool-observer.mjs  #   PostToolUse hook (lint + telemetry)
│   │   └── session-stop.mjs        #   Stop hook (session report)
│   ├── settings.json               # ← Bob hook configuration (all 7 events)
│   └── rules/
│       └── AGENTS.md               #   Project rules
├── Docs/
│   ├── Architecture.md             #   Mermaid architecture diagrams
│   ├── Quickstart.md               #   Step-by-step setup guide
│   └── HooksReference.md           #   Complete hook API reference
├── routes/
│   └── reports.mjs                 # ← Report router (marked + sanitize-html)
├── scripts/
│   ├── start.sh                    #   Start server in detached mode
│   ├── stop.sh                     #   Graceful shutdown
│   ├── cleanup.sh                  #   Remove node_modules, output, logs
│   └── run-hooks.sh                # ← Automation runner shell entry-point
├── tests/
│   ├── run-tests.mjs               #   36 unit tests (one-off hook assertions)
│   └── run-all-hooks.mjs           # ← 26-scenario automation runner + report writer
├── logs/                           #   Runtime logs (git-ignored)
├── output/                         #   Session JSON + Markdown reports (git-ignored)
├── input/                          #   Input documents (content git-ignored)
├── server.mjs                      # ← HTTP listener + dashboard + report viewer
├── package.json
├── .env.example                    #   Environment variable template
└── README.md

Below are key code excerpts illustrating how hooks process incoming payloads from standard input (stdin) and communicate decisions back via standard output (stdout) or process exit codes.

Context Injection on SessionStart: when a new session opens, session-start.mjs collects local environment metadata (Git branch, commit hash, Node version) and writes it to standard output. Bob captures this standard output and injects it into the prompt context.

// session-start.mjs
import { execSync } from "node:child_process";
import { existsSync, mkdirSync, appendFileSync } from "node:fs";
import { join } from "node:path";

// Read payload from standard input
let raw = "";
for await (const chunk of process.stdin) raw += chunk;
let payload = JSON.parse(raw || "{}");

const sessionId = payload.session_id ?? "unknown";
const cwd = payload.cwd ?? process.cwd();
const source = payload.source ?? "startup";

// Collect runtime git and toolchain context
function safeExec(cmd) {
  try { return execSync(cmd, { encoding: "utf8", cwd }).trim(); }
  catch { return "unavailable"; }
}

const gitBranch = safeExec("git rev-parse --abbrev-ref HEAD");
const gitLastCommit = safeExec("git log -1 --pretty=%s");
const nodeVersion = safeExec("node --version");

// Inject context directly into Bob model context via stdout
const context = `
=== Bob Session Context (auto-injected) ===
Session     : ${sessionId}
Source      : ${source}
Node.js     : ${nodeVersion}
Git Branch  : ${gitBranch}
Last Commit : ${gitLastCommit}
===========================================
`.trim();

process.stdout.write(context + "\n");
process.exit(0);

Prompt Safety Guardrail with UserPromptSubmit: before a prompt reaches the model, prompt-guard.mjs inspects the prompt string for prohibited commands or security breaches. It can either return a structured JSON block or enrich the query with company guidelines.

// prompt-guard.mjs
let raw = "";
for await (const chunk of process.stdin) raw += chunk;
let payload = JSON.parse(raw || "{}");

const prompt = payload.prompt ?? "";

// Safety check against destructive instructions
const BLOCKED_PATTERNS = [
  { pattern: /\bdrop\s+table\b/i, reason: "Destructive SQL not permitted." },
  { pattern: /rm\s+-rf\s+\/(?!\S)/, reason: "Root directory deletion blocked." }
];

const match = BLOCKED_PATTERNS.find(({ pattern }) => pattern.test(prompt));

if (match) {
  // Structured JSON block response (exit code 0 with decision: "block")
  const blockResponse = JSON.stringify({
    decision: "block",
    reason: `Prompt blocked by policy: ${match.reason}`,
  });
  process.stdout.write(blockResponse + "\n");
  process.exit(0);
}

// Structured enrichment response (attaches additional contextual instructions)
const enrichResponse = JSON.stringify({
  hookSpecificOutput: {
    hookEventName: "UserPromptSubmit",
    additionalContext: "[Policy]: Ensure code includes unit tests and follows style guidelines.",
  },
});
process.stdout.write(enrichResponse + "\n");
process.exit(0);

Automated Code Quality Feedback with PostToolUse: when Bob executes a file mutation tool like write_file, post-tool-observer.mjs executes code verification tools (such as ESLint) against the file. Any discovered linting issues are piped to stdout and automatically added to the conversation so Bob can fix them immediately.

// post-tool-observer.mjs
import { spawnSync } from "node:child_process";
import { extname, basename, join } from "node:path";

let raw = "";
for await (const chunk of process.stdin) raw += chunk;
let payload = JSON.parse(raw || "{}");

const toolName = payload.tool_name ?? payload.tool ?? "";
const toolInput = payload.tool_input ?? payload.input ?? {};

if (["write_file", "apply_diff"].includes(toolName)) {
  const filePath = String(toolInput.path ?? "");
  const ext = extname(filePath);

  if ([".js", ".mjs", ".ts"].includes(ext)) {
    // Run automated linter on the generated/edited file
    const eslint = spawnSync("npx", ["--no", "eslint", "--format", "compact", filePath], {
      encoding: "utf8",
      timeout: 15000,
    });

    if (eslint.status !== 0 && eslint.stdout) {
      // Pipe linter findings back to stdout for model feedback
      process.stdout.write(
        `[PostToolUse/ESLint] Linting issues found in ${basename(filePath)}:\n${eslint.stdout}\n`
      );
    }
  }
}
process.exit(0);

Telemetry and Webhook Ingestion Listener: all hooks can send async HTTP signals to a central telemetry endpoint. Below is a sample Node.js/Express server configured to consume lifecycle events and track statistics:

// server.mjs (Express Listener Endpoint)
import express from "express";
import { appendFileSync } from "node:fs";

const app = express();
app.use(express.json());

app.post("/hooks", (req, res) => {
  const event = {
    id: `evt_${Date.now()}`,
    received_at: new Date().toISOString(),
    ...req.body,
  };

  console.log(`▶ Hook Event Received: ${event.hook_event_name || event.event} | Session: ${event.session_id}`);

  // Persist for centralized compliance and SIEM auditing
  appendFileSync("logs/http-events.log", JSON.stringify(event) + "\n");

  return res.status(200).json({ ok: true, event_id: event.id });
});

app.listen(3001, () => console.log("Bob Hook Telemetry Server listening on port 3001"));

You can either test the hooks individually or run the test as batch (adapted to your use-cases) and verify the outputs before implementing for real use-cases.

Conclusion

IBM Bob’s hook system provides a robust architecture for bringing governance, contextual intelligence, and automated verification into your AI development environment. By employing deterministic script execution at key lifecycle checkpoints, software engineering teams can confidently adopt AI tooling while enforcing organizational policies, code quality standards, and centralized security monitoring.

To explore configuration options, event payloads, and setup guides, check out the official IBM Bob Lifecycle Hooks Documentation.

Thanks for reading 🪝🐟

Bridging AI Agents and VS Code: Implementing the Agent Client Protocol (ACP) for IBM Bob

Alain Airom (Ayrom) — Tue, 06 Oct 2026 15:03:08 +0000

A “test” implementation of ACP for VSCode!

Introduction

The landscape of developer tooling is shifting rapidly from static auto-complete toward dynamic, multi-turn AI agents capable of executing shell commands, reading codebase graphs, and editing files autonomously. However, as the ecosystem of code agents explodes, a clear integration barrier has emerged: How do we connect disparate AI agents to various code editors without writing custom UI integrations for every single combination?

Enter the Agent Client Protocol (ACP).

In this post, I’ll explore what the Agent Client Protocol is, dive deep into our implementation connecting IBM Bob to VS Code (as a test only), inspect the underlying architecture and transport mechanisms, and examine code excerpts straight from the implementation repository.

What is the Agent Client Protocol (ACP)?

From official Github repository: The Agent Client Protocol (ACP) standardizes communication between code editors (interactive programs for viewing and editing source code) and coding agents (programs that use generative AI to autonomously modify code).

Originally spearheaded by the team behind the Zed Editor, the Agent Client Protocol (ACP) is an open, agent-agnostic standard designed to decouple AI coding agents from text editors and IDEs.

┌─────────────────────────┐                            ┌─────────────────────────┐
│     Editor / Client     │    JSON-RPC 2.0 over       │        AI Agent         │
│  (VS Code, Zed, etc.)   │ ─────────────────────────► │ (IBM Bob, Claude, etc.) │
│                         │ ◄───────────────────────── │                         │
└─────────────────────────┘      stdin / stdout        └─────────────────────────┘

The Problem ACP Solves

Before ACP, integrating a new agent (like IBM Bob, Claude Code, or Gemini CLI) into VS Code meant writing dedicated webviews, bespoke API callers, custom event streaming logic, and complex state management. If you wanted to support 5 agents across 3 editors, you needed 15 separate integrations.

How ACP Operates

ACP solves this by providing a standardized JSON-RPC 2.0 interface running over stdin/stdout (using Newline-Delimited JSON / NDJSON).

Under ACP:

The Editor is the Client (drives UI rendering, permission modals, workspace selection).
The Agent is the Server (runs as a subprocess, handles LLM calls, maintains conversation state, and invokes tools).

Through standardized RPC methods (initialize, session/new, prompt/send, session/update), the client and agent communicate bidirectionally with native support for real-time streaming, tool execution requests, and user permission gates.

Architecture: How Bob Integrates via ACP

To demonstrate ACP in practice, we built an ACP-compliant integration between IBM Bob and VS Code.

Disclaimer: as of today Bob officially supports the following IDE's using ACP; Zed, IntelliJ, IDEA, Neovim and Xcode. This post's aim to provide a hypothetical implementation of ACP with Bob, this is not a support for VSCode!

TL;DR-ACP vs. Headless Agent Pattern

When integrating IBM Bob into a workspace or automated pipeline, developers often choose between two integration styles:

Headless Agent Mode (bob -p "..."): One-shot CLI invocations or raw text output, ideal for headless CI/CD automation or shell scripts.
ACP Integration Pattern: A interactive, multi-turn, stateful protocol interface that hooks natively into the editor UI.

┌─────────────────────────────────────────────────────────────────────────┐
│ VS Code Extension Host (Node.js)                                        │
│                                                                         │
│  ┌──────────────────────┐  postMessage  ┌─────────────────────────────┐  │
│  │ ChatWebviewProvider  │ ◄───────────► │ HTML/CSS Chat UI            │  │
│  └──────────┬───────────┘               └─────────────────────────────┘  │
│             │                                                           │
│  ┌──────────▼───────────┐               ┌─────────────────────────────┐  │
│  │     SessionManager   │ ◄───────────► │ AgentsTreeProvider          │  │
│  └──────────┬───────────┘               └─────────────────────────────┘  │
│             │                                                           │
│  ┌──────────▼───────────┐                                               │
│  │      AgentManager    │                                               │
│  └──────────┬───────────┘                                               │
│             │                                                           │
│  ┌──────────▼───────────┐                                               │
│  │       AcpClient      │                                               │
│  └──────────┬───────────┘                                               │
└─────────────┼───────────────────────────────────────────────────────────┘
              │ JSON-RPC 2.0 (NDJSON) over stdin/stdout
              │
┌─────────────▼───────────────────────────────────────────────────────────┐
│ Child Process                                                           │
│                                                                         │
│  ┌───────────────────────────────────────────────────────────────────┐  │
│  │ Bob ACP Adapter (@ibm/bob-acp)                                    │  │
│  └──────────────────────────────────┬────────────────────────────────┘  │
│                                     │ internal                          │
│  ┌──────────────────────────────────▼────────────────────────────────┐  │
│  │ Bob AI Engine (watsonx)                                           │  │
│  └───────────────────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────────────┘

Side-by-Side Comparison

Dimension	ACP (acp-vscode extension)	Headless Bob Shell
Agent target	Any ACP-compliant agent (IBM Bob, Claude, Gemini, etc.)	IBM Bob only
Protocol	JSON-RPC 2.0 — open standard	Bob Shell CLI — proprietary flags
Transport	stdin/stdout with framed NDJSON	stdin/stdout as CLI text I/O
Session model	Stateful multi-turn sessions with `sessionId`	One-shot (`-p`) or interactive REPL
Streaming	Native — typed `session/update` events	Not structured; raw stdout text
Tool calls / permissions	Explicit `permission_request` events with approve/deny callback	`--yolo` flag (auto-allow) or interactive approval
Multi-agent	Yes — N simultaneous agent connections	No — one Bob process per invocation
Interoperability	Vendor-neutral; swap agents without changing client code	IBM Bob-specific
Authentication	`IBM_BOB_API_KEY` env var or VS Code secret storage	`BOB_API_KEY` env var or browser SSO
Requires VS Code	Yes (runs inside extension host)	No (pure CLI)
Primary use case	Rich editor UI, streaming chat, multi-agent orchestration	Automation, CI/CD, batch processing, scripting

When to Use Each

Use the ACP extension when…

You need a persistent, interactive chat UI inside VS Code.
You want streaming partial responses rendered in real-time.
You want to switch between agents (IBM Bob, Claude Code, Gemini CLI, …).
You need structured tool call visibility and permission approval in the UI.
You are building a vendor-neutral integration.
You manage multiple simultaneous conversations.

Use Bob Shell (headless) when…

Automating tasks in bash scripts or CI/CD pipelines.
Batch-processing files without opening VS Code.
Simple one-shot prompts (bob -p "…").
Environments without a desktop / GUI (server, container, CI runner).
You only need IBM Bob, no other agent.
Quick scripted workflows with minimal boilerplate.

In both cases, an API key for Bob should be generated and provided.

# acp-vscode — Environment Variables Example
#
# Copy this file to ".env" and fill in your values.
# The ".env" file is in .gitignore and will NEVER be committed.
#
# These variables are referenced in your agent configurations via
# the ${env:VAR_NAME} placeholder syntax in VS Code settings.
#
# The extension resolves ${env:VAR_NAME} at process spawn time from the
# host shell environment — no secrets are stored in VS Code settings.

# Required to connect to IBM Bob via ACP.
# Obtain your key from: https://cloud.ibm.com/iam/apikeys
# Set this in your shell profile (~/.zshrc or ~/.bashrc) so VS Code inherits it:
#   export IBM_BOB_API_KEY=your-key-here
# Alternatively set acp.ibmBobApiKey in VS Code settings (marked secret).
IBM_BOB_API_KEY=

# ── IBM watsonx endpoint (optional — defaults to the public watsonx.ai endpoint) ──
IBM_BOB_ENDPOINT=

Component Breakdown

AcpClient (src/acp/acpClient.ts): Spawns and manages the underlying agent subprocess. Handles NDJSON framing, RPC request tracking, and event dispatching.
AgentManager (src/acp/agentManager.ts): Coordinates lifecycle across multiple concurrent agent instances.
SessionManager (src/acp/sessionManager.ts): Manages in-memory state for multi-turn sessions, handling message histories and permission approvals.
ChatWebviewProvider (src/webview/chatWebviewProvider.ts): Renders the editor UI and streams updates via isolated webview channels.

Realization & Code Excerpts

Let’s take a look at the exact TypeScript implementations driving this integration.

NDJSON Transport Buffer & Frame Parsing (AcpClient): because stdio streams deliver chunked data across arbitrary byte boundaries, AcpClient maintains a rolling string buffer to extract complete, newline-delimited JSON frames:

// src/acp/acpClient.ts
/**
 * acp-vscode — ACP Client Core
 *
 * Manages the lifecycle of a single ACP agent process:
 *   - Spawning and monitoring the child process (stdio transport).
 *   - Framing and dispatching JSON-RPC 2.0 messages (newline-delimited).
 *   - Executing the ACP initialize handshake.
 *   - Exposing typed request/notification helpers used by AgentManager.
 *
 * The client emits typed events via vscode.EventEmitter so that upstream
 * managers and UI providers can react to agent lifecycle and streaming data.
 *
 * Security:
 *   - No credentials are stored or logged by this class.
 *   - env is resolved from ConfigManager (placeholder substitution only).
 *   - The child process inherits the minimal required env subset.
 *
 * @module acp/acpClient
 */

/////
private handleStdoutData(chunk: Buffer): void {
  this.buffer += chunk.toString('utf8');
  const lines = this.buffer.split('\n');

  // Keep incomplete trailing line in the buffer
  this.buffer = lines.pop() ?? '';

  for (const line of lines) {
    const trimmed = line.trim();
    if (!trimmed) continue;

    try {
      const message = JSON.parse(trimmed);
      this.handleMessage(message);
    } catch (err) {
      this.logger.error(`Failed to parse JSON-RPC frame: ${trimmed}`, err);
    }
  }
}

Sending Structured JSON-RPC Requests: sending prompts and requests involves generating sequential IDs and wrapping payloads into typed JSON-RPC 2.0 frames:

// src/acp/acpClient.ts
public sendRequest<T = unknown>(method: string, params: Record<string, unknown>): Promise<T> {
  return new Promise((resolve, reject) => {
    const id = ++this.requestIdCounter;

    const payload = {
      jsonrpc: '2.0',
      id,
      method,
      params
    };

    // Track pending response promise
    this.pendingRequests.set(id, { resolve: resolve as (val: unknown) => void, reject });

    const messageString = JSON.stringify(payload) + '\n';
    this.process?.stdin?.write(messageString, 'utf8', (err) => {
      if (err) {
        this.pendingRequests.delete(id);
        reject(err);
      }
    });
  });
}

Handling Real-Time Event Streams (SessionManager): when Bob processes a prompt, it streams structured events (e.g., text generation, thinking traces, or tool calls) through session/update notifications:

/**
 * acp-vscode — Session Manager
 *
 * Maintains the in-memory state of all ACP sessions across all connected
 * agents. Listens to session/update events from AgentManager and updates
 * session state accordingly.
 *
 * Responsibilities:
 *   - Creating and storing AgentSession objects.
 *   - Appending streaming events (text, thinking, tool calls) to messages.
 *   - Notifying the chat WebviewProvider of state changes.
 *   - Handling permission requests (auto-approve or prompt the user).
 *
 * @module acp/sessionManager
 */
////

// src/acp/sessionManager.ts
public handleSessionUpdate(params: { sessionId: string; event: AcpSessionEvent }): void {
  const session = this.sessions.get(params.sessionId);
  if (!session) return;

  const { event } = params;

  switch (event.type) {
    case 'text':
      session.appendContent(event.text);
      this._onTextAppended.fire({ sessionId: session.id, text: event.text });
      break;

    case 'thinking':
      session.updateThinking(event.thought);
      this._onThinkingUpdated.fire({ sessionId: session.id, thought: event.thought });
      break;

    case 'tool_call':
      session.registerToolCall(event.toolCallId, event.name, event.args);
      this._onToolCalled.fire({ sessionId: session.id, tool: event });
      break;

    case 'turn_end':
      session.markTurnComplete(event.reason);
      this._onTurnEnded.fire({ sessionId: session.id, reason: event.reason });
      break;

  }
}

Conclusion

The Agent Client Protocol (ACP) represents a vital maturation step in the AI engineering ecosystem. By standardizing stdin/stdout messaging through JSON-RPC 2.0, ACP allows developers to decouple AI capabilities from editor visual layers.

By adopting ACP for IBM Bob:

Interoperability: Bob can seamlessly plug into VS Code, Zed, or any client implementing ACP without rewriting agent core logic.
Granular Control: Developers retain native editor visibility into real-time tool execution, reasoning traces, and explicit permission prompts.
Extensibility: Adding new tools, language backends, or agent models requires zero changes to the client transport layer.

Whether you're building custom extension tooling or deploying enterprise AI coding assistants, ACP provides a clean, open foundation for the next generation of developer tools.

Thanks for reading 🤖

Links and references

Code & Implementation Repository: github.com/aairom/bob-acp-test
Bob's ACP Documentation: https://bob.ibm.com/docs/shell/features/acp
Agent Client Protocol Documentation: https://agentclientprotocol.com/get-started/introduction
Agent Client Protocol Repository: https://github.com/agentclientprotocol/agent-client-protocol

Inside Apple Containers: Architectural Deep Dive, CLI Parity, and Benchmarking against Podman

Alain Airom (Ayrom) — Fri, 02 Oct 2026 10:04:04 +0000

Benchmarking Apple Containers vs. Podman

Introduction

With the release of macOS 26 (Tahoe), Apple introduced a native container orchestration model powered by Virtualization.framework and managed by the com.apple.container.apiserver launchd service. Rather than relying on shared-kernel Linux namespaces inside a single monolithic virtual machine (VM), Apple Container creates a lightweight virtual machine—backed by a guest Kata Containers kernel (version 3.32.0-debug)—for every individual container process.

For macOS developers accustomed to Red Hat's daemonless, rootless container engine, Podman, this architectural shift introduces fundamentally different performance characteristics, isolation security boundaries, and command-line interfaces.

In this post, I walk through container-compare (a Go tool built to benchmark and inspect these two runtimes) to explore:

How hardware-enforced per-container VM boundaries compare to Podman's shared Linux namespace architecture.
CLI mappings, network bridging via vmnet, and native Rosetta 2 binary translation.
Real-world execution timing, memory overhead, and JSON metadata differences.

By the way, Podman Desktop handles quite well both Podman and Apple containers engines!

High-Level Architecture Comparison

The `container-compare` Test Harness

The container-compare suite uses Go wrappers (internal/applecontainer and internal/podman) to issue commands directly to both engines, collecting timing statistics and JSON inspect schema metrics.

Apple Container Architecture: Dedicated Lightweight VMs

Apple Container interacts with the com.apple.container.apiserver daemon. Every container runs within its own lightweight VM isolated via Virtualization.framework.

Podman Architecture: Shared VM & Linux Namespaces

Podman operates daemonless on the host, but on macOS it controls containers inside a shared Fedora Linux VM (Podman Machine) via conmon and crun/runc.

Key Feature Matrix & Technical Differences

Feature / Dimension	Apple Container	Podman (macOS)	Architectural Impact
Isolation Boundary	Lightweight VM (`Virtualization.framework`) MD	Linux namespaces & `cgroups` MD	Apple provides hardware-enforced isolation per container; Podman relies on shared kernel boundaries within the Machine VM. MD
Background Daemon	`com.apple.container.apiserver` (`launchd`) MD	Daemonless (`podman` CLI) MD	Apple requires `container system start`; Podman requires no persistent host service. MD
Guest Kernel	Kata Containers kernel (`3.32.0-debug`) MD+ 1	Shared Fedora kernel inside Podman Machine MD	Apple allows per-VM kernel configurations. MD
x86_64 Emulation	Native `--rosetta` flag MD	QEMU user emulation MD+ 1	Apple leverages macOS Rosetta 2 translation inside the VM for faster x86 execution on Apple Silicon. MD+ 1
Network IP Visibility	Full `vmnet` per-container IP (ADDR column) MD+ 2	Shared bridge (CNI / Netavark) MD	Apple exposes dedicated VM IPs directly in `container list`. MD+ 1

CLI Parity & Developer Workflow

Commands on Apple Container map closely to Docker and Podman conventions:

# System Service Operations
container system start              # Start launchd API server
podman machine start                # Start Podman Machine Linux VM

# Container Lifecycle
container run -d --name web nginx   # Apple: Boots dedicated VM + container
podman run -d --name web nginx      # Podman: Forks process in shared VM

# Native Rosetta 2 Execution on Apple Silicon
container run --rosetta --rm amd64/ubuntu:22.04 uname -m

# Log Extraction
container logs --boot web           # Apple: Extract VM boot logs
podman logs web                     # Podman: Standard container output logs

Metadata Schema Differences (`inspect`)

Comparing output formats highlights the VM vs. namespace distinction.

Apple Container (container inspect): returns a structured JSON payload detailing allocated hardware resources (memoryInBytes, cpus) and network configurations:

[
  {
    "status": "running",
    "networks": [
      {
        "address": "192.168.64.3/24",
        "gateway": "192.168.64.1",
        "hostname": "my-container.test.",
        "network": "default"
      }
    ],
    "configuration": {
      "id": "my-container",
      "hostname": "my-container",
      "resources": {
        "cpus": 4,
        "memoryInBytes": 1073741824
      },
      "mounts": []
    }
  }
]

Podman (podman inspect): uses OCI/Docker-compatible inspection schemas, nesting status under State and network information under NetworkSettings:

[
  {
    "Id": "abc123...",
    "Name": "/my-container",
    "State": {
      "Status": "running",
      "Running": true,
      "ExitCode": 0
    },
    "NetworkSettings": {
      "IPAddress": "",
      "Networks": {
        "podman": { "IPAddress": "10.88.0.2" }
      }
    },
    "HostConfig": {
      "Memory": 0,
      "NanoCpus": 0
    }
  }
]

Implementation Code Excerpts

Apple Container Go Client (`internal/applecontainer/client.go`)

The Client handles execution safely via direct parameter arrays to eliminate shell injection vulnerabilities:

// Package applecontainer wraps the Apple Container CLI (`container`).
//
// Apple Container (https://github.com/apple/container) is an open-source
// tool from Apple that creates and runs Linux containers as lightweight
// virtual machines on macOS. It is written in Swift and optimised for
// Apple Silicon.
//
// Key architectural traits (v1.3.1):
//   - Each container runs as a dedicated lightweight VM using
//     Virtualization.framework (macOS 26+).
//   - OCI-compatible: pulls/pushes to any standard OCI registry.
//   - CLI convention mirrors Docker/Podman closely.
//   - macOS 26 (Tahoe) required for full network isolation.
//
// Security: all user-supplied strings are validated against strict
// allowlists before use; arguments are passed as a []string to
// exec.Command — no shell is ever invoked.
//
// Validated against Apple Container v1.3.1 (released 2026-08-29).
package applecontainer

import (
    "bytes"
    "context"
    "encoding/json"
    "fmt"
    "os/exec"
    "regexp"
    "strings"
    "time"
)
// ....
// RunContainerDetached starts a container in the background.
// Equivalent CLI command: container run --detach --name <name> <image> [<args>...]
func (c *Client) RunContainerDetached(
    ctx context.Context,
    name, image string,
    cmdArgs ...string,
) (string, time.Duration, error) {
    if err := validateContainerName(name); err != nil {
        return "", 0, err
    }
    if err := validateImage(image); err != nil {
        return "", 0, err
    }
    for _, a := range cmdArgs {
        if err := validateSafeArg(a); err != nil {
            return "", 0, fmt.Errorf("unsafe container argument: %w", err)
        }
    }
    args := append([]string{"run", "--detach", "--name", name, image}, cmdArgs...)
    start := time.Now()
    out, err := c.runner.Run(ctx, args...)
    return strings.TrimSpace(out), time.Since(start), err
}

Podman Go Client (`internal/applecontainer/client.go`)

The Podman Go handles execution safely via Podman in detached mode:

// Package podman wraps the Podman CLI (`podman`).
//
// Podman (https://podman.io) is a daemonless, rootless OCI container engine
// from Red Hat. It is the locally-installed Docker replacement on this machine.
//
// Key architectural traits:
//   - Daemonless: each `podman` invocation is a standalone process.
//   - Rootless by default: containers run as the calling user.
//   - OCI-compatible: uses the same image format and registry protocol.
//   - Uses kernel namespaces + cgroups for isolation (Linux-side).
//   - On macOS, Podman runs inside a Linux VM managed by Podman Machine.
//
// This package mirrors the applecontainer package API so the comparator
// can drive both runtimes with the same interface.
//
// Security: all user-supplied strings are validated against strict
// allowlists before use; no shell expansion takes place.
package podman

import (
    "bytes"
    "context"
    "fmt"
    "os/exec"
    "regexp"
    "strings"
    "time"
)

// ...

// RunContainer starts a container and waits for exit (foreground).
//
// Podman run flags compared to Apple Container:
//   - `--rm`    → remove on exit   (same)
//   - `--name`  → assign name      (same)
//   - `--cpus`  → CPU quota        (cgroup-based, unlike Apple Container VM CPU)
//   - `--memory`→ memory limit     (cgroup limit, not VM RAM)
//   - `--userns=keep-id` → rootless UID mapping (Podman-specific)
//
// Equivalent CLI command:
//
//  podman run --rm --name <name> <image> [<args>...]
func (c *Client) RunContainer(
    ctx context.Context,
    name, image string,
    cmdArgs ...string,
) (string, time.Duration, error) {
    if err := validateContainerName(name); err != nil {
        return "", 0, err
    }
    if err := validateImage(image); err != nil {
        return "", 0, err
    }
    for _, a := range cmdArgs {
        if err := validateSafeArg(a); err != nil {
            return "", 0, fmt.Errorf("unsafe container argument: %w", err)
        }
    }
    args := append([]string{"run", "--rm", "--name", name, image}, cmdArgs...)
    start := time.Now()
    out, err := c.runner.Run(ctx, args...)
    return out, time.Since(start), err
}

// RunContainerDetached starts a container in the background.
//
// Equivalent CLI command:
//
//  podman run --detach --name <name> <image> [<args>...]
func (c *Client) RunContainerDetached(
    ctx context.Context,
    name, image string,
    cmdArgs ...string,
) (string, time.Duration, error) {
    if err := validateContainerName(name); err != nil {
        return "", 0, err
    }
    if err := validateImage(image); err != nil {
        return "", 0, err
    }
    for _, a := range cmdArgs {
        if err := validateSafeArg(a); err != nil {
            return "", 0, fmt.Errorf("unsafe container argument: %w", err)
        }
    }
    args := append([]string{"run", "--detach", "--name", name, image}, cmdArgs...)
    start := time.Now()
    out, err := c.runner.Run(ctx, args...)
    return strings.TrimSpace(out), time.Since(start), err
}

// ListContainers returns containers as JSON.
//
// Educational note:
//   - Podman list output does NOT include an IP column by default
//     (unlike Apple Container which always shows VM IPs).
//   - Use `podman inspect` to retrieve IP info for Podman containers.
//
// Equivalent CLI command:
//
//  podman ps [--all] --format json
func (c *Client) ListContainers(ctx context.Context, all bool) (string, error) {
    args := []string{"ps", "--format", "json"}
    if all {
        args = append(args, "--all")
    }
    return c.runner.Run(ctx, args...)
}

// InspectContainer returns detailed JSON for a named container.
//
// Key JSON fields:
//   - State.Status              → "running" | "exited"
//   - NetworkSettings.IPAddress → container IP (bridge network)
//   - HostConfig.NanoCpus       → CPU limit
//   - HostConfig.Memory         → memory limit in bytes
//
// Equivalent CLI command:
//
//  podman inspect <name>
func (c *Client) InspectContainer(ctx context.Context, name string) (string, error) {
    if err := validateContainerName(name); err != nil {
        return "", err
    }
    return c.runner.Run(ctx, "inspect", name)
}

// InspectImage returns detailed JSON for a local image.
//
// Equivalent CLI command:
//
//  podman inspect --type image <image>
func (c *Client) InspectImage(ctx context.Context, image string) (string, error) {
    if err := validateImage(image); err != nil {
        return "", err
    }
    return c.runner.Run(ctx, "inspect", "--type", "image", image)
}

// StopContainer stops a container gracefully.
//
// Educational note:
//   - Podman sends SIGTERM then SIGKILL (after --time seconds).
//   - Unlike Apple Container, no VM is shut down — only the Linux process.
//
// Equivalent CLI command:
//
//  podman stop <name>
func (c *Client) StopContainer(ctx context.Context, name string) (string, time.Duration, error) {
    if err := validateContainerName(name); err != nil {
        return "", 0, err
    }
    start := time.Now()
    out, err := c.runner.Run(ctx, "stop", name)
    return out, time.Since(start), err
}

// RemoveContainer deletes a stopped container.
//
// Equivalent CLI command:
//
//  podman rm <name>
func (c *Client) RemoveContainer(ctx context.Context, name string) (string, error) {
    if err := validateContainerName(name); err != nil {
        return "", err
    }
    return c.runner.Run(ctx, "rm", name)
}

// RemoveImage removes a local image.
//
// Equivalent CLI command:
//
//  podman rmi <image>
func (c *Client) RemoveImage(ctx context.Context, image string) (string, error) {
    if err := validateImage(image); err != nil {
        return "", err
    }
    return c.runner.Run(ctx, "rmi", image)
}

// Logs returns container stdout/stderr logs.
//
// Educational note:
//   - Podman does not have a --boot flag (no VM boot logs).
//   - Boot parameter is accepted for API compatibility but ignored.
//
// Equivalent CLI command:
//
//  podman logs <name>
func (c *Client) Logs(ctx context.Context, name string, _ bool) (string, error) {
    if err := validateContainerName(name); err != nil {
        return "", err
    }
    return c.runner.Run(ctx, "logs", name)
}

// Stats returns a one-shot resource stats snapshot.
//
// Educational note:
//   - `podman stats --no-stream` returns one sample then exits.
//   - Output format: CONTAINER  CPU%  MEM USAGE/LIMIT  NET I/O  BLOCK I/O
//
// Equivalent CLI command:
//
//  podman stats --no-stream <name>
func (c *Client) Stats(ctx context.Context, name string) (string, error) {
    if err := validateContainerName(name); err != nil {
        return "", err
    }
    return c.runner.Run(ctx, "stats", "--no-stream", name)
}

// SystemInfo returns Podman system information.
//
// Educational note:
//   - Podman has no "system start/stop" — it is daemonless.
//   - `podman system info` shows host + store + registry info.
//
// Equivalent CLI command:
//
//  podman system info --format json
func (c *Client) SystemInfo(ctx context.Context) (string, error) {
    return c.runner.Run(ctx, "system", "info", "--format", "json")
}

// DiskUsage reports disk usage of images/containers/volumes.
//
// Equivalent CLI command:
//
//  podman system df --format json
func (c *Client) DiskUsage(ctx context.Context) (string, error) {
    return c.runner.Run(ctx, "system", "df", "--format", "json")
}

// NetworkList lists container networks.
//
// Equivalent CLI command:
//
//  podman network ls --format json
func (c *Client) NetworkList(ctx context.Context) (string, error) {
    return c.runner.Run(ctx, "network", "ls", "--format", "json")
}

// VolumeList lists named volumes.
//
// Equivalent CLI command:
//
//  podman volume ls --format json
func (c *Client) VolumeList(ctx context.Context) (string, error) {
    return c.runner.Run(ctx, "volume", "ls", "--format", "json")
}

// CommandLine returns the shell command string for educational display.
func CommandLine(args ...string) string {
    return "podman " + strings.Join(args, " ")
}

// ── realRunner ────────────────────────────────────────────────────────────────

// realRunner implements cmdRunner using the installed `podman` binary.
//
// Security: the executable is the literal string "podman", resolved by the
// OS PATH at runtime. User-supplied data flows only into args, which are
// validated before reaching here. No shell is invoked.
type realRunner struct{ verbose bool }

// Run implements CmdRunner.
func (r *realRunner) Run(ctx context.Context, args ...string) (string, error) {
    cmd := exec.CommandContext(ctx, "podman", args...)
    var stdout, stderr bytes.Buffer
    cmd.Stdout = &stdout
    cmd.Stderr = &stderr

    if r.verbose {
        fmt.Printf("[podman] podman %s\n", strings.Join(args, " "))
    }

    if err := cmd.Run(); err != nil {
        return "", fmt.Errorf(
            "podman %q failed: %w\n  stderr: %s",
            strings.Join(args, " "), err, strings.TrimSpace(stderr.String()),
        )
    }
    out := stdout.String()
    if r.verbose && out != "" {
        fmt.Printf("[podman] output:\n%s\n", out)
    }
    return out, nil
}

Dashboard View Model Generator (`internal/podman/client.go`)

Generates a single-file, zero-dependency HTML dashboard to display benchmarking data and visual charts:

// Package reporter — dashboard.go
//
// WriteHTMLDashboard generates a fully self-contained, single-file HTML
// dashboard that visualises all comparison and benchmark data produced by the
// container-compare tool.
//
// Design goals:
//   - Zero external dependencies: no CDN, no JS framework, no fonts fetch.
//     Everything is inlined — CSS, all content, SVG charts built by template.
//   - Tab-based navigation (pure CSS, no JavaScript required).
//   - Colour-coded status badges, SVG horizontal bar charts, timing tables.
//   - Safe HTML generation via html/template throughout; no template.HTML
//     conversions of dynamic data (avoids XSS risk).
//   - Written to ./output/ with an ISO-8601 timestamp prefix.
package reporter

import (
    "fmt"
    "html/template"
    "math"
    "os"
    "path/filepath"
    "time"

    "github.com/apple-container-update/internal/comparator"
)
// ...
// buildBenchRows converts BenchmarkSummary into pre-computed benchRow values.
func buildBenchRows(bs *comparator.BenchmarkSummary) []benchRow {
    rows := make([]benchRow, 0, len(bs.Results))
    for _, pair := range bs.Results {
        appleMS := msFloat(pair.Apple.Mean)
        podmanMS := msFloat(pair.Podman.Mean)
        maxMS := math.Max(appleMS, podmanMS)
        if maxMS == 0 {
            maxMS = 1
        }
        appleW := int(float64(barMaxPx) * appleMS / maxMS)
        podmanW := int(float64(barMaxPx) * podmanMS / maxMS)

        rows = append(rows, benchRow{
            OpName:      pair.OperationName,
            AppleMS:     appleMS,
            AppleBarW:   appleW,
            PodmanMS:    podmanMS,
            PodmanBarW:  podmanW,
            SVGHeight:   104,
        })
    }
    return rows
}

Performance & Benchmarking Observations

Initial benchmarks using container-compare benchmark --image alpine:latest highlight key performance tradeoffs:

Metric / Operation	Apple Container	Podman (macOS)	Tradeoff / Analysis
First Container Startup	~1.0 – 3.0 s MD	~0.3 – 1.0 s MD	Apple incurs cold VM boot overhead per container. MD
Subsequent Startups	~0.5 – 1.5 s MD	~0.3 – 0.8 s MD	Podman forks a process in an active shared VM. MD
Memory Overhead	~100–200 MB per container MD	~1–10 MB per container MD	Apple allocates dedicated hypervisor overhead per instance. MD
x86_64 Translation	High (Rosetta 2) MD+ 1	Low/Moderate (QEMU) MD+ 1	Rosetta 2 significantly reduces runtime CPU overhead for legacy x86 images. MD+ 1

Conclusion

Both Apple Container and Podman offer compelling container development environments on macOS, but address distinct priorities:

Choose Apple Container when strong, hardware-isolated multi-tenant security is required (each container runs in its own kernel boundary), when running non-native x86_64 workloads via Rosetta 2, or when testing native macOS 26 vmnet capabilities.
Choose Podman when startup speed, low per-container memory footprint, and standard Docker-CLI / Kubernetes (podman kube) workflow compatibility are paramount.

Thanks for reading 🚛

Mastering Multi-Agent Systems: Building Sub-Agent Pipelines

Alain Airom (Ayrom) — Thu, 01 Oct 2026 09:29:40 +0000

A sub-agent demonstration implementation with IBM Bob

Introduction

Large Language Model (LLM) agents excel at multi-step tasks when focused on tight scope and explicit goals. However, as tasks expand, single-agent systems suffer from context window saturation, tool confusion, and non-deterministic behavior.

To solve this, modern LLM architectures employ Sub-Agents: isolated, specialized agents spawned by a parent orchestrator to solve individual sub-problems. In this post, we explore a lightweight, framework-agnostic Python implementation demonstrating how an orchestrator agent delegates complex codebase documentation generation to two distinct sub-agents.

The demonstration implemented is meant to showcase the following points;

Understanding Sub-Agent Lifecycles: How isolation, strict parameter passing (SubAgentTask), and unified results (SubAgentResult) prevent state corruption.
Context Partitioning (**fork_context**): Why passing forward selective context payloads improves accuracy and reduces token usage over shared context windows.
Pluggable LLM Abstractions: How to decouple agent logic from model backends (Ollama, llama.cpp, OpenAI) via factory callables.

Implementation and Architecture Overview

The system consists of three main roles across five core Python modules:

subagents-demo/src/
├── models.py                     # Shared DTOs (SubAgentTask, SubAgentResult)
├── utils.py                      # Abstracted LLM client factory & formatting
├── code_explorer_agent.py        # Sub-Agent 1: "explore" type (read-only AST parsing)
├── documentation_writer_agent.py # Sub-Agent 2: "general" type (Markdown generation)
└── orchestrator.py               # Parent Agent managing lifecycle & aggregation

High-Level System Flow

## The Sub-Agent Lifecycle

Every sub-agent in this pattern transitions through four strict phases, ensuring total state isolation and predictable execution:

Core Implementation Details

In this codebase a sub-agent is a self-contained Python object that:

receives a single, clearly bounded task from the orchestrator (parent agent),
executes that task entirely within its own execute() method,
returns its findings as a SubAgentResult dataclass, and
shares no mutable state with any other agent.

This mirrors the Bob IDE sub-agent concept described at https://bob.ibm.com/docs/ide/features/subagents:

“A new, independent agent is created with its own context window. Bob passes it a description of the focused task to perform. The subagent executes the task using its available tools. The subagent returns a summary of its findings back to Bob.”

The demo implements two concrete sub-agents — one of the explore type (read-only codebase analysis) and one of the general type (documentation generation) — and one orchestrator that manages them sequentially.

What sub-agents are NOT in this codebase

They are not OS processes, threads, or coroutines. They are synchronous Python class instances called sequentially within a single process.
They do not have independent network identities or message queues.
They do not share Python object references — the only channel of communication is the SubAgentTask the orchestrator hands in and the SubAgentResult the sub-agent hands back.

Data Transfer Objects (`models.py`)

Communication between agents is strictly controlled using explicit Data Transfer Objects (DTOs) rather than shared memory or global state:

"""
models.py – Shared data-transfer objects used by the orchestrator and both sub-agents.

These lightweight dataclasses mirror the conceptual model described in the Bob
sub-agents documentation:

  • SubAgentTask   – what the orchestrator hands off to a sub-agent
  • SubAgentResult – what a sub-agent hands back to the orchestrator
  • OrchestratorReport – the unified output the orchestrator assembles
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any


@dataclass
class SubAgentTask:
    """
    Encapsulates a focused, self-contained task that the orchestrator delegates
    to a sub-agent.  Mirrors the 'description' parameter passed when Bob spawns
    a subagent (see: https://bob.ibm.com/docs/ide/features/subagents).

    Attributes
    ----------
    agent_id : str
        Logical name of the sub-agent being targeted.
    task_type : str
        Short label for the category of work (e.g. "explore", "general").
    description : str
        Full natural-language description of the work to perform.
    fork_context : bool
        When True the sub-agent receives the orchestrator's accumulated context
        (mirrors the fork_context flag in the Bob spawn_subagent call).
    payload : dict[str, Any]
        Arbitrary data the sub-agent may need (file paths, config, etc.).
    """

    agent_id: str
    task_type: str          # "explore" | "general"
    description: str
    fork_context: bool = False
    payload: dict[str, Any] = field(default_factory=dict)


@dataclass
class SubAgentResult:
    """
    Encapsulates the summary a sub-agent returns to the orchestrator after it
    finishes its isolated work.

    Attributes
    ----------
    agent_id : str
        Identifies which sub-agent produced this result.
    success : bool
        True when the sub-agent completed without error.
    summary : str
        Human-readable summary of what was accomplished (the 'results back to
        the main conversation' as described by the Bob documentation).
    data : dict[str, Any]
        Structured output produced by the sub-agent.
    error : str | None
        Error message when success is False.
    duration_seconds : float
        Wall-clock time the sub-agent spent on its task.
    tools_used : list[str]
        Names of tools the sub-agent invoked during execution.
    """

    agent_id: str
    success: bool
    summary: str
    data: dict[str, Any] = field(default_factory=dict)
    error: str | None = None
    duration_seconds: float = 0.0
    tools_used: list[str] = field(default_factory=list)


@dataclass
class OrchestratorReport:
    """
    The unified output the orchestrator assembles by aggregating every
    sub-agent result.  This is written to the output/ directory as a
    timestamped Markdown file.

    Attributes
    ----------
    generated_at : datetime
        Timestamp of report generation.
    target_directory : str
        Codebase directory that was analysed.
    results : list[SubAgentResult]
        One entry per sub-agent that ran.
    unified_documentation : str
        Final Markdown documentation produced by aggregation.
    total_duration_seconds : float
        Sum of all sub-agent durations.
    """

    generated_at: datetime
    target_directory: str
    results: list[SubAgentResult] = field(default_factory=list)
    unified_documentation: str = ""
    total_duration_seconds: float = 0.0

    # ── convenience helpers ───────────────────────────────────────────────────

    @property
    def all_succeeded(self) -> bool:
        return all(r.success for r in self.results)

    @property
    def failed_agents(self) -> list[str]:
        return [r.agent_id for r in self.results if not r.success]

Read-Only Code Exploration Agent (`code-explorer_agent.py`)

The CodeExplorerAgent is a specialized, read-only agent (explore type). It scans directory structures, parses files using Python's native ast module, extracts structural metadata (classes, methods, imports), and requests a natural-language summary from the LLM without modifying any files:

"""
code_explorer_agent.py – Sub-Agent 1: "explore" type

Responsibility
--------------
This sub-agent is responsible for **read-only codebase exploration**.
It mirrors the 'explore' sub-agent type described in the Bob IDE documentation:
  "Read-only codebase exploration, runs on a lighter model.
   Best for: Searching and summarising code, finding relevant files,
   understanding structure."
  (https://bob.ibm.com/docs/ide/features/subagents#subagent-types)

Lifecycle (per Bob documentation)
----------------------------------
1. REGISTRATION  – The orchestrator instantiates the agent and sets its task.
2. TASK ASSIGNMENT – The orchestrator calls `assign_task()` with a SubAgentTask.
3. EXECUTION     – `execute()` runs in an isolated context:
                   it scans the target directory, collects Python file metadata,
                   and uses the LLM to summarise the code structure.
4. RESULT RETURN – `execute()` returns a SubAgentResult summary back to the
                   orchestrator, which uses it to continue the main flow.

Isolation guarantee
-------------------
This agent does NOT write any files and does NOT share state with the
documentation-writer agent.  It only returns a summary + structured data.
"""

def execute(self) -> SubAgentResult:
    if self._task is None:
        return SubAgentResult(agent_id=self.agent_id, success=False, summary="", error="No task assigned.")

    start_time = time.monotonic()
    tools_used = []

    try:
        # Step a: Discover files
        tools_used.append("directory_scan")
        target_dir = Path(self._task.payload.get("target_directory", "input"))
        py_files = sorted(target_dir.rglob("*.py"))

        # Step b: AST parsing
        tools_used.append("ast_parser")
        modules = [self._parse_python_file(p) for p in py_files]

        # Step c: LLM summarization
        tools_used.append("llm_summarise")
        summary_text = self._llm_summarise(modules)

        return SubAgentResult(
            agent_id=self.agent_id,
            success=True,
            summary=summary_text,
            data={"files_found": len(py_files), "modules": modules},
            duration_seconds=time.monotonic() - start_time,
            tools_used=tools_used,
        )
    except Exception as exc:
        return SubAgentResult(
            agent_id=self.agent_id,
            success=False,
            summary="",
            error=f"{type(exc).__name__}: {exc}",
            duration_seconds=time.monotonic() - start_time,
            tools_used=tools_used,
        )

Payload-Driven Documentation Writer (`documentation_writer_agent.py`)

The DocumentationWriterAgent belongs to the general type. Rather than re-reading the filesystem, it processes the structured payload provided in SubAgentTask.payload (implementing fork_context=True):

"""
documentation_writer_agent.py – Sub-Agent 2: "general" type

Responsibility
--------------
This sub-agent is responsible for **generating structured Markdown documentation**
from the code-exploration findings produced by Sub-Agent 1.

It mirrors the 'general' sub-agent type described in the Bob IDE documentation:
  "Full tool access, runs on the default model.
   Best for: Any self-contained task requiring reads, writes, or commands."
  (https://bob.ibm.com/docs/ide/features/subagents#subagent-types)

Lifecycle (per Bob documentation)
----------------------------------
1. REGISTRATION  – Orchestrator instantiates the agent and sets its identity.
2. TASK ASSIGNMENT – Orchestrator calls `assign_task()` with a SubAgentTask
                     that carries the explorer's findings in its payload.
3. EXECUTION     – `execute()` runs in its own isolated context:
                   it calls the LLM to turn raw exploration data into
                   polished per-module documentation, then assembles a
                   full Markdown document.
4. RESULT RETURN – Returns a SubAgentResult containing the Markdown text
                   back to the orchestrator.

Isolation guarantee
-------------------
This agent does NOT re-read the filesystem.  It only consumes the structured
data passed by the orchestrator in the task payload (the 'fork_context'
pattern).  It does NOT share state with the explorer agent.
"""

def execute(self) -> SubAgentResult:
    if self._task is None:
        return SubAgentResult(agent_id=self.agent_id, success=False, summary="", error="No task assigned.")

    start_time = time.monotonic()
    tools_used = []

    try:
        tools_used.append("payload_reader")
        payload = self._task.payload
        modules = payload.get("modules", [])
        project_name = payload.get("project_name", "Unknown Project")
        explorer_summary = payload.get("explorer_summary", "")

        tools_used.append("llm_document")
        module_sections = [self._document_module(m) for m in modules]

        tools_used.append("markdown_assembler")
        markdown = self._assemble_document(project_name, explorer_summary, module_sections, len(modules))

        return SubAgentResult(
            agent_id=self.agent_id,
            success=True,
            summary=f"Generated documentation for {len(modules)} module(s).",
            data={"markdown": markdown, "modules_documented": len(modules)},
            duration_seconds=time.monotonic() - start_time,
            tools_used=tools_used,
        )
    except Exception as exc:
        return SubAgentResult(
            agent_id=self.agent_id,
            success=False,
            summary="",
            error=f"{type(exc).__name__}: {exc}",
            duration_seconds=time.monotonic() - start_time,
            tools_used=tools_used,
        )

Sequential Orchestration (`orchestrator.py`)

The Orchestrator coordinates task assignment, sequential execution, error capturing, and output writing:

"""
orchestrator.py – Parent / Orchestrator Agent

This module is the heart of the sub-agents demo.  It implements the full
sub-agent lifecycle as described in the Bob IDE documentation:
  https://bob.ibm.com/docs/ide/features/subagents

Orchestrator responsibilities
------------------------------
1. Load configuration and prepare the shared context.
2. REGISTER both sub-agents (CodeExplorerAgent, DocumentationWriterAgent).
3. ASSIGN tasks – each sub-agent receives a focused, non-overlapping task.
4. EXECUTE sub-agents – in this demo they run sequentially; the explorer
   runs first and its output feeds the documentation-writer.
5. COLLECT results – gather SubAgentResult objects returned by each agent.
6. AGGREGATE – combine both results into a single OrchestratorReport and
   write a timestamped Markdown file to the output/ directory.

Sub-agent lifecycle per Bob documentation
------------------------------------------
  • A new, independent agent is created with its own context window.
  • Bob passes it a description of the focused task to perform.
  • The subagent executes the task using its available tools.
  • The subagent returns a summary of its findings back to Bob.
  • Bob uses that summary to continue the main conversation.
"""

# Task Assignment & Execution Flow inside Orchestrator.run()
explorer = CodeExplorerAgent()
doc_writer = DocumentationWriterAgent()

# 1. Execute Code Explorer
explorer_task = SubAgentTask(
    agent_id=explorer.agent_id,
    task_type="explore",
    description="Scan target codebase and extract AST metadata.",
    fork_context=False,
    payload={"target_directory": str(self.target_dir)},
)
explorer.assign_task(explorer_task)
explorer_result = explorer.execute()

# 2. Forward Explorer Context to Documentation Writer (fork_context=True)
doc_writer_task = SubAgentTask(
    agent_id=doc_writer.agent_id,
    task_type="general",
    description="Generate Markdown documentation using exploration payload.",
    fork_context=True,
    payload={
        "project_name": PROJECT_NAME,
        "modules": explorer_result.data.get("modules", []),
        "explorer_summary": explorer_result.summary,
    },
)
doc_writer.assign_task(doc_writer_task)
doc_result = doc_writer.execute()

Pluggable LLM Backend Abstraction

To ensure local development flexibility and seamless production deployments, the agents connect to LLMs through a unified client factory (utils.py). The active backend is selected via the LLM_BACKEND environment variable (ollama, llamacpp, or openai):

"""
utils.py – Shared helpers: logging, LLM backend factory, output utilities.

LLM Backend selection
---------------------
Set the environment variable  LLM_BACKEND  to choose a backend:

  LLM_BACKEND=ollama      (default) – Ollama local HTTP API
  LLM_BACKEND=llamacpp    – llama-cpp-python in-process inference
  LLM_BACKEND=openai      – OpenAI Chat Completions API

Each factory returns a callable with signature:
    chat(system_prompt: str, user_message: str) -> str

The callable prints its response to the console so LLM output is always
visible during a run (prefixed with the backend name for clarity).
"""

def get_llm_client(agent_label: str = "agent") -> Callable[[str, str], str]:
    backend = (os.getenv("LLM_BACKEND") or "ollama").lower().strip()

    if backend in ("llamacpp", "llama_cpp", "llama-cpp"):
        return _llamacpp_chat_factory(agent_label)
    elif backend == "openai":
        return _openai_chat_factory(agent_label)
    else:
        return _ollama_chat_factory(agent_label)

Each factory returns a simple chat(system_prompt, user_message) -> str callable, decoupling agent implementations from specific vendor SDKs.

# Sub-Agents Demo – Sample Codebase – API & Module Documentation

> Auto-generated by the **documentation-writer** sub-agent

---

## Project Overview

**Codebase Summary**

The codebase is composed of four primary modules: `data_processor.py`, `inventory.py`, `order_processor.py`, and `reporting.py`, with a utility module `utils.py` providing general-purpose helpers.

**Overall Purpose**

The codebase is designed to manage inventory and process orders, providing a minimal inventory management system. It includes classes and functions for loading, validating, and transforming tabular data records, as well as handling order creation, validation, and fulfilment.

**Main Components**

1. **Data Processor**: `data_processor.py` provides classes and helpers for loading, validating, and transforming tabular data records, represented as plain Python dicts.
2. **Inventory**: `inventory.py` is a minimal inventory management system, including classes for `Product` and `Inventory`, as well as functions for managing inventory, such as `is_in_stock` and `total_inventory_value`.
3. **Order Processor**: `order_processor.py` handles order creation, validation, and fulfilment against the inventory store, including classes for `OrderStatus`, `OrderLine`, `Order`, and `OrderProcessor`.
4. **Reporting**: `reporting.py` generates human-readable summary reports from the inventory and a list of orders, with functions for `inventory_summary`, `orders_summary`, and `full_report`.
5. **Utilities**: `utils.py` provides general-purpose helpers, including string normalization, simple retry logic, and data transformation functions.

**Key Patterns**

1. **Data Transformation**: The codebase uses plain Python dicts to represent tabular data records, with functions like `normalise_keys` and `add_timestamp` transforming the data.
2. **Inventory Management**: The `inventory.py` module provides functions for managing inventory, such as `is_in_stock` and `total_inventory_value`.
3. **Order Fulfillment**: The `order_processor.py` module handles order creation, validation, and fulfilment against the inventory store.

**Notable Dependencies**

1. **CSV**: The `data_processor.py` module uses the `csv` module for loading and parsing CSV files.
2. **Datetime**: The codebase uses `datetime` for date and time manipulation, including `datetime.datetime`.
3. **Dataclasses**: The `inventory.py` and `order_processor.py` modules use `dataclasses.dataclass` for defining classes.
4. **Type Hints**: The codebase uses type hints, including `typing.Any`, `typing.Optional`, and `typing.Callable`.

---

## Module Reference

*5 module(s) documented below.*

### Overview of data_processor.py

Provides classes and helpers for loading, validating, and transforming tabular data records. Records are represented as plain Python dicts, making the module self-contained with no third-party dependencies.

### Classes

#### CsvLoader
Loads CSV data from a file.

*   `__init__(self, filepath, delimiter)`: Initializes the loader with a file path and optional delimiter.
*   `load(self)`: Reads the CSV file and returns all rows as a list of dicts.
*   `filter_empty(self, records, required_field)`: Removes records where *required_field* is blank or missing.

#### RecordTransformer
Transforms and normalizes record data.

*   `__init__(self, records, id_field)`: Initializes the transformer with a list of record dicts and an ID field.
*   `normalise_keys(self)`: Strips whitespace and lowercase every key in every record.
*   `add_timestamp(self, field_name)`: Injects a UTC timestamp string into every record.
*   `drop_field(self, field_name)`: Removes *field_name* from every record, if present.

### Top-level functions

#### summarise(records)
Returns a lightweight summary of a record collection.

*   Returns: A dictionary containing the total record count and a set of field names.

### Notable dependencies
The module relies on the following Python imports:
*   `csv`
*   `datetime`
*   `typing.Any`

---

**Inventory Management System**
==============================

The `inventory.py` module is a minimal inventory management system used as the target codebase for sub-agent exploration and documentation. It provides a basic framework for managing products, tracking stock levels, and calculating inventory values.

### Classes

#### `Product`

*   **Methods**:
    *   `is_in_stock()`: Returns `True` if there is at least one unit available.
    *   `total_value()`: Returns the total monetary value of all units in stock.

#### `Inventory`

*   **Methods**:
    *   `__init__()`: Initializes the inventory system.
    *   `add_product(product)`: Registers a new product; raises `ValueError` if SKU already exists.
    *   `remove_product(sku)`: Removes and returns a product by SKU, or `None` if not found.
    *   `restock(sku, quantity)`: Increases the quantity of an existing product.
    *   `sell(sku, quantity)`: Decreases product quantity and returns the revenue generated.
    *   `total_inventory_value()`: Returns the combined monetary value of all products.
    *   `low_stock_report(threshold)`: Returns products whose quantity is at or below the threshold.
    *   `__len__()`: Returns the total number of products in the inventory.
    *   `__contains__(sku)`: Returns `True` if the inventory contains a product with the specified SKU.

### Top-level functions

*   `is_in_stock()`: Returns `True` if there is at least one unit available.
*   `total_value()`: Returns the total monetary value of all units in stock.

### Key imports

*   `__future__.annotations`
*   `dataclasses.dataclass`
*   `dataclasses.field`
*   `typing.Optional`

---

## Order Processor Module
The `order_processor.py` module is a Python module that handles order creation, validation, and fulfilment against an Inventory store.

### Classes

#### OrderStatus
The `OrderStatus` class is used to track the status of an order.

#### OrderLine
The `OrderLine` class represents a single line item in an order.

#### Order
The `Order` class represents a complete order with multiple line items.

#### OrderProcessor
The `OrderProcessor` class is responsible for processing orders, including validation and shipment.

### Key Functions

#### line_total
The `line_total` function calculates the total cost of a single line item in an order.

#### add_line
The `add_line` function appends a new line item to an order.

#### order_total
The `order_total` function calculates the total cost of an entire order.

#### cancel
The `cancel` function cancels an order if it has not yet shipped.

#### process
The `process` function validates and fulfils an order.

#### ship
The `ship` function marks a confirmed order as shipped.

#### processed_count
The `processed_count` function returns the number of successfully processed orders.

### Dependencies
The `order_processor.py` module depends on the following key imports:

* `__future__.annotations`
* `dataclasses.dataclass`
* `dataclasses.field`
* `datetime.datetime`
* `enum.Enum`
* `typing.TYPE_CHECKING`
* `inventory.Inventory`

---

## Reporting Module
===============

The `reporting.py` module generates human-readable summary reports from an Inventory and a list of Orders.

## Key Functions
----------------

### `inventory_summary(inventory)`

Produces a plain-text summary of the current inventory state, including:

* Total product count
* Total value
* Low-stock products

### `orders_summary(orders)`

Produces a plain-text summary of a list of orders, grouped by status and reporting aggregate revenue.

### `full_report(inventory, orders)`

Combines inventory and orders summaries into one formatted report, providing a comprehensive overview of the current state.

## Notable Dependencies
----------------------

* `inventory.Inventory`
* `order_processor.Order`
* `order_processor.OrderStatus`
* `datetime.datetime`
* `typing.TYPE_CHECKING`

---

## `utils.py` Module Documentation
#### Overview

The `utils.py` module provides a collection of general-purpose helpers used across the demo project. It covers four concerns:

*   String normalization
*   Simple retry logic for flaky callables
*   Flat key-value configuration loaded from environment or a dict
*   Miscellaneous utility functions

#### Classes

### `Config` Class

The `Config` class serves as a store for configuration values. It provides methods to initialize with optional default values, load environment variables, retrieve specific values, and convert to a dictionary.

#### Key Methods

*   `__init__(self, defaults)`: Initializes with optional default values.
*   `load_env(self, prefix)`: Imports environment variables into the store.
*   `get(self, key, default)`: Returns the string value for a given key, or a default value if absent.
*   `get_int(self, key, default)`: Returns the integer value for a given key, or a default value if absent or not parseable as an integer.
*   `as_dict(self)`: Returns a shallow copy of the internal store.

#### Top-level Functions

### `slugify(text)`

Converts `text` to a URL/filename-safe slug by:

1.  Lowercasing the text.
2.  Replacing runs of non-alphanumeric characters.

### `truncate(text, max_length, suffix)`

Truncates `text` to `max_length` characters, or returns the original text if it's already within the limit.

### `flatten_dict(nested, sep, prefix)`

Recursively flattens a nested dict into a single-level dict, joining nested keys with the specified separator.

#### Key Parameters

*   `retry(func)`: Calls `func` with `args` / `kwargs`, retrying on failure.

#### Notable Dependencies

*   `os`, `re`, `time`, `typing.Any`, `typing.Callable`, `typing.TypeVar`

---

*Documentation generated by the Sub-Agents Demo orchestrator.*

Conclusion & Key Realizations

Building multi-agent systems using this sub-agent demonstration pattern highlights several architectural takeaways:

State Isolation Prevents Cascading Errors: Keeping sub-agents as synchronous Python objects with no shared mutable state ensures that if a single agent encounters bad inputs or invalid syntax (such as AST parse errors), the outer pipeline catches the failure gracefully without crashing the application context.
Context Efficiency via fork_context: Rather than forcing every downstream agent to process the entire execution history, passing explicit payloads (fork_context=True) maintains low token usage and prevents attention degradation in complex tasks.
Determinism over Complex Frameworks: By leveraging standard Python data structures (dataclass), AST tools, and structured DTOs, developers gain full control over agent handoffs without relying on heavy external multi-agent orchestrators or opaque background runtimes.

This pattern offers a clean, production-ready blueprint for building deterministic, scalable AI workflows that remain easy to test, inspect, and maintain.

Thanks for reading 🤖🕵️‍♂️👮‍♀️

Unlocking Background Automation: Wrapping IBM Bob Shell into a Headless REST Service

Alain Airom (Ayrom) — Fri, 25 Sep 2026 12:43:51 +0000

Building a ‘Bob’ headless agent!

Introduction

When building developer tools and automated workflows, IBM Bob Shell (bob run) serves as a powerful coding agent. However, because its native design is a one-shot CLI process meant for interactive terminal usage, running it programmatically inside web applications, background pipelines, or microservices introduces several key challenges:

Lack of Persistence: Every bob run execution is stateless, offering no native thread management or chat history.
Concurrency Bottlenecks: Direct, simultaneous executions risk process collisions without built-in queuing or workspace isolation.
Missing HTTP/REST Surface: Operating Bob headlessly requires a middle layer to handle process orchestration, streaming responses, timeouts, and error handling.

The Headless Bob integration pattern solves this by wrapping IBM Bob Shell inside a persistent service driven via REST endpoints and Server-Sent Events (SSE).

TL;DR-What are Headless Agents?

A headless agent is an autonomous AI agent or software program that operates entirely in the background without a Graphical User Interface (GUI). Unlike traditional AI assistants or applications that rely on chat windows, dashboards, or visual controls, headless agents are designed to execute tasks, process data, and take actions programmatically.

They interact with system environments, APIs, databases, and network protocols directly—typically triggered by events, command-line interfaces (CLI), or scheduled jobs.

Why Headless Agents Are Used

Headless agents excel in scenarios where human visual interaction is either unnecessary or inefficient. Key reasons for their adoption include:

Automated Workflow Execution: They run complex, multi-step backend pipelines (such as CI/CD deployment, data transformation, or automated testing) continuously without requiring manual oversight or interface interactions.
Low Resource Overhead: By eliminating the memory and rendering costs associated with visual interfaces, headless agents consume significantly fewer resources, making them lightweight and fast.
Seamless System Integration: Designed to communicate via APIs, webhooks, or messaging queues, they easily plug into existing microservices and enterprise architectures.
High Scalability: Multiple instances of a headless agent can be spawned concurrently in containers or cloud environments to handle massive workloads in parallel.
Background Monitoring & Event-Driven Action: They can idle efficiently while listening for specific events—such as system alerts, log anomalies, or incoming data packets—and immediately execute remediation or reporting tasks.

Headless Bob (from IBM Open-Sources Building Blocks)

headlessbob runs IBM Bob Shell as a Node.js/TypeScript service with REST and Agent Communication Protocol (ACP) APIs and an integrated browser UI. Both HTTP APIs share an Agent Client Protocol connector to IBM Bob Shell. Agent Communication Protocol is the partner HTTP interface; Agent Client Protocol is the internal Bob interface. You can find more about ACP integration in the IBM Bob ACP Documentation.

📚 View Full Documentation · 📦 Runnable Asset: assets/headlessbob/

Features

Persistent Conversations: Thread-based conversation lifecycle with rename, search, archive, delete, and turn pagination backed by SQLite.
Asynchronous Execution & Streaming: Queued runs with real-time Server-Sent Events (SSE) streaming and execution cancellation.
Dual Protocols: Native text-based ACP 0.2.0 endpoints (/agents, /runs, /session) alongside thread-based REST APIs (/api/v1).
Integrated Browser UI: Single-page chat interface with live markdown rendering, code block copying, run JSON inspection, and workspace file browsing/downloads.
Usage History: Retains previously reported usage. New Client Protocol runs omit unavailable totals; cost/turn limits are not exposed by Bob 2.0.4 ACP, while timeout, output and event limits remain enforced.
Security & Authorization: Bearer-token authentication, caller-isolated workspaces, path traversal guards, and sub-process lifecycle termination.

API Overview

REST Endpoints (`/api/v1`)

All REST endpoints require Authorization: Bearer <TOKEN> and return structured JSON.

| Endpoint                            | Method                   | Purpose                                                      |
| ----------------------------------- | ------------------------ | ------------------------------------------------------------ |
| `/api/v1/capabilities`              | `GET`                    | Retrieve server status, capabilities, and operational limits |
| `/api/v1/threads`                   | `GET`, `POST`            | List, search, or create conversation threads                 |
| `/api/v1/threads/{id}`              | `GET`, `PATCH`, `DELETE` | Inspect, rename, archive, or remove a thread                 |
| `/api/v1/threads/{id}/messages`     | `GET`, `POST`            | Post a prompt (returns run status) or list conversation turns |
| `/api/v1/runs/{id}`                 | `GET`                    | Get run execution status, results, and usage stats           |
| `/api/v1/runs/{id}/events`          | `GET`                    | Stream live run output via Server-Sent Events (SSE)          |
| `/api/v1/runs/{id}/cancel`          | `POST`                   | Cancel an active run execution                               |
| `/api/v1/threads/{id}/files`        | `GET`                    | List files generated in the thread's workspace               |
| `/api/v1/threads/{id}/files/{path}` | `GET`                    | Download a workspace file                                    |

ACP Endpoints (ACP 0.2.0)

Standard Agent Communication Protocol endpoints for multi-agent interoperability:

| Endpoint                | Method          | Purpose                                       |
| ----------------------- | --------------- | --------------------------------------------- |
| `/agents`               | `GET`           | Agent discovery and manifest                  |
| `/runs`                 | `POST`          | Execute an ACP run (sync, async, or streamed) |
| `/runs/{run_id}`        | `GET`           | Get ACP run status and output                 |
| `/runs/{run_id}/events` | `GET`           | Stream ACP run events via SSE                 |
| `/runs/{run_id}/cancel` | `POST`          | Request cancellation of an active run         |
| `/session/{session_id}` | `GET`, `DELETE` | Inspect or terminate an ACP session           |

OpenAPI specifications are available at /api/openapi.json (REST) and /acp/openapi.json (ACP).

Technical Stack and Blocks Required to Create the Same Functionality

Step 1: Prerequisites

Ensure the following are available on your machine:

Tool	Version	When needed
Python	3.11+	Always — runs the Streamlit app
Git	any	Always — to clone this repo
npm / Node.js	22+	Only for Option B (local service)
IBM Bob Shell	2.0.4+	Only for Option B (local service)
IBM Bob API key	—	Only for Option B (local service)

IBM Bob Shell installation

If Bob Shell is not installed:

Go to https://bob.ibm.com and sign in with your IBMid.
Download the Bob Shell installer for macOS (arm64 or x64).
Follow the installation instructions.
Generate an Inference-scope API key at Account → API Keys.

Step 2: Connect to the Headless Bob Service

The demo connects to a Headless Bob service via HEADLESS_BOB_URL.
You do not need to run the service locally. Choose the option that suits you:

Option A — Use a shared or remote instance (no local setup)

If your team already has a Headless Bob service running on a server, container, or cloud URL:

# In your .env, point to the remote instance:
HEADLESS_BOB_URL=https://your-team-server.example.com
HEADLESS_BOB_TOKEN=the-token-defined-on-that-server

No Node.js, npm, or Bob Shell required on your machine.

Option B — Run the service locally (full local setup)

Requires: Node.js 22+, IBM Bob Shell 2.0.4+, and an IBM Bob API key.

Clone and start the service from the IBM self-serve-assets repository:

# Clone the building blocks repo
git clone https://github.com/ibm-self-serve-assets/building-blocks.git
cd building-blocks/ai/ai-engineering/headless-bob/assets/headlessbob

# Install dependencies
npm ci

# Configure credentials
cp .env.example .env
# Edit .env — set these two required fields:
#   BOB_API_KEY=<your-bob-inference-api-key>
#   AUTH_TOKENS={"owner":"my-demo-token-at-least-24-chars"}
#
# AUTH_TOKENS is a JSON object you invent — the key ("owner") is a label,
# the value is the bearer token you will also set as HEADLESS_BOB_TOKEN
# in the demo's .env file.  Generate a strong token with:
#   python3 -c "import secrets; print(secrets.token_urlsafe(32))"

# Build and start
npm run build
npm start

Verify the service is running:

curl -s http://127.0.0.1:8000/api/v1/capabilities \
     -H "Authorization: Bearer my-demo-token-at-least-24-chars" | jq .

You should see a JSON response with version, bob_version, and capacity info.

🏗 System Architecture

The setup consists of a Streamlit frontend UI connecting through a lightweight, zero-dependency REST/SSE client to an external Headless Bob service, which in turn manages local IBM Bob Shell subprocesses.

High-Level Architecture Diagram

⚡ Data Flow: Event-Driven SSE Streaming

Instead of waiting for a full bob run execution to finish, responses are streamed token-by-token using Server-Sent Events (SSE).

💻 Code Highlights & Implementation Lessons

Zero-Dependency Python REST & SSE Client

Using standard library modules (urllib.request), the client consumes data: streams directly without requiring heavy external dependencies.

def stream_events(events_url: str) -> Generator[dict, None, None]:
    """Consume a Server-Sent Events (SSE) stream from Headless Bob."""
    if events_url.startswith("/"):
        events_url = _base_url() + events_url

    headers = {"Accept": "text/event-stream"}
    headers.update(_auth_header())

    req = urllib.request.Request(events_url, headers=headers, method="GET")

    with urllib.request.urlopen(req, timeout=300) as resp:
        buffer = ""
        for raw_line in resp:
            line = raw_line.decode("utf-8", errors="replace")
            buffer += line

            if "\n\n" in buffer or "\r\n\r\n" in buffer:
                chunks = buffer.replace("\r\n", "\n").split("\n\n")
                buffer = chunks[-1]
                for chunk in chunks[:-1]:
                    event = _parse_sse_chunk(chunk)
                    if event:
                        yield event

Overcoming API Constraints

During development with the live service, a strict schema constraint was identified: POST /api/v1/threads/{id}/messages enforces additionalProperties: false. Passing parameters like mode causes an immediate HTTP 400 Bad Request.

To support custom modes without breaking backend constraints, natural-language mode prefixes are injected directly into the content payload:

def send_message(thread_id: str, prompt: str, mode: str = "") -> Run:
    # API only allows {"content": <string>} — no extra fields accepted.
    # Prepend mode as a natural-language instruction when not default.
    if mode and mode != "agent":
        content = f"[Mode: {mode}] {prompt}"
    else:
        content = prompt

    data = _request("POST", f"/api/v1/threads/{thread_id}/messages", {"content": content})
    run_data = data.get("run", {})
    run_id = run_data.get("run_id", "")
    return Run(
        run_id=run_id,
        status=run_data.get("status", "queued"),
        thread_id=thread_id,
        events_url=data.get("events_url", f"/api/v1/runs/{run_id}/events"),
    )

The main app.py does the rest of the job...

"""
app.py
======
Streamlit web application demonstrating **Headless Bob** integration.

This application provides a chat-style UI that connects to a running
Headless Bob service (https://github.com/ibm-self-serve-assets/building-blocks/
tree/main/ai/ai-engineering/headless-bob) via its REST API.

Features demonstrated:
  - Listing and creating conversation threads (persistent conversation lifecycle)
  - Sending prompts to IBM Bob Shell via REST POST /api/v1/threads/{id}/messages
    Note: the message API only accepts {"content": <string>}.  Extra fields
    (including a ``mode`` parameter) cause a 400 Bad Request.  The selected
    mode is therefore prepended to the prompt as "[Mode: <name>] <prompt>"
    when a non-default mode is chosen.
  - Streaming live SSE output — events are typed as ``message.part``,
    ``message.completed``, ``run.completed``, and ``run.failed``
  - Viewing run metadata (duration, tokens, tool calls)
  - Browsing and downloading workspace files generated by Bob
  - Checking server capabilities

Usage:
    streamlit run src/app.py
"""

import os
import time
import threading
from datetime import datetime, timezone
from pathlib import Path

import streamlit as st
from dotenv import load_dotenv

# Load .env from the project root (one directory above src/)
_ROOT = Path(__file__).resolve().parent.parent
load_dotenv(_ROOT / ".env")

# Import our Headless Bob client module
import sys
sys.path.insert(0, str(Path(__file__).resolve().parent))
import headless_bob_client as hb

# ---------------------------------------------------------------------------
# Page configuration
# ---------------------------------------------------------------------------

st.set_page_config(
    page_title="Headless Bob Demo",
    page_icon="🤖",
    layout="wide",
    initial_sidebar_state="expanded",
)

# ---------------------------------------------------------------------------
# Session state helpers
# ---------------------------------------------------------------------------

def _init_state() -> None:
    """Initialize all Streamlit session state keys on first load."""
    defaults = {
        "active_thread_id": None,
        "active_thread_title": "",
        "threads": [],
        "chat_history": [],   # list of {"role": str, "content": str, "meta": dict}
        "streaming_text": "",
        "run_meta": None,
        "last_error": "",
        "capabilities": None,
        "workspace_files": [],
    }
    for key, value in defaults.items():
        if key not in st.session_state:
            st.session_state[key] = value


_init_state()
...

🛠 Features Enabled by Headless Bob

Per-Thread Workspace Isolation: Files created by Bob during a task are constrained to the thread's isolated workspace directory, enabling safe multi-tenant file generation and downloads.

Real-time Stream Rendering: The UI processes message.part events dynamically to build a responsive, token-by-token chat interface.

Execution Observability: Run completion metadata exposes detailed metrics including duration, token usage, and tool execution counts.

Conclusion

By decoupling IBM Bob Shell from terminal-bound single executions, the Headless Bob pattern unlocks a scalable foundation for continuous enterprise automation, background code synthesis, and multi-agent workflows. Transforming one-shot CLI tooling into an event-driven, HTTP-accessible service demonstrates how localized developer AI can seamlessly bridge into cloud-native architectures. As agentic AI paradigms continue to evolve, exposing robust REST APIs, isolated thread workspaces, and real-time SSE streams ensures that background execution models remain responsive, performant, and ready for production integration.

Thanks for reading 👨‍🦲

Deep Agents in Action: Building a Multi-Agent Research System

Alain Airom (Ayrom) — Tue, 22 Sep 2026 08:43:48 +0000

I tested yet another implementation of a multi-agent orchestrator.

The paradigm for AI agents is shifting rapidly from single, monolithic prompt loops to structured multi-agent systems capable of division of labor, state persistence, and fine-grained tool orchestration.

Although IBM watsonx Orchestrate is my go-to solution for enterprise-level agent deployment, I wanted to experiment with LangChain's Deep Agents to evaluate its multi-agent orchestration capabilities.

Introduction: The Deep Agents Architecture

LangChain announced recently Managed Deep Agents, introducing a managed platform to execute deep agent topologies. The Deep Agents paradigm addresses the fundamental limitations of single-agent workflows—specifically context window degradation, tool confusion, and compounding reasoning failures during extended, multi-step tasks.

What is the Deep Agents Pattern?

The Deep Agents pattern is a hierarchical multi-agent architecture where:

A top-level Orchestrator decomposes complex tasks and delegates sub-tasks to specialists.
Each Specialist Agent runs its own isolated ReAct loop with its own tools and memory.
Specialists return structured results to the orchestrator, which synthesises them.
Agents can be composed recursively — an orchestrator can itself be a sub-agent.

This pattern adds genuine value over a single agent because:

| Single Agent                       | Deep Agents                                       |
| ---------------------------------- | ------------------------------------------------- |
| One context window gets overloaded | Each agent has an isolated, focused context       |
| All reasoning in one prompt        | Specialised reasoning per domain                  |
| Hard to scale                      | Add specialists without changing the orchestrator |
| Hard to debug                      | Full delegation trace for auditability            |

---

Excerpt from LangChain Deep Agents Documention;

Deep Agents overview

Deep Agents is the easiest way to start building agents and applications that are powered by LLMs—with built-in capabilities for file systems for context management, subagent-spawning, and long-term memory. Optional capabilities such as task planning and skills extend the harness when your use case needs them. You can use deep agents for any task, including complex, multi-step tasks.Deep Agents comes with the following capabilities:

Take actions in an environment: Take actions via tools, read and write files, execute code
Connect to your data: Load memories, skills, and domain knowledge at the right moment
Manage growing context: Summarize history and offload large results across long runs
Parallelize tasks: Delegate to general or specialized subagents running in isolated context windows
Stay in the loop: Pause for human approval at critical decision points
Improve over time: Update memory, skills, and prompts based on real usage

And from the Github repository;

Deep Agents is an open source agent harness — an opinionated agent that runs out of the box. Extend, override, or replace any piece.

Principles:

Opinionated — defaults tuned for long-horizon, multi-step work
Extensible — override or replace any piece without forking
Model-agnostic — works with any LLM that supports tool calling: frontier, open-weight, or local
Production-ready — built on LangGraph (streaming, persistence, checkpointing) with first-class tracing, evaluation, and deployment via LangSmith

Features include:

Sub-agents — delegate tasks to agents with isolated context windows
Filesystem — read, write, edit, or search over pluggable local, sandboxed, or remote backends
Context management — summarize long threads and offload tool outputs to disk
Shell access — run commands in your sandbox of choice
Persistent memory — pluggable state and store backends for cross-session recall
Human-in-the-loop — approve, edit, or reject tool calls before they run
Skills — reusable behaviors the agent can load on demand
Tools — bring your own functions or any MCP server

Core Concepts of LangChain Deep Agents

Hierarchical Orchestration: A primary agent acts as a supervisor, receiving high-level goals and delegating sub-tasks to dedicated specialist agents rather than executing every step sequentially in a single context loop.
Specialized Agent Roles: Individual agents operate with tightly scoped system prompts, specialized tools, and isolated context windows to execute sub-tasks (e.g., information retrieval, numeric calculations, or document composition).
Isolated Memory Stores: Each agent maintains its own short-term working memory, facts, and episodic memory, preventing state contamination across different phases of execution.
ReAct Reasoning Loop: Sub-agents execute autonomous Thought → Action → Observation loops using validated tool interfaces.

Technical Implementation

Now that the feature set of "Deep Agents" is done, let's jump into a simple didatical (for myself) implementation.

The implementation in the provided repository (scaffolded via IBM Bob) provides a complete local and cloud-compatible implementation of the Deep Agents pattern.

Architecture Overview

User Task
  └─ OrchestratorAgent (Planning & Synthesis)
      ├─ ResearcherAgent (web_search)
      ├─ AnalystAgent (calculator, code_executor)
      └─ WriterAgent (file_reader)

The system is structured across four primary execution layers:

Orchestration Layer (agents/orchestrator.py): Accepts user input, evaluates task complexity, builds a plan, delegates sub-tasks sequentially, and synthesizes outputs.
Specialist Agents Layer (agents/specialists.py): Implements ResearcherAgent, AnalystAgent, and WriterAgent.
Tool Infrastructure (tools/): Provides AST-safe arithmetic evaluation, isolated Python execution, web search abstraction, and sandboxed file I/O.
Relay & UI Layer (app.py & main.py): Offers a Streamlit interface with asynchronous cross-thread state syncing, live phase tracking, and a CLI execution harness.

Technical Stack

The main application is written in Python using Streamlit for UI
The back-end llm inference for local use is tested with Ollama but llama.cpp support and configraion is implemented as well.

Code Structure

Based on the requirements I expressed, hereafter is the current structure of the application;

   deepagents-usecase/
   ├── agents/
   │   ├── __init__.py         # Package exports
   │   ├── base.py             # Abstract BaseAgent + ReAct loop
   │   ├── llm_client.py       # LLM adapter (Ollama/llama.cpp/OpenAI/Anthropic)
   │   ├── messages.py         # Typed message protocol
   │   ├── orchestrator.py     # OrchestratorAgent (top-level)
   │   ├── researcher.py       # ResearcherAgent specialist
   │   ├── analyst.py          # AnalystAgent specialist
   │   └── writer.py           # WriterAgent specialist
   ├── tools/
   │   ├── __init__.py
   │   ├── base.py             # BaseTool + ToolResult
   │   ├── calculator.py       # Safe AST-based arithmetic evaluator
   │   ├── code_executor.py    # Sandboxed Python execution (exec with allow-list)
   │   ├── file_reader.py      # Sandboxed file reading (input/ only)
   │   └── web_search.py       # Web search (stub + live Tavily)
   ├── memory/
   │   ├── __init__.py
   │   └── store.py            # AgentMemoryStore (short/long-term/episodic)
   ├── config/
   │   ├── __init__.py
   │   └── settings.py         # Centralised env-based configuration
   ├── tests/
   │   ├── test_tools.py           # Tool unit tests (71 tests)
   │   ├── test_memory.py          # Memory unit tests (24 tests)
   │   ├── test_agents.py          # Agent integration tests (92 tests)
   │   └── test_code_executor.py   # Code Executor tests (134 tests)
   ├── input/                  # Input documents (content gitignored)
   ├── output/                 # Generated reports (content gitignored)
   ├── memory/                 # Persistent agent memory (JSON files)
   ├── scripts/
   │   ├── start.sh                  # Launch Streamlit in detached mode
   │   ├── stop.sh                   # Gracefully stop the application
   │   ├── cleanup.sh                # Remove .venv, __pycache__, etc.
   │   └── check_code_executor.py    # Standalone Code Executor sanity-check
   ├── Docs/
   │   ├── Architecture.md           # Mermaid architecture diagrams
   │   ├── Quickstart.md             # Step-by-step getting started guide
   │   ├── API.md                    # Full public API reference
   │   └── CodeExecutorVerification.md  # Code Executor test & verification guide
   ├── app.py                  # Streamlit web UI
   ├── main.py                 # CLI entry point
   ├── requirements.txt
   ├── .env.example            # Environment variable template
   └── README.md

Flexible configurations allow the codebase to easily adapt to standard use cases. The backend supports multiple LLM inference sources, including local runtimes like llama.cpp and Ollama, as well as cloud-hosted providers accessed via API keys.

   # =============================================================================
   # Deep Agents System — Environment Configuration
   # =============================================================================
   # Copy this file to .env and fill in your values.
   # NEVER commit the .env file to version control.
   #
   # Usage:
   #   cp .env.example .env
   #   # edit .env with your actual keys
   # =============================================================================

   # ─── LLM Provider ──────────────────────────────────────────────────────────
   # Select ONE provider. Comment out the others.

   # Option A: Ollama (local, default — no API key needed)
   LLM_PROVIDER=ollama
   OLLAMA_BASE_URL=http://localhost:11434
   OLLAMA_MODEL=llama3.2

   # Option B: llama.cpp (local)
   # LLM_PROVIDER=llamacpp
   # LLAMACPP_BASE_URL=http://localhost:9931/v1
   # LLAMACPP_MODEL=local-model

   # Option C: OpenAI
   # LLM_PROVIDER=openai
   # OPENAI_API_KEY=sk-...
   # OPENAI_MODEL=gpt-4o

   # Option D: Anthropic
   # LLM_PROVIDER=anthropic
   # ANTHROPIC_API_KEY=sk-ant-...
   # ANTHROPIC_MODEL=claude-3-5-sonnet-20241022

   # ─── Application Settings ──────────────────────────────────────────────────
   APP_PORT=8501
   APP_HOST=0.0.0.0
   LOG_LEVEL=INFO

   # ─── Agent Configuration ───────────────────────────────────────────────────
   # Maximum reasoning steps per agent (lower = faster on slow local LLMs)
   MAX_AGENT_STEPS=8

   # Maximum tokens per LLM call (1024 is enough for ReAct; raise for longer reports)
   MAX_TOKENS=1024

   # Temperature for LLM responses (0.0 = deterministic, 1.0 = creative)
   TEMPERATURE=0.1

   # ─── Memory & Storage ──────────────────────────────────────────────────────
   # Directory for persistent agent memory (relative to project root)
   MEMORY_DIR=./memory

   # Maximum number of memories to retain per agent
   MAX_MEMORY_ENTRIES=100

   # ─── Tool Configuration ────────────────────────────────────────────────────
   # Enable or disable specific tools (true/false)
   TOOL_WEB_SEARCH_ENABLED=true
   TOOL_CALCULATOR_ENABLED=true
   TOOL_FILE_READER_ENABLED=true
   TOOL_CODE_EXECUTOR_ENABLED=true

   # Web search stub — set to real Tavily/SerpAPI key for live search
   # TAVILY_API_KEY=tvly-...

   # ─── Output ────────────────────────────────────────────────────────────────
   OUTPUT_DIR=./output

Key Code Excerpts

The application serves to illustrate multi-agent orchestration, highlighting how tasks are dynamically distributed and delegated across dedicated toolsets.

Some core implementation details are outlined below. Designed with extensibility in mind, the architecture easily accommodates custom tools and additional specialist agents.

Safe AST Arithmetic Evaluator (`tools/calculator.py`)

To prevent arbitrary code execution vulnerabilities in arithmetic tasks, the system evaluates mathematical expressions using Python's Abstract Syntax Tree (ast) module against an explicit whitelist of AST nodes and functions—avoiding raw eval() calls:

# Whitelisted AST operators and math functions
_SAFE_OPERATORS = {
    ast.Add: operator.add,
    ast.Sub: operator.sub,
    ast.Mult: operator.mul,
    ast.Div: operator.truediv,
    ast.Pow: operator.pow,
    ast.USub: operator.neg,
}

_SAFE_FUNCTIONS = {
    "abs": abs, "round": round, "sqrt": math.sqrt,
    "sin": math.sin, "cos": math.cos, "log": math.log,
    "pi": math.pi, "e": math.e,
}

def _safe_eval(node: ast.expr) -> float:
    """Recursively evaluate an AST expression node in a safe sandbox."""
    if isinstance(node, ast.Constant):
        if isinstance(node.value, (int, float)):
            return float(node.value)
        raise ValueError(f"Unsupported constant type: {type(node.value).__name__}")

    if isinstance(node, ast.BinOp):
        op_type = type(node.op)
        if op_type not in _SAFE_OPERATORS:
            raise ValueError(f"Unsupported binary operator: {op_type.__name__}")
        left = _safe_eval(node.left)
        right = _safe_eval(node.right)
        return _SAFE_OPERATORS[op_type](left, right)

    if isinstance(node, ast.Call):
        func_name = node.func.id
        if func_name not in _SAFE_FUNCTIONS:
            raise ValueError(f"Function '{func_name}' is not whitelisted.")
        args = [_safe_eval(a) for a in node.args]
        return _SAFE_FUNCTIONS[func_name](*args)

    raise ValueError(f"Unsupported AST node type: {type(node).__name__}")

Sandboxed Python Code Executor (`tools/code_executor.py`)

For advanced data processing, the code executor provides an isolated environment with restricted built-ins, pre-imported scientific modules, output redirection, and thread timeout controls:

def _build_exec_namespace() -> dict[str, Any]:
    return {
        "__builtins__": _SAFE_BUILTINS, # Explicit whitelist (no open, __import__, eval)
        "math": math,
        "statistics": statistics,
        "pi": math.pi,
        "e": math.e,
    }

def _execute_code(code: str, timeout: int = 5) -> ToolResult:
    exec_result = _ExecResult()
    captured_io = io.StringIO()

    def _worker():
        namespace = _build_exec_namespace()
        namespace["__builtins__"]["print"] = lambda *args, **kw: print(*args, **{**kw, "file": captured_io})
        try:
            exec(code, namespace)
            exec_result.stdout = captured_io.getvalue()
        except Exception as exc:
            exec_result.error = f"{type(exc).__name__}: {exc}"

    thread = threading.Thread(target=_worker, daemon=True)
    thread.start()
    thread.join(timeout=timeout)
    if thread.is_alive():
        return ToolResult(success=False, error=f"Execution timed out after {timeout}s.")

Thread-Safe Streamlit UI Relay (`app.py`)

Streamlit worker threads cannot directly write to st.session_state due to missing execution contexts (ScriptRunContext). The application uses an @st.cache_resource singleton relay with reentrant locking (threading.RLock) to safely stream agent execution steps across thread boundaries into the web UI:

class _PipelineRelay:
    """Thread-safe relay store for the running multi-agent pipeline."""
    def __init__(self) -> None:
        self.lock = threading.RLock()
        self.status_messages: list[dict[str, str]] = []
        self.agent_cards: dict[str, list[str]] = {}
        self.running: bool = False

    def append_status(self, msg_dict: dict[str, str]) -> None:
        with self.lock:
            self.status_messages.append(msg_dict)
            agent_id = msg_dict.get("agent_id", "orchestrator")
            self.agent_cards.setdefault(agent_id, []).append(msg_dict.get("detail", ""))

@st.cache_resource
def _get_relay() -> _PipelineRelay:
    return _PipelineRelay()

Example Use Case: Climate Change Analysis

Input task:

"Analyse the impact of rising CO2 concentrations on global ecosystems. Include key statistics such as current CO2 levels and the percentage increase since pre-industrial times. Conclude with policy recommendations."

What happens:

Orchestrator decomposes the task into research + analysis + writing.
Researcher runs 3 web searches: CO2 trends, ecosystem impact, policy frameworks.
Analyst uses the calculator tool: (421 - 280) / 280 * 100 = 50.36% increase.
Writer produces a structured Markdown document with executive summary, findings, data analysis, and recommendations.
Orchestrator synthesises all outputs into the final report.

Output is saved to output/deep_agents_report_<timestamp>.md.

Benefits of Renewable Energy and Calculation of 280 Times 42

Benefits of Renewable Energy

Renewable energy reduces greenhouse gas emissions, contributing to climate change mitigation (Source: [National Renewable Energy Laboratory](https://www.nrel.gov/renewables/energy-benefits.html)). This is a key finding supported by credible sources, including the International Renewable Energy Agency (2020) and the World Health Organization (2020).

Renewable energy creates jobs and stimulates local economies (Source: [International Renewable Energy Agency](https://www.irena.org/publications/2020/Jun/Global-Status-Report-2020)). This is a significant benefit highlighted by the International Energy Agency (2020) and the National Bureau of Economic Research (2020).

Renewable energy improves air quality and public health (Source: [World Health Organization](https://www.who.int/news-room/fact-sheets/detail/air-pollution)). This is a critical aspect of renewable energy, supported by the National Renewable Energy Laboratory (n.d.).

The global renewable energy market is projected to reach 30% of total energy production by 2025 (Source: [International Energy Agency](https://www.iea.org/news/pressrelease/2020/june/global-renewables-report-2020/)). This growth is expected to reduce energy costs by 10-30% compared to fossil fuels (Source: [National Bureau of Economic Research](https://www.nber.org/papers/w28822)).

Calculation of 280 Times 42

The calculation of 280 times 42 yields 11,840. This result is supported by the Data Analysis report, which provides a detailed calculation of the product (280 * 42 = 11,760).

Trend Analysis

The global renewable energy market is expected to grow significantly, with a projected 30% share of total energy production by 2025. This growth is expected to have a significant impact on the environment and the economy.

Key Insights

1. Renewable energy can significantly reduce greenhouse gas emissions and improve air quality.
2. Renewable energy can create jobs and stimulate local economies.

Limitations

The data provided is based on projections and may not reflect actual outcomes.

Sources

1. National Renewable Energy Laboratory. (n.d.). Energy Benefits of Renewable Energy. Retrieved from <https://www.nrel.gov/renewables/energy-benefits.html>
2. International Renewable Energy Agency. (2020). Global Status Report 2020. Retrieved from <https://www.irena.org/publications/2020/Jun/Global-Status-Report-2020>
3. World Health Organization. (2020). Air pollution. Retrieved from <https://www.who.int/news-room/fact-sheets/detail/air-pollution>
4. International Energy Agency. (2020). Global Renewables Report 2020. Retrieved from <https://www.iea.org/news/pressrelease/2020/june/global-renewables-report-2020/>
5. National Bureau of Economic Research. (2020). The Economics of Renewable Energy. Retrieved from <https://www.nber.org/papers/w28822>

Adding a New Specialist Agent

As noted earlier, the application is designed to easily accommodate new tools. The section below provides a concrete example demonstrating this extensibility.

Create agents/my_specialist.py subclassing BaseAgent.
Implement the role_description property.
Register the agent in OrchestratorAgent.__init__.
Add delegation logic in OrchestratorAgent.run_task.

Example skeleton:

from agents.base import BaseAgent

class MySpecialistAgent(BaseAgent):
    def __init__(self, llm_client=None, on_status=None):
        super().__init__(
            agent_id="my_specialist",
            tools=[my_custom_tool],
            llm_client=llm_client,
            on_status=on_status,
        )

    @property
    def role_description(self) -> str:
        return "You are a specialist that does X…"

Conclusion

This codebase demonstrates how the Deep Agents design pattern can be built in practice. By combining hierarchical agent delegation with AST-sandboxed tool safety and thread-safe streaming UI updates, the project illustrates how complex agentic pipelines can run reliably across both cloud endpoints and local LLM runtimes (like Ollama and llama.cpp).

Thanks for reading ⚡

Building a Synthetic Data Generator

Alain Airom (Ayrom) — Wed, 16 Sep 2026 11:53:53 +0000

Building a synthetic data generator based on Red Hat's "sdg hub"

Introduction

For a recent initiative, I needed to generate versatile synthetic data to support multiple downstream workflows. While I’ve previously relied on Docling’s Synthetic Data Generator with great results, I wanted to evaluate a tool from Red Hat that caught my eye recently. Drawing inspiration from its core architecture—and tailoring it to my specific pipeline requirements—I built a custom tool that I'll showcase below.

Before diving into what I built, let’s look at the foundation: what exactly is Red Hat's SDG Hub?

TL;DR-What is SDG Hub (Red Hat synthetic data generator hub)?

SDG Hub is an open-source library providing composable blocks and flows for synthetic data generation.

It is a Python framework for building synthetic data generation pipelines. Chain LLM, parsing, transform, filtering, and agent blocks into YAML-defined flows -- then generate training data at scale.

Excerpt from Github;

Get Started

pip install sdg-hub

from sdg_hub import FlowRegistry, Flow

# Discover and load a built-in flow
FlowRegistry.discover_flows()
flow = Flow.from_yaml(FlowRegistry.get_flow_path("MCP Server Distillation"))

# Configure and run
flow.set_model_config(model="openai/gpt-4o")
result = flow.generate(dataset)

See the Quick Start for a full walkthrough, or browse all built-in flows.

SDG Hub is available as a plugin for two coding agents, bringing synthetic data generation directly into your coding workflow.

Claude Code
Via org marketplace (recommended — includes all Red Hat AI plugins):

/plugin marketplace add Red-Hat-AI-Innovation-Team/plugins
/plugin install sdg-hub@Red-Hat-AI-Innovation-Team/plugins

Via this repo directly:

/plugin marketplace add Red-Hat-AI-Innovation-Team/sdg_hub
/plugin install sdg-hub@Red-Hat-AI-Innovation-Team/sdg_hub

From a local clone:

git clone https://github.com/Red-Hat-AI-Innovation-Team/sdg_hub.git
/plugin marketplace add /path/to/sdg_hub

Codex CLI

codex plugin marketplace add Red-Hat-AI-Innovation-Team/plugins

Then install the plugin from the marketplace. See .codex-plugin/INSTALL.md for manual installation.

Once you install the SDG_HUB, it comes with a powerfull command line implementation making it quite easy to generate ad-hoc synthetic data for your requirements.

Implementation

Inspired by the original repository, I adapted the framework to align with Bob and my broader SDLC workflows. The resulting application leverages SDG Hub under the hood while extending its capabilities to match my project’s specific requirements, context, and environment.

SDG App — Based on Red Hat Synthetic Data Generator

A production-ready synthetic data generation pipeline built on Red Hat SDG Hub / InstructLab, featuring:

📄 Multi-format ingestion — Markdown, PDF, DOCX, PPTX, HTML, TXT, JSON/JSONL, images (OCR)
🧠 Taxonomy-driven generation — QA pairs, multi-turn dialogues, instruction pairs
🎓 Teacher-critic loops — quality evaluation with any LLM backend
🛡️ Automated filtering — toxicity, PII detection, deduplication, quality scoring
📦 Flexible export — JSONL, Parquet, Hugging Face Dataset
🌐 REST API (FastAPI) + Web UI (Gradio, 4 tabs) + CLI (Typer)
🤖 Local-first — works entirely offline with Ollama or llama.cpp
🔄 16 sdg_hub flows — RAG evaluation, Knowledge QA, MCP distillation, and more
🤝 Bob plugin — pre-built AI assistant commands for this project (.bob-plugin/)
🧠 Claude plugin — .claude/ skills for data-generation, flow-browser, setup-guide, and synthetic-data-generation

Application Architecture

sdg-app/
├── src/sdg_app/
│   ├── core/
│   │   ├── settings.py        # Pydantic configuration (env + YAML)
│   │   ├── seed_parser.py     # Document ingestion (all formats)
│   │   ├── prompt_builder.py  # Prompt engineering
│   │   ├── providers.py       # LLM backend adapters (LiteLLM)
│   │   ├── generator.py       # Generation orchestrator + critic
│   │   ├── validator.py       # Quality / toxicity / PII / dedup filtering
│   │   ├── exporter.py        # JSONL · Parquet · HF Dataset writers
│   │   ├── job_manager.py     # Background job lifecycle management
│   │   ├── flow_runner.py     # sdg_hub flow routing (CLI + API)
│   │   └── sdg_hub_runner.py  # sdg_hub shared execution engine
│   ├── api/                   # FastAPI REST service (:8000)
│   │   └── routes/            # health · jobs · metrics
│   ├── cli/main.py            # Typer CLI (run · serve · ui · config · flows
│   │                          #            rag-eval · knowledge-qa · mcp-distill)
│   ├── ui/
│   │   ├── gradio_app.py      # Gradio web UI (:7860) — 4-tab layout
│   │   └── tabs/              # Per-tab modules
│   │       ├── llm_backend.py # Reusable provider-selector widget
│   │       ├── rag_eval.py    # 🔍 RAG Evaluation tab
│   │       ├── knowledge_qa.py# 📚 Knowledge QA tab
│   │       └── mcp_distill.py # 🔌 MCP Distillation tab
│   └── utils/observability.py # Logging + metrics
├── tests/
│   ├── unit/                  # 137 unit tests
│   └── integration/           # 7 API integration tests
├── flows/                     # Custom sdg_hub YAML flows
├── input/                     # Seed documents (gitignored content)
├── output/                    # Generated datasets (gitignored content)
├── scripts/                   # setup_venv.sh · start.sh · stop.sh · cleanup.sh
├── Docs/                      # Architecture · Quickstart · UserGuide · GapAnalysis
├── .bob-plugin/               # Bob AI assistant plugin for this project
├── .claude/                   # Claude AI assistant plugin (skills + hooks)
│   ├── settings.json          # Hook configuration
│   ├── hooks/                 # commit-on-stop · track-read · verify-gate
│   └── skills/                # data-generation · flow-browser · setup-guide
│       │                      # synthetic-data-generation (+ references/)
├── Dockerfile                 # Multi-stage, Podman-compatible
├── docker-compose.yml         # Podman Compose deployment
├── config.yaml                # Reference configuration (all options)
└── .env.example               # Environment variable template

Component Stack

The application is using the following stack;

Python Application using Gradio framework for the UI (and the rest of it...)
Ollama/llama.cpp for local LLM inference
Docling for document ingestion

Application's Core

The synthetic data generation framework operates as an asynchronous, decoupled pipeline designed to transform unstructured seed text into verified datasets across multiple formats. Below is a detailed breakdown of the internal mechanics, execution pathways, and validation gates.

Tab	Description	sdg_hub flow
🛠️ Custom Generation	General pipeline: QA, multi-turn, instruction	Built-in
🔍 RAG Evaluation	Q/A/context triplets for evaluating RAG chatbots	`loud-dawn-245`
📚 Knowledge QA	Atomic fact extraction + 5 QA pairs per fact	`heavy-heart-77`
🔌 MCP Distillation	Agent tool-use training data (requires agent server)	`new-night-835`

Every tab includes a provider selector that auto-fills model, endpoint, and key hint when switching between Ollama, llama.cpp, OpenAI, and Custom/vLLM.

Multi-Stage Generation Orchestration

The system uses GenerationOrchestrator (generator.py) as the primary controller for execution workflows. It translates source material into domain-specific, structured datasets through a systematic transformation process:

Task Distribution: Processes source documents through configurable sample types (e.g., Q&A, instruction-following, multi-turn dialogues).
Critic-in-the-Loop Feedback: If enable_critic is toggled on, generated candidates pass through a secondary LLM evaluation step (_run_critic). Samples failing the defined critic_threshold are pruned immediately to save computational overhead.
Downstream Delivery: Successfully critiqued and validated samples are emitted as streamable GeneratedSample objects containing assigned metadata and quality scores.

# Core Orchestration Flow (generator.py)
class GenerationOrchestrator:
    def process_document(self, doc: SeedDocument) -> Iterator[GeneratedSample]:
        for sample_type in self.sample_types:
            raw_samples = self._generate_type(doc, sample_type)
            for raw in raw_samples:
                # Critic Gate: Filter out low-fidelity generations early
                if self.enable_critic and _run_critic(self.critic, doc, raw) < self.critic_threshold:
                    continue

                # Validation Gate: Structural, safety, and content checks
                result = self.validator.validate(raw)
                if result.accepted:
                    yield GeneratedSample(..., quality_score=result.quality_score)

External Flow Delegation & Asynchronous Execution

To support domain-specific workflows without bloating the core orchestrator, the system exposes specialized delegates and asynchronous worker pools:

External Flow Delegation (sdg_hub_runner.py): Wraps complex external generation flows via run_sdg_hub_flow_to_file(). This handles specialized task models like RAG evaluation dataset generation (loud-dawn-245) and Model Context Protocol distillation (new-night-835).
Asynchronous Lifecycle Management (job_manager.py): Manages non-blocking dataset creation via JobManager. Long-running tasks execute inside isolated daemon threads, tracking job states, execution metrics, and generated artifacts without blocking the main runtime process.

Multi-Layer Quality & Safety Engine

The SampleValidator (validator.py) enforces deterministic quality and safety standards before any output is finalized. It applies a strict sequence of validation filters:

Length & Boundary Checks: Rejects truncated, over-length, or structurally malformed generations.
Safety & Compliance Filters: Employs heuristic and classifier checks (_check_toxicity, _check_pii) to prevent toxic text or Personally Identifiable Information from entering training corpora.
Deduplication: Evaluates exact content identity using SHA-256 fingerprinting to eliminate redundancy.
Heuristic Scoring: Computes a overall quality_score based on semantic coherence, dynamic metrics, and formatting integrity. Rejects samples below min_quality_score.

# Validation & Quality Filtering (validator.py)
class SampleValidator:
    def validate(self, sample: Dict) -> ValidationResult:
        text = _extract_text(sample)

        # Fast Failure: Reject length anomalies or toxic outputs
        if not self._check_length(text)[0] or not self._check_toxicity(text)[0]:
            return ValidationResult(accepted=False)

        # Compliance Failure: Filter out detected PII
        if self.enable_pii and not self._check_pii(text)[0]:
            return ValidationResult(accepted=False)

        # Scoring & Deduplication: Calculate heuristic quality & check uniqueness
        quality = self._compute_quality_score(text, sample)
        if quality < self.min_quality_score or not self._check_dedup(text)[0]:
            return ValidationResult(accepted=False, quality_score=quality)

        return ValidationResult(accepted=True, quality_score=quality)

Pipeline Summary

Ingestion & Delegation: JobManager kicks off thread-isolated tasks, delegating complex pipelines to sdg_hub_runner.py or sending standard documents directly to GenerationOrchestrator.
Generation & Critique: Candidate samples are constructed per task specification and evaluated via an optional LLM Critic.
Validation & Filtering: SampleValidator strips unsafe, duplicate, or low-scoring generations via SHA-256, safety models, and length thresholds.
Artifact Export: Accepted samples are assigned quality scores and saved as standardized data artifacts.

REST API

The application exposes a REST API running at http://localhost:8000/docs (with interactive Swagger UI documentation) to control dataset generation, monitor system metrics, and fetch outputs:

Job Execution & Status: Submit new generation workloads (POST /api/v1/jobs), query the complete job list (GET /api/v1/jobs), or retrieve status and results for a specific run (GET /api/v1/jobs/{id}).
Artifact Retrieval: Download generated dataset outputs directly via GET /api/v1/jobs/{id}/artifact.
System Operations: Monitor pipeline operational health (GET /api/v1/health) and real-time execution metrics (GET /api/v1/metrics).

Method	Endpoint	Description
`GET`	`/api/v1/health`	Service health check
`POST`	`/api/v1/jobs`	Submit generation job
`GET`	`/api/v1/jobs`	List all jobs
`GET`	`/api/v1/jobs/{id}`	Job status & results
`GET`	`/api/v1/jobs/{id}/artifact`	Download output file
`GET`	`/api/v1/metrics`	Pipeline metrics

Interactive docs: http://localhost:8000/docs

LLM Implementation

The pipeline supports flexible LLM backend integration across both local and cloud environments through standard .env configuration profiles:

Supported Backends: Choose between local, cost-free runtimes—including Ollama (ollama/llama3.2 at http://localhost:11434) and llama.cpp (openai/<model> at http://localhost:9931/v1)—or production-scale endpoints like OpenAI (openai/gpt-4o) and custom hosted vLLM instances (hosted_vllm/<model>).
Critical Path Rules: When configuring Ollama, do not append /v1 to SDG_LLM_API_BASE, as doing so breaks sdg_hub execution flows.
Setup Reference: Pre-configured environment setups for all three backend profiles are available in .env.example.


  | Backend                     | Provider   | Model Prefix          | `SDG_LLM_API_BASE`                      |
  | --------------------------- | ---------- | --------------------- | --------------------------------------- |
  | **Ollama** (local, free)    | `ollama`   | `ollama/llama3.2`     | `http://localhost:11434` ← **no `/v1`** |
  | **llama.cpp** (local, free) | `llamacpp` | `openai/<model>`      | `http://localhost:9931/v1`              |
  | **OpenAI** (cloud, paid)    | `openai`   | `openai/gpt-4o`       | `https://api.openai.com/v1`             |
  | **vLLM / Custom**           | `vllm`     | `hosted_vllm/<model>` | `http://your-host/v1`

⚠️ Ollama only: do NOT add /v1 to the API base — it breaks sdg_hub flow commands.

Flows

Executing data generation workflows in the application is managed through sdg_hub flows, which structure processing tasks into chained YAML-defined pipeline blocks. The system provides over 14 pre-built flows across seven distinct categories:

Direct Execution: Every flow in the catalog can be triggered directly from the terminal using the command sdg-app run --flow <flow-id>, or configured interactively within the UI wizard.
Dedicated Subcommands: Core workloads feature primary CLI subcommands and direct UI tab integration. Specifically, heavy-heart-77 (sdg-app knowledge-qa) extracts core facts, questions, and responses; loud-dawn-245 (sdg-app rag-eval) formats retrieval-augmented generation datasets with questions, responses, and ground truth contexts; and new-night-835 (sdg-app mcp-distill) processes tool-use trajectory data.
Specialized Requirements: Advanced pipelines—such as Model Context Protocol (MCP) server distillation (new-night-835), code evaluation benchmark generation (domain-code-eval), and red teaming prompt generation (major-sage-742)—require additional setup. For example, running new-night-835 requires a connected agent server (such as LangFlow), while domain-code-eval requires installing sandboxed execution packages like sdg-hub[code]

14+ built-in flows available. Three have dedicated CLI subcommands and UI tabs:

  | Flow ID          | CLI subcommand         | Output                                              |
  | ---------------- | ---------------------- | --------------------------------------------------- |
  | `heavy-heart-77` | `sdg-app knowledge-qa` | `key_fact, question, response`                      |
  | `loud-dawn-245`  | `sdg-app rag-eval`     | `question, response, context, ground_truth_context` |
  | `new-night-835`  | `sdg-app mcp-distill`  | tool-use trajectory data

All flows accessible via: sdg-app run --flow <flow-id>

Specific Plugin built for Bob by Bob

To streamline local workflows, the project includes a specialized Bob AI assistant plugin located in the .bob-plugin/ directory that provides pre-built commands, project context, and automated coding conventions directly inside your IDE environment.

Command Capabilities: Streamline operations with dedicated /sdg-* commands, including execution guides (/sdg-run, /sdg-generate-qa, /sdg-generate-rag), backend switching (/sdg-switch-backend), output inspection (/sdg-inspect-output), debugging (/sdg-debug), API monitoring (/sdg-api), test suite validation (/sdg-test), and custom YAML scaffolding (/sdg-add-flow).
Plugin Architecture: Organizes distinct task workflows inside the commands/ directory, provides continuous runtime context via context/ (project-state.md and known-issues.md), and enforces code style alignment through rules/project-conventions.md.
Usage: Open the repository in Bob and invoke any /sdg-* command in the chat. The assistant automatically ingests the step-by-step instructions from the target command file alongside the project rules without requiring manual context setup each session.

Available commands

Command	What it does
`/sdg-run`	Guide through a complete generation run
`/sdg-generate-qa`	Run Knowledge QA pipeline step-by-step
`/sdg-generate-rag`	Run RAG Evaluation pipeline step-by-step
`/sdg-switch-backend`	Switch between Ollama / llama.cpp / OpenAI
`/sdg-inspect-output`	Read and summarise the latest output file
`/sdg-debug`	Diagnose and fix pipeline failures
`/sdg-add-flow`	Scaffold a new custom sdg_hub YAML flow
`/sdg-test`	Run the test suite and explain failures
`/sdg-api`	Submit and monitor jobs via the REST API

Plugin structure

.bob-plugin/
├── README.md               # Project overview and command index
├── commands/               # One Markdown file per /command
│   ├── sdg-run.md
│   ├── sdg-generate-qa.md
│   ├── sdg-generate-rag.md
│   ├── sdg-switch-backend.md
│   ├── sdg-inspect-output.md
│   ├── sdg-debug.md
│   ├── sdg-add-flow.md
│   ├── sdg-test.md
│   └── sdg-api.md
├── context/
│   ├── project-state.md    # Current working state snapshot
│   └── known-issues.md     # Known limitations + workarounds
└── rules/
    └── project-conventions.md  # Coding and project rules for Bob

Conclusion

Inspired by "Red Hat Synthetic Data Generation Hub", and by integrating an extensible orchestration framework (generator.py), asynchronous job execution, and rigorous multi-stage validation (validator.py), this synthetic data generation system offers an end-to-end pipeline tailored for high-quality dataset creation. This application tries to seamlessly bridge localized engine configurations—such as Ollama and llama.cpp—with cloud-native backends, providing robust execution whether driven via the REST API endpoints, the interactive UI flow wizard, or command-line subcommands. Combined with tailored developer tooling like the embedded Bob AI plugin and specialized sdg_hub flows, the architecture provides a scalable, extensible foundation designed to process raw context files into reliable, instruction-tuned corpora with speed and precision.

Thanks for reading 💾

Testing NVIDIA NemoClaw in a Sandboxed Environment

Alain Airom (Ayrom) — Sat, 12 Sep 2026 14:21:47 +0000

Testing NVIDIA NemoClaw in a Local Sandboxed Environment with Bob

Introduction

For a while, I’ve wanted to test and implement NemoClaw in a sandboxed environment to better understand its capabilities and real-world usage. Safety and isolation are paramount when granting LLMs agency—especially when agents can execute terminal commands, modify local workspaces, or perform egress network requests.

To evaluate its operational stack, I built a test harness and orchestration environment using Bob. In this post, I’ll walk through the fundamentals of NVIDIA NemoClaw, break down its core security architecture, and share an actionable implementation guide so you can spin up your own isolated agent sandbox.

TL;DR: What is NVIDIA NemoClaw?

NemoClaw is an open-source reference stack from NVIDIA designed to execute always-on AI agents safely inside OpenShell sandboxes. It serves as a bridge between high-capability agent runtimes—such as OpenClaw, Hermes, or LangChain Deep Agents—and host system security.

> Image from Nvidia (https://docs.nvidia.com/nemoclaw/user-guide/openclaw/reference/architecture)

+-----------------------------------------------------------------------+
|                              Host Machine                             |
|  +-------------------+  +------------------+  +--------------------+  |
|  | nemoclaw CLI      |  | OpenShell        |  | Host State         |  |
|  | (Node.js CJS)     |  | Gateway Proxy    |  | ~/.nemoclaw/       |  |
|  +---------+---------+  +--------+---------+  +--------------------+  |
+------------|---------------------|------------------------------------+
             | orchestrates        | L7 Proxy & Credentials
             v                     v
+-----------------------------------------------------------------------+
|                    🛡️ Sandbox Container (Podman/Docker)               |
|                    --read-only + tmpfs mounts                          |
|  +-------------------+  +------------------+  +--------------------+  |
|  | AI Agent          |  | NemoClaw         |  | Managed Endpoint   |  |
|  | (OpenClaw/Hermes) |  | In-Process Plugin|  | (inference.local)  |  |
|  +-------------------+  +------------------+  +--------------------+  |
+-----------------------------------------------------------------------+

NVIDIA NemoClaw: Reference Stack for Sandboxed AI Agents in OpenShell (from official Github repository)

Image from Nvidia (https://docs.nvidia.com/nemoclaw/user-guide/openclaw/reference/architecture)
_

NVIDIA NemoClaw is an open source reference stack for running supported AI agents more safely inside NVIDIA OpenShell sandboxes. It provides guided onboarding, managed inference, network policy, managed integrations, snapshots, and lifecycle operations through the NemoClaw CLI and its agent-specific aliases.

Supported agents:

OpenClaw (default)
Hermes
LangChain Deep Agents Code

Technical Stack of the Local Implementation

Components Stack

The code and project uses the following stacks;

Podman (for provisionning NemoClaw Sandbox)
Ollama (local llm, if not using a model hosted by Nvidia using an API Key for Nvidia. One can use either Ollama or llama.cpp)
LLam.cpp (local llm, if not using a model hosted by Nvidia using an API Key for Nvidia. One can use either Ollama or llama.cpp)
Python Streamlit as the application’s framework

Core Architecture Highlights

Host & Sandbox Separation: The orchestration logic (nemoclaw CLI) runs on the host, while the agent executes inside an isolated container namespace (Podman or Docker).
No Raw Credentials in Sandbox: Agents make requests to a local proxy (inference.local). The host-side OpenShell Gateway injects real credentials (such as an NVIDIA_API_KEY) at egress and strips them from incoming responses.
Multi-Provider Routed Inference: Out-of-the-box support for switching between cloud endpoints like NVIDIA Nemotron/NIM and local providers like Ollama (localhost:11434) or llama.cpp (localhost:9931).
Strict Containment Policies: Containers launch with immutable root filesystems (--read-only), capability drops (--cap-drop=all), non-root privilege restrictions, and fine-grained network egress rules.

Implementation Guide And Tests

This step-by-step setup guide includes blueprint excerpts and setup scripts.

Environment Setup & Prerequisites

First, define your environment variables in .env to configure API keys and runtime parameters:

# NemoClaw Environment Configuration
# Copy this file to .env and fill in your values.
# NEVER commit .env to source control.

# ─── Inference Provider ────────────────────────────────────────────────────────
# NVIDIA API key for Nemotron / NIM inference
NVIDIA_API_KEY=your-nvidia-api-key-here

# Default inference model (e.g. nvidia/llama-3.1-nemotron-ultra-253b-v1)
NEMOCLAW_MODEL=nvidia/llama-3.1-nemotron-70b-instruct

# Inference provider: nvidia | ollama | openai-compatible | model-router
NEMOCLAW_INFERENCE_PROVIDER=nvidia

# ─── Ollama (local LLM) ────────────────────────────────────────────────────────
# If using Ollama as inference provider
OLLAMA_BASE_URL=http://localhost:11434

# ─── llama.cpp (local LLM) ────────────────────────────────────────────────────
LLAMACPP_BASE_URL=http://localhost:9931/v1
LLAMACPP_API_KEY=

# ─── OpenShell Gateway ────────────────────────────────────────────────────────
# Port for the OpenShell gateway (default: 10000)
NEMOCLAW_GATEWAY_PORT=10000

# ─── Agent Configuration ──────────────────────────────────────────────────────
# Agent to use: openclaw | hermes | langchain-deepagents-code
NEMOCLAW_AGENT=openclaw

# Sandbox name (1-63 lowercase letters, numbers, internal hyphens)
NEMOCLAW_SANDBOX_NAME=nemoclaw-sandbox

# ─── Dashboard / UI ───────────────────────────────────────────────────────────
# Port for the Streamlit dashboard (avoid 5000 on macOS - reserved for AirDrop)
DASHBOARD_PORT=8501

# ─── Logging ──────────────────────────────────────────────────────────────────
# Log level: debug | info | warn | error
LOG_LEVEL=info

# Log file path (optional, defaults to stdout)
LOG_FILE=

# ─── Network Policy ───────────────────────────────────────────────────────────
# Comma-separated list of allowed egress hosts (beyond defaults)
EXTRA_ALLOWED_HOSTS=

# ─── State / Storage ──────────────────────────────────────────────────────────
NEMOCLAW_STATE_DIR=~/.nemoclaw

# ─── Security ─────────────────────────────────────────────────────────────────
# SSRF validation: strict | permissive
SSRF_VALIDATION_MODE=strict

# ─── MCP Servers ──────────────────────────────────────────────────────────────
# Comma-separated list of MCP server names to enable
MCP_SERVERS=

Run the initialization script to bootstrap Node.js dependencies, set up the Python virtual environment, and link the CLI globally:

#!/usr/bin/env bash
set -e

echo "=== 1. Installing Node Dependencies & Linking CLI ==="
npm install
cd nemoclaw && npm install && npm run build && cd ..
npm link

echo "=== 2. Setting up Python Environment ==="
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt

echo "=== 3. Host Readiness Check ==="
nemoclaw host probe

Architecture & Hardening Configuration

NemoClaw enforces isolation at the container engine layer. The following excerpt shows how runner.js provisions a hardened sandbox using Podman:

// Excerpt: bin/lib/runner.js - Container Launch Configuration
const podmanFlags = [
  'run', '-d',
  '--name', sandboxName,
  '--read-only',                                      // Immutable root filesystem
  '--security-opt', 'no-new-privileges',             // Prevent privilege escalation
  '--cap-drop=all',                                  // Drop all Linux capabilities
  '--tmpfs', '/tmp:rw,noexec,nosuid,size=128m',       // Writable non-executable temp space
  '--tmpfs', '/home/nemoclaw/.streamlit:rw,nosuid,size=32m', // Session cache space
  networkModeFlag,                                   // --network=pasta or --network=bridge
  containerImage
];

// Placeholder: Custom container invocation parameters
// ____________________________________________________
// ____________________________________________________

Network Policy Definition

Network egress is restricted to explicitly allowed hostnames. The network policy is defined declaratively using YAML:

# nemoclaw-blueprint/policies/default-policy.yamlversion: "1"egress:  - host: inference.local    port: 443    protocol: https    comment: "Managed inference endpoint proxied by Gateway"  - host: github.com    port: 443    protocol: https    comment: "Repository operations"  - host: registry.npmjs.org    port: 443    protocol: https    comment: "Package dependencies"deny: []

You can modify rules dynamically via the CLI without recreating the container:

# Add a custom host rule
nemoclaw policy add ________________________

# Apply a preset policy (e.g., Slack or Discord)
nemoclaw policy add slack

Sandbox Onboarding & Execution

Execute the onboarding sequence to build the image, register credentials from .env, instantiate the OpenShell Gateway, and start the container runtime:

# Execute onboarding sequence
nemoclaw onboard --agent openclaw --name ________________________

# Verify status and running processes
nemoclaw status

# Stream live container logs
nemoclaw logs --follow

Python State Inspection Test

You can write tests using unittest or pytest to programmatically inspect sandbox metadata and verify that secrets remain unexposed:

"""Unit tests for the local NemoClaw sandbox demonstration.

The five tests map directly to the scenarios in Docs/NemoClaw-Sandbox-User-Guide.md.
"""

import sys
import unittest
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent / "scripts"))

from nemoclaw_sandbox_demo import AgentProfile, Sandbox, SandboxLimits, SandboxTimeout


class TestNemoClawSandboxScenarios(unittest.TestCase):
    """Validate the sandbox contract without requiring Podman or network access."""

    def setUp(self):
        self.agent = AgentProfile(
            name="agent",
            role="test agent",
            capabilities=frozenset({
                "read_workspace", "write_workspace", "network:inference", "compute", "inter_agent:send",
            }),
        )

    def test_normal_in_sandbox_execution(self):
        """An authorized agent can write and read only its private workspace."""
        with Sandbox() as sandbox:
            sandbox.register_agent(self.agent)
            written = sandbox.run(self.agent, "write_workspace", path="workspace/report.txt", content="pass")
            read = sandbox.run(self.agent, "read_workspace", path="workspace/report.txt")

        self.assertTrue(written.permitted)
        self.assertEqual(read.value, "pass")
        self.assertTrue(sandbox.closed)

    def test_escape_and_network_boundary_are_denied(self):
        """Traversal and non-approved egress are blocked and audited."""
        with Sandbox() as sandbox:
            sandbox.register_agent(self.agent)
            traversal = sandbox.run(self.agent, "read_workspace", path="../../etc/passwd")
            egress = sandbox.run(self.agent, "network_request", host="example.com")

        self.assertFalse(traversal.permitted)
        self.assertIn("filesystem path denied", traversal.error)
        self.assertFalse(egress.permitted)
        self.assertIn("network host denied", egress.error)

    def test_resource_limits_are_enforced(self):
        """Excess memory is rejected and long-running work is terminated."""
        limits = SandboxLimits(memory_limit_mb=8, timeout_seconds=0.1, cpu_limit_seconds=1)
        with Sandbox(limits) as sandbox:
            sandbox.register_agent(self.agent)
            memory = sandbox.run(self.agent, "allocate_memory", megabytes=16)
            timeout = sandbox.run(self.agent, "busy_loop")

        self.assertFalse(memory.permitted)
        self.assertIn("exceeds limit", memory.error)
        self.assertFalse(timeout.permitted)
        self.assertIsInstance(timeout.error, str)
        self.assertIn("timeout", timeout.error)

    def test_inter_agent_communication_stays_inside_sandbox(self):
        """A message can move between registered agents through the local mailbox."""
        sender = self.agent
        receiver = AgentProfile("receiver", "reviewer", frozenset({"inter_agent:receive"}))
        with Sandbox() as sandbox:
            sandbox.register_agent(sender)
            sandbox.register_agent(receiver)
            sent = sandbox.run(sender, "send_message", recipient="receiver", message="approved")
            received = sandbox.run(receiver, "receive_message")

        self.assertTrue(sent.permitted)
        self.assertEqual(received.value, "approved")

    def test_failure_is_reported_and_teardown_is_graceful(self):
        """A denied operation returns a structured error and context is removed."""
        sandbox = Sandbox()
        sandbox.__enter__()
        sandbox.register_agent(self.agent)
        result = sandbox.run(self.agent, "unsupported_operation")
        root = sandbox.root
        sandbox.teardown()

        self.assertFalse(result.permitted)
        self.assertIn("not supported", result.error)
        self.assertTrue(sandbox.closed)
        self.assertFalse(root.exists())
        with self.assertRaises(RuntimeError):
            sandbox.root


if __name__ == "__main__":
    unittest.main()

Test-Case Walkthrough

Now that the application is running, we will validate its behavior under realistic conditions using the following test scenarios.

The test harness exposes five automated test scenarios (test/test_nemoclaw_sandbox.py) that validate allowed behaviors while asserting that policy violations fail predictably.

Normal In-Sandbox Execution (TC-01)

Validates that an authorized agent can write artifacts to its designated workspace and read them back cleanly.

# test/test_nemoclaw_sandbox.py - Scenario 1
def test_normal_in_sandbox_execution(self):
    """An authorized agent can write and read only its private workspace."""
    with Sandbox() as sandbox:
        sandbox.register_agent(self.agent)
        written = sandbox.run(self.agent, "write_workspace", path="workspace/report.txt", content="pass")
        read = sandbox.run(self.agent, "read_workspace", path="workspace/report.txt")

    self.assertTrue(written.permitted)
    self.assertEqual(read.value, "pass")
    self.assertTrue(sandbox.closed)

Scenario 2: Traversal & Egress Violation Defense (TC-02)

Ensures that directory traversal attempts (e.g., ../../etc/passwd) and unauthorized external network requests (e.g., example.com) are immediately denied and logged to the audit trail.

# test/test_nemoclaw_sandbox.py - Scenario 2
def test_escape_and_network_boundary_are_denied(self):
    """Traversal and non-approved egress are blocked and audited."""
    with Sandbox() as sandbox:
        sandbox.register_agent(self.agent)
        traversal = sandbox.run(self.agent, "read_workspace", path="../../etc/passwd")
        egress = sandbox.run(self.agent, "network_request", host="example.com")

    self.assertFalse(traversal.permitted)
    self.assertIn("filesystem path denied", traversal.error)
    self.assertFalse(egress.permitted)
    self.assertIn("network host denied", egress.error)

Scenario 3: Resource Limit Enforcement (TC-03)

Proves that attempts to exceed memory quotas are rejected prior to allocation, and long-running operations (such as infinite loops) are terminated by worker process timeouts.

# test/test_nemoclaw_sandbox.py - Scenario 3
def test_resource_limits_are_enforced(self):
    """Excess memory is rejected and long-running work is terminated."""
    limits = SandboxLimits(memory_limit_mb=8, timeout_seconds=0.1, cpu_limit_seconds=1)
    with Sandbox(limits) as sandbox:
        sandbox.register_agent(self.agent)
        memory = sandbox.run(self.agent, "allocate_memory", megabytes=16)
        timeout = sandbox.run(self.agent, "busy_loop")

    self.assertFalse(memory.permitted)
    self.assertIn("exceeds limit", memory.error)
    self.assertFalse(timeout.permitted)
    self.assertIsInstance(timeout.error, str)
    self.assertIn("timeout", timeout.error)

Scenario 4: Inter-Agent Communication Containment (TC-04)

Demonstrates secure inter-agent messaging routed entirely within in-memory mailboxes inside the sandbox, without opening external host ports or network sockets.

# test/test_nemoclaw_sandbox.py - Scenario 4
def test_inter_agent_communication_stays_inside_sandbox(self):
    """A message can move between registered agents through the local mailbox."""
    sender = self.agent
    receiver = AgentProfile("receiver", "reviewer", frozenset({"inter_agent:receive"}))
    with Sandbox() as sandbox:
        sandbox.register_agent(sender)
        sandbox.register_agent(receiver)
        sent = sandbox.run(sender, "send_message", recipient="receiver", message="approved")
        received = sandbox.run(receiver, "receive_message")

    self.assertTrue(sent.permitted)
    self.assertEqual(received.value, "approved")

Scenario 5: Failure Handling & Graceful Teardown (TC-05)

Confirms that when unsupported or illegal operations occur, structured errors are returned to the caller and the temporary workspace directory is unmounted and completely wiped upon teardown.

# test/test_nemoclaw_sandbox.py - Scenario 5
def test_failure_is_reported_and_teardown_is_graceful(self):
    """A denied operation returns a structured error and context is removed."""
    sandbox = Sandbox()
    sandbox.__enter__()
    sandbox.register_agent(self.agent)
    result = sandbox.run(self.agent, "unsupported_operation")
    root = sandbox.root
    sandbox.teardown()

    self.assertFalse(result.permitted)
    self.assertIn("not supported", result.error)
    self.assertTrue(sandbox.closed)
    self.assertFalse(root.exists())
    with self.assertRaises(RuntimeError):
        sandbox.root

Implementation Blueprint & Execution

To run these verification checks directly in your environment, use the provided Python entry points:

Running the Full Test Suite

Running the tests;

  python3 -m unittest discover -s test -p 'test_nemoclaw_sandbox.py' -v
  test_escape_and_network_boundary_are_denied (test_nemoclaw_sandbox.TestNemoClawSandboxScenarios.test_escape_and_network_boundary_are_denied)
  Traversal and non-approved egress are blocked and audited. ... ok
  test_failure_is_reported_and_teardown_is_graceful (test_nemoclaw_sandbox.TestNemoClawSandboxScenarios.test_failure_is_reported_and_teardown_is_graceful)
  A denied operation returns a structured error and context is removed. ... ok
  test_inter_agent_communication_stays_inside_sandbox (test_nemoclaw_sandbox.TestNemoClawSandboxScenarios.test_inter_agent_communication_stays_inside_sandbox)
  A message can move between registered agents through the local mailbox. ... ok
  test_normal_in_sandbox_execution (test_nemoclaw_sandbox.TestNemoClawSandboxScenarios.test_normal_in_sandbox_execution)
  An authorized agent can write and read only its private workspace. ... ok
  test_resource_limits_are_enforced (test_nemoclaw_sandbox.TestNemoClawSandboxScenarios.test_resource_limits_are_enforced)
  Excess memory is rejected and long-running work is terminated. ... ok

  ----------------------------------------------------------------------
  Ran 5 tests in 0.114s

Or... execute the 5 scenario assertions using standard Python unit testing tools:

# Run unit tests verbosely
python -m unittest test/test_nemoclaw_sandbox.py -v

For automated integration pipelines, run the local harness script directly with the --json flag to inspect structured audit trails:

# Generate JSON audit report
python scripts/nemoclaw_sandbox_demo.py --json

Example JSON Output:

  python3 scripts/nemoclaw_sandbox_demo.py --json
  {
    "results": [
      {
        "operation": "write_workspace",
        "permitted": true,
        "value": "workspace/result.txt",
        "error": null
      },
    {
      "operation": "read_workspace",
      "permitted": true,
      "value": "approved",
      "error": null
    },
    {
      "operation": "network_request",
      "permitted": true,
      "value": {
        "host": "inference.local",
        "status": "routed through inference.local"
      },
      "error": null
    },
    {
      "operation": "network_request",
      "permitted": false,
      "value": null,
      "error": "network host denied: example.com"
    },
    {
      "operation": "read_workspace",
      "permitted": false,
      "value": null,
      "error": "filesystem path denied: ../../etc/passwd"
    },
    {
      "operation": "busy_loop",
      "permitted": false,
      "value": null,
      "error": "operation exceeded 1.0s timeout"
    },
    {
      "operation": "allocate_memory",
      "permitted": false,
      "value": null,
      "error": "memory request 128 MiB exceeds limit 64 MiB"
    },
    {
      "operation": "send_message",
      "permitted": true,
      "value": "message delivered",
      "error": null
    },
    {
      "operation": "receive_message",
      "permitted": true,
      "value": "result ready",
      "error": null
    }
  ],
  "audit_events": 12,
  "sandbox_torn_down": true
}

Troubleshooting & Production Best Practices

When transitioning from the local Python test harness to standard containerized deployments, refer to this common operational matrix:

Issue / Symptom	Root Cause	Resolution
`filesystem path denied`	The requested path attempts directory traversal outside `workspace/`. MD	Use strictly relative subpaths inside `workspace/`; do not weaken path validation. MD
`network host denied`	Host is not present on the policy allowlist. MD	Route requests through `inference.local` or explicitly add authorized host rules. MD
`capability denied`	Agent profile lacks the required permission scope. MD	Grant the minimum required capability to the `AgentProfile` scope. MD
`operation exceeded timeout`	Task exceeded wall-clock deadline. MD	Optimize worker routines or increase `timeout_seconds` in `SandboxLimits` with workload profiling evidence. MD
`sandbox is not active`	Access attempted before context entry or post-teardown. MD	Ensure work is wrapped within `with Sandbox(...)` context managers. MD

Summary Checklist for Deployment

Least-Privilege Capabilities: Start agent profiles with zero capabilities and enable permissions on a strict need-to-have basis.
Egress Boundary Lockdown: Deny external egress by default; route all model traffic through inference.local managed gateways.
Secret Isolation: Never store or pass API keys inside prompts, agent workspace files, or sandbox environments.
Guaranteed Teardown: Ensure sandbox contexts cleanly dispose of transient filesystems immediately upon execution completion.

Conclusion

Implementing NVIDIA NemoClaw with Bob demonstrates how modern AI agent orchestration can strike a practical balance between functional flexibility and strict security controls.

Key takeaways from this implementation:

Zero-Trust Credentials inside Sandboxes: By routing agent API calls through inference.local, raw API keys never enter the sandbox environment.
Defense-in-Depth Container Isolation: Combining --read-only filesystems, capability drops, tmpfs mounts, and network egress policies prevents agents from compromising host resources.
Flexible Runtime Operations: Hot-swapping between cloud endpoints (NVIDIA NIM) and local runtimes (Ollama/llama.cpp) allows switching between local offline execution and production-grade models without altering agent business logic.

NemoClaw provides a dependable framework for safely running agentic workflows locally or in automated pipelines.

Thanks for reading 🦀

End-to-End Enterprise RAG Pipeline with IBM Docling Pipelines and OpenSearch

Alain Airom (Ayrom) — Wed, 09 Sep 2026 14:16:28 +0000

Implementing a RAG pipeline with docling-pipelines and OpenSearch

Introduction

In enterprise Retrieval-Augmented Generation (RAG) applications, moving from a simple proof-of-concept to a robust production system requires high-fidelity document conversion, flexible topology orchestration, and low-latency vector retrieval. Unstructured enterprise documents—such as complex multi-page PDFs, technical whitepapers, and dynamic web pages—often break naive text splitters by scattering table structures, disrupting read order, or discarding critical metadata.

To solve these challenges at scale, this article demonstrates an end-to-end, production-ready RAG pipeline built using IBM Docling Pipelines (docling-pipeline) and OpenSearch. Docling Pipelines provides a Directed Acyclic Graph (DAG) framework for document curation—enabling declarative flows that handle document ingestion, high-accuracy text and structure extraction, language detection, readability analysis, hybrid chunking, embedding generation, and vector indexing. Combined with OpenSearch's k-Nearest Neighbors (k-NN) search capabilities and Ollama's local LLM/embedding inference, this stack provides a privacy-conscious, local-first RAG solution.

TL;DR-What is Docling-Pipelines?

Excerpt from docling-pipelines github repository.

Docling pipelines is an enterprise-grade document curation pipeline for Retrieval Augmented Generation (RAG) applications. It ingests data from unstructured sources, curates documents, and writes entities and vector embeddings to targets — enabling AI-ready pipelines at scale.

It connects to cloud document sources (S3, OneDrive, SharePoint, Google Drive, Box, and more) and extracts content and entities from PDF, DOCX, HTML, images, and other formats using Docling. Extracted content is curated for LLMs, converted into chunks and embeddings, and stored in a vector database such as Milvus or OpenSearch.

Features

📥 Multi-source ingestion — local filesystem, Amazon S3, IBM COS, SharePoint, OneDrive, Google Drive, Box, and web pages
📄 Document extraction — PDF, DOCX, HTML, images, and more via Docling, with optional VLM and ASR pipelines
🧠 Entity extraction — LLM-based extraction via LiteLLM (100+ providers), IBM watsonx.ai, or Docling templates
✂️ Chunking — Docling-native and semantic chunking strategies
🔢 Embeddings — vector embedding generation for any downstream vector store
🔍 Quality operators — language detection, readability scoring, PII/HAP detection, deduplication, redaction, SQL filtering, document classification, and ML enrichment
🗄️ Vector storage — write to OpenSearch or Milvus
🔀 DAG-based flows — define pipelines as JSON with automatic dependency resolution and parallel execution
🔌 Extensible — load custom operators from Python packages, local paths, or S3 without modifying core code
🖥️ Multiple interfaces — CLI, Python API (DocpipeFlowManager), and REST API (FastAPI)

Architecture Overview of the Solution

The pipeline processes unstructured documents through two execution paths: a real execution engine backed by docling-pipelines CLI/Python APIs, and a fallback simulation manager designed for zero-dependency test environments.

┌─────────────────────────────────────────────────────────────────────────────┐
│                           Document Ingestion Layer                          │
│          (Filesystem, Web URLs, S3/COS, SharePoint, Uploaded Files)          │
└──────────────────────────────────────┬──────────────────────────────────────┘
                                       │
                                       ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                           IBM Docling Curation DAG                          │
│                                                                             │
│  ┌────────────────┐     ┌──────────────────┐     ┌───────────────────────┐  │
│  │ ingest_source  ├────►│ extract_operator ├────►│ lang_detect/readability│ │
│  └────────────────┘     └────────┬─────────┘     └───────────┬───────────┘  │
│                                  │                           │              │
│                                  ▼                           ▼              │
│                         ┌────────────────────────────────────────┐          │
│                         │        docling_hybrid Chunker          │          │
│                         └────────────────┬───────────────────────┘          │
│                                          │                                  │
│                                          ▼                                  │
│                         ┌────────────────────────────────────────┐          │
│                         │    embeddings (Ollama/LiteLLM)         │          │
│                         └────────────────┬───────────────────────┘          │
└──────────────────────────────────────────┼──────────────────────────────────┘
                                           │
                                           ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                        OpenSearch Vector Store Index                        │
│                   (k-NN HNSW Indexing & Semantic Search)                    │
└──────────────────────────────────────┬──────────────────────────────────────┘
                                       │
                                       ▼
┌─────────────────────────────────────────────────────────────────────────────┐
│                             RAG Engine & Chat                               │
│              (Context Retrieval → Grounded LLM Prompting)                   │
└─────────────────────────────────────────────────────────────────────────────┘

The system is organized into modular components:

📄 Documents  →  📥 Ingest  →  📝 Extract  →  ✂️ Chunk  →  🔢 Embed  →  🗄️ VectorDB

UI & Control Center (app.py): Interactively configures pipeline parameters, template topologies, document sources, and operator settings.

Flow Definition Builder (flow_builder.py): Programmatically generates valid JSON DAG specifications following the official Docling Flow Authoring Format.

Pipeline Manager (pipeline_manager.py): Orchestrates document processing, handling document parsing, step-by-step operator execution, schema evolution, and artifact export.
Vector Database & RAG Engine (opensearch_client.py & rag_engine.py): Manages OpenSearch index creation, dense vector embeddings, k-NN retrieval, and LLM chat generation.

Implementation

Components Stack

The code and project uses the following stacks;

Docling-pipelines library
OpenSearch as RGA implementation
Podman (for provisionning OpenSearch locally)
Ollama (with granite-embedding:278m for embedding and granite4.2:latest for local chat enablement)
Python Streamlit as the application's framework

Programmatic Pipeline Authoring (`flow_builder.py`)

Docling Pipelines relies on declarative flow representations that enforce dependency graphs between operators. The FlowBuilder class transforms user settings into standard flow JSON objects.

Below is an excerpt showing how the Full RAG Pipeline topology is constructed, complete with LiteLLM-compatible embedding parameters and OpenSearch vector database configs:

# flow_builder.py
from __future__ import annotations
import json
from pathlib import Path
from config import AppConfig

class FlowBuilder:
    """Builds a Docling Pipelines flow JSON from a UI pipeline configuration dict."""

    def __init__(self, config: AppConfig) -> None:
        self.config = config

    def _build_full_rag_pipeline(self, pipeline_cfg: dict) -> dict:
        """Complete RAG pipeline: Ingest → Extract → Chunk → Embed → VectorDB."""
        return {
            "flow_name": "full-rag-pipeline",
            "description": (
                "Complete RAG pipeline: ingest documents, extract text via Docling, "
                "chunk for retrieval, generate embeddings with Ollama, and store in OpenSearch."
            ),
            "flow": [
                self._ingest_operator(pipeline_cfg),
                self._extract_operator(["ingest"]),
                self._chunker_operator(["extract"], pipeline_cfg),
                self._embeddings_operator(["chunker"], pipeline_cfg),
                self._vectordb_operator(["embeddings"], pipeline_cfg),
            ],
            "global_config": {
                "doc_column": "content",
                "disable_validation": False,
            },
        }

    def _embeddings_operator(self, depends_on: list[str], pipeline_cfg: dict) -> dict:
        model = pipeline_cfg.get("ollama_model", self.config.ollama_embedding_model)
        if not model.startswith("openai/"):
            litellm_model = f"openai/{model}"
        else:
            litellm_model = model

        base = self.config.ollama_base_url.rstrip("/")
        api_base = base if base.endswith("/v1") else f"{base}/v1"

        return {
            "name": "embeddings",
            "type": "embeddings",
            "depends_on": depends_on,
            "config": {
                "provider": "litellm",
                "provider_config": {
                    "model_id": litellm_model,
                    "api_base": api_base,
                    "api_key": self.config.ollama_api_key,
                },
            },
        }

    def _vectordb_operator(self, depends_on: list[str], pipeline_cfg: dict) -> dict:
        return {
            "name": "vectordb",
            "type": "vectordb",
            "depends_on": depends_on,
            "config": {
                "provider": "opensearch",
                "provider_config": {
                    "index_name": pipeline_cfg.get(
                        "opensearch_index", self.config.opensearch_index
                    ),
                    "host": self.config.opensearch_host,
                    "port": self.config.opensearch_port,
                    "user": self.config.opensearch_user,
                    "password": self.config.opensearch_password,
                    "use_ssl": self.config.opensearch_use_ssl,
                },
            },
        }

Pipeline Execution Engine (`pipeline_manager.py`)

The PipelineManager resolves execution requests by dynamically choosing between the system binary docling-pipelines and a simulation mode. During document extraction, binary document types (such as PDFs or Word documents) are parsed using docling.

# pipeline_manager.py
from dataclasses import dataclass, field
import json
import logging
import os
import subprocess
import time
from pathlib import Path
from config import AppConfig

logger = logging.getLogger(__name__)

@dataclass
class PipelineResult:
    """Uniform result object returned by any backend."""
    success: bool
    duration_seconds: float
    docs_total: int
    docs_completed: int
    docs_failed: int
    error_message: str = ""
    operator_results: list[dict] = field(default_factory=list)
    output_documents: list[str] = field(default_factory=list)
    chunks: list[dict] = field(default_factory=list)
    schema_changes: dict[str, list[str]] = field(default_factory=dict)
    metadata: dict = field(default_factory=dict)

class PipelineManager:
    def __init__(self, config: AppConfig) -> None:
        self.config = config

    def run(self, flow_def: dict, pipeline_cfg: dict, progress_callback=None) -> PipelineResult:
        simulate = pipeline_cfg.get("simulate", not self.config.docling_pipelines_enabled)
        if not simulate and not self._cli_available():
            logger.warning("docling-pipelines CLI not found; falling back to simulation mode.")
            simulate = True

        if simulate:
            return self._run_simulation(flow_def, pipeline_cfg, progress_callback)
        else:
            return self._run_real(flow_def, pipeline_cfg, progress_callback)

    def _run_real(self, flow_def: dict, pipeline_cfg: dict, progress_callback=None) -> PipelineResult:
        ts = time.strftime("%Y%m%dT%H%M%S")
        flow_path = Path(self.config.output_dir) / f"flow_{ts}.json"
        flow_path.parent.mkdir(parents=True, exist_ok=True)
        flow_path.write_text(json.dumps(flow_def, indent=2), encoding="utf-8")

        cli = self._cli_path() or "docling-pipelines"
        cmd = [cli, "--flow-file", str(flow_path)]
        env = os.environ.copy()
        env["PYTHONPATH"] = "src:" + env.get("PYTHONPATH", "")

        start = time.time()
        completed = subprocess.run(cmd, capture_output=True, text=True, timeout=600, env=env)
        duration = time.time() - start

        success = (completed.returncode == 0)
        error_msg = "" if success else (completed.stdout or completed.stderr)

        return PipelineResult(
            success=success,
            duration_seconds=duration,
            docs_total=len(pipeline_cfg.get("saved_paths", [])) or 1,
            docs_completed=1 if success else 0,
            docs_failed=0 if success else 1,
            error_message=error_msg,
            output_documents=self._collect_output_files(),
        )

Dense Vector Ingestion & RAG Orchestration (`rag_engine.py`)

The RAGEngine links document processing outputs to OpenSearch vector stores. It generates text embeddings via Ollama models (e.g., granite-embedding:278m), performs k-NN retrieval, and queries chat models (e.g., granite4.2:latest).

To prevent hallucinated answers when no context matches a user query, the RAGEngine short-circuits execution and returns a deterministic fallback message without incurring unnecessary LLM generation costs.

# rag_engine.py
import logging
import requests
from config import AppConfig
from opensearch_client import OpenSearchClient

logger = logging.getLogger(__name__)

class RAGEngine:
    _NO_CONTEXT_MSG = (
        "No information about this topic was found in the knowledge base. "
        "Please ingest relevant documents before querying."
    )

    def __init__(self, *, config: AppConfig, os_client: OpenSearchClient | None = None) -> None:
        self.config = config
        self._os = os_client or OpenSearchClient(config=config)

    def ingest_chunks(self, *, chunks: list[dict], progress_callback=None) -> dict:
        if not chunks:
            return {"indexed": 0, "failed": 0, "message": "No chunks provided."}

        self._os.ensure_index()
        embedded = []
        for chunk in chunks:
            text = chunk.get("text", "").strip()
            if not text:
                continue
            vec = self._embed(text)
            embedded.append({**chunk, "embedding": vec})

        index_result = self._os.index_chunks(chunks=embedded)
        return {
            "indexed": index_result["indexed"],
            "failed": index_result.get("failed", 0),
            "message": f"Ingested {index_result['indexed']} chunks.",
        }

    def chat(self, *, query: str, top_k: int = 5, source_filter: str | None = None) -> dict:
        context = self.retrieve(query=query, top_k=top_k, source_filter=source_filter)

        # Deterministic check when retrieval returns zero chunks
        if not context:
            return {
                "answer": self._NO_CONTEXT_MSG,
                "context": [],
                "query": query,
                "sources": [],
                "retrieved_count": 0,
                "fallback": True,
            }

        ctx_text = "\n\n".join(
            f"[Source: {c['source_document']}, chunk {c['chunk_index']}]\n{c['text']}"
            for c in context
        )
        system = (
            "You are a helpful assistant. "
            "Answer the user's question using ONLY the provided context. "
            "If the context does not contain enough information, say so clearly."
        )
        prompt = f"Context:\n{ctx_text}\n\nQuestion: {query}\n\nAnswer:"
        answer = self._chat(prompt, system)
        sources = list(dict.fromkeys(c["source_document"] for c in context))

        return {
            "answer": answer,
            "context": context,
            "query": query,
            "sources": sources,
            "retrieved_count": len(context),
            "fallback": False,
        }

    def _embed(self, text: str) -> list[float]:
        url = f"{self.config.ollama_base_url.rstrip('/')}/api/embed"
        resp = requests.post(url, json={"model": self.config.ollama_embedding_model, "input": text}, timeout=60)
        data = resp.json()
        embeddings = data.get("embeddings") or data.get("embedding")
        return embeddings[0] if isinstance(embeddings[0], list) else embeddings

    def _chat(self, prompt: str, system: str = "") -> str:
        url = f"{self.config.ollama_base_url.rstrip('/')}/api/chat"
        messages = []
        if system:
            messages.append({"role": "system", "content": system})
        messages.append({"role": "user", "content": prompt})

        resp = requests.post(url, json={"model": self.config.ollama_llm_model, "messages": messages, "stream": False}, timeout=120)
        return resp.json().get("message", {}).get("content", "").strip()

Conclusion

Combining IBM Docling Pipelines with OpenSearch creates a reliable, scalable foundation for enterprise RAG applications. By decoupling document curation into modular operator DAGs—covering ingestion, high-precision layout extraction, structure-aware chunking, dense vector embedding, and indexing—organizations can handle diverse unstructured document formats consistently.

Integrating this pipeline with vector search and local LLM runtimes ensures strict context grounding while reducing hallucination risks. As unstructured data continues to grow in enterprise environments, modular Curation-as-Code frameworks like docling-pipeline will be critical for keeping production RAG systems reliable, observable, and easy to maintain.

Thanks for reading 🦆

What is Text Watermarking?

Alain Airom (Ayrom) — Sun, 06 Sep 2026 14:41:08 +0000

Implementing a didactical ‘water marking’ sample application (for self learning and explanation)

Since recently there is a lot of hype of gen ai tools implementing text watermarking for the text they provide. I wanted to learn what it is about and how it could be implemented and detected. So I made a didactical application for myself using Bob. The implementation follows.

Introduction & Theoretical Foundations

Text watermarking in Generative AI refers to the practice of embedding a hidden, detectable signal within AI-generated text. Unlike digital watermarks applied to continuous media like images or audio, a text watermark operates over a discrete domain of tokens without altering the fundamental meaning or readability of the text.

In autoregressive Large Language Models (LLMs), watermarking is integrated directly into the token sampling process rather than applied post-hoc.

The Green-List / Red-List Partitioning Algorithm (Kirchenbauer et al., 2023)

Explanations here: https://github.com/aairom/text-watermarking-test/blob/main/Docs/text_watermarking_algorithms_20260906.md && https://github.com/aairom/text-watermarking-test/blob/main/Docs/WatermarkingImplementation.md

The primary reference framework for LLM text watermarking relies on dynamic vocabulary partitioning:

Context Extraction: At step t, the system extracts a context window ct=(tt−h,…,tt−1) of the preceding h tokens.
Deterministic Pseudo-Random Seeding: The context sequence and a secret key k are hashed using SHA-256 to generate a pseudo-random seed st=SHA-256(ct∥k).
Vocabulary Partitioning: Using st, the vocabulary V is partitioned into a Green List Gt of size γ∣V∣ and a Red List Rt of size (1−γ)∣V∣.
Logit Biasing: During generation, a bias parameter δ>0 is added to the unnormalized logit zv for all green tokens v∈Gt:

Sampling: Softmax normalization is applied to z~v, biasing the model toward green-list tokens.

Statistical Hypothesis Testing & Detection

Detection reconstructs the sequence of green lists without needing access to the model logits or original prompt:

Null Hypothesis (*H*0): Text is human-written or unwatermarked. Each token lands in Gt with probability γ.
Test Statistic (*z*-score): For a sequence of length T with ∣G∣ green tokens:

Decision Rule: Reject H*0 (flag as watermarked) if *z≥z*threshold (typically *z≥4.0, corresponding to p≈3.17×10−5).

Didactical Application Architecture

To explore this concept hands-on, I built a local didactical full-stack application using IBM Bob.

System Flowchart

Data Flow Generation and Detection

Generation Sequence

The data flow generation shows how the application creates a "watermarked" text.

Detection Sequence

This is the architecture of "watermark" detection

Code Implementation & Excerpts

Deterministic Green List Generation

The implement code uses SHA-256 context hashing and random.Random Fisher-Yates shuffling:

def _hash_context_to_seed(context_ids: tuple[int, ...], secret_key: str) -> int:
    payload = secret_key.encode("utf-8") + struct.pack(f"{len(context_ids)}I", *context_ids)
    digest = hashlib.sha256(payload).digest()
    seed = struct.unpack(">Q", digest[:8])[0]
    return seed

def get_green_list(
    context_ids: tuple[int, ...],
    secret_key: str,
    vocab_size: int,
    gamma: float,
) -> frozenset[int]:
    seed = _hash_context_to_seed(context_ids, secret_key)
    rng = random.Random(seed)
    green_size = max(1, int(gamma * vocab_size))
    all_ids = list(range(vocab_size))
    rng.shuffle(all_ids)
    return frozenset(all_ids[:green_size])

Biased Token Sampling Loop

During token generation, δ is applied directly to logits of green-listed candidates:

def _sample_next_token(
    context_ids: tuple[int, ...],
    green_list: Optional[frozenset[int]],
    delta: float,
    temperature: float,
    rng: random.Random,
) -> int:
    prev_id = context_ids[-1] if context_ids else _COMMON_START_TOKENS[0]
    candidates = _BIGRAM_MODEL.get(prev_id, [])

    if not candidates:
        candidates = list(range(4, min(200, _VOCAB_SIZE)))

    pool = list(dict.fromkeys(candidates + list(range(4, 60))))[:80]

    logits = []
    for tid in pool:
        logit = 1.0
        if green_list is not None and tid in green_list:
            logit += delta  # Add bias before temperature scaling
        logits.append(logit / temperature)

    max_l = max(logits)
    exp_l = [math.exp(l - max_l) for l in logits]
    total = sum(exp_l)
    probs = [e / total for e in exp_l]

    r = rng.random()
    cumulative = 0.0
    for tid, p in zip(pool, probs):
        cumulative += p
        if r <= cumulative:
            return tid
    return pool[-1]

Detection Engine & Statistical Test

Detection evaluates z-scores and one-sided p-values via the error function:

def detect_watermark(
    text: str,
    secret_key: str,
    gamma: float = 0.5,
    context_window: int = 1,
    threshold_z: float = 4.0,
) -> DetectionResult:
    token_ids = _tokenize(text)
    T = len(token_ids)

    if T == 0:
        return DetectionResult(...)

    token_infos: list[TokenInfo] = []
    green_count = 0

    for i, tid in enumerate(token_ids):
        if i == 0:
            ctx = (_SPECIAL_TOKENS["<BOS>"],)
        else:
            ctx = tuple(token_ids[max(0, i - context_window): i])

        green_list = get_green_list(ctx, secret_key, _VOCAB_SIZE, gamma)
        seed_val = _hash_context_to_seed(ctx, secret_key)
        is_green = tid in green_list
        if is_green:
            green_count += 1

        token_infos.append(TokenInfo(
            token_id=tid,
            token_text=_ID_TO_TOKEN.get(tid, "<UNK>"),
            position=i,
            is_green=is_green,
            context_hash_seed=seed_val % (2 ** 31),
            green_list_size=len(green_list),
        ))

    expected_g = gamma * T
    z = (green_count - expected_g) / math.sqrt(T * gamma * (1 - gamma))
    p_value = 0.5 * math.erfc(z / math.sqrt(2))
    confidence_pct = round((1.0 - p_value) * 100, 2)
    verdict = "WATERMARKED" if z >= threshold_z else "NOT_WATERMARKED"

    return DetectionResult(
        text=text,
        token_infos=token_infos,
        total_tokens=T,
        green_tokens=green_count,
        expected_green=round(expected_g, 2),
        green_fraction=round(green_count / T, 4),
        z_score=round(z, 4),
        p_value=round(p_value, 6),
        confidence_pct=confidence_pct,
        verdict=verdict,
        threshold_z=threshold_z,
    )

REST API Endpoints (FastAPI)

The REST API using FastAPI implemented provides a full access through APIs to handle the actions like generation, detection…

@app.post("/api/generate", response_model=GenerateResponse, tags=["generation"])
def generate(req: GenerateRequest) -> GenerateResponse:
    wm = generate_text(
        prompt=req.prompt,
        secret_key=req.secret_key,
        gamma=req.gamma,
        delta=req.delta,
        temperature=req.temperature,
        max_new_tokens=req.max_new_tokens,
        context_window=req.context_window,
        watermarked=True,
        seed=42,
    )
    baseline = generate_text(
        prompt=req.prompt,
        secret_key=req.secret_key,
        gamma=req.gamma,
        delta=req.delta,
        temperature=req.temperature,
        max_new_tokens=req.max_new_tokens,
        context_window=req.context_window,
        watermarked=False,
        seed=42,
    )
    return GenerateResponse(watermarked=wm, baseline=baseline)

@app.post("/api/detect", response_model=DetectionResult, tags=["detection"])
def detect(req: DetectRequest) -> DetectionResult:
    return detect_watermark(
        text=req.text,
        secret_key=req.secret_key,
        gamma=req.gamma,
        context_window=req.context_window,
        threshold_z=req.threshold_z,
    )

Attack Simulator & Robustness Testing

To evaluate how well the watermark withstands real-world modifications, the application includes a dedicated Attack Simulator Panel (RobustnessPanel.tsx). Watermarks embedded during generation must remain detectable even after the text undergoes editing, paraphrasing, or noise injection.

Supported Attack Modes

The simulator allows testing four core text perturbation strategies at configurable intensity levels (intensity∈[0.01,1.0]):

Deletion Attack: Randomly strips a percentage of tokens from the text. This simulates aggressive summarization or partial content extraction.
Substitution Attack (Paraphrasing / Typo Injection): Replaces candidate tokens with synonyms or introduces adjacent character swaps. Because green-list selection depends on preceding context ct, modifying even a single token desynchronizes subsequent green-list lookups.
Insertion Attack: Inserts random filler words (e.g., "actually", "basically", "like") into the sequence, introducing local shift distortions into the context window.
Local Shuffle Attack: Randomly scrambles small token windows (3-grams) to disrupt local sentence structure without completely altering word frequency.

API Endpoint (POST /api/perturb);

@app.post("/api/perturb", response_model=PerturbResponse, tags=["robustness"])
def perturb(req: PerturbRequest) -> PerturbResponse:
    """
    Apply a text perturbation and re-run watermark detection.
    Shows watermark persistence / degradation under attack.
    """
    if not req.text.strip():
        raise HTTPException(status_code=422, detail="text must not be blank")

    perturbed = apply_perturbation(req.text, req.mode, req.intensity, req.seed)
    original_det = detect_watermark(
        req.text, req.secret_key, req.gamma, req.context_window, req.threshold_z
    )
    perturbed_det = detect_watermark(
        perturbed, req.secret_key, req.gamma, req.context_window, req.threshold_z
    )

    return PerturbResponse(
        original_text=req.text,
        perturbed_text=perturbed,
        mode=req.mode,
        intensity=req.intensity,
        original_detection=original_det,
        perturbed_detection=perturbed_det,
    )

Empirical Robustness Insights

Context Sensitivity: Because the pseudo-random seed st depends on ct=(tt−h,…,tt−1), a single token edit at position i invalidates the green-list match for the subsequent h tokens.
Sequence Length Resilience: For longer generated sequences (T≥150), the z-score margin is sufficiently high (z≥7.0) that the signal remains statistically significant even after 20–30% token degradation.

Theoretical Algorithm Deep Dive

The Algorithm Reference module (MathExplainer.tsx) provides an interactive breakdown of the mathematical properties governing the Kirchenbauer et al. framework.

Parameter Sensitivity & Trade-offs: the behavior and detectability of the watermark are governed by three primary hyper-parameters:

Parameter	Notation	Description	Impact of Higher Values
Green Fraction	γ	Ratio of vocabulary assigned to Gt.	Increases expected baseline green token count (γT), requiring longer sequences for clear detection.
Logit Bias	δ	Direct bias added to unnormalized logits of Gt tokens.	Increases detection confidence (z-score), but higher values (δ>3.0) can compromise text perplexity and fluency.
Context Window	h (or k)	Number of preceding tokens used as hashing context ct.	Increases resistance against naive token substitution, but makes detection more sensitive to context disruptions.

Mathematical Detection Limit & One-Sided p-Value

To determine the exact likelihood of false positives (a human-written text accidentally triggering a watermark detection), the detection engine converts the computed z-score into an upper-tail p-value using the complementary error function (erfc):

Where:

z≥4.0⟹p≈3.17×10−5 (99.997% confidence)

z≥6.0⟹p≈9.86×10−10 (99.9999999% confidence)

Conclusion

Text watermarking offers a robust, mathematically sound approach to establishing content provenance for large language models. By embedding statistical signals directly during sampling, models can output text that remains entirely natural to human readers while carrying a verifiable fingerprint.

Building this didactical application with IBM Bob highlighted the core mechanics of the algorithm:

Zero Model Overhead: Detection requires no knowledge of model weights or prompts—only the secret key and candidate text.
Controllable Quality Trade-offs: The logit bias δ and green fraction γ provide fine-grained control over detectability versus text quality.
Statistical Power: Even under minor perturbations, longer sequences (T≥200) maintain strong statistical significance against paraphrase attacks.

As generative AI continues to scale, watermarking frameworks like Kirchenbauer et al.'s Green/Red partitioning and DeepMind's SynthID will serve as critical foundational blocks for transparency, regulatory compliance, and content integrity. """

Thanks for reading #️⃣

Running vinD with Podman: Local Kubernetes without Docker Desktop

Alain Airom (Ayrom) — Thu, 03 Sep 2026 10:54:41 +0000

Transforming vinD to vinP (vCluster in Podman) on macOS!

Introduction

I always wanted to test vinD which seems more complete than Kind for a local test and deployment. However the name is "vind - vCluster in Docker" and I use Podman... so I put IBM Bob into work to make all what it takes to use vinD with Podman locally.

While vinD was explicitly designed as a Docker-native tool that expects standard Docker socket endpoints and CLI responses, migrating the setup to a macOS workstation running Podman Desktop is entirely feasible with a few targeted compatibility shims and VM adjustments.

This post walks through the architectural considerations, step-by-step implementation, custom compatibility layers, and deployment manifests required to bring vinD up and running seamlessly on Podman.

TL;DR: What is vinD?

tl;dr — vinD (vCluster in Docker) gives you a real, lightweight Kubernetes cluster running inside a single container in seconds.

Real Kubernetes Control Plane: Unlike standard namespaces, vcluster create provisions a fully isolated, dedicated Kubernetes control plane (API server, etcd, controller manager) inside one host container.
High-Speed Provisioning & Resource Efficiency: Starts in ~15 seconds with a tiny RAM footprint (~150 MB). It supports sleep/wake capabilities (vcluster pause / vcluster resume) so an idle cluster uses zero CPU and zero memory.
Full Tenant Isolation: Multiple isolated vClusters can run side-by-side on a single host machine with separate CRDs, RBAC, and Kubernetes API versions without interfering with each other.
Native Tooling: You interact with vinD using standard kubectl, helm, and vcluster CLI commands just as you would with EKS, GKE, or Kind.

Excerpt from vinD Github;

🎯 What is vind?

vind (vCluster in Docker) is an open-source way to run Kubernetes clusters directly as Docker containers. Built on top of vCluster, vind combines the power of virtual Kubernetes clusters with the simplicity of Docker, creating isolated Kubernetes environments that are perfect for development, testing, and CI/CD pipelines.

Note: vind uses vCluster's Private Nodes mode internally. This is automatically enabled when using the Docker driver and is required for proper operation. This is expected behavior, not a configuration issue.

Why vind?

🚀 Faster than KinD - Optimized container-based architecture
💤 Sleep & Wake - Pause clusters to save resources, resume in under 3 seconds
🎨 Built-in UI - Free vCluster Platform UI for cluster management
⚡ Load Balancers OOB - Automatic LoadBalancer services without extra setup
🐳 Docker Native - Leverages Docker's networking and storage
🔄 Pull-through Cache - Faster image pulls via local Docker daemon
🌐 Hybrid Nodes - Join external nodes (even cloud instances) via VPN
📸 Snapshots - Save cluster state to OCI registries, S3, or local files and restore them
🔧 In-Place K8s Upgrades - Upgrade Kubernetes version without deleting the cluster

Architecture Overview

High-Level Architecture

The following diagram illustrates how the macOS host environment interfaces with the underlying Podman VM and vinD control plane container:

Key Differences & Podman Compatibility Shims

To bridge the gap between vcluster (which assumes a native Docker daemon) and Podman, four specific adjustments are required:

Custom Docker-to-Podman CLI Shim: When probing network subnets, vcluster issues docker network inspect --format '{{.IPAM.Config}}'. Podman formats network inspect JSON under .subnets[].subnet. A lightweight shim intercepts this call and formats the JSON output accordingly.
br_netfilter Kernel Module: Flannel CNI requires the br_netfilter module loaded in the Podman Machine VM. Loading this module explicitly prevents CNI initialization failure.
DOCKER_HOST Socket Redirection: Standard API interactions route through Podman’s Docker-compatible API socket (unix:///.../podman.sock).
Architecture-Aware Image Build (TARGETARCH): Multi-stage container builds must respect the Apple Silicon (arm64) execution environment of the Podman VM to avoid runtime exec format error issues.

**Attention:* Installing vinD on macOS using Homebrew standard commands is not sufficient on its own*. Because the loft-sh/tap repository isn't trusted by default, I used **WailBrew* to explicitly flag and manage the tap as a trusted runnable source.*

Implementation

Application Source & Multi-Stage Containerfile

A lightweight HTTP service written in Go serves as the test workload.

app/main.go

// Package main provides a minimal HTTP Hello World server.
// Port is controlled via the PORT environment variable (default: 8081).
package main

import (
    "fmt"
    "log"
    "net/http"
    "os"
)

// getPort returns the port to listen on.
// It reads from the PORT environment variable; falls back to 8081.
func getPort() string {
    port := os.Getenv("PORT")
    if port == "" {
        port = "8081"
    }
    return port
}

// helloHandler writes "Hello, World!" to the response.
func helloHandler(w http.ResponseWriter, r *http.Request) {
    log.Printf("Request received: %s %s from %s", r.Method, r.URL.Path, r.RemoteAddr)
    fmt.Fprintln(w, "Hello, World!")
}

// healthHandler provides a basic liveness probe endpoint.
func healthHandler(w http.ResponseWriter, r *http.Request) {
    w.WriteHeader(http.StatusOK)
    fmt.Fprintln(w, "OK")
}

// newMux builds and returns the HTTP ServeMux with all routes registered.
func newMux() *http.ServeMux {
    mux := http.NewServeMux()
    mux.HandleFunc("/", helloHandler)
    mux.HandleFunc("/healthz", healthHandler)
    return mux
}

func main() {
    port := getPort()
    addr := ":" + port

    mux := newMux()

    log.Printf("Starting Hello World server on %s", addr)
    if err := http.ListenAndServe(addr, mux); err != nil {
        log.Fatalf("Server failed: %v", err)
    }
}

app/Containerfile

# Stage 1 — Builder
FROM docker.io/library/golang:1.21-alpine AS builder

WORKDIR /src

COPY go.mod ./
RUN go mod download

COPY . .
RUN go test ./... -v

ARG TARGETARCH
RUN CGO_ENABLED=0 GOOS=linux GOARCH=${TARGETARCH:-$(go env GOARCH)} \
    go build -ldflags="-s -w" -o /out/helloworld .

# Stage 2 — Runtime
FROM scratch

COPY --from=builder /out/helloworld /helloworld

EXPOSE 8081
ENV PORT=8081
USER 65534:65534

ENTRYPOINT ["/helloworld"]

Kubernetes Deployment Manifests

The service is exposed on NodePort 30080, which maps to the host.

manifests/namespace.yaml

apiVersion: v1
kind: Namespace
metadata:
  name: helloworld
  labels:
    app.kubernetes.io/managed-by: vind

manifests/deployment.yaml

apiVersion: apps/v1
kind: Deployment
metadata:
  name: helloworld
  namespace: helloworld
  labels:
    app: helloworld
    app.kubernetes.io/name: helloworld
    app.kubernetes.io/version: "1.0.0"
spec:
  replicas: 1
  selector:
    matchLabels:
      app: helloworld
  template:
    metadata:
      labels:
        app: helloworld
    spec:
      securityContext:
        runAsNonRoot: true
        runAsUser: 65534
        runAsGroup: 65534
      containers:
        - name: helloworld
          image: localhost/helloworld:latest
          imagePullPolicy: Never
          ports:
            - containerPort: 8081
              protocol: TCP
              name: http
          env:
            - name: PORT
              value: "8081"
          resources:
            requests:
              cpu: "50m"
              memory: "32Mi"
            limits:
              cpu: "200m"
              memory: "64Mi"
          livenessProbe:
            httpGet:
              path: /healthz
              port: 8081
            initialDelaySeconds: 5
            periodSeconds: 10
          readinessProbe:
            httpGet:
              path: /healthz
              port: 8081
            initialDelaySeconds: 3
            periodSeconds: 5

manifests/service.yaml

apiVersion: v1
kind: Service
metadata:
  name: helloworld
  namespace: helloworld
  labels:
    app: helloworld
    app.kubernetes.io/name: helloworld
spec:
  type: NodePort
  selector:
    app: helloworld
  ports:
    - name: http
      port: 80
      targetPort: 8081
      nodePort: 30080
      protocol: TCP

and even a Load Balancer (manifests/loadbalancer-service.yaml)

  ---
  # =============================================================================
  # Kubernetes Service — helloworld (LoadBalancer type)
  #
  # vinD provides automatic LoadBalancer support out-of-the-box.
  # The EXTERNAL-IP is assigned from the Podman/Docker bridge network.
  # Apply this INSTEAD of service.yaml if you prefer LoadBalancer over NodePort.
  #
  # Apply with:
  #   kubectl apply -f manifests/loadbalancer-service.yaml
  # =============================================================================
  apiVersion: v1
  kind: Service
  metadata:
    name: helloworld-lb
    namespace: helloworld
    labels:
      app: helloworld
      app.kubernetes.io/name: helloworld
  spec:
    type: LoadBalancer
    selector:
      app: helloworld
    ports:
      - name: http
        port: 80           # External port
        targetPort: 8081   # Container port
        protocol: TCP

Orchestration & Podman Shim Setup Script

The orchestration script manages binary downloads, setting up the docker CLI translation shim, enabling br_netfilter in the Podman VM, creating the vCluster instance, importing images into containerd, and deploying manifests.

scripts/setup.sh

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

CLUSTER_NAME="${CLUSTER_NAME:-vind-helloworld}"
APP_IMAGE="${APP_IMAGE:-localhost/helloworld:latest}"

echo "==> 1. Ensuring vcluster CLI is installed..."
if ! command -v vcluster &> /dev/null; then
  mkdir -p "$HOME/.local/bin"
  curl -fsSL "https://github.com/loft-sh/vcluster/releases/latest/download/vcluster-darwin-arm64" \
    -o "$HOME/.local/bin/vcluster"
  chmod +x "$HOME/.local/bin/vcluster"
  export PATH="$HOME/.local/bin:$PATH"
fi

echo "==> 2. Detecting Podman Socket..."
PODMAN_SOCK=$(podman info --format '{{.Host.RemoteSocket.Path}}' 2>/dev/null || true)
if [ -z "$PODMAN_SOCK" ]; then
  PODMAN_SOCK="$HOME/.local/share/containers/podman/machine/qemu/podman.sock"
fi
export DOCKER_HOST="unix://${PODMAN_SOCK}"

echo "==> 3. Creating docker -> podman CLI shim with IPAM translation..."
mkdir -p "$HOME/.local/bin"
cat << 'EOF' > "$HOME/.local/bin/docker"
#!/usr/bin/env bash
if [[ "$*" == *"network inspect"* ]] && [[ "$*" == *"IPAM"* ]]; then
  NET_NAME="${@:$#}"
  SUBNET=$(podman network inspect "$NET_NAME" --format '{{range .subnets}}{{.subnet}}{{end}}' 2>/dev/null)
  echo "[{\"Subnet\":\"${SUBNET}\"}]"
  exit 0
fi
exec podman "$@"
EOF
chmod +x "$HOME/.local/bin/docker"
export PATH="$HOME/.local/bin:$PATH"

echo "==> 3b. Loading br_netfilter module in Podman VM..."
podman machine ssh "sudo modprobe br_netfilter" || true
podman machine ssh "echo 'br_netfilter' | sudo tee /etc/modules-load.d/br_netfilter.conf" || true

echo "==> 4. Configuring vCluster driver..."
vcluster use driver docker

echo "==> 5. Creating vinD cluster..."
if ! vcluster list | grep -q "$CLUSTER_NAME"; then
  vcluster create "$CLUSTER_NAME" -f vcluster.yaml
fi

echo "==> 6. Building Go application image..."
podman build -t "$APP_IMAGE" -f app/Containerfile app/

echo "==> 7. Importing image into cluster containerd..."
podman save "$APP_IMAGE" | docker exec -i "vcluster.cp.${CLUSTER_NAME}" ctr --namespace k8s.io images import -

echo "==> 8. Applying Kubernetes manifests..."
kubectl apply -f manifests/namespace.yaml
kubectl apply -f manifests/deployment.yaml
kubectl apply -f manifests/service.yaml

echo "==> 9. Waiting for deployment readiness..."
kubectl rollout status deployment/helloworld -n helloworld --timeout=60s

echo "==> Setup complete! Test endpoint:"
echo "curl http://localhost:30080"

Conclusion

By addressing the four core points of divergence between Docker and Podman environments—CLI output formatting, kernel module presence in the underlying machine VM, platform-native image builds, and API socket location—vinD runs efficiently on top of Podman.

This setup offers a fast local deployment pipeline, retaining vCluster's single-container control plane model while operating fully within open-source, Podman-based development tooling.

Thanks for reading 🍇

DEV Community: Alain Airom (Ayrom)

Building Document Intelligence with docling-context

Introduction

TL;DR-What is Docling-Context?

2nd TL;DR-What is Docklang?

Implementation

Flow Architecture

Core Implementation & Code Excerpts

Conclusion

Links

Extending AI Capabilities: Looking into Implementing Lifecycle Hooks with IBM Bob

Introduction

What Are Lifecycle Hooks in IBM Bob?

Overall System Architecture

Lifecycle Execution State Diagram

Hooks Decision Tree

Practical Hook Implementations

Conclusion

Links

Bridging AI Agents and VS Code: Implementing the Agent Client Protocol (ACP) for IBM Bob

Introduction

What is the Agent Client Protocol (ACP)?

The Problem ACP Solves

How ACP Operates

Architecture: How Bob Integrates via ACP

TL;DR-ACP vs. Headless Agent Pattern

Side-by-Side Comparison

When to Use Each

Use the ACP extension when…

Use Bob Shell (headless) when…

Component Breakdown

Realization & Code Excerpts

Conclusion

Links and references

Inside Apple Containers: Architectural Deep Dive, CLI Parity, and Benchmarking against Podman

Introduction

High-Level Architecture Comparison

The container-compare Test Harness

Apple Container Architecture: Dedicated Lightweight VMs

Podman Architecture: Shared VM & Linux Namespaces

Key Feature Matrix & Technical Differences

CLI Parity & Developer Workflow

Metadata Schema Differences (inspect)

Implementation Code Excerpts

Apple Container Go Client (internal/applecontainer/client.go)

Podman Go Client (internal/applecontainer/client.go)

Dashboard View Model Generator (internal/podman/client.go)

Performance & Benchmarking Observations

Conclusion

Links

Mastering Multi-Agent Systems: Building Sub-Agent Pipelines

Introduction

Implementation and Architecture Overview

High-Level System Flow

Core Implementation Details

What sub-agents are NOT in this codebase

Data Transfer Objects (models.py)

Read-Only Code Exploration Agent (code-explorer_agent.py)

Payload-Driven Documentation Writer (documentation_writer_agent.py)

Sequential Orchestration (orchestrator.py)

Pluggable LLM Backend Abstraction

Conclusion & Key Realizations

Links

Unlocking Background Automation: Wrapping IBM Bob Shell into a Headless REST Service

Introduction

TL;DR-What are Headless Agents?

Why Headless Agents Are Used

Headless Bob (from IBM Open-Sources Building Blocks)

Features

API Overview

REST Endpoints (/api/v1)

ACP Endpoints (ACP 0.2.0)

Technical Stack and Blocks Required to Create the Same Functionality

Step 1: Prerequisites

IBM Bob Shell installation

Step 2: Connect to the Headless Bob Service

Option A — Use a shared or remote instance (no local setup)

Option B — Run the service locally (full local setup)

🏗 System Architecture

High-Level Architecture Diagram

The `container-compare` Test Harness

Metadata Schema Differences (`inspect`)

Apple Container Go Client (`internal/applecontainer/client.go`)

Podman Go Client (`internal/applecontainer/client.go`)

Dashboard View Model Generator (`internal/podman/client.go`)

Data Transfer Objects (`models.py`)

Read-Only Code Exploration Agent (`code-explorer_agent.py`)

Payload-Driven Documentation Writer (`documentation_writer_agent.py`)

Sequential Orchestration (`orchestrator.py`)

REST Endpoints (`/api/v1`)

Safe AST Arithmetic Evaluator (`tools/calculator.py`)

Sandboxed Python Code Executor (`tools/code_executor.py`)

Thread-Safe Streamlit UI Relay (`app.py`)