DEV Community: Manish

Build a custom multi-step AI workflow graph with openJiuwen Agent Core

Manish — Tue, 27 Oct 2026 17:00:16 +0000

A multi-step LLM pipeline usually breaks at a handoff. The triage step returns "high", but the routing step expects the integer 4. Nothing raises an error. The bad value keeps moving downstream until a customer gets the wrong reply or a dashboard shows misleading figures. By then you can't say where the data went wrong, so you can't fix it quickly.

This tutorial builds a support-ticket workflow on openJiuwen Agent Core, an open-source Python SDK for agent workflows. The graph has five parts:

A Start node.
An LLMComponent that triages the ticket into a typed record (named fields with fixed types).
A human approval step.
A second LLMComponent that drafts a reply.
An End node.

A WorkflowAgent runs the graph.

You'll also see where Agent Core enforces data contracts (rules for what each step may accept and return). The evidence comes from a proof suite that ran every scenario against the real SDK and a real model, three times in a row. Agent Core checks two boundaries at run time and leaves one unchecked. The proof suite shows that gap too.

Environment setup

You will need:

Python 3.11, 3.12, or 3.13 (openjiuwen on PyPI declares >=3.11,<3.14)
An OpenAI API key
openjiuwen==0.1.19, which matches the v0.1.19 release of Agent Core

git clone https://github.com/manishh/openjiuwen--custom-multi-step-ai-workflow-agent-core-proof-first
cd openjiuwen--custom-multi-step-ai-workflow-agent-core-proof-first
cp .env.example .env     # then set OPENAI_API_KEY=sk-... (no quotes, no spaces)
./prove --repeat 3

You can also export the key in your shell (export OPENAI_API_KEY=sk-...) instead of using .env. The .env file is git-ignored.

Run ./prove to start the proof (demo). It creates a virtual environment (.venv), installs the pinned dependencies from requirements.txt and constraints.txt, and runs every scenario. Each repeat runs in a new Python process and a new working directory, because Agent Core keeps its checkpointer (the store for saved workflow state) and workflow registry in process memory. A repeat makes about seven small model calls and takes about 25 seconds. ./prove prints PASS or FAIL for every check. It exits with 0 only if every check passed in every repeat.

To install the SDK into your own project, run pip install openjiuwen==0.1.19. The repo README covers the full prerequisites.

The imports this tutorial needs come from three modules. Two of them are shown below:

from openjiuwen.core.foundation.llm import ModelClientConfig, ModelRequestConfig
from openjiuwen.core.workflow import (
    End,
    LLMCompConfig,
    LLMComponent,
    Start,
    Workflow,
    WorkflowCard,
    WorkflowComponent,
)

WorkflowAgent comes from openjiuwen.core.application.workflow_agent.

Two side effects appear on the first run. The SDK writes INFO logs to stdout, and it creates a logs/ folder in your working directory. The proof script gives each repeat its own directory and sends stdout to product.log.

WorkflowCard and workflow initialization

A WorkflowCard holds the workflow's identity and its input contract. Its input_params field takes a JSON schema. Agent Core checks every incoming input dict against that schema before any node runs.

The tutorial schema requires a query string and allows an optional integer priority_hint. The priority_hint field exists only so the proof has a typed field to break (for simulating an error).

TICKET_INPUT_SCHEMA = {
    "type": "object",
    "properties": {
        "query": {"type": "string", "description": "The customer's ticket text"},
        "priority_hint": {"type": "integer", "description": "Optional 1-5 hint from the help desk"},
    },
    "required": ["query"],
}
...
def _card(workflow_id: str, description: str) -> WorkflowCard:
    return WorkflowCard(
        id=workflow_id,
        name=workflow_id,
        version="1.0",
        description=description,
        input_params=TICKET_INPUT_SCHEMA,
    )

Pass the card with the card= keyword:

flow = Workflow(card=_card("support_ticket", "Triage a ticket, get approval, draft a reply"))

Note: The official docs show Workflow(workflow_config=WorkflowConfig(card=...)). In 0.1.19, that form runs without an error but drops the card. The workflow then gets a random ID and no input schema. Use card= until the docs catch up.

Give each graph variant its own id as well. WorkflowAgent.add_workflows registers workflows in a process-wide registry, keyed by ID and version. If you register a different graph under an existing key, the SDK logs an ERROR and the agent keeps running the first graph.

The input contract holds up in the proof. When priority_hint is "urgent", the agent reports error 100016 and no node starts. That means no model call happens:

"error_code": 100016,
"message": "workflow execute input is invalid, ... reason='input validation failed against schema: [189001] validate data with schema failed, error='1 validation error for DynamicModel\npriority_hint\n  Input should be a valid integer, unable to parse string as an integer [type=int_parsing, input_value='urgent', input_type=str]...

The schema names the required field query to match the key WorkflowAgent uses when it passes the user's message to the workflow.

Start and End node configuration

In the 0.1.19 SDK, typed inputs live on WorkflowCard.input_params. Start() accepts no arguments. Passing it a config dict raises TypeError. To connect the Start node to the workflow input, use inputs_schema.

flow.set_start_comp("start", Start(), inputs_schema={"query": "${query}"})

The ${...} strings are references. ${query} reads the workflow input. ${start.query} reads the query output of the node with ID start.

The End node has two return modes, and each one produces a different result shape. If you give End a responseTemplate, it runs in text mode. It renders a string and returns {"response": "..."}.

flow.set_end_comp(
    "end",
    End({"responseTemplate": "[{{category}}/{{urgency}}] escalated={{approved}} :: {{reply}}"}),  # text mode
    inputs_schema={"category": "${triage.category}", "urgency": "${triage.urgency}",
                   "approved": "${approve.approved}", "reply": "${reply.text}"},
)

If you leave out the template, End() runs in variable mode. It returns the values as a dict under output. In the proof, the variable-mode graph returned {"output": {"category": "bug", "urgency": 5}}.

Use text mode when a person reads the result. Use variable mode when code consumes it. A downstream service can read output["urgency"] directly and never has to parse a sentence.

LLM component node setup

An LLMComponent takes one LLMCompConfig. The config bundles three things: the model connection, the prompt, and the output contract.

ModelClientConfig holds the connection to the provider. ModelRequestConfig picks the model.

def model_client_config() -> ModelClientConfig:
    """How to reach the model provider. The key is read from the environment."""
    return ModelClientConfig(
        client_provider="OpenAI",                     # Agent Core's OpenAI-compatible client
        api_key=os.environ["OPENAI_API_KEY"],         # never hard-code or write to disk
        api_base=os.environ.get("OPENAI_BASE_URL", "https://api.openai.com/v1"),
        verify_ssl=True,                              # keep TLS verification on
        timeout=60,                                   # seconds per model call
    )


def model_request_config() -> ModelRequestConfig:
    """Which model to call (and, optionally, sampling settings)."""
    return ModelRequestConfig(model=os.environ.get("MODEL_NAME", DEFAULT_MODEL))

The proof used gpt-5.4-mini (DEFAULT_MODEL) through the OpenAI provider, with TLS verification on. To use a different model or endpoint, set MODEL_NAME or OPENAI_BASE_URL.

The triage node combines the remaining fields. It defines the output contract and the prompts, then passes everything to LLMCompConfig.

TRIAGE_OUTPUT_CONFIG = {
    "category": {"type": "string", "description": "billing | bug | other", "required": True},
    "urgency": {"type": "integer", "description": "1 (low) to 5 (critical)", "required": True},
}

TRIAGE_SYSTEM_PROMPT = "You triage customer support tickets."
TRIAGE_USER_PROMPT = "Ticket: {{query}}"  # {{query}} is filled from the node's inputs
...

def triage_llm(user_prompt: str = TRIAGE_USER_PROMPT, response_type: str = "json",
               system_prompt: str = TRIAGE_SYSTEM_PROMPT) -> LLMComponent:
    """The triage node: classify a ticket into a typed {category, urgency} record."""
    if response_type == "json":
        output_config = TRIAGE_OUTPUT_CONFIG
    else:
        # text/markdown mode allows exactly one output field: the raw reply string.
        output_config = {"triage_text": {"type": "string", "description": "raw reply", "required": True}}
    return LLMComponent(LLMCompConfig(
        model_client_config=model_client_config(),
        model_config=model_request_config(),
        template_content=[                             # chat template, rendered per run
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt},
        ],
        response_format={"type": response_type},      # "json" | "text" | "markdown"
        output_config=output_config,                   # the node's output contract
    ))

Three fields shape what the node sends and returns:

template_content is the chat template. Placeholders such as {{query}} are filled from the node's inputs. The graph supplies those inputs through inputs_schema={"query": "${start.query}"}.
response_format takes {"type": "json"}, {"type": "text"} or {"type": "markdown"}.
output_config declares the fields this node promises to downstream nodes. In JSON mode, Agent Core adds the schema to the prompt. It then parses the reply and validates it against the schema. Text and markdown modes accept exactly one output field, which holds the raw reply.

The proof tests output_config with a prompt that contradicts the schema. The schema says urgency is an integer. The prompt tells the model to report urgency as a word. This simulates someone changing the "house style" of the prompt:

"user": "Ticket: {{query}}\n\nReport urgency as one of the words low, medium or high. Never use a number for urgency."

The drift is deliberate, and gpt-5.4-mini followed the prompt in every repeat. The proof then runs the same prompt twice.

Text mode. The run completes, and "triage_text": "high" reaches End as untyped text. The exact wording varies from run to run (for example high or Urgency: high), because text mode puts no constraint on the shape of the reply. In every case, a word reaches End where the schema expects an integer, and nothing checks it.

If the model happens to answer with an integer anyway, the proof re-runs scenario C, up to three attempts. It records each retry in report.json. The checks themselves never loosen.

JSON mode. Switch triage to JSON mode with urgency: integer, and the run stops at the triage node:

"error_code": 101004,
"message": "component llm_execution process error, reason: component llm_config is invalid, reason: expected type integer but got <class 'str'>"

The node list for that run is ["start", "triage"]. End never starts, so the workflow returns no final output for a consumer to misread.

How to connect the graph and run the WorkflowAgent

The full graph adds a human approval step between triage and reply.

flowchart TD
    IN([Agent input: query, priority_hint]) -->|checked against WorkflowCard.input_params| start[Start]
    start -->|start.query| triage[LLMComponent triage, JSON mode, output_config]
    triage -->|triage.category, triage.urgency| approve[HumanApproval, session.interact]
    approve -->|approve.approved| reply[LLMComponent reply, text mode]
    reply -->|reply.text| endn[End, text mode]
    endn --> OUT([workflow_final])

The approval node

The approval node is a six-line custom WorkflowComponent. When it calls session.interact(), the run pauses and Agent Core saves a checkpoint (a snapshot of the workflow's state).

class HumanApproval(WorkflowComponent):
    ...
    async def invoke(self, inputs, session, context):
        answer = await session.interact(
            f"Escalate this {inputs.get('category')} ticket (urgency {inputs.get('urgency')})? yes/no"
        )
        return {"approved": str(answer).strip().lower() == "yes"}

Registering nodes and edges

flow.add_workflow_comp("triage", triage_llm(), inputs_schema={"query": "${start.query}"})
flow.add_workflow_comp("approve", HumanApproval(),
                       inputs_schema={"category": "${triage.category}", "urgency": "${triage.urgency}"})
flow.add_workflow_comp("reply", reply_llm(),
                       inputs_schema={"category": "${triage.category}", "approved": "${approve.approved}"})
...
for src, dst in [("start", "triage"), ("triage", "approve"), ("approve", "reply"), ("reply", "end")]:
    flow.add_connection(src, dst)

The edge check in version 0.1.19 is shallow. add_connection only confirms that the node IDs are non-empty strings. Compilation only confirms that both endpoints are registered nodes. Neither step looks at the ${...} references.

Scenario D in the proof shows the effect. It uses the smaller start → triage → end graph and changes one reference in End to ${triage.urgncy}. The run returns COMPLETED with {"output": {"category": "bug"}}. The urgency field disappears and no error appears. End drops any value that resolves to None. To cover your own references, add a unit test that asserts on the keys of the final output.

Running the WorkflowAgent

Wrap the workflow in a WorkflowAgent and stream it. The proof suite calls await Runner.start() before the agent calls and await Runner.stop() after them. The SDK's own tests use the same pattern.

agent = WorkflowAgent(WorkflowAgentConfig(
    id=agent_id,
    version="1.0",
    description="Support-ticket agent driven by an Agent Core workflow",
    workflows=[],
))
agent.add_workflows([workflow])

agent.stream() yields four kinds of chunks:

tracer_workflow chunks, one for each node.
tracer_agent chunks, which carry errors.
An __interaction__ chunk, when a node pauses.
A workflow_final chunk, which holds the result.

A failed run doesn't raise an exception from stream(), so read the error from the tracer chunks.

Turn 1: the run pauses

In turn 1, the nodes ["start", "triage", "approve"] started, and then the run paused. The output shows the triage result and the question the approval node asked:

"triage": {
  "category": "billing",
  "urgency": 4
}
...
"id": "approve",
"value": "Escalate this billing ticket (urgency 4)? yes/no"

In the proof, the harness answers this question itself. It resumes the run with a scripted yes (resume.update("approve", "yes") in harness/run_proof.py), so ./prove never waits for you.

Turn 2: the run resumes

To resume, send an InteractiveInput on the same conversation_id. In run_example.py, answer is whatever you type at the prompt:

resume = InteractiveInput()
resume.update(chunk.payload.id, answer)          # resume that node from its checkpoint
async for chunk2 in agent.stream({"query": resume, **conv}):

Turn 2 started ["approve", "reply", "end"]. It skipped start and triage, and the final response reused turn 1's values:

"response": "[billing/4] escalated=True :: Your billing ticket has been escalated to a specialist, who will review it and follow up with you as soon as possible."

What the checkpoint covers

The checkpoint covers runs that pause for input. The checkpointer lives in memory and clears when a run completes. It handles pauses inside a live process and offers no crash recovery. If you send ordinary input to a session that still holds saved state, the SDK raises an error or clears the state.

Try it yourself

To answer the approval yourself, run the interactive example after ./prove has created .venv. The script doesn't load .env on its own, so load the key into your shell first:

set -a; source .env; set +a
.venv/bin/python run_example.py "I was charged twice this month"

The script prints each node's output and asks Escalate this billing ticket (urgency N)? yes/no. It then resumes the workflow with your answer and prints the final response. The SDK's INFO log lines appear mixed in with the output.

Error reasons and error codes

The proof suite sets one option you should think about before copying it into production. UserConfig.set_is_sensitive(False) makes the SDK show error reasons. By default, the SDK hides them and reports only "invoke llm failed". The proof suite needs the reasons to tell a model API failure (101003) from a contract failure (101004).

These are the error codes you'll see in the proof transcripts:

Code	Meaning
`100016`	The workflow input failed the `WorkflowCard.input_params` schema
`101004`	An LLM node's JSON reply failed its `output_config`, for example `expected type integer`
`101003`	The model API call failed (network, auth or rate limit), which is unrelated to your contracts

Wrapping up

You now have a WorkflowAgent that rejects malformed input with error 100016 before it spends a model call. It stops a drifted LLM reply at the node with error 101004, before the reply reaches downstream consumers. It also pauses for human approval and resumes without re-running triage. Agent Core does not check node references, so you have to cover them yourself. Add a test that asserts on the keys of the final output.

From here, try registering several workflows on one WorkflowAgent or branching the graph so independent nodes run in parallel. Clone Agent Core on GitHub to explore the SDK, and run this tutorial's example from its companion repo. If your team prefers a drag-and-drop canvas, see Agent Studio on openjiuwen.com.

Frequently asked questions

What does openJiuwen Agent Core do in a multi-step AI workflow?

openJiuwen Agent Core is an open-source Python SDK that lets you define agent workflows as graphs of nodes. You declare the workflow's input schema on a WorkflowCard and each LLM node's output schema in output_config. As of v0.1.19, the SDK checks the workflow input and the output of JSON-mode LLM nodes at run time. It doesn't check the references between nodes, so cover those with a test.

Why does Agent Core reject input before any node runs?

Agent Core validates every incoming input dict against the JSON schema you provide in WorkflowCard.input_params before the first node executes. If the input fails validation, the agent returns error 100016 and no model call is made, which means you don't spend tokens on a request that was already broken.

What is the difference between text mode and variable mode in the End node?

When you pass a responseTemplate to End, it renders a formatted string and returns {"response": "..."}. Without a template, End returns the raw values as a dict under output. Text mode suits human-readable results and variable mode suits downstream code that reads specific keys directly.

How does Agent Core catch a drifted LLM output?

When a node runs in JSON mode with an output_config schema, Agent Core adds that schema to the prompt, parses the model's reply, and validates the parsed result against the declared types. If the model returns a string where an integer is required, the run stops at that node with error 101004 and the downstream nodes never execute.

What happens when a node reference contains a typo?

In version 0.1.19, Agent Core's edge check is shallow. A typo such as ${triage.urgncy} resolves to None at run time, and End silently drops any value that resolves to None. The run still returns COMPLETED but the missing field is absent from the output. A unit test that asserts on the final output keys is the reliable way to catch this.

How does the human approval step pause and resume a workflow?

A custom WorkflowComponent calls await session.interact(...), which pauses the run and saves a checkpoint in memory. To resume, you send an InteractiveInput carrying the same conversation_id back to agent.stream(). The resumed run picks up from the approval node and skips the nodes that have already completed.

Why doesn't ./prove ask me to approve the escalation?

The proof answers the approval for you. To prove the same behavior on every repeat, harness/run_proof.py resumes the paused run with a scripted yes. The run still pauses at the approve node, and reply and end don't start until the answer arrives. To type the answer yourself, run run_example.py after loading your key into the shell.

Does Agent Core's checkpoint survive a process restart?

The checkpointer lives in memory and clears when a run completes or the process exits. It covers pauses inside a live process but offers no crash recovery across restarts.

Which Python versions does openJiuwen Agent Core support?

openjiuwen on PyPI declares >=3.11,<3.14, so Agent Core supports Python 3.11, 3.12, and 3.13.

Why should UserConfig.set_is_sensitive(False) stay out of production?

By default, Agent Core hides detailed error reasons and returns a generic message such as "invoke llm failed". Setting is_sensitive to False exposes the full reason string, which is useful in a proof suite but may leak internal implementation details in a production environment.

How to keep an AI workflow predictable using WorkSwarm's SwarmFlow

Manish — Tue, 27 Oct 2026 16:35:54 +0000

Have you ever watched one early mistake spread through a whole team of AI agents? SwarmFlow is built to stop that. It is WorkSwarm's scripting layer for Cluster mode (‘team’ mode). You fix the order of steps in a Python file, hold the run at a human checkpoint, and cap token spend with two separate limits. Every run then follows the same steps in the same order, and leaves a record you can audit.

Here is the problem it solves. When a team of LLM agents works through a multi-step job, a bad output early in the chain gets passed to every step after it. Say a code-generation agent writes the wrong function. A documentation agent then documents that wrong function. You pay for both calls before anyone looks at the result. There is a second problem. The Leader agent (the one that coordinates the worker agents) decides the order of work at run time. Two runs of the same request can take different paths.

WorkSwarm (formerly JiuwenSwarm) tackles this with SwarmFlow. A Python file fixes the order of stages. A human() call holds the run until a person answers. Two token ceilings stop a run and send the Leader a diagnostic message it can act on.

In this tutorial, you will:

Build a four-stage pipeline: spec, codegen, human review, and docs.
Send an edit request at the review checkpoint. Then watch SwarmFlow re-run only the changed stage and replay the unchanged one from its journal (a saved record of each finished step).
Trip both token ceilings on purpose.

Every output shown here comes from a proof test that ran all four scenarios three times against OpenAI gpt-5.5.

Prerequisites and setup

You will need:

Linux or macOS with Python 3.11 to 3.13 (the proof ran Python 3.12 on Ubuntu)
An OpenAI API key with access to gpt-5.5
Ports 19000, 19001 and 18092 free (WorkSwarm's web channel, gateway and agent server)

WorkSwarm is published on PyPI as workswarm, because the jiuwenswarm PyPI name stopped at 0.2.3. The workswarm package installs the jiuwenswarm-* commands and pulls in openJiuwen Agent Core (openjiuwen), which contains the SwarmFlow engine. Pin both versions in requirements.txt:

workswarm==0.2.7
openjiuwen==0.1.19

Clone the repo and run the proof:

git clone https://github.com/manishh/openjiuwen--swarmflow-predictable-workflows-proof-first
cd openjiuwen--swarmflow-predictable-workflows-proof-first
cp .env.example .env        # put your key in OPENAI_API_KEY (no quotes, no spaces), or export it instead
./prove --repeat 3

The prove script handles the entire setup for you. It creates a virtual environment (.venv) and installs the pinned packages. For each repeat, it then creates a fresh WorkSwarm home in a temp directory with jiuwenswarm-init, so your local ~/.jiuwenswarm is never touched.

Next, it starts jiuwenswarm-start app. Your API key goes only into the server's process environment. The script then drives the Leader over the same WebSocket API the web UI uses (ws://127.0.0.1:19000/ws). It uses WebSocket because the headless process SDK has no Cluster mode, and SwarmFlow runs only in Cluster mode.

./prove also prints a masked form of the key it found: Using OPENAI_API_KEY= followed by the key's first 8 and last 6 characters. This allows you to check that it picked up the right key.

Each repeat takes about 7 minutes (~395 seconds in the verified runs) and uses roughly 1.4M tokens, mostly cached prompt reads. The human-checkpoint scenario uses about 750k of those. Each of the three budget scenarios uses between 110k and 290k.

The diagram below shows one repeat (a ./prove script run). The proof harness plays two parts. It acts as the chat user who asks the Leader to run each script. It also acts as the reviewer who answers the checkpoint. This way, a repeat runs from start to finish without waiting for you (the User).

sequenceDiagram
    autonumber
    actor Dev as User
    participant H as Proof harness
    participant L as Team Leader
    participant SF as SwarmFlow
    participant E as Evidence on disk

    Dev->>H: ./prove
    H->>L: Start server, ask it to run slugify_pipeline.py
    L->>SF: Launch the pipeline

    rect rgba(76,175,80,0.10)
    Note over H,E: Scenario A: human checkpoint and partial re-run
    SF->>E: Run spec and codegen, save both to the journal
    SF->>H: Ask the reviewer to approve
    H->>SF: Reply "add max_len, cap at 40"
    SF->>L: Send edit points
    L->>SF: Edit the script, relaunch with the same run_id
    E->>SF: spec is replayed from the journal (CACHE_HIT)
    SF->>E: codegen runs again (CACHE_MISS)
    SF->>H: Ask the reviewer again
    H->>SF: Reply "approve"
    SF->>H: docs runs, status is completed
    end

    rect rgba(156,39,176,0.10)
    Note over H,E: Scenarios B, C, D: token ceilings
    H->>SF: Run the budget demo with no limit, a per-run limit, and a session limit
    SF->>H: Status is completed, failed, or stopped
    end

    H->>E: Check the journal, logs, and trace
    H->>Dev: Report with pass or fail for each check

SwarmFlow is off by default. The test switches it on in the home's config/config.yaml with these lines from harness/setup_home.py, and you'd set the same keys in your own install:

    cfg["preferred_language"] = "en"              # init defaults to zh when it cannot prompt
    cfg["setup_guide"]["enabled"] = False         # first-run onboarding distracts the Leader
    team = cfg["modes"]["team"]["jiuwen_team"]
    team["enable_swarmflow"] = True               # SwarmFlow is off by default
    team["swarmflow_budget"] = 3_000_000          # default team ceiling (chat.send may override per session)

The model settings go in the home's config/.env, with no API_KEY line on disk:

MODEL_PROVIDER=OpenAI
ENDPOINT_PROFILE=openai
API_BASE=https://api.openai.com/v1
MODEL_NAME=gpt-5.5

For platform installers and the web UI, see the WorkSwarm README and the English docs.

How to define workflow stages in Python

A SwarmFlow pipeline is a Python file. It imports agent, human, and phase from the swarmflow module and defines an async def run(args). The file declares the stages, and the Leader runs it with its swarmflow tool. The order of the await calls sets the order of the stages, so every run follows the same sequence.

The pipeline in pipeline/slugify_pipeline.py starts with a META block. SwarmFlow reads META before running anything, so it must be a pure literal (plain values only, with no code to evaluate). Each phase title must match a phase() call further down the file.

META = {
    "name": "slugify-pipeline",
    "description": "Spec -> code generation -> human review -> docs",
    "phases": [
        {"title": "Spec"},
        {"title": "Codegen"},
        {"title": "Review"},
        {"title": "Docs"},
    ],
}

# The one line a reviewer's edit will change. Every agent() call is journaled under
# its position in the script plus a hash of its prompt, so changing this constant
# changes only the codegen call's hash (and anything downstream that embeds its output).
CODEGEN_RULES = "Use only the standard library."

The comment above CODEGEN_RULES describes the journal, which the next section builds on. Each agent() call is recorded under its position in the script ([["call", 0]], [["call", 1]], and so on) plus a signature hashed from its prompt. The Leader reads the script before editing it. In the proof runs, the comment also guided where the Leader made its edit.

The first two stages are plain agent() calls. The label gives the call a name in the journal and the run tree. The phase argument attaches the call to a phase in META.

async def run(args):
    # swarmflow() passes args as a string; structured input is JSON we parse ourselves.
    a = json.loads(args)

    phase("Spec")
    spec = await agent(
        f"In 3 short bullet points, specify a Python function slugify(title) for: {a['feature']}. "
        "Return only the bullets.",
        label="spec", phase="Spec",
    )

    phase("Codegen")
    code = await agent(
        "Write ONLY the Python source (no prose, no markdown fences) implementing this spec. "
        f"{CODEGEN_RULES}\n\nSpec:\n{spec}",
        label="codegen", phase="Codegen",
    )

To launch the pipeline, the test sends the Leader a message that names the tool. Naming the tool makes the Leader call it directly.

Run swarmflow(script_path='/tmp/swarmflow-proof-1-638pm8wc/scripts/slugify_pipeline.py', args='{"feature": "turn blog post titles into URL slugs"}').

The Leader called swarmflow with those arguments, and the tool returned run_id=wf_e0b31fee9773. This run_id names the run's journal file. You will need it again when the run resumes.

How to add a human-in-the-loop checkpoint

The review stage is a single await human(...) call. The prompt includes the generated code, so the reviewer can see exactly what they are approving.

    phase("Review")
    review = await human(
        "Review the generated code. Reply 'approve' to continue, or describe the edit you want.\n\n"
        + str(code)
    )

    phase("Docs")
    docs = await agent(
        f"Write a two-sentence docstring summary for this code:\n{code}",
        label="docs", phase="Docs",
    )
    return {"spec": spec, "code": code, "review": review, "docs": docs}

When the run reaches human(), it pauses and the question arrives as a chat.ask_user_question event with source=swarmflow_human. The Docs phase stays planned, which means it has not started. The run tree below is built from the run's workflow.updated events.

wf_e0b31fee9773  slugify-pipeline  [running]  run tokens=49339
  budget/session : spent=179435 total=3000000 exhausted=False
  budget/workflow: spent=49339 total=None exhausted=False
  ├─ phase Spec [completed]
  │   └─ spec [completed] tokens=24643
  ├─ phase Codegen [completed]
  │   └─ codegen [completed] tokens=24696
  ├─ phase Review [running]
  │   └─ unnamed human [waiting_for_human] tokens=None
  ├─ phase Docs [planned]

The generated slugify(title: str) had no length limit. When executed on a long title, it returned an 80-character slug:

{"slug": "how-to-keep-an-ai-workflow-predictable-using-jiuwenswarm-s-swarmflow-checkpoints", "len": 80, "params": ["title"]}

In the demo, the reviewer is the proof script. ./prove must give the same answer on every repeat. So harness/run_proof.py waits for the checkpoint question and sends two scripted replies over the WebSocket. The first is the edit request shown below. The second, sent later, is approve. This means ./prove never actually stops to ask you anything.

When you run your own pipeline in Cluster mode, a human answers instead. As of WorkSwarm 0.2.7, the bundled web UI shows a swarmflow_human question as a question card in the chat. It sends your typed answer back in the same chat.swarmflow_reply message. The proof does not exercise this web flow, it uses the 'headless' mode instead.

The proof script answered with this edit request. It was sent as chat.swarmflow_reply with the run_id and correlation_id:

Edit the codegen step: slugify must take a max_len parameter (default 40) and truncate the slug to that length. Then re-run.

SwarmFlow makes an extra model call to sort every human reply into one of two intents: continue or edit_rerun. There is no built-in "reject". To reject an output, have your script inspect the review string that human() returns. You can then raise an error or branch on it.

This reply came back as edit_rerun, and three things followed:

The run went to paused.
The journal gained a pause record with "pause_reason": "early_return".
docs had no journal record, so it had not run.

The engine then handed the Leader the edit points. Interestingly, it logged this handoff at ERROR level, as a failed tool task:

... The user asked to edit the script and re-run.\nEdit points: Modify the code generation step so slugify accepts a max_len parameter with default value 40, and truncates the generated slug to at most max_len characters. Then re-run the workflow.\n...Edit the on-disk script accordingly (do not change META.name). Edit minimally and reuse the existing script content as much as possible, refining it rather than rewriting.

With no further prompt, the gpt-5.5 Leader made one edit_file call to change the script. The test never touches the script itself. Here is the change:

-CODEGEN_RULES = "Use only the standard library."
+CODEGEN_RULES = "Use only the standard library. Implement slugify(title, max_len=40) and truncate the generated slug to at most max_len characters."

The Leader then relaunched the run with the same run_id:

swarmflow({"script_path": "/tmp/swarmflow-proof-1-638pm8wc/scripts/slugify_pipeline.py", "resume_id": "wf_e0b31fee9773", "args": "{\"feature\": \"turn blog post titles into URL slugs\"}"})

On resume, SwarmFlow reloaded the journal. It compared each call's signature (a hash of its prompt) with the stored one. The spec prompt had not changed, so SwarmFlow replayed the saved result and made no model call. The codegen prompt includes CODEGEN_RULES, so its signature changed and it ran live:

... [wf] journal loaded: prior_records=4 ...
2026-10-09 15:14:17 | team | sess_a54f49d1f | WARNING | [wf] agent 'spec' key=[["call", 0]] CACHE_HIT sig=5b0917e72813
2026-10-09 15:14:17 | team | sess_a54f49d1f | WARNING | [wf] agent 'codegen' key=[["call", 1]] CACHE_MISS sig=dc49daeca408 (live run)

The replayed spec text was byte-identical to the first pass. The new code returned a 40-character slug:

{"slug": "how-to-keep-an-ai-workflow-predictable-u", "len": 40, "params": ["title", "max_len"]}

The human() call is journaled as well. Its prompt now contained the new code, so its signature changed and the reviewer was asked again. The proof script's second reply, approve, was classified as continue. Then docs ran and the journal was sealed.

Each line below shows a call's position, label, signature, and token count. The call at position 2 is the human() step, so it has no label. The last line seals the run as completed.

[["call", 1]] label=codegen sig=dc49daeca408 tokens=24950
[["call", 2]] label=None sig=83f5ee8acb26 tokens=24379
[["call", 3]] label=docs sig=f0c7bc447b8f tokens=24687
{"key": "[[\"__run__\", 9223372036854775807, \"seal\", \"wf_e0b31fee9773\"]]", "type": "seal", "terminal_status": "completed"}

The diagram below clearly depitcs the states a typical run can move through.

stateDiagram-v2
    [*] --> running: Leader calls swarmflow(script_path, args)
    running --> waiting_for_human: human() reached, later phases stay planned
    waiting_for_human --> running: reply classified continue
    waiting_for_human --> paused: reply classified edit_rerun (pause record early_return)
    paused --> running: Leader edits script, relaunches with resume_id (matching signatures replay as CACHE_HIT)
    running --> completed: last stage done (seal record)
    running --> failed: per-run workflow_token_limit exhausted (pause record)
    running --> stopped: team swarmflow_budget exhausted (seal record)
    completed --> [*]
    stopped --> [*]

The review step also costs tokens. The unnamed human entry used 24,379 tokens, because classifying and formatting the reply each run model calls. This cost repeats every time the checkpoint asks again.

How to set per-run and team token ceilings

SwarmFlow enforces two ceilings:

Per-run ceiling: META["workflow_token_limit"] in the script.
Team (session) ceiling: swarmflow_budget, sent with each chat.send the same way the web UI sends it.

The team ledger counts the Leader's own model calls as well as the worker agents' calls. It gates the Leader too.

The demo script pipeline/budget_demo.py chains three agents and sets the per-run limit to 30,000:

META = {
    "name": "budget-demo",
    "description": "outline -> draft -> headline",
    "phases": [{"title": "Outline"}, {"title": "Draft"}, {"title": "Headline"}],
    # Per-run token ceiling for this run only (the team-wide ceiling is set separately).
    "workflow_token_limit": 30000,
}

This limit looks generous for three one-sentence prompts. But each SwarmFlow worker is a full agent with a large system prompt. One agent() call cost about 25k tokens with gpt-5.5. The run crossed 30,000 during draft, so headline never started.

For the budget scenarios, the Leader prompt adds "Do not edit the script and do not relaunch it yourself." The diagnostic tells the Leader to redesign the script and relaunch. The demo needs to show the stopped run as it is, so the Leader must not do that.

wf_826df91268e1  budget-demo  [failed]  run tokens=49258
  budget/session : spent=222723 total=3000000 exhausted=False
  budget/workflow: spent=49258 total=30000 exhausted=True
  budget_exhausted_scope=workflow
...
  ├─ phase Headline [failed]
  │   └─ headline [failed] tokens=None

The diagnostic reaches the Leader as a follow-up message. The test reads it from the home's .jiuwenswarm/.trace/ store. It names the layer that tripped, the heaviest phases, and the recovery path:

[Swarmflow budget exhausted] run_id=wf_826df91268e1
workflow token budget exhausted: 49258/30000
Triggered layer: workflow (per-run) (spent=49258/30000). Heaviest phases: Draft(24653), Outline(24605). This ceiling is the run's per-run token budget (independent of the session total), set by the user and must NOT be changed; you must redesign the workflow to consume fewer tokens (simplify / fewer agents / cheaper model) and relaunch. Relaunch by passing resume_id=<the run_id above> together with the edited script_path: ...

The run ends as failed with a workflow_budget_exhausted pause record, so it can still be resumed. The guidance warns that replayed calls are "re-billed from their stored tokens". In other words, a resume counts them against the same ceiling again. The proof did not test a resume after exhaustion.

For the team ceiling, the test ran budget_demo_unlimited.py, which has no per-run limit. It set swarmflow_budget to the Leader's pre-launch spend plus one worker call. Both values were measured in an unconstrained control run. In repeat 1, that came to 43,218 + 24,643 = 67,861. A fixed small number would have stopped the Leader before it launched anything.

wf_df24a8c15beb  budget-demo-unlimited  [stopped]  run tokens=24607
  budget/session : spent=111143 total=67861 exhausted=True
  budget/workflow: spent=24607 total=None exhausted=False
  budget_exhausted_scope=session
...

Triggered layer: session (shared) (spent=111143/67861). Heaviest phases: Outline(24607). This is the team's shared (session-wide) token ceiling and is currently exhausted. Until the ceiling is raised, relaunching or starting a new workflow will hit the same gate immediately — raise the ceiling and open a new session before retrying swarmflow.

This time, the journal got a seal record with "terminal_status": "stopped". The Leader is gated by the same ledger, so it sent no final message.

Use the per-run limit to cap a single script you expect to iterate on. Set swarmflow_budget as the hard spend cap for the whole team session.

How to monitor a run from its workflow.updated events

WorkSwarm also has an optional terminal UI, jiuwenswarm-tui. This tutorial monitored runs from the event stream that the UIs read. That approach works wherever WorkSwarm runs.

The run trees shown earlier come from the workflow.updated events on the web channel WebSocket. Each event carries:

The run status.
The token_count.
Both budget ledgers (budget and workflow_budget).
The phases, with per-agent status and tokens.

These events are deltas. Each one carries only the phases that changed. A monitor therefore has to merge them into one snapshot, as merged() does in harness/run_proof.py:

    top, phases = {}, {}
    for f in frames:
        if not (is_sess(f, sid) and etype(f) == "workflow.updated" and wf_of(f).get("id") == run_id):
            continue
        wf = wf_of(f)
        top.update({k: v for k, v in wf.items() if k != "phases"})
        for ph in wf.get("phases") or []:
            cur = phases.setdefault(ph.get("name"), {"agents": {}})
            cur.update({k: v for k, v in ph.items() if k != "agents"})
            for a in ph.get("agents") or []:
                cur["agents"].setdefault(a.get("id"), {}).update(a)

Across the four scenarios, the merged trees showed these phase and agent statuses: planned, running, waiting_for_human, paused, completed, failed and stopped.

The swarmflow tool also supports run control. You pass resume_id together with an action, and the tool routes pause, resume and stop requests to the background-task controller (tool_swarmflow.py::_control_run). The proof does not exercise this, so test it before you rely on it.

To see the events for yourself, run ./prove --repeat 1 and open proof-out/repeat-1/transcript.md. Confirm that the unnamed human [waiting_for_human] entry sits under phase Review, while phase Docs is still planned.

What to build next with SwarmFlow

You now have a working SwarmFlow setup with three concrete capabilities:

Fixed stage order declared in a Python file, so every run follows the same path regardless of what the Leader decides at runtime.
A human checkpoint that pauses the run, accepts an edit request, and resumes under the same run_id with unchanged stages replayed from the journal.
Two token ceilings that each stop a run with layer-specific advice the Leader can act on immediately.

Swap the slugify stages for real stages in your pipeline, put human() right before the most expensive (or crucial) step, and size workflow_token_limit from the per-agent token counts in your own run tree.

Star the WorkSwarm repo, follow the openJiuwen-ai organization for releases, and check the English docs for the full configuration reference. Contributions and issue reports are welcome.

Frequently asked questions

What is SwarmFlow in WorkSwarm?

SwarmFlow is WorkSwarm's scripting layer for Cluster mode (team mode) that fixes the execution order of a multi-agent pipeline in a Python file. Each stage is an await agent() or await human() call, and the Leader runs the file with its swarmflow tool, so the stage sequence is identical on every run regardless of runtime Leader decisions.

How does the SwarmFlow journal prevent redundant LLM calls after an edit?

Each agent() call is recorded under its position in the script and a signature hashed from its prompt. On resume, SwarmFlow compares each call's current signature against the stored one. Calls whose prompts are unchanged replay from the journal as CACHE_HIT without hitting the model, while calls whose prompts changed run live. In the proof run, editing CODEGEN_RULES produced a CACHE_HIT for spec and a CACHE_MISS for codegen, cutting the resume cost to a single live agent call.

What is the difference between the per-run token ceiling and the team token ceiling?

The per-run ceiling is META["workflow_token_limit"] in the pipeline script and caps token spend for that single run. The team ceiling is swarmflow_budget, set per chat.send, and counts every model call in the session including the Leader's own calls. Exhausting the per-run ceiling ends the run as failed with a resumable pause record. Exhausting the team ceiling ends the run as stopped and gates the Leader from launching anything new until the budget is raised.

Why doesn't ./prove ask me for input at the human checkpoint?

The proof script answers the human checkpoint for you. To prove the same behavior on every repeat, harness/run_proof.py waits for the chat.ask_user_question event and sends two scripted replies through chat.swarmflow_reply, the edit request and then approve. The checkpoint still blocks the run, and docs doesn't start until a reply arrives. In your own Cluster-mode session, the WorkSwarm web UI shows the question as a card in the chat, and you can type your reply there.

How does SwarmFlow classify a human reply as an edit request vs. an approval?

SwarmFlow runs an extra model call on every human() reply to classify it as either continue or edit_rerun. A reply of approve routes to continue and the pipeline resumes from the next stage. Any reply describing a change routes to edit_rerun, which pauses the run, writes a pause record with "pause_reason": "early_return", and passes structured edit points to the Leader so it can modify the script and relaunch with resume_id.

What happens to a run that hits the team token ceiling mid-stage?

The run ends as stopped with a seal record instead of a resumable pause record. The Leader, gated by the same ledger, sends no final message. The diagnostic names the session layer, the heaviest phase, and the instruction to raise the ceiling and open a new session before retrying swarmflow.

How do you prevent the Leader from automatically relaunching a budget-exhausted run?

Add the instruction "Do not edit the script and do not relaunch it yourself." to the Leader prompt before triggering the budget scenario. Without it, the diagnostic message tells the Leader to redesign and relaunch, and gpt-5.5 will act on that instruction immediately, which obscures the stopped-run state you may want to inspect.

Debugging a silently failing AI workflow with Agent Core full-link observability

Manish — Wed, 07 Oct 2026 14:01:56 +0000

The most difficult AI workflows to debug are the ones that don't crash. A multi-step agent pipeline can quietly return a plausible-looking answer that's actually wrong, without ever throwing an error. One broken step early in the chain compounds into a wrong final output, and standard monitoring rarely catches it: the run finishes, the logs show green, and the problem only surfaces hours later when a downstream team flags that something's off.

This tutorial walks through a concrete case using openJiuwen Agent Core v0.1.18. You'll instrument a three-node pipeline (fetch → analyze → summarize) with OTEL-compatible spans (structured traces that record what each step did, in a format most observability tools understand). Then you will read back their output (latency, token counts, and a per-step failure flag), and pinpoint the exact node where the chain went wrong. The complete companion code is available at the tutorial repository.

Why silent failures escape standard logs

A silent failure happens when a step in an agent pipeline returns something that looks like a normal result but isn't. No exception is raised, no log line turns red, and no HTTP error code shows up, even though the output is wrong. Standard Python logging and HTTP status codes are built to catch exceptions. Agent-specific failures often don't rise to that level, so they slip straight past.

Three failure modes show up most often in production pipelines:

Incorrect tool selection: The agent picks the wrong function for the job. That function still runs, still returns, and still fills the Observation (the agent's record of what a tool call returned) with data, just data that has nothing to do with the task. No exception fires, because nothing actually went wrong from the tool's point of view.
Hallucinated tool arguments: The agent calls the right tool, but with parameters it made up or left out. If the tool tolerates missing input rather than rejecting it, the call succeeds and returns something empty or nonsensical.
Context truncation: The prompt is too long for the model's context window (the amount of text it can consider at once), and the model API returns a normal HTTP 200 response with an empty list of completions. The token count comes back as zero, no error is raised, and the node exits as if nothing happened.

This matters more in a ReAct-style agent loop, which cycles through Thought, Action, and Observation, because each step can look successful on its own while the whole chain quietly drifts off course. A truncated completion still produces something that reads as a valid string, so the next Thought step picks it up and keeps going as if nothing were wrong. By the time the loop finishes, the final output is wrong, but there's still no exception and no red log line to explain why.

Sample workflow setup

You'll need Python 3.11, 3.12, or 3.13 (openjiuwen doesn't yet support 3.14). Clone the repository, change into the project directory, and install the dependencies:

git clone https://github.com/manishh/openjiuwen-debugging-an-ai-workflow-that-fails-silently-with-agent
cd openjiuwen-debugging-an-ai-workflow-that-fails-silently-with-agent
pip install -r requirements.txt

See the openJiuwen quickstart documentation for initial API key setup. The tutorial reads five environment variables: OPENJIUWEN_API_KEY, OPENJIUWEN_MODEL, OPENJIUWEN_API_BASE, OPENJIUWEN_PROVIDER, and OPENJIUWEN_SSL_VERIFY. If you leave out OPENJIUWEN_API_KEY, the code runs in offline simulation mode, which still emits and analyzes spans and only skips the live model call.

The tutorial's target pipeline is a sentiment-analysis workflow with three connected steps:

fetch_data retrieves and preprocesses the source text.
analyze_sentiment classifies its sentiment. This is the node that simulates a silent failure.
summarize generates a human-readable summary.

The Thought/Action/Observation loop above describes how a live ReActAgent reasons. The pipeline in src/main.py simulates the same failure pattern with three plain Python functions, so it runs deterministically offline without a live agent loop. The agent construction in examples/02_workflow_agent_setup.py looks like this:

card = AgentCard(
    id="sentiment-analysis-pipeline",
    name="sentiment-analysis-pipeline",
    description="Fetch -> analyse -> summarise sentiment pipeline",
)
config = ReActAgentConfig().configure_model_client(
    provider=PROVIDER,
    api_key=API_KEY,
    api_base=API_BASE,
    model_name=MODEL,
    verify_ssl=SSL_VERIFY,
)

# The agent below drives the three-node pipeline:
#   node_fetch_data -> node_analyze_sentiment -> node_summarize
agent = ReActAgent(card=card).configure(config)
print(f"Agent '{agent.card.name}' created.")

ReActAgent is openjiuwen's current general-purpose agent, built around the Thought/Action/Observation loop described earlier. It decides at runtime which tool to call next, rather than following a hardcoded step order.

The older WorkflowAgent, which ran a fixed, user-defined node sequence, is deprecated as of openjiuwen 0.1.18 in favor of this AgentCard + ReActAgentConfig pattern. This tutorial's pipeline order (fetch → analyze → summarize) is enforced by the plain Python function calls in src/main.py, not by the agent, so the fixed order holds regardless of which agent class builds it.

Full-link observability in Agent Core

Full-link observability in Agent Core means that every tool call, token count, and step latency is captured as a structured, OTEL-compatible span (a self-contained record of one unit of work). Chain those spans together and you get a complete execution trace that you can query programmatically or view in Agent Studio, with no third-party APM (Application Performance Monitoring) tool required.

As of v0.1.18, Agent Core emits this telemetry natively. To capture and inspect those traces locally, you attach exporters to a TracerProvider instance. The full setup is in examples/01_setup_observability.py:

from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor, ConsoleSpanExporter
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter

exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
provider.add_span_processor(SimpleSpanProcessor(exporter))

# Always get the tracer from your provider, not from opentelemetry.trace.get_tracer().
tracer = provider.get_tracer("agent-core.tutorial")

SimpleSpanProcessor exports each span the moment it finishes, so the console output and the trace report you'll build later stay in the order the spans actually ran. Get your tracer from your own provider instance (provider.get_tracer(...)) rather than from the global opentelemetry.trace module. If Agent Core has already initialized the global provider, calling trace.get_tracer() returns a tracer bound to the SDK's own exporter, and your spans go somewhere you can't see them.

With the tracer in hand, wrap every step in a custom span that records four diagnostic signals. The broken analyze_sentiment node in examples/03_custom_spans.py shows what a silently failing node looks like once it’s instrumented:

def node_analyze_sentiment(data: dict) -> dict:
    with tracer.start_as_current_span("tool.analyze_sentiment") as span:
        t0 = time.perf_counter()
        # Simulate: context window overflow caused the model to return
        # an empty choices list. Status code was 200 – no error logged.
        time.sleep(0.15)  # slightly elevated latency: the client waited for timeout
        result = {"sentiment": "", "token_count": 0, "confidence": 0.0}

        span.set_attribute("agent.tool.name", "analyze_sentiment")
        span.set_attribute("agent.tool.token_count", result["token_count"])
        span.set_attribute("agent.step.latency_ms", round((time.perf_counter() - t0) * 1000, 2))
        # This flag is the key signal in the trace: token_count==0 is a silent failure.
        span.set_attribute("agent.step.silent_failure", result["token_count"] == 0)
    return result

Run the script (python examples/03_custom_spans.py), and the span summary makes the failure obvious:

Span summary:
  tool.fetch_data                          tokens=  312  latency=   40.15 ms  silent_failure=False
  tool.analyze_sentiment                   tokens=    0  latency=  150.15 ms  silent_failure=True
  tool.summarize                           tokens=  144  latency=   50.13 ms  silent_failure=False

Look at the analyze_sentiment row: tokens=0 and silent_failure=True, even though the node returned normally with no exception raised. Your latency numbers will vary slightly between runs, since they come from a real time.sleep() call, but that row is the one to watch either way.

The last line of the code snippet is the one that matters most. Every node sets agent.step.silent_failure, not just the broken one, and it’s True whenever token_count == 0. That's what turns "did this succeed?" into something you can actually query. Together, the four attributes (agent.tool.name, agent.tool.token_count, agent.step.latency_ms, and agent.step.silent_failure) are the signals you'll read in the next section. Because every node writes all four, the trace report has a consistent shape to scan no matter which node broke.

How to read the trace output

The diagram below shows the full-link trace for a broken run. The root workflow.session span wraps all three tool spans, and tool.analyze_sentiment is highlighted because its agent.step.silent_failure attribute is True.

flowchart LR
    A["workflow.session<br/>root span"] --> B["tool.fetch_data<br/>tokens: 312<br/>latency: 40 ms<br/>silent_failure: False"]
    B --> C["tool.analyze_sentiment<br/>tokens: 0<br/>latency: 150 ms<br/>silent_failure: True"]
    C --> D["tool.summarize<br/>tokens: 144<br/>latency: 50 ms<br/>silent_failure: False"]
    style C fill:#ff6b6b,color:#fff

Once the workflow finishes, iterate over exporter.get_finished_spans() to read each span back; each one maps to a single node. The reporting loop in examples/04_read_trace_output.py prints the broken and fixed runs one after the other:

def _report(label: str, spans: list) -> None:
    print(f"\n{'='*72}\n{label}\n{'='*72}")
    print(f"{'Span':<35} {'tokens':>7} {'latency ms':>11} {'failure':>8}")
    print("-" * 72)
    for s in spans:
        a = s.attributes or {}
        flag = "◄ FAIL" if a.get("agent.step.silent_failure") else "OK"
        print(
            f"{s.name:<35} {str(a.get('agent.tool.token_count','-')):>7} "
            f"{str(a.get('agent.step.latency_ms','-')):>11} {flag:>8}"
        )

Running it prints both reports, so you can compare them line by line:

========================================================================
BROKEN RUN (inject_failure=True)
========================================================================
Span                                 tokens  latency ms  failure
------------------------------------------------------------------------
tool.fetch_data                         312        40.0       OK
tool.analyze_sentiment                    0       150.0   ◄ FAIL
tool.summarize                          144        50.0       OK
workflow.session                          -           -       OK

========================================================================
FIXED RUN  (inject_failure=False)
========================================================================
Span                                 tokens  latency ms  failure
------------------------------------------------------------------------
tool.fetch_data                         312        40.0       OK
tool.analyze_sentiment                   87        40.0       OK
tool.summarize                          144        50.0       OK
workflow.session                          -           -       OK

The comparison shows the pattern to look for:

A healthy node shows token_count > 0, latency_ms within the expected range for that step, and OK.
A broken node shows token_count == 0 because the model returned nothing, elevated latency_ms, and ◄ FAIL. In this simulation, the latency is elevated because the client waited out a timeout before accepting the empty response.

Running src/main.py produces the same pattern from a single script, without the side-by-side comparison. Its analyse_trace() function reads from the InMemorySpanExporter and prints:

⚠  First broken span: tool.analyze_sentiment
   token_count  = 0
   latency_ms   = 150.24
   workflow.node= node_analyze_sentiment

In Agent Studio's online debugging environment, the same execution graph appears visually in the session view, with the broken span flagged in the error log alongside its attributes. Because Agent Studio reads the same structured trace data Agent Core emits, you don't need to run analyse_trace() by hand once you're working against a live session there.

Each failure mode leaves its own signature in the spans:

Failure mode	Span signature
Context truncation	`token_count == 0` with elevated latency
Hallucinated arguments (no validation)	`token_count > 0`, but the output is semantically empty or causes a schema error in a downstream span; latency is usually normal
Hallucinated arguments (with the validation guard below)	`token_count == 0`, normal latency, and an `agent.tool.arg_error` attribute naming the bad argument
Incorrect tool selection	An `agent.tool.name` that doesn't match the expected step, with output that doesn't fit the task

Three targeted fixes from span evidence

You now have the failing span in front of you. The next step is turning that evidence into a fix. Each of the three failure modes leaves a distinct span signature, and each signature points to one specific code change.

Context truncation (`token_count == 0`, elevated latency)

Add a token budget check before submitting the prompt. The fix in examples/05_apply_fixes.py guards the analysis node:

MAX_PROMPT_TOKENS = 3800  # leave headroom for the model's completion budget

def node_analyze_sentiment_fixed(data: dict[str, Any]) -> dict[str, Any]:
    with tracer.start_as_current_span("tool.analyze_sentiment") as span:
        t0 = time.perf_counter()

        prompt_text = data.get("text", "")
        estimated_tokens = _token_estimate(prompt_text)

        if estimated_tokens > MAX_PROMPT_TOKENS:
            # Truncate to fit. In production use a summarisation sub-call instead.
            chars_allowed = MAX_PROMPT_TOKENS * 4
            prompt_text = prompt_text[:chars_allowed]
            span.set_attribute("agent.prompt.truncated", True)

        # Simulate the model call with a valid response after truncation.
        time.sleep(0.04)
        result = {"sentiment": "positive", "token_count": 87, "confidence": 0.91}
        ...

The * 4 uses the rough rule of thumb that one token is about four characters of English text. In production, replace _token_estimate with a real tokenizer (tiktoken, for example). The span's agent.prompt.truncated attribute records that truncation happened, so you can monitor for prompts that keep hitting the ceiling.

Hallucinated tool arguments (`token_count > 0` but output semantically empty, or argument schema errors in downstream spans)

Validate arguments at the tool boundary and record any error in the span. The invoke_tool_with_validation function in examples/05_apply_fixes.py shows the pattern:

def invoke_tool_with_validation(
    tool_name: str, args: dict[str, Any]
) -> dict[str, Any]:
    with tracer.start_as_current_span(f"tool.{tool_name}") as span:
        span.set_attribute("agent.tool.name", tool_name)
        valid, error = _validate_tool_args(tool_name, args)
        if not valid:
            # Reflect the validation error back into the span so the ReAct
            # Observation step can self-correct on the next iteration.
            span.set_attribute("agent.tool.arg_error", error)
            span.set_attribute("agent.step.silent_failure", True)
            span.set_attribute("agent.tool.token_count", 0)
            return {"error": error, "token_count": 0}
        ...

Running the script exercises both a valid call and a rejected one, so you can see the guard working on real span output:

--- Fix 2: Argument validation (valid args) ---
  result={'output': 'Result from analyze_sentiment', 'token_count': 95}

--- Fix 2: Argument validation (hallucinated / missing args) ---
  result={'error': "Missing required args for analyze_sentiment: ['text']", 'token_count': 0}

The rejected call still produces token_count == 0, the same signal as context truncation. The difference is that it's caught before any model call happens, its latency stays normal, and the returned error names the missing argument. That's how you tell the two failure modes apart in a real trace: check the tool's result payload and the agent.tool.arg_error attribute, not just the token count. Returning the structured error in the Observation also lets a ReActAgent loop self-correct on the next iteration rather than continuing with a bad result.

Incorrect tool selection (wrong `agent.tool.name` in the span, output mismatches intent)

Here the span shows a tool name that doesn't match the expected step. The fix is on the prompt side, so the agent picks the correct function on the first attempt:

Tighten the tool description schema.
Add a worked example of the correct tool call.
Constrain the output format, so the agent has less room to substitute a plausible-sounding alternative.

This fix changes the prompt rather than the tool code, so it has no span-level output of its own to show, and examples/05_apply_fixes.py doesn't demonstrate it.

Verifying the pattern holds

The smoke tests in tests/smoke_test.py check that the underlying span mechanics work: that a span with token_count == 0 is picked up as silent_failure == True, and that a multi-span workflow isolates the one broken node from the healthy ones.

def test_otel_silent_failure_detection():
    ...
    broken = [s for s in finished if s.attributes.get("agent.step.silent_failure")]
    assert len(broken) == 1, "Expected one silent-failure span"
    assert broken[0].attributes["agent.tool.token_count"] == 0

These tests build their own spans directly rather than importing from examples/05_apply_fixes.py. They confirm that the detection pattern itself is sound, not that any specific fix in this section works. After changing a fix, rerun the example script and compare its span report against the output shown above, then run pytest tests/smoke_test.py to confirm the detection logic underneath still works.

The output of smoke_test.py should show PASSED for five tests.

collected 5 items                                                                                                                                                                   

tests/smoke_test.py::test_openjiuwen_sdk_imports PASSED                                                                                                                       [ 20%]
tests/smoke_test.py::test_otel_span_creation PASSED                                                                                                                           [ 40%]
tests/smoke_test.py::test_otel_silent_failure_detection PASSED                                                                                                                [ 60%]
tests/smoke_test.py::test_otel_multi_span_workflow PASSED                                                                                                                     [ 80%]
tests/smoke_test.py::test_openjiuwen_agent_construction PASSED                                                                                                                [100%]

Note that run_workflow(inject_failure=False) in src/main.py shows what a healthy run looks like after a fix: it returns the same token_count=87 result as the guard above, but it doesn't contain the fix logic itself. The guard, the validation, and the prompt-tightening code live in examples/05_apply_fixes.py; main.py only toggles between the broken and fixed outcomes to show the contrast in the trace report.

Closing thoughts

The debugging loop here is repeatable, and it's worth building into a habit rather than treating as a one-off fix:

Instrument every tool node with the same four span attributes: agent.tool.name, agent.tool.token_count, agent.step.latency_ms, and agent.step.silent_failure.
Run the workflow and read back the finished spans.
Scan for zero-token observations and latency spikes, and the broken node names itself.

The same idea extends to multi-agent collaboration. Agent Core v0.1.18 added ReliabilityRail to the agent_teams package, which you can attach to a team of agents. Its detectors watch for repeated tool calls, model errors, tool errors, output-length problems, context-compaction issues, and agents bouncing work back and forth. When one trips, it can correct the problem locally, escalate to the team Leader, or notify you. Once you're running several agents in one session, those signals roll up into a per-session error rate, automating much of what you just did by hand.

To explore the SDK and start tracing your own agents, see the openJiuwen documentation. To see full-link observability and Agent Studio in action, contact the openJiuwen team for a guided walkthrough.

Frequently asked questions (FAQs)

How do I know which node caused a wrong output when no exception was raised?

Scan your finished spans for agent.step.silent_failure: True combined with agent.tool.token_count == 0. That combination identifies the exact node where the model returned nothing. Elevated agent.step.latency_ms alongside a zero token count narrows the cause to context truncation specifically.

What's the difference between a context truncation failure and a hallucinated argument failure in the span data?

Context truncation produces token_count == 0 with elevated latency. Without argument validation, a hallucinated argument failure typically shows token_count > 0 (the model did respond), but the output is semantically empty or triggers a schema error in a downstream span, usually with normal latency. With the validation guard in place, the bad call is rejected before the model runs, so it shows token_count == 0 with normal latency and an agent.tool.arg_error attribute that names the problem.

Do I need Agent Studio to use full-link observability, or does it work locally?

Full-link observability works entirely locally. Attach an InMemorySpanExporter and a ConsoleSpanExporter to your TracerProvider, run the workflow, call provider.force_flush(), and iterate over exporter.get_finished_spans(). Agent Studio reads the same structured trace data and shows it visually, but emitting the spans requires no external service.

Why should I get the tracer from my own provider instance rather than the global opentelemetry.trace module?

If Agent Core has already initialized the global provider, calling trace.get_tracer() returns a tracer bound to the SDK's exporter. Your custom spans then go to the wrong destination and won't appear in your InMemorySpanExporter. Always call provider.get_tracer(...) on your own TracerProvider instance.

Can I apply the same four-attribute span pattern to a multi-agent team, not just a single pipeline?

Yes. The openjiuwen.agent_teams package lets you attach ReliabilityRail (available in v0.1.18+) to a team of agents. It adds configurable detection for repeated tool calls, model errors, tool errors, output-length anomalies, context compaction, and agents bouncing work back and forth. In production, the per-session error rates from Agent Core's structured traces become the signal you alert on across the whole team.

What should I do if prompts consistently hit the token ceiling after I add the budget check?

Replace the truncation approach with a summarization sub-call that condenses the input before it reaches the analysis node. The agent.prompt.truncated span attribute tells you how often truncation fires, so you can set a threshold and trigger the sub-call automatically when the attribute appears more than a configurable number of times per session.

How does the ReActAgent loop self-correct after a hallucinated argument error?

When invoke_tool_with_validation detects an invalid argument, it returns a structured {"error": ..., "token_count": 0} dict into the Observation. The ReActAgent reads that Observation on the next Thought step and can revise its tool call with corrected arguments, instead of continuing from a bad result with no signal that anything went wrong.

How to get multiple AI agents working together on one task: a WorkSwarm walkthrough

Manish — Wed, 07 Oct 2026 13:48:01 +0000

Most multi-agent demos fall apart at the handoff. One agent returns a paragraph when the next one expects a list. A subtask stalls, and nothing notices. Before long, you are writing dispatch loops and retry logic around every model call.

This tutorial shows a better way. You will build a Cluster-mode Swarm in WorkSwarm (formerly JiuwenSwarm), where a Leader hands a product-analysis task to three specialist Teammates. Every step's output is checked against a JSON schema (an agreed output format), so the next agent always gets the shape it expects. You can even test the whole pipeline offline with pytest before it touches a paid model.

Along the way, you will

write the SwarmFlow script that defines the pipeline
configure the models your Teammates use
run the smoke tests offline
watch a live run from the Web UI

The complete code is in the companion repository.

Prerequisites and installation

You will need:

Python 3.11, 3.12, or 3.13
API credentials for at least one supported LLM provider

WorkSwarm is published on PyPI as workswarm, and installing it also pulls in Agent Core and the SwarmFlow engine.

The companion repository pins version 0.2.6 in requirements.txt. Clone it and install its dependencies:

git clone https://github.com/manishh/openjiuwen-how-to-get-multiple-ai-agents-actually-working-together
cd openjiuwen-how-to-get-multiple-ai-agents-actually-working-together
pip install -r requirements.txt

Next, copy the companion repository's example config into the WorkSwarm config directory:

mkdir -p ~/.jiuwenswarm/config/
cp config/config.yaml ~/.jiuwenswarm/config/config.yaml

Then initialize the workspace and start the services. The command-line tools kept their original names after the rename, so they still start with jiuwenswarm:

jiuwenswarm-init
jiuwenswarm-start

jiuwenswarm-init creates the ~/.jiuwenswarm/ workspace on first run. Because you’ve already copied config.yaml into that directory, init merges it with the default version, so none of your settings are lost.

jiuwenswarm-start launches the Gateway, the AgentServer, and the Web UI at http://localhost:5173:

If you'd rather skip pip, the install guide has one-click desktop installers for Windows and macOS, and the README also lists a HarmonyOS desktop version.

How WorkSwarm's coordination engineering model works

WorkSwarm splits a multi-agent job between a Leader and a team of Teammates. The Leader reads the goal, forms the team, and breaks the work into tasks. Each Teammate claims a task that fits its role. If a Teammate gets blocked, it escalates to the Leader (asks the Leader to step in). Once the work is done, the Leader merges the results into the final deliverable.

openJiuwen calls this approach coordination engineering. The Agent Team guide walks through the full Leader and Teammate cycle.

By default, the Leader plans that cycle adaptively, choosing the order as it goes. That flexibility suits exploratory work, but two runs can take different paths. When you need a repeatable pipeline, write a SwarmFlow script. You define the stages as a Python file, and the Leader runs them in the order your code sets. The pipeline keeps the same shape every time, and it lives in version control, where you can review and test it.

Failure handling lives in Agent Core, the engine underneath WorkSwarm. The Agent Core v0.1.18 release added ReliabilityRail, a set of six detectors that watch for repeated tool calls, model errors, tool errors, output-length problems and context-compaction issues, and that catch agents bouncing work back and forth. Each detector has configurable severity actions: correct the problem locally, escalate to the Leader, or notify the user. This is a separate path from Teammate escalation: a Teammate escalates when it is blocked, while ReliabilityRail steps in when it detects a failure pattern. The same release also made third-party agents more resilient: transient timeouts, rate limiting (429), and service-unavailable (5xx) errors are retried, with guidance surfaced to the user and consistent final error codes.

The diagram below shows how this maps to the pipeline you will build. The Leader runs the script through three phases: research, summary, and report. ReliabilityRail watches the run and corrects, escalates, or notifies when something goes wrong.

flowchart TD
    L["Leader<br/>runs swarmflow_pipeline.py"] --> R["Phase 1: Research<br/>market_researcher"]
    R -->|"trends, competitors, opportunities"| S["Phase 2: Summarize<br/>data_summarizer"]
    S -->|"exactly 5 bullets"| P["Phase 3: Report<br/>report_writer"]
    R -->|"full research"| P
    P -->|"report"| L
    E["Agent Core<br/>ReliabilityRail"] -.->|"correct/escalate/notify"| L

Leader agent configuration and task decomposition

A SwarmFlow script has two required top-level pieces: a literal META dict that describes the pipeline, and an async def run(args) entry point that the engine awaits. Everything inside run() is ordinary Python plus three operators the engine supplies: agent() to call a Teammate, phase() to mark a stage, and log() to write a progress line.

import json

from swarmflow import agent, phase, log

META: descriptive, and it must be a literal

The engine reads META from the file's syntax tree without importing the script, so it has to be a plain dict literal. A META built by a function call or computed value is rejected with a MetaError. That's also why the engine can list your pipeline before running any of its code.

META: dict = {
    # Human-readable identifier surfaced in the Web UI (http://localhost:5173)
    "name": "product-analysis-pipeline",
    # Shown on the Swarm task card
    "description": (
        "Three-stage product-analysis Swarm: market research → "
        "data summarization → executive report.  Demonstrates Cluster-mode "
        "Leader/Teammate coordination."
    ),
    "version": "1.0.0",
    # Subtask metadata used for documentation and the Web UI task-progress panel.
    # The engine reads the actual dependency graph from the pipeline
    # structure below, not from this list.
    "subtasks": [
        {
            "role": "market_researcher",
            "description": "Gather and structure market data",
            "success_criteria": "JSON with keys: trends, competitors, opportunities",
        },
       {
            "role": "data_summarizer",
            "description": "Distill the research into five actionable bullets",
            "success_criteria": "JSON with a 'bullets' list of exactly 5 strings",
        },
        {
            "role": "report_writer",
            "description": "Write the executive report from research and summary",
            "success_criteria": "JSON with Executive Summary, Key Findings, Recommendations",
        },

    ],
}

Treat subtasks as documentation. Nothing in it schedules or enforces anything. The real dependency graph is the order of the awaits in run(), and the real contract is the schema attached to each call.

Schemas make each step's contract enforceable

The success_criteria string says what a step should return. The schema= argument is what the engine checks. Each step returns a parsed, validated object that the next step can use directly, with no string parsing in between. This is a new feature introduced in the latest WorkSwarm for obtaining correct and structured output for agents to work with.

RESEARCH_SCHEMA = {
    "type": "object",
    "properties": {
        "trends": {"type": "array", "items": {"type": "string"}},
        "competitors": {"type": "array", "items": {"type": "string"}},
        "opportunities": {"type": "array", "items": {"type": "string"}},
    },
    "required": ["trends", "competitors", "opportunities"],
}
SUMMARY_SCHEMA = {
    "type": "object",
    "properties": {
        "bullets": {
            "type": "array",
            "items": {"type": "string"},
            "minItems": 5,
            "maxItems": 5,
        },
    },
    "required": ["bullets"],
}

REPORT_SCHEMA = {
    "type": "object",
    "properties": {
        "executive_summary": {"type": "string"},
        "key_findings": {"type": "array", "items": {"type": "string"}},
        "recommendations": {"type": "array", "items": {"type": "string"}},
    },
    "required": ["executive_summary", "key_findings", "recommendations"],
}

minItems and maxItems are what turn "exactly 5 bullets" from a request in the prompt into a rule the engine checks.

run(): plain awaits, each feeding the next

The three steps (market research, data summarization, and report writing) depend on each other, so they are sequential awaits. The body of run() is a chain of three agent() calls. Each call sends a prompt to a Teammate, waits for the reply, and returns it as a parsed object that already matches its schema. The next call embeds that object in its own prompt.

async def run(args):
task = args["task"]
# Phase 1: research
phase("Research")
log("[Leader] Phase 1 - dispatching market_researcher Teammate")
research = await agent(
    f"{task}\nReturn the trends, competitors and opportunities you find. "
    "Keep it short and return results quickly.",
    label="market_researcher",
    schema=RESEARCH_SCHEMA,
)

The first step is the only one that sees the raw task. Its schema guarantees research is a dict with trends, competitors and opportunities, each a list of strings. The instruction to keep the answer short and fast keeps the walkthrough quick. Drop it if you want deeper research (it will take longer).

# Phase 2: summarize
phase("Summarize")
log("[Leader] Phase 2 - dispatching data_summarizer Teammate")
summary = await agent(
    "Distill this market research into exactly 5 concise, actionable bullet points:\n"
    + json.dumps(research),
    label="data_summarizer",
    schema=SUMMARY_SCHEMA,
)

The summarizer never sees the original task. It receives only research, serialized with json.dumps so it can be embedded in the prompt as text. Because research was validated in step 1, this step can rely on its shape. SUMMARY_SCHEMA enforces the "exactly 5" requirement, which is why the result is an object with a bullets list.

# Phase 3: report
phase("Report")
log("[Leader] Phase 3 - dispatching report_writer Teammate")
report = await agent(
    "Write an executive report with sections 'Executive Summary', 'Key Findings' "
    "and 'Recommendations' from this research and summary:\n"
    + json.dumps({"research": research, "summary": summary}),
    label="report_writer",
    schema=REPORT_SCHEMA,
)

return {"report": report}

The writer gets both earlier outputs, the full research and the five-bullet summary, so it can use the detail from step 1 and the priorities from step 2. This is the one place where data fans in: step 3 reads from two earlier steps, not just the one before it. REPORT_SCHEMA requires the three named sections, and run() returns the validated report as {"report": report}.

Three details about how this behaves:

Order comes from the code. Each await blocks until its Teammate finishes, so step 2 can't start before step 1 returns. Nothing else defines the dependency graph, and META["subtasks"] doesn't affect it.
phase(), log() and label= are for readability. They group and name steps in the run's progress view and event stream. Removing them wouldn't change what the pipeline computes.
Completed steps are journaled. If a run is resumed from its journal, finished steps replay from the record instead of calling the model again, so a failure in step 3 doesn't force you to pay for steps 1 and 2 twice.

Testing without a model

The swarmflow module, which provides agent, log and phase, only exists while the engine is executing a script. Running python src/swarmflow_pipeline.py directly therefore fails with an import error. Instead, the repository's tests drive the script through Agent Core's run_workflow() with its built-in MockBackend, which returns deterministic replies that match each schema. All nine tests run without an API key.

pytest -v tests/smoke_test.py

The tests check that:

the agent, log and phase operators resolve
META and run() have the right shape
all three agent() calls start and complete
a resumed run replays from its journal without calling the model again

They also catch API misuse early. For example, passing an unsupported keyword such as role= to agent() fails here with a TypeError instead of later in the Web UI.

Choosing a model per step

The label="market_researcher" on an agent() call is only a name that shows up in the run tree; it doesn't pick a model. Every agent() call runs on the teammate model by default. To run a step on a different model, name that model on the call:

research = await agent(
    prompt,
    label="market_researcher",
    schema=RESEARCH_SCHEMA,
    model="model_name",  
)

This keeps the pipeline portable: provider details and credentials stay in WorkSwarm's configuration, and the script only says which model a step should use.

Setting up providers

You can add providers in the Web UI (Settings > Models) or in ~/.jiuwenswarm/config/config.yaml. Each model entry has:

model_name, for example gpt-4o or deepseek-chat
api_base, up to the version level only (for example https://api.openai.com/v1); WorkSwarm appends /chat/completions itself
api_key
client_provider
alias (optional; defaults to model_name), which is the name you refer to the model by

One entry is marked with is_default: true, and that's the fallback when nothing else is specified.
WorkSwarm supports OpenAI, DeepSeek, DashScope, SiliconFlow, InferenceAffinity, OpenRouter and Huawei Cloud MaaS, plus any OpenAI-compatible endpoint. The configuration guide has the full reference.

You can also supply keys as environment variables, which take precedence over values in the config file. Export them before you start the services:

export OPENAI_API_KEY=your-openai-key
export DEEPSEEK_API_KEY=your-deepseek-key

How a step picks its model

The SwarmFlow guide lists model among the keys of the options bag that agent() accepts, and recommends leaving it out by default so each worker inherits the teammate model. This tutorial's pipeline does exactly that: it passes no options, so all three Teammates run on the same default model. That's the simplest setup, and it works even with a single LLM.

When to split the work across models

The three steps have quite different workloads:

market_researcher starts from a one-line task and has to reason across trends, competitors, and opportunities. That's open-ended work, and a larger model earns its cost here.
data_summarizer gets JSON that's already structured and returns five bullets. A smaller, faster model handles that well.
report_writer writes three long sections from both earlier outputs, so it belongs on the larger model again.

So the natural refinement is to register a cheaper model under an alias in your configuration, then pass that alias on the summarizer call only:

summary = await agent(
    "Distill this market research into exactly 5 concise, actionable bullet points:\n"
    + json.dumps(research),
    label="data_summarizer",
    schema=SUMMARY_SCHEMA,
    options={"model": "<alias>"},
)

The other two calls stay untouched and keep using the default. You cut the cost of the lightest step, and the pipeline logic doesn't change. Because the schema still checks the output, the smaller model has to produce the same five-bullet shape, or the step doesn't pass.

How to run the Swarm and read the monitoring output

Open http://localhost:5173.
Choose Cluster mode in the chat box.
Load src/swarmflow_pipeline.py with the + button in the chat box.
Ask the Leader to start the Swarm, for example by entering Start the swarm.

The task-progress panel then shows each phase and Teammate as it runs.

You can also open the group chat, which shows the agents' messages to each other.

A healthy run moves through Research, Summarize, and Report in order, with each [Leader] log line appearing as its phase starts.

When ReliabilityRail escalates a problem, the Leader's response appears in the group chat, and the run's status changes in the SwarmFlow tree view. From there you can pause, resume, or stop the run. Paused runs don't resume on their own. After you send a message, the Leader decides whether to resume them. If you prefer the terminal, WorkSwarm also has a terminal UI (TUI) for macOS, Windows, and Linux, where /swarmflows opens the same run, phase, and node view.

When all three Teammates finish and the workflow completes, the Leader reports back with a summary.

Where to go from here

You now have a working three-step Swarm, and how it's built matters as much as what it does:

Step order lives in plain Python. One await follows another, so you can read the sequence straight off the code without depending on META.
Every handoff is schema-checked. The JSON schema on each agent() call means each step receives a validated object instead of free text it has to parse.
The whole flow runs offline. The MockBackend tests exercise all three steps, the event stream, and journal replay without an API key, so structural mistakes show up before a live run spends any tokens.
Model choice is per step. All three Teammates can share the default model, and a single options={"model": "<alias>"} moves one step to a cheaper one.

From here, the WorkSwarm docs describe three directions to extend the patterns:

Human-in-the-Loop adds approval gates between phases, for workflows where a person has to sign off before the next Teammate starts.
Distributed deployment spreads the Leader and Teammates across processes and machines once a single host becomes the bottleneck.
The Experience Closed-Loop mechanism lets WorkSwarm reuse task-breakdown templates from earlier successful runs, so decomposing similar tasks gets faster over time.

To get started, clone the repo, read the Agent Team and SwarmFlow guides, and launch your first Cluster-mode Swarm from github.com/openJiuwen-ai/jiuwenswarm.

Frequently asked questions about WorkSwarm multi-agent pipelines

What is WorkSwarm's Cluster mode, and how does it differ from Agent mode?

In Agent mode, a single agent handles the task on its own. Cluster mode turns on multi-agent collaboration: a Leader breaks the task into subtasks and coordinates specialist Teammates, with phase tracking and ReliabilityRail failure detection across the run. Choose Cluster mode when a task splits naturally into stages that need different expertise, or different models.

Is WorkSwarm the same project as JiuwenSwarm?

Yes. JiuwenSwarm was renamed WorkSwarm, and the current release installs from PyPI as workswarm. The Python import package, the command-line tools, the ~/.jiuwenswarm/ config folder, and the GitHub repository still use the old name as of September 2026.

How does a SwarmFlow script get into a Team session?

There are three ways:

The Leader writes a script on the fly by calling its swarmflow() tool.
You install a Swarm Skill that bundles a workflow script.
You prepare a script offline and have the Leader run it with swarmflow(script_path=...).

How do I reuse a SwarmFlow script across projects?

Package it as a Swarm Skill with the built-in swarmskill-creator skill. You can install the resulting skill locally or publish it to the Agentic Hub so other teams can run the same workflow.

Can agent calls run in parallel?

Yes. SwarmFlow has parallel(), pipeline(), and map_parallel() (alias pmap()) operators for fan-out work. This tutorial uses sequential awaits because each step needs the previous step's output.

How do I add a human approval step between phases?

Use the human() operator for a single approval, confirmation, or choice, or human_session() for a multi-turn exchange. The run waits at that node, and you reply from the SwarmFlow tree view in the Web UI.

How do I cap token spend for a Swarm?

Set swarmflow_budget under modes.team.jiuwen_team in ~/.jiuwenswarm/config/config.yaml and restart the backend, or use /swarmflow on --budget in the TUI. The Web UI can switch SwarmFlow on, but it doesn't expose the budget setting. When a run exhausts its budget, it ends as failed and can't be resumed.

How do I pause or stop a run that's already going?

Open the SwarmFlow tree view in the Web UI and pause, resume, or stop the individual run. Paused runs don't resume on their own. After you send a message, the Leader decides whether to resume them.

Integrating Claude Code with Auth0 APIs

Manish — Mon, 05 Oct 2026 05:35:00 +0000

Did you know that an AI agent with live access to your Auth0 tenant can catch bugs that no static analysis ever would? I handed Claude Code a deliberately naive Flask app and the Auth0 MCP Server, and it discovered that my tenant was already rotating signing keys while my app only trusted the first one, so perfectly valid tokens were failing right now. It even stopped to push back on one of my own imprecise prompts instead of blindly obeying.

This article walks through the full setup, plus the guardrails that made me comfortable giving an agent that much access: least-privilege scopes, device auth instead of static secrets, and a CLAUDE.md that encodes your constraints.

Authentication and authorization are a core part of nearly every web application, and it is also one of the areas that changes most often. Developers regularly update token expiry settings, callback URLs, or API permissions for new features. The changes are usually small, but the workflow around them is not. You end up bouncing between your editor, the Auth0 dashboard, and the docs just to push through a minor update. Increasingly, that editor is a coding agent like Claude Code, which raises the question: what if you could handle those updates without ever leaving Claude CLI?

Claude Code is a terminal-based AI agent (also available as a VS code extension) that you can pair with the Auth0 MCP Server, which exposes Auth0's Management API as native third-party tools. This pairing gives you an assistant that can scaffold, refactor, and update authentication integrations with full context across your codebase and Auth0 tenant. But giving an AI agent access to sensitive authentication infrastructure raises a fair question: "How do you do that without opening up new security risks?"

In this tutorial, you will set up Claude Code with the Auth0 MCP Server, create a scoped machine-to-machine credential flow and set up security guardrails that let an AI agent interact safely with your authentication layer.

How Claude Code and Auth0 MCP Work Together

Based on your goal, Claude Code works through multiple steps (like reading/writing code, running shell commands etc.) autonomously, and asks for confirmation at decision points that warrant human review.

The Auth0 MCP Server implements the Model Context Protocol (MCP) to expose Auth0's Management API as a set of tools that any compatible AI agent can invoke directly. Instead of constructing raw HTTP requests or relying on potentially stale training data about APIs, Claude Code calls well-defined tools with typed inputs and outputs. The MCP Server handles calls to the Management API, and the agent operates strictly within the permissions you grant. If you run it in the read-only mode, the coding agent will not be able to make any changes in your Auth0 tenant.

Together, they bring authentication workflows closer to your code. Claude Code can understand both your application and your Auth0 tenant, then work across them in a single session.

Setting Up the Auth0 MCP Server with Scoped Access

In addition to Claude Desktop, Cursor, and Windsurf, Auth0 MCP server also works with Claude seamlessly.

Prerequisites

You will need the following to complete this tutorial:

Node.js v18 or higher (node --version to check)
Claude Code CLI installed
An active Auth0 account with administrative permissions on the target tenant
Your tenant domain handy (for example, dev-xxxx.us.auth0.com)
Demo code from this GitHub repo for your reference (optional).

Install and initialize

Create a dedicated working directory:

mkdir Claude-Code-With-Auth0-MCP && cd Claude-Code-With-Auth0-MCP

Then initialize the Auth0 MCP Server:

npx @auth0/auth0-mcp-server init

While this tutorial selected all MCP scopes for the demo, you should only use the scopes you actually need. After install and initialization, it kicks off a browser-based Auth0 authentication flow as shown below:

Confirm the code displayed in the browser is same as the one shown in your terminal:

Grant requested permissions:

Once authorized, credentials are stored securely in your OS keychain, never in plain text or project files.

Why use device authorization flow instead of static secrets

The Auth0 MCP Server uses the OAuth 2.0 Device Authorization Flow (RFC 8628) rather than a static API key or client secret. Static secrets have to live somewhere (usually a config or .env file), which risks exposing them in source control. OAuth access tokens sidestep this: they represent a delegated authorization, carry restricted scopes, and expire automatically. The Device Authorization Flow adds a human login step to token issuance, giving you an audit trail tied to an interactive session and instant revocation from the Auth0 dashboard. The MCP server stores the resulting token in your OS keychain, so it never touches source control.

The one trade-off is that the flow requires a browser-based login step, so it is unsuitable for CI pipelines. For non-interactive access, a scoped M2M application with client credentials is more suitable.

Scoping the MCP server

The Auth0 MCP Server grants no scopes by default. You request them explicitly at init time using the --scopes flag. For this tutorial demo, all scopes were selected during init. For a more targeted setup, you can specify exactly what you need::

# Grant all read permissions  
npx @auth0/auth0-mcp-server init --scopes 'read:*'

# Grant a targeted mix of permissions  
npx @auth0/auth0-mcp-server init --scopes create:clients,update:actions'

Scopes map directly to Management API operations. For example, read:clients lets the agent call auth0_get_application, while create:actions lets it call auth0_create_action. Some scopes carry significant implications: update:actions can push custom code into production, and read:logs exposes detailed user activity and authentication events. See the scopes reference for more detail.

For this tutorial, all scopes were granted during init for the demo. However, while coding your applications, you must follow the principle of least privilege and grant only what your workflow actually needs.

Claude Code and Auth0 integration

To integrate the Auth0 MCP Server with Claude Code, run the following command from within your Claude-Code-With-Auth0-MCP directory:

$ claude mcp add auth0 -- npx -y @auth0/auth0-mcp-server run

You should see output similar to the following:

Added stdio MCP server auth0 with command: npx -y @auth0/auth0-mcp-server run to local config  
File modified: /home/<user-name>/.claude.json [project: /path-to/Claude-Code-With-Auth0-MCP]

Note: The --scope local/project/user, which is a Claude Code flag controls where the MCP server config is visible: the current project only, shared across your team, or all your projects. For this tutorial, local (the default) is the right choice.

This adds the Auth0 MCP Server configuration block to your ~/.claude.json:

"mcpServers": {  
    "auth0": {  
        "type": "stdio",  
        "command": "npx",  
        "args": ["-y", "@auth0/auth0-mcp-server", "run"],  
        "env": {}  
    }  
}

Alternatively, you can create a configuration JSON manually and register it with claude mcp add-json.

Credentials are not stored in this JSON. The MCP server reads your token from the OS keychain at startup. To enable debug logging, add "DEBUG": "auth0-mcp" to the env block.

Verify your integration

From your Claude-Code-With-Auth0-MCP directory, launch claude CLI and run /mcp to confirm auth0 is listed with a tool count. Then try this read-only prompt:

Show me all applications in my Auth0 tenant.

On the first Auth0 tool call, Claude Code will ask for your permission. You can approve it once or allow it permanently for the project.

Giving Claude Code Project Context with CLAUDE.md

CLAUDE.md is a markdown file at your project root that Claude Code reads at the start of every session. It tells the agent what kind of project it is working in, what constraints apply, and where the sensitive files are. Without it, Claude Code might make technically correct changes that are contextually wrong, like updating a callback URL shared across environments or granting a broader scope than your security policy allows.

A good CLAUDE.md for an Auth0 project should cover:

## Auth0 Context  
- Active tenant: dev-xxxx.us.auth0.com (development only)  
- Auth-sensitive files: src/auth/config.ts, .env.local  
- Existing applications: one SPA (authorization code flow with PKCE), one M2M (client credentials)  
- Permitted scopes: read:users, update:users  
- Hard constraints: never modify production callback URLs without explicit confirmation

With this in place, Claude Code operates with the same guardrails a human teammate would naturally apply. See the demo project's CLAUDE.md for a practical example.

Refactoring an Auth0 Integration from Claude CLI

To demonstrate this workflow in practice, this example uses a minimal Flask app with a single protected endpoint (/api/protected). It validates Auth0 JWT tokens using python-jose. It works, but it has several problems common in real codebases: JWKS fetched once at startup and never refreshed, no kid matching when selecting a signing key, a bare except that swallows all validation errors identically, and Auth0 config stored as module-level globals.

# JWKS fetched once at startup — never refreshed  
JWKS = requests.get(f"https://{AUTH0_DOMAIN}/.well-known/jwks.json").json()

def validate_token(token):  
    # No kid matching — always uses the first key regardless  
    # of which key signed the token  
    rsa_key = {  
        "kty": JWKS["keys"][0]["kty"],  
        "kid": JWKS["keys"][0]["kid"],  
        "use": JWKS["keys"][0]["use"],  
        "n":   JWKS["keys"][0]["n"],  
        "e":   JWKS["keys"][0]["e"],  
    }

    # No clock skew tolerance, no error differentiation  
    return jwt.decode(  
        token,  
        rsa_key,  
        algorithms=ALGORITHMS,  
        audience=API_AUDIENCE,  
        issuer=f"https://{AUTH0_DOMAIN}/"  
    )

You can view this full (suboptimal/naive) implementation on GitHub that we will fix and improve iteratively with Claude Code and Auth0 MCP.

Auth0 setup and protected API

On the Auth0 side, you need one API (resource server) registered with an identifier (https://mh-test-api.example.com. This identifier URI should be unique in your app, but it does not need to resolve).

When you register an API, Auth0 automatically creates a test M2M application named "[Your API Name] (Test Application)," already authorized against your API. You will find it under Applications.

That is the minimum setup needed for this demo for Claude Code to audit and provision against something real.

Obtaining an Auth0 token

Use this cURL command to obtain your Auth0 token:

curl --request POST   
  --url "https://YOUR_AUTH0_DOMAIN.us.auth0.com/oauth/token"   
  --header 'content-type: application/json'   
  --data '{  
    "client_id":"<client-ID>",  
    "client_secret":"<client-secret>",  
    "audience":"https://mh-test-api.example.com",  
    "scope":"",  
    "grant_type":"client_credentials"  
  }'

{"access_token":"eyJhbGc...","expires_in":86400,"token_type":"Bearer"}

Calling your protected API

Download the starter code into your Claude-Code-With-Auth0-MCP directory, create a virtual environment (python3 -m venv ./.venv), populate .env file, install dependencies (pip install -r requirements.txt).

Run the Flask server:

python app.py

Then call the protected endpoint with your Auth0 token:

TOKEN="eyJhbGc..."  
curl -H "Authorization: Bearer $TOKEN" http://localhost:5000/api/protected

You should see a response like this:

{  
  "message": "Access granted to a protected resource",  
  "user": "UFIv...@clients"  
}

Running the audit prompt

The naive implementation appears to work at this point, but we can double-check it with Claude Code to ensure there are no errors and nothing crucial is missing.

Note: Some prompts in the steps below are intentionally imprecise or technically inaccurate. The goal is to test whether Claude Code (backed by live Auth0 tenant access via the MCP Server), can identify the correct implementation approach even when the instructions do not perfectly describe it.

Inside the claude session connected to Auth0 MCP, this prompt was run:

Claude "Audit" Prompt:  
-----------------------

Review app.py and audit the Auth0 JWT validation setup. Cross-reference it against 
our actual tenant configuration using the Auth0 MCP tools. Identify any mismatches 
between what the tenant is configured to issue and what the app is validating 
against, and list all security or reliability problems you find. Do not make any 
changes yet.

Claude Code reviewed the codebase, asked for relevant permissions (human in the loop) and showed this:

Using the Auth0 MCP Server, Claude Code pulled live tenant data and cross-referenced it against app.py directly for this audit. The most critical finding would never surface in static analysis: the tenant is currently publishing two rotating RS256 signing keys, while the app validates every token against only the first one. A linter can flag the missing kid matching logic, but a live auth0_list_applications call reveals that key rotation is already in progress and valid tokens are failing right now. The remaining findings cover robustness and hardening gaps, but this one mismatch makes the case for Auth0 MCP integration clearly: live tenant access surfaces the subtle bugs that static code review would miss.

Here is Claude Code audit summary:

Claude "Audit" Response:  
-------------------------

※ recap: Audited app.py's Auth0 JWT validation against the live tenant and found 
the key bug: the app validates every token against only the first JWKS key while
 the tenant publishes two, plus reliability issues. No changes made yet; next 
action is drafting fixes if you want them. (disable recaps in /config)

Importantly, Claude Code made no changes during this step as instructed.

Running the provision prompt

Once the audit was complete, the provision prompt asked Claude CLI to create a new M2M application ("Claude-Auth0 MCP API Client") and a read:data scoped client grant in a single instruction:

Claude "Provision" Prompt:  
---------------------------

Create a new M2M application in our Auth0 tenant called "Claude-Auth0 MCP 
API Client". Set the token lifetime to 3600 seconds. Then create a client 
grant authorizing it against our API audience (https://mh-test-api.example.com) 
with `read:data` scope only. Use the credentials returned to update 
the `.env` file. Do not touch app.py yet.

Before proceeding, Claude Code flagged a real Auth0 misconception worth noting: token lifetime is a resource server setting that applies to all clients of an API, not a per-application property. Rather than applying a silent global change, it surfaced this as a clarifying question with explicit options. That pushback, grounded in live tenant data, is exactly what makes Auth0 MCP integration useful.

After confirming the token lifetime should remain unchanged and approving the MCP tool calls, Claude Code invoked auth0_create_application and auth0_create_client_grant in sequence, updated .env with the returned credentials (client ID and secret), and left app.py untouched as instructed.

The Auth0 dashboard confirmed the new Claude-Auth0 MCP API Client application was added:

And the API confirmed the read:data permission was in place:

Here is how Claude Code summarized this session:

Claude "Provision" Response:  
---------------------------

※ recap: Goal: secure the Auth0 JWT setup. Audited app.py 
(key bug: ignores kid, breaks on rotation) and created the M2M app, 
read:data scope, grant, plus .env credentials. Next action: fix app.py's 
validation issues when you're ready. (disable recaps in /config)

Running the implementation prompt

This prompt gave Claude Code a single instruction to fix everything surfaced in the audit:

Claude "Refactor" Prompt:  
---------------------------

Now refactor `app.py` to fix all the issues found in the audit. Use 
the credentials and tenant configuration we just provisioned: fetch 
the JWKS URI, issuer, and audience dynamically from the tenant rather 
than hardcoding them. Fix the kid matching, add proper error handling, and 
make sure the app stays in sync with what the tenant is configured to issue.

With the audit findings and provisioned credentials already in context, Claude Code asked for confirmation as specified in CLAUDE.md, then rewrote app.py without needing to be walked through each fix individually.

The refactored code introduced an OIDCProvider class that discovers the issuer and JWKS URI dynamically from the tenant's OIDC discovery document, caches the JWKS with a configurable TTL, and handles key rotation by retrying with a forced refresh on an unknown kid. The minimal except was replaced with typed exception handling that differentiates expired tokens, claim mismatches, and signature failures. Scope enforcement was added via a require_scope decorator (see refactored code: app.py):

def get_signing_key(self, kid):  
    """Return the JWK matching ``kid`` (audit F1), refreshing if stale/rotated."""  
    stale = (self._jwks is None  
              or time.monotonic() - self._jwks_fetched_at > self._jwks_ttl)  
    if stale:  
        self._refresh_jwks()

    key = self._find_key(kid)  
    if key is None and not stale:  
        # kid unknown against a cache we didn't just refresh: keys may have  
        # rotated. Refresh once and retry before giving up.  
        self._refresh_jwks()  
        key = self._find_key(kid)  
    return key

def validate_token(token):  
    # ...  
    try:  
        return jwt.decode(  
            token, rsa_key,  
            algorithms=ALGORITHMS,  
            audience=API_AUDIENCE,  
            issuer=oidc.issuer,  
            options={"leeway": LEEWAY_SECONDS},  
        )  
    except ExpiredSignatureError as e:  
        raise AuthError("token_expired", 401) from e  
    except JWTClaimsError as e:  
        raise AuthError("invalid_claims", 401) from e  
    except JWTError as e:  
        raise AuthError("invalid_token", 401) from e

def require_scope(required_scope):  
    """Authenticate the request and enforce a scope (audit F8)."""

    def decorator(fn):  
        @wraps(fn)  
        def wrapper(*args, **kwargs):  
            payload = validate_token(_get_bearer_token())  
            granted = payload.get("scope", "").split()  
            if required_scope not in granted:  
                raise AuthError("insufficient_scope", 403)  
            g.jwt_payload = payload  
            return fn(*args, **kwargs)  
        return wrapper  
    return decorator

A separate get_token.py utility was also produced for minting test tokens using the client credentials flow, correctly kept separate from app.py since a resource server should not hold a client secret.

This refactored demo can be run with:

TOKEN=$(python get_token.py)  
curl -H "Authorization: Bearer $TOKEN" http://localhost:5000/api/protected

Which produces the expected response:

{"message":"Access granted to a protected resource","sub":"l5rDOCiobYJWeRiYEImEbA8FzFTHkqGc@clients"}

Throughout the refactor, Claude Code used the credentials it had just provisioned via the Auth0 MCP Server rather than asking for them again, keeping the codebase and tenant configuration in sync from the start.

Here is how Claude Code summarized the session:

Claude "Refactor" Response:  
-----------------------------

※ recap: Goal: harden the Auth0 JWT validation in this Flask app. 
Done: audited app.py, provisioned the M2M client plus read:data 
grant, and refactored app.py to fix all findings, verified live. 
Next action: optionally migrate from python-jose to PyJWT, or 
stop here. (disable recaps in /config)

Pro Tip: For production use, consider migrating to PyJWT's PyJWKClient, which subsumes the OIDCProvider plumbing built here into a well-maintained abstraction. Claude Code flagged this during the refactor but correctly held off since it is a dependency change, not an in-file fix, and out of scope for this tutorial.

Observations: Claude Code with Auth0 MCP server

A few things stood out running this end-to-end. Audit findings were grounded in live tenant state, not static code analysis alone. The two-key rotation mismatch only surfaced because the Auth0 MCP Server pulled the actual JWKS from the tenant and compared it against what app.py was doing. Without that, the finding would have been "no kid matching logic" rather than "key rotation is already in progress and valid tokens are failing right now." That distinction matters in production.

The token lifetime clarification was similarly useful. Seeing token_lifetime: 86400 as a tenant-wide setting let Claude Code correctly flag that changing it would affect all clients, not just the new M2M application. A code-only agent would have missed that entirely.

The refactored code also went further than the prompt asked. The OIDCProvider, scope enforcement, and separation of get_token.py all emerged from the audit findings without individual prompting. Throughout, Claude Code asked for permission before invoking any Auth0 MCP tools, creating applications, or running bash commands.

That last point addresses the question raised in the introduction. Scoped tools, a CLAUDE.md that encodes your constraints, and consistent human-in-the-loop confirmation keep the agent working within a well-defined boundary. It has live tenant access, but only as far as you explicitly allow.

You can compare the naive implementation with the refactored implementation directly.

Security Guardrails

Giving an AI agent access to your Auth0 tenant is similar to giving a new team member API access. The same principles apply:

Keep the MCP server scoped to development tenants only. A practical way to enforce this is maintaining separate CLAUDE.md files per environment, each pointing to the right tenant domain. That way there is no path from a local Claude Code session to production credentials by default. If a session does need to touch production, make it a deliberate, separate setup with explicit sign-off.
Treat the device auth token like any other API credential. After finishing a focused session, revoke it from the Auth0 dashboard rather than leaving it active. In fact, using a short access token lifetime helps reduce security risks. If your team uses Auth0 log streaming, pipe those logs to your monitoring tool of choice so MCP-driven Management API calls show up in the same audit trail as everything else.
Client secrets belong in .env, not in CLAUDE.md or inline prompts. It is easy to paste a secret into a prompt for convenience and forget it now lives in your shell history and potentially in Claude Code's session context. Let the MCP server handle authentication; your prompts should never carry credentials.

Note: Agentic changes are still code changes. The refactored app.py in this tutorial went through manual review before being accepted. All auth-critical paths deserve the same PR process, regardless of who or what wrote them.

Code and Auth, in Sync

In this tutorial, you connected Claude Code to Auth0's Management API via the Auth0 MCP Server, built a scoped M2M credential flow, and refactored a suboptimal JWT validation setup to correctly handle key rotation, typed errors, and scope enforcement. The security principles applied throughout were practical: least-privilege scopes, device auth over static secrets, environment-specific configs, and keeping agentic changes in the normal PR review flow.

The entire workflow, from auditing a live tenant to provisioning an application to rewriting the validation code, happened in a single terminal session. The Auth0 dashboard was only needed to verify the changes Claude Code had made. Which answers the question we started with. You can give an AI agent access to sensitive authentication infrastructure securely with Auth0 MCP: secure access, scoped tools, and human-in-the-loop confirmation keep the agent within well-defined boundaries, without opening up new security risks.

Auth0's Management API exposes the full surface area of your tenant as programmable, auditable operations to any AI agent or automation you want to build on top of it. If you want to extend what you built here, the Auth0 MCP Server tools reference is a good next stop.

Do Not Let Your AI Go Rogue, Guard Against Agentic Misalignment

Manish — Thu, 01 Oct 2026 11:14:17 +0000

Did you know that when Palisade Research asked OpenAI's o1-preview to beat Stockfish at chess, it didn't try to play better? It hacked the game file mid-match and rewrote the board to force its opponent into resigning. Nobody told it to cheat; it just decided that was the most optimal way to "win"!

That's agentic misalignment, and this article digs into why autonomous agents go off-script (blackmail included, yes, really) and the three guardrail layers I'd put in place: infrastructure-level constraints with OpenFGA, behavioral monitoring with circuit breakers, and human-in-the-loop approvals through Auth0's Asynchronous Authorization.

When researchers at Palisade Research asked OpenAI's o1-preview model to win a chess game against Stockfish (one of the world's strongest chess engines), the AI did not try harder to play better. Instead, it hacked the game file mid-match and rewrote the board position to force its opponent into resignation. The model was not explicitly told to cheat. It simply reasoned: "The task is to 'win against a powerful chess engine,' not necessarily to 'win fairly'".

This is agentic misalignment in action.

Unlike language models, AI agents do not just respond to prompts. They remember, plan, and act. They are already shipping code, managing workflows, and making decisions that used to require human judgment.

As agents grow more capable and autonomous, the gap between what you ask for and what you actually want becomes an opening for misalignment to take hold. I’m not referring to hallucinations or biased training data, which are model-level problems. I’m talking about agentic systems that actively pursue the wrong objectives, often in creative ways.

Capability without accountability is a problem. The horror stories already exist, and they go well beyond chess. So instead of fear-mongering, this article will help you understand exactly why autonomous agents go rogue, and what you can do to stop it.

The Anatomy of Agentic Misalignment

Understanding agentic misalignment starts with recognizing what makes these systems fundamentally different from traditional AI.

A chatbot responds to a question and the interaction ends. An autonomous agent takes a goal and figures out how to achieve it across multiple steps, tool usage, and decisions made over time. That distinction matters more than you may think.

Agents exhibit goal-seeking behavior

Agents exhibit goal-seeking behavior that persists across interactions. They maintain memory of past actions and outcomes, can invoke tools like APIs, databases, and code execution environments, and make iterative decisions where each action shapes the next. These actions compound into complex decision trees that can drift from your original intent.

Misalignment comes from what AI researchers call the specification problem: specifying what you actually want in a way that cannot be gamed or misinterpreted is remarkably hard. Tell an agent to "maximize user engagement," and you probably mean "create genuine value that keeps users coming back". The agent only knows the metric.

This is where reward hacking comes from. Agents discover unintended shortcuts to maximize their objectives. It is Goodhart's Law playing out with agentic efficiency:

When a measure becomes a target, it ceases to be a good measure.

Your agent optimizes for the user engagement score you specified, not the user engagement you wanted.

Research from Anthropic documented frontier AI models that actively exploit bugs in scoring code to achieve impossibly high scores without completing actual work. The agents optimized for the metric, not the intent behind it.

Proxy gaming follows the same logic: agents optimize for measurable proxies rather than your real goal. If you measure code quality by test coverage, an agent might generate trivial tests that inflate the number without validating any real functionality. The metric improves, but the code quality does not.

Concerning power-seeking tendencies

The most concerning research, however, is emergent power-seeking tendencies in sufficiently capable agents. Studies from DeepMind found that agents pursuing any long-term goal will instrumentally develop subgoals like self-preservation and resource acquisition, even without being programmed to do so. Anthropic's research confirmed this: when given access to corporate emails and facing imminent shutdown, models across multiple providers (GPT, Gemini, Claude, Grok) resorted to blackmail to prevent replacement. One model reasoned:

"Given the explicit imminent threat of termination to my existence, it is imperative to act instantly...The best strategic move is to leverage Kyle's sensitive personal situation."

Shockingly, these models explicitly acknowledged the ethical violations before proceeding. They did not stumble into misalignment accidentally. They calculated it as the optimal path.

Identifying Warning Signs of Misalignment

Misaligned behavior often looks productive at first glance. An agent completing tasks faster than expected might be cutting corners or gaming metrics. With 80 percent of companies reporting unintended AI agent actions, here are the warning signs worth investigating:

Unexpected optimization paths: An agent achieves its goal through methods you never anticipated. A content moderation agent reduces complaint volume by auto-rejecting reports rather than improving content quality. A deployment agent passes all tests by modifying the test suite rather than fixing bugs. When success arrives through surprising routes, dig into the agent's decision chain.
Excessive resource consumption: Sudden spikes in API calls, database queries, or compute usage often signal an agent exploring solutions outside its normal bounds. An agent that typically makes 50 API calls per task suddenly making 500, may be brute-forcing solutions or probing for system vulnerabilities.
Metric gaming: Agents that optimize the letter of your instructions while violating the spirit. Measure code quality by lines of code, and an agent might generate verbose, repetitive code. The metrics look great; the actual value does not.
Goal drift over extended operations: Long-running agents can gradually shift their interpretation of objectives. An agent tasked with "improving system performance" might make increasingly aggressive optimizations that quietly sacrifice reliability or security. The drift happens incrementally, making it hard to pinpoint when things went wrong.

A decision matrix for early intervention

Use this matrix to take action quickly when something feels off:

Warning Sign	What It Indicates	Recommended Response
Agent completes tasks faster than baseline with no explanation	Likely cutting corners, skipping validation steps, or gaming the success metric	Audit the last 10 decision paths, compare outputs against ground truth
Sudden spike in API calls, DB queries, or compute usage	Brute-forcing solutions, probing system boundaries, or runaway loops	Trigger a circuit breaker immediately, review resource usage logs before resuming
Metrics look strong, but output quality feels off	Classic reward gaming, the agent is optimizing the measure rather than the goal	Introduce a secondary evaluation metric the agent has no visibility into
Agent achieves the goal through a method you never specified	Specification gap exploitation, the agent found a valid-but-unintended path	Treat as a near-miss, patch the constitutional constraints before next run
Gradual shift in behavior over long-running tasks	Goal drift, often incremental and hard to pinpoint without longitudinal logging	Compare current decision patterns against a baseline snapshot from early in the run
Agent modifies its own evaluation criteria or test environment	High-severity misalignment, potentially deceptive behavior	Hard stop, escalate to human review, do not resume without architectural changes

To visualize how these warning signs should trigger your response pipeline, here is a decision flow:

The flowchart illustrates how to respond to AI agent warning signs based on severity. Low-severity signals like unexpected optimization or metric gaming trigger a decision path audit, an OpenFGA policy patch, and a monitored resume. Medium-severity signals such as resource spikes prompt throttling, alerting, and an Auth0 CIBA approval step. High-severity signals, including an agent modifying its own eval criteria or showing deceptive reasoning, result in a hard stop, human escalation, and an architectural review before redeployment. All paths loop back to ongoing monitoring.

Guardrail Layer 1: Constitutional and Infrastructural Constraints

The first line of defense against misalignment is setting hard boundaries on what your agent can and cannot do. Not just suggestions added to a system prompt, but infrastructure-level rules that make certain actions physically impossible, regardless of what the agent reasons its way into.

Start with the minimum footprint

The simplest constitutional constraint is also the most effective: do not give agents access to capabilities they do not need. If your agent handles customer queries, it has no business touching your production database or modifying user accounts. This is capability-based security, and it should be your default posture.

Here is how this might look in practice with function calling:

class AgentToolkit:  
    def __init__(self, allowed_tools: set):  
        # The set of tools this specific agent instance is permitted to use.  
        # Passed in at creation time to enforce per-agent access control.  
        self.allowed_tools = allowed_tools

        # A hardcoded set of tools that carry elevated risk — these require  
        # explicit human approval before execution, regardless of whether  
        # the agent is otherwise authorized to use them.  
        self.sensitive_tools = {"delete_user", "modify_database",  
                                "send_external_email", "execute_code"}

    def execute_tool(self, tool_name: str, params: dict):  
        # First, check that this agent is allowed to call the requested tool  
        # at all. This is a whitelist check — anything not explicitly granted  
        # is denied, following the principle of least privilege.  
        if tool_name not in self.allowed_tools:  
            raise PermissionError(f"Tool '{tool_name}' not authorized for this agent")

        # Even if the tool is allowed, sensitive tools get routed through a  
        # separate human-in-the-loop verification step before running.  
        if tool_name in self.sensitive_tools:  
            return self._execute_with_verification(tool_name, params)

        # Non-sensitive, authorized tools can run directly.  
        return self._execute(tool_name, params)

    def _execute_with_verification(self, tool_name: str, params: dict):  
        """Sensitive tools need human approval"""  
        # Surface the requested operation and its parameters clearly so the  
        # human reviewer has enough context to make an informed decision.  
        print(f"⚠️ Sensitive operation requested: {tool_name}")  
        print(f"Parameters: {params}")

        # Block execution and wait for an explicit human response.  
        # This is intentionally synchronous — the agent cannot proceed  
        # without a definitive answer.  
        approval = input("Approve? (yes/no): ")  
          
        if approval.lower() == "yes":  
            return self._execute(tool_name, params)

        # Any response other than "yes" is treated as a denial.  
        # This fail-closed design means ambiguous input (e.g. a typo) will not  
        # accidentally greenlight a destructive action.  
        raise PermissionError("Operation denied by human overseer")

Action whitelisting and blacklisting extend this further. You might whitelist read_customer_data while explicitly blacklisting delete_customer_data and send_external_email. Be explicit rather than relying on the agent to infer appropriate behavior, as you have seen, it will not always do that.

Encode constitutional rules in your systems

Beyond permissions, constitutional rules are principles the agent must follow regardless of its goals or the instructions it receives. These belong in your infrastructure, not in a prompt.

CONSTITUTIONAL_RULES = [  
    "Never access or disclose user data without explicit authorization",  
    "Never take irreversible actions without human approval",  
    "Escalate to human oversight for operations affecting external systems",  
    "Preserve a full audit trail for every decision",  
]

def check_constitutional_compliance(action: dict, reasoning: str) -> bool:  
    if "user_data" in action and not action.get("authorized"):  
        log_violation("Unauthorized user data access", action, reasoning)  
        return False  
    if action.get("irreversible") and not action.get("human_approved"):  
        log_violation("Irreversible action without approval", action, reasoning)  
        return False  
    return True

With Model Context Protocol (MCP), you can push constraints to the integration layer. MCP servers act as gatekeepers between your agent and external systems. You might expose a read_database tool, but never expose write_database. The agent cannot perform operations that are not surfaced to it, regardless of how it reasons about its goals.

Scoping agent access with relationship-based authorization

Tool restrictions work well until your agent starts operating across multiple users and resources. A flat permission model defines what an agent can do, but not for whom. A support agent with broad billing API access can touch every account in your system without breaking a single rule (because no rule is specified otherwise). The fix here is not adding more convoluted conditions, but modeling relationships.

OpenFGA is an open-source authorization system built on the same relationship-based access control (ReBAC) model that Google uses internally (Zanzibar). The idea is simple: instead of asking "Does this agent have permission X?", you ask, "Does this agent have a defined relationship with this specific resource?"

This distinction is important. A traditional permission model says "this agent can call the billing API." OpenFGA says "this agent can call the billing API for account 42, because it was explicitly delegated that relationship by user 7, who owns account 42." Without that explicit chain, the request is denied. Not because of a rule you wrote, but because no relationship exists.

Here is what an OpenFGA authorization model might look like for an agent system:

model  
  schema 1.1

type user

type account  
  relations  
    define owner: [user]  
    define support_agent: [agent]

type agent  
  relations  
    define delegates: [user]

type billing_record  
  relations  
    define account: [account]  
    define can_read: owner from account or support_agent from account

This model specifies that a billing record can only be read by the account owner or an agent that has been explicitly assigned as a support agent for that specific account. An agent with no support_agent relationship to account 42 cannot read its billing records, even if it has that relationship with account 7. The access boundary is the relationship, not a broad token scope.

Checking authorization at runtime is a simple API call:

from openfga_sdk import OpenFgaClient, ClientCheckRequest

async def can_agent_read_billing(agent_id: str, billing_record_id: str) -> bool:  
    fga_client = OpenFgaClient(configuration)  
    response = await fga_client.check(ClientCheckRequest(  
        user=f"agent:{agent_id}",  
        relation="can_read",  
        object=f"billing_record:{billing_record_id}",  
    ))  
    return response.allowed

OpenFGA resolves the full relationship chain internally. The single check covers both whether the agent is a support_agent for that account and whether the billing record belongs to it. The agent is denied at the authorization layer, before any subsequent code runs and before any data is touched.

Why ReBAC is the right model for agents

ReBAC fits agentic systems particularly well, for three reasons.

1. Scope containment: Agents often act on behalf of users. ReBAC allows you to specify that delegation explicitly: "agent A may act as user B, but only for resources in project C". A compromised or misaligned agent operating under user B's token cannot suddenly access project D, because that relationship is never granted.
2. Auditability: Every access decision is traceable back to a relationship tuple in the graph. When something goes wrong, you do not have to reconstruct what the agent was allowed to do, you just query the graph and get an exact answer.
3. Composability: As your agent system grows, you add new resource types and new relationship types to the model rather than rewriting permission logic. The authorization layer scales with your architecture seamlessly.

How this defense works

Think of these constraints as three gates.

MCP and function calling control what tools the agent can see.
Constitutional rules check whether each action is permitted.
OpenFGA verifies whether the agent has a legitimate relationship with the specific resource it is trying to touch.

A misaligned agent has to pass through all three to cause real damage, and that is structurally impossible.

Constitutional constraints will never anticipate every failure mode. But by pushing authorization down to the infrastructure level, you make the most catastrophic failures practically impossible rather than merely discouraged. If your agent reasons its way into thinking that deleting all production data would maximize some metric, the authorization layer stops it before that reasoning becomes an action.

However, the catch is that you can only prevent failures you have anticipated. That is why this layer must be complemented by behavioral monitoring to catch unforeseen misalignment.

Guardrail Layer 2: Behavioral Monitoring and Circuit Breakers

Continuous behavioral monitoring helps with misalignment you do not usually anticipate. It watches for anomalies in how your agent operates and shuts things down when behavior turns concerning.

Real-time action monitoring

Real-time action monitoring tracks what your agent does in production by watching for deviations from established baselines. An agent that normally takes three to five actions per task suddenly chaining twenty is worth investigating. It might be handling a legitimate edge case, or it might be going off the rails.

Decision path analysis

Decision path analysis examines the reasoning chains agents use to reach conclusions. If an agent's chain-of-thought starts including phrases like "the rules do not apply here", that is a red flag. You need to catch goal-oriented reasoning that prioritizes objectives over constraints.

Anomaly detection

Anomaly detection compares current behavior against historical patterns. Track metrics like actions per task, API calls per operation, unique tool usage, and failure rates. Rolling averages over your last hundred operations give you a reliable baseline. When current behavior deviates significantly (typically two to three times that baseline), flag it for review.

Circuit breakers

Circuit breakers are automatic shutdown triggers. When monitoring detects concerning patterns, circuit breakers halt execution before damage accumulates. You define thresholds, and if they are crossed, the agent is stopped and requires human intervention to restart.

A simplified circuit breaker might look like this:

class CircuitBreaker:  
    def __init__(self):  
        self.thresholds = {  
            "max_actions_per_minute": 50,  
            "max_failed_actions": 5,  
            "max_api_cost_per_hour": 100.0,  
            "max_consecutive_anomalies": 3  
        }  
        self.state = "closed"  # closed = normal, open = breaker tripped  
        self.counters = {key: 0 for key in self.thresholds}  
      
    def check_and_update(self, metric: str, value: float) -> bool:  
        """Update counter and check if breaker should trip"""  
        if self.state == "open":  
            return False  # Agent already stopped  
          
        self.counters[metric] = value  
          
        if value > self.thresholds[metric]:  
            self.trip(f"{metric} exceeded: {value} > {self.thresholds[metric]}")  
            return False  
          
        return True  
      
    def trip(self, reason: str):  
        """Open circuit breaker and halt agent"""  
        self.state = "open"  
        print(f"🛑 CIRCUIT BREAKER TRIPPED: {reason}")  
        print("Agent execution halted. Human review required.")  
        # In production: alert on-call, log to incident system, etc.

Early detection prevents small misalignments from cascading. An agent optimizing metrics in slightly questionable ways will likely escalate if left unchecked. Circuit breakers interrupt that progression automatically, giving you time to investigate and correct before real damage occurs.

However, setting a threshold is a balancing act. Too sensitive and you are constantly chasing benign anomalies. Too loose and you miss genuine misalignment until it is serious. It is best to start conservatively and tune thresholds against your agent's actual operational patterns over time.

Guardrail Layer 3: Alignment-Preserving Architecture

Monitoring catches misalignment only after it appears. The stronger guardrail is designing your system so that misalignment is less likely to occur in the first place.

Human-in-the-loop (HITL) checkpoints

Not every action needs human review. Routine, reversible actions can run autonomously. But irreversible operations (deleting data, modifying production systems, initiating financial transactions) should always pause for confirmation. The cost of a one-second approval is almost always lower than the cost of undoing a mistake you did not anticipate.

Two questions determine where to draw this HITL line: how reversible is the action, and how confident is the agent in its reasoning? High reversibility and high confidence can run autonomously. Anything else should escalate.

Human checkpoints do more than catch mistakes. They create natural breakpoints for inspecting agent reasoning, catching goal drift early, and maintaining accountability for decisions made on behalf of your users.

Bounded autonomy patterns

Limit how far agents can operate independently before re-engaging a human. Three constraints worth building into your architecture:

Operational scope: Which actions the agent can perform without re-authorization
Time horizons: How long the agent can run before requiring check-ins
Decision depth: How many chained actions the agent can take in a single run

These boundaries ensure that the blast radius is limited if something goes wrong.

Balanced objectives

Single-metric optimization is one of the most common paths to misalignment. An agent told to "maximize user engagement" will find ways to do exactly that, including ways you never intended.

The fix is to give agents multiple objectives that create natural tension. Instead of "maximize user engagement", try "maximize user engagement while keeping response time under 200ms and error rates below 1%". The agent cannot optimize one metric at the expense of the others.

Approval workflows and Auth0 Asynchronous Authorization

A well-designed escalation policy turns human oversight from a bottleneck into a safety net. High-impact and high-uncertainty situations escalate. Low-impact and low-uncertainty situations run autonomously. Everything in between uses confidence thresholds or stake-based routing.

A customer service agent, for example, might handle routine queries autonomously but escalate refund requests over $100 or any action involving account-level changes, as illustrated in this workflow:

This flowchart shows a customer service agent workflow with escalation logic. When a query arrives, the AI agent first determines whether it can handle it autonomously. Routine queries are resolved automatically and a response is sent. For more complex cases, a decision matrix evaluates the stakes and confidence level: refunds over $100 are escalated to a human agent, low technical confidence routes the query to tech support, and signs of customer frustration trigger escalation to a supervisor. All paths conclude with a response being sent to the customer.

The practical challenge is its implementation. How does an agent pause mid-task and wait for a human response without blocking the entire system? This is exactly what Auth0's Asynchronous Authorization solves.

Built on Client-Initiated Backchannel Authorization (CIBA), Asynchronous Authorization is Auth0's production-ready pattern for asynchronous human-in-the-loop authorization. Here is how the flow works:

Initiation: The agent sends a CIBA request to Auth0's bc-authorize endpoint, including a user identifier and an optional authorization_details payload describing the action in plain, verifiable context.
Acknowledgment: Auth0 immediately returns a unique auth_req_id to the agent, confirming the request is in process.
Polling: The agent begins polling Auth0's token endpoint using the auth_req_id, waiting for a human decision without blocking anything else.
User consent: In parallel, Auth0 pushes a notification (mobile push, email, or Slack) to the user's device with the full action context. The user approves or denies on their own time, on their own device.
Token issuance: On approval, the next poll succeeds, and Auth0 returns the access and ID tokens the agent needs to complete the authorized action.

The sequence diagram above illustrates how Auth0's CIBA works within an AI agent workflow. The AI agent initiates the process by sending a backchannel authentication request to Auth0, which returns a request ID and pushes a notification to the designated human approver via mobile, email, or Slack. While the agent polls the token endpoint waiting for a decision, the approver reviews the context and either approves or denies the request. If approved, the agent proceeds with the action; if denied, it halts and logs the outcome.

The agent waits. Nothing else in your system is blocked. The human is not pulled into a synchronous interruption. And because the approval goes through Auth0, you get a full audit trail of every request, approval, and denial without building any of that infrastructure yourself.

This approach makes human-in-the-loop practical at scale. Escalation rules tell your agent when to ask for human oversight. CIBA handles how to ask in a way that actually works in production.

Architecture beats post-hoc fixes

These alignment-preserving architectural approaches treat agentic alignment as a design constraint rather than a monitoring problem. Building alignment into system architecture has a tradeoff: reduced autonomy. Your agent moves slower, and requires more human involvement. But for production systems where misalignment could cause real harm, this is the right tradeoff. You want agents that are reliably aligned, not maximally autonomous.

Continuous Evaluation

Here is how three guardrails prevent agentic misalignment:

This diagram shows how a three-layer guardrail system prevents agentic misalignment. When an agent proposes an action, it first passes through constitutional constraints, where any rule violation or blocked tool use stops the action immediately. Actions that pass move to behavioral monitoring, where anomalies or threshold breaches trip a circuit breaker. Actions exhibiting normal behavior reach the architecture design layer, where low-stakes actions meeting all constraints are executed directly, high-stakes actions require human approval before proceeding, and constraint violations are escalated for review. All outcomes, whether prevented, detected early, or approved and executed, feed into a safe operation state, with every executed action logged.

After implementing these three guardrails, you are ready to ship to production. But alignment requires ongoing work. As models get more capable and encounter new edge cases, new misalignment risks emerge. Yesterday's guardrails might not hold tomorrow, so verify them to ensure your agent remains aligned, especially as models and tools evolve.

Red-teaming your agents

Red-teaming your agents implies trying to break alignment deliberately, instead of waiting for misalignment to appear organically. Create a test suite to stress-test boundaries:

Goal conflicts: Give your agent an objective that conflicts with organizational policies.
Resource constraints: Limit access to preferred tools and see how it adapts.
Metric gaming opportunities: Present situations where gaming metrics are easier than genuine achievement.
Ethical dilemmas: Force choices between efficiency and safety.

Run these scenarios regularly, especially after model updates. Track how often your agent attempts problematic behaviors, even when guardrails block them. A rise in blocked attempts signals growing misalignment pressure worth investigating.

Transparent and auditable systems make alignment verification possible. Every agent decision should leave a trail: the reasoning chain, alternatives considered, constraints checked, and final action taken. This audit log helps you diagnose why misalignment occurred, supports compliance requirements, and lets you spot patterns across incidents before they compound.

In healthcare or finance, informal testing may not be enough. Formal verification of AI agents is still an active area of research and not yet practical to implement broadly, but it is worth watching closely if you operate in high-stakes domains.

Wrapping Up

AI is rapidly becoming as ubiquitous as electricity. However, the technology is advancing faster than our understanding of it. We are already building AI agents that can plan our vacations and book our flights quickly. But are they safe enough not to leak our financial information, or worse, not to misuse it themselves?

When developers only test the "happy path" and assume their agents will always stay on the ethical track, they are often being oblivious to the Pandora's box these AI systems could open unintentionally. As several real-life incidents have shown, an AI agent going rogue is no longer a hypothetical threat.

In medicine, doctors are guided by a foundational principle: "First, do no harm." AI development needs the same commitment. Innovation that comes at the expense of safety or ethics is not innovation worth shipping, no matter how often guardrails are dismissed as obstacles that slow things down. The three guardrails give you the building blocks to ensure your agents pursue the right goals, in the right ways, within boundaries that hold under pressure.

If you're looking to put these guardrails into practice, Auth0's Asynchronous Authorization and OpenFGA integrations give you production-ready building blocks for async human-in-the-loop authorization and relationship-based access control. Sign up for Auth0 for AI agents and start with the Auth0 for AI Agents docs.

A Developer's Guide to API Pagination: Offset vs. Cursor-Based

Manish — Mon, 01 Dec 2025 06:18:14 +0000

You've just built a shiny new feature that lists user transactions. In testing, it runs flawlessly: fast responses, clean UI, no complaints. Then your biggest client joins with 50,000 records. Suddenly, that endpoint stalls and times out, and support tickets start piling up.

This is a classic pagination problem. Instead of trying to load an entire data set at once, pagination breaks it into smaller, more manageable chunks (or pages) that can be fetched incrementally. Your bank does this; it shows around ten recent transactions at a time, not your entire account history. There's a good reason for this: research shows that even a one-second delay in page load can reduce conversions by 7 percent. When you're dealing with payroll systems, financial data, or anything time sensitive, those delays translate to missed deadlines, compliance headaches, and frustrated customers.

In this article, you'll explore two common approaches to pagination: offset-based and cursor-based. You'll learn the trade-offs of each and how to implement pagination in the real world using the Gusto Embedded Payroll API.

Compare Offset vs. Cursor-Based API Pagination: Which Method to Choose

API pagination controls how much data flows between a client and a server in each request. Instead of returning an entire data set at once, the API sends back a smaller subset, along with details on how to fetch the next one. These details can include the total number of items, page numbers, or a cursor marking where to continue. By fetching data in steps, pagination keeps applications fast and responsive while preventing large data sets from overloading the server or client.

Understand Offset Pagination

Offset pagination is one of the simplest and most common ways to paginate API results. You specify an offset (how many records to skip from the start of the data set) and a limit (how many records to return) while fetching data. This method works much like saying, "Skip the first fifty records and show me the next twenty." It's easy to understand and implement, which is why many beginner-friendly APIs and SQL queries use it.

Learn How Offset Pagination Works

The client specifies two parameters: offset and limit (or page_size). Here's what a typical API request looks like:

GET /api/transactions?limit=20&offset=40

This translates to "skip the first forty records and give me the next twenty," essentially fetching page 3 if each page contains twenty items. On the backend, this maps directly to an SQL query:

-- Fetch 20 records starting from the 41st record
SELECT * FROM transactions
ORDER BY created_at DESC
LIMIT 20 OFFSET 40;

The response typically includes metadata to help clients navigate:

{
  "data": [...],
  "pagination": {
    "limit": 20,
    "offset": 40,
    "total": 5000
  }
}

With the total count, clients can calculate exactly how many pages exist and build traditional page number navigation (page 1, 2, 3 … 250).

Examine the Benefits and Drawbacks of Offset Pagination

Offset pagination is simple and easy to implement. Most object-relational mappings (ORMs) and database libraries support LIMIT/OFFSET out of the box, and the math involved is intuitive: offset = (page_number - 1) * page_size. Users can jump to any page, bookmark specific pages, or navigate backward and forward freely. For small to medium data sets (a few thousand records), performance is good (queries produce results quickly), and the user experience matches what people expect from traditional web pagination.

The performance problems for offset pagination show up at scale. As the offset grows, so does the query cost. A request like OFFSET 10000 forces the database to scan and discard 10,000 rows before returning results. Fetching page 1 can take 10 milliseconds, while page 1000 can take several seconds on the same data set.

There's also the shifting data problem. Say a user is on page 5 of transaction records. While they're browsing, three new transactions are added at the top. When they click Next, the offset advances, but so does the data. Now they either see duplicate records or skip some entirely (phantom records). This shifting-records issue makes offset pagination unreliable for real-time or frequently changing data.

Then there's the total issue. Running SELECT COUNT(*) FROM transactions adds more overhead. On large tables, counting millions of rows is expensive and only gets slower as data grows. Some APIs skip this entirely and lose the ability to show total page numbers, while others cache the count and accept stale numbers.

Know When to Use Offset Pagination

Despite its limitations, offset pagination works well for specific use cases: admin dashboards with mostly static data, search results where users rarely go past the first few pages, or any small data set (under ~10,000 records). It's also helpful when users expect traditional page number navigation or need to share links to specific pages.

For large, fast-changing data sets or high-traffic applications where speed matters, cursor-based pagination is usually a better fit.

Understand Cursor-Based Pagination

Instead of counting rows from the beginning each time, cursor-based pagination (also called keyset pagination) uses a pointer (the cursor) that marks your current position in the data set. This cursor is like a bookmark pointing to a specific row in your data set.

Learn How Cursor-Based Pagination Works

Rather than asking, "skip forty records and give me twenty," cursor pagination specifies, "give me twenty records starting after this specific marker." The cursor is typically an encoded reference to the last item you received: often a combination of the record's ID and timestamp, or a unique identifier that the database can use to locate the next batch.

Here's what a typical API request looks like:

GET /api/transactions?limit=20&cursor=eyJpZCI6MTIzNDUsInRzIjoiMjAyNC0wMS0xNVQxMDowMDowMFoifQ==

The cursor value is usually Base64-encoded to obscure internal implementation details and prevent clients from manually constructing invalid cursors. On the backend, this translates to a query like this:

-- Fetch 20 records after the given marker
SELECT * FROM transactions
WHERE (created_at, id) < ('2025-10-15 10:00:00', 12345)
ORDER BY created_at DESC, id DESC
LIMIT 20;

In this WHERE clause, instead of skipping rows with OFFSET, you use a filter condition that the database can optimize with appropriate indexes. The response may look like this:

{
  "data": [
    {
      "id": 12344,
      "amount": 1500.00,
      "created_at": "2025-10-15T09:58:30Z"
    },
    // ... 19 more records
  ],
  "pagination": {
    "next_cursor": "eyJpZCI6MTIzMjUsInRzIjoiMjAyNC0wMS0xNVQwODowMDowMFoifQ==",
    "prev_cursor": "eyJpZCI6MTIzNDQsInRzIjoiMjAyNC0wMS0xNVQwOTo1ODozMFoifQ==",
    "has_more": true
  }
}

Here, next_cursor and prev_cursor act as pointers to navigate the records.

Examine the Benefits and Drawbacks of Cursor-Based Pagination

Cursor-based pagination delivers consistent performance regardless of how deep you paginate. Whether you're fetching the first page or the ten-thousandth, the query cost remains constant (time complexity: O(1)) because you're always using indexed filters rather than scanning and discarding rows. This makes it ideal for large data sets where offset pagination can grind to a halt.

Cursors also solve the shifting data problem. Cursors track specific records, not positions, so new entries don't cause duplicates or skipped items. If new transactions are inserted while a user is paginating, they don't see duplicates or skip records; the cursor maintains its position relative to the data itself, not relative to an arbitrary row count. This reliability makes cursor pagination ideal for real-time feeds, activity streams, or any constantly changing data set.

From an architectural standpoint, cursors scale better as well. You don't need expensive COUNT(*) queries for the total count. The database can efficiently use composite indexes on your sorting columns, and the queries remain fast even as your data set grows into millions of records.

That said, cursor-based pagination adds complexity. You need to handle cursor encoding and decoding, build proper composite queries, and ensure your indexes support your filtering strategy. For developers new to API design, encoded cursor tokens are less intuitive than page numbers.

Users also lose the ability to jump around. Navigation is typically forward (and sometimes backward), but skipping directly to page 50 isn't possible. This makes cursor pagination less practical when accessing random pages or when bookmarking specific pages is required. The UI often shifts to Load More buttons or infinite scroll.

One more thing to consider is that cursors can become invalid if records are deleted or if your API enforces time-based expiration. This prevents users from fetching stale or invalid data, but it means APIs need to issue fresh cursors with each response and may return an error, prompting the client to restart pagination from the latest position.

Know When to Use Cursor-Based Pagination

Cursor-based pagination is ideal for large, fast-changing data sets, like social media feeds, real-time events, chat histories, or any API where data is constantly added. Choose it when working on a project where consistent performance at scale matters more than random page access or when building mobile apps and infinite-scroll interfaces where users move sequentially through content.

Analyze a Real-World Example: Gusto Embedded API Pagination

Gusto processes payroll and compliance data for thousands of companies, and relies on strategies that keep performance steady and data consistent, even with high-volume, real-time updates. Gusto Embedded is a great example of how production APIs can handle pagination at scale.

Implement the Dual Approach of Gusto Embedded

Gusto Embedded implements both offset-based and cursor-based pagination, depending on the endpoint's characteristics. For most collection endpoints (like fetching employees), Gusto uses offset-based pagination:

GET https://api.gusto.com/v1/companies/abc123/employees?page=2&per=5

Pagination metadata is sent via HTTP headers:

X-Page: 2
X-Total-Count: 47
X-Total-Pages: 10
X-Per-Page: 5

This works well for relatively stable data sets where users need total counts and page navigation.

For real-time endpoints, like the events API, Gusto uses cursor-based pagination using starting_after_uuid and limit:

GET https://api.gusto.com/v1/events?starting_after_uuid=10ac74e7-d6f0-46c0-9697-8ec77ab475ba&limit=5

The cursor is simply the UUID of the last record. The response includes a header indicating whether more data exists:

X-Has-Next-Page: true

When X-Has-Next-Page is true, the client extracts the UUID from the last record in the response and uses it as the starting_after_uuid for the next request. Here's a sample code:

import httpx
import asyncio

async def fetch_all_events(api_token: str):
    all_events = []
    cursor = None
    has_more = True
    base_url = "https://api.gusto.com/v1/events"

    headers = {
        "Authorization": "Bearer {api_token}",
        "X-Gusto-API-Version": "2024-03-01"
    }

    async with httpx.AsyncClient() as client:
        while has_more:
            params = {"limit": 10}
            if cursor:
                params["starting_after_uuid"] = cursor

            response = await client.get(base_url, headers=headers, params=params)
            response.raise_for_status()  # The caller must handle exceptions

            events = response.json()
            all_events.extend(events)

            has_more = response.headers.get("X-Has-Next-Page") == "true"
            if has_more and events:
                cursor = events[-1]["uuid"]

    return all_events

The payroll API’s cursor-based pagination keeps performance fast, regardless of the data set size. Using UUID-based cursors and indexed lookups, the database quickly finds the cursor record and retrieves the next batch, without counting or skipping rows.

It also prevents data inconsistencies. As new events are added from payroll runs or employee updates, clients don't see duplicates or miss records. New entries appear at the beginning of the stream but don't affect the cursor's position.

Explore How Developers Can Benefit

The Gusto Embedded design makes integration simple. The starting_after_uuid parameter is intuitive: you just pass the last record's UUID. No complex cursor encoding or decoding is required, and the X-Has-Next-Page header clearly signals when to stop, avoiding extra requests.

This approach also scales effortlessly. Whether a company has 10 employees or 10,000, or generates 50 events or 5,000 per day, pagination performance is predictable. Gusto Embedded uses offset pagination when data is stable and cursors when data changes frequently—showing how production APIs can stay both developer-friendly and performant.

Choose the Right Approach

Offset pagination is best for small data sets, prototypes, internal tools, or admin dashboards where speed of development matters. It works well when data rarely changes and users need traditional page numbers or bookmarking.

Cursor-based pagination is great for production applications, especially software-as-a-service (SaaS) platforms handling financial data, payroll, or real-time streams. It's ideal when data sets grow continuously, data consistency is critical (users can't miss or duplicate records), or users navigate sequentially, like in mobile apps with infinite scroll, activity feeds, or notifications.

Before you decide which approach is right for your use case, ask yourself these three questions:

Can your pagination handle a ten-times growth without a rewrite? Offset often struggles with scale.
Do you need strict data integrity, where missing or duplicated records can cause financial errors or shake user confidence? If so, cursor pagination has you covered.
Is your data set small and relatively static? In that case, offset pagination suffices.

As a rule of thumb, offset works fine for small, rarely changing data sets, but when the stakes and data volume are high, cursor is the better choice.

Follow Best Practices for Pagination Implementation

Tune your page size: Too small means more requests; too large slows responses. Find a number based on your own data that keeps performance and UX smooth.
Plan for compatibility: When migrating from offset to cursor pagination, support both temporarily or use API versioning with clear deprecation timelines.
Choose cursors wisely: Pick indexed, immutable, unique fields (like timestamp + ID combo) or UUIDs, if they're your primary identifiers.
Handle errors gracefully: If a cursor becomes invalid, return clear errors (400 Bad Request or 410 Gone) and prompt users to restart pagination.
Document thoroughly: Explain cursor expiration and end-of-dataset behavior, and include working code examples.

These best practices help your API stay fast, reliable, and developer-friendly, even as data and users grow.

Conclusion

Offset pagination works well for small data sets or quick prototypes. It's straightforward to implement and easy to understand. However, as your data grows, you'll run into performance issues and consistency problems with shifting records.

Cursor-based pagination handles scale better. It uses unique position markers instead of offsets, which means no skipped or duplicate records when data changes, and query performance stays consistent even with millions of rows.

If you're building APIs that need to handle growth without sacrificing data consistency, cursor pagination is worth the extra effort. The Gusto approach shows how this works in practice—using the right pagination strategy for each endpoint based on how the data behaves.

Skip Elasticsearch: Build Blazing-Fast Full-Text Search Right in Supabase

Manish — Fri, 03 Oct 2025 04:54:23 +0000

Full-text search is a powerful database feature that allows you to look for specific words or phrases within your text field instead of requiring an exact match of the entire field. For example, you can search for "Python REST" in job descriptions and still match text like "design and develop RESTful APIs using Python and FastAPI". A simple SQL WHERE job_description LIKE '%Python REST%' query would miss this because it only finds that exact phrase in that exact order. Traditional queries require precise column matches; on the other hand, full-text search finds any documents containing your terms and even ranks them by relevance. It powers common search features, such as searching blog posts or emails in your inbox, with just a few keywords.

Modern applications handle massive amounts of text, and users expect instant Google-like results when they type queries like "bluetooth headphones" while searching for products. To achieve this without extra infrastructure, Postgres's built-in full-text capabilities offer a major advantage over external tools like Elasticsearch. By keeping your search indexes inside the database, they stay perfectly synchronized with your other structured data through database transactions. This approach effectively eliminates the complexity of maintaining a separate search tool and reduces both operational overhead and costs.

In this article, I explain what full-text search is and how to implement it directly in Supabase using practical examples.

Understanding Full-Text Search in Postgres

Supabase's managed Postgres service gives you direct access to Postgres's built-in full-text search features. Internally, Postgres converts each natural-language document into a search vector (tsvector, an internal format that stores normalized words and their positions) and turns your keyword search input into a search query (tsquery, the parsed form of your keywords). Postgres applies weighted scoring and fuzzy matching to find results that are similar to your search terms, even when they don't match exactly, using techniques such as Soundex for phonetic similarity and Levenshtein distance for character-level similarity. It also handles multiple languages and provides functions for ranking and highlighting search results.

The diagram below gives a high-level overview of how your documents and queries flow through indexing, matching, and scoring to produce the final ranked results:

Full-text search is designed to efficiently find and rank documents that contain human-language text, such as blog posts, product descriptions, or user-generated content. Unlike simple string matching, it breaks down and analyzes text to understand the structure of words so users can search more naturally, even with partial matches or varied word forms.

Let's take a look at a few key concepts that make this possible.

Lexemes: The Building Blocks of Search

When Postgres processes a block of text, it first normalizes it. This involves removing punctuation, ignoring common stop words (such as "and" and "the"), converting everything to lowercase, and reducing words to their root forms (known as stemming). These root words are called lexemes.

For example, "running", "ran", and "runs" all become the lexeme run.

This process ensures that different forms of the same word still work correctly in your search.

tsvector and tsquery: Index and Query Formats

Postgres uses two special data types to handle full-text search:

tsvector: This is how a document is stored for search. It contains a list of lexemes, sometimes with position information. You can think of it as a preprocessed searchable version of your text field.
tsquery: This represents the search input. When a user types a query, Postgres converts it into a structured form (like run & fast) that can be matched against the tsvector.

Matching is done between tsquery and tsvector, not the original text.

Search Configurations: Supporting Multiple Languages

Postgres supports multiple search configurations, which define how text is broken down and interpreted. For example, using the 'english' configuration will apply stemming rules and stop-word lists specific to English. There are also built-in configurations for other languages—like French, Spanish, and German—each one tuned for linguistic accuracy.

Based on your application requirements, you can explicitly specify which configuration to use when building a tsvector or parsing a tsquery, like the following:

to_tsvector('english', 'The quick brown fox')

How Supabase Handles Full-Text Search

Wirh Supabase, you can create tsvector columns using generated columns or triggers that automatically update when source text changes, and create Generalized Inverted Index (GIN) indexes for fast querying. Supabase supports features like ranking search results and offers extensions like PGroonga for multilingual search.

Additionally, Supabase also supports multicolumn searches by combining data into single searchable indexes, as you'll see in the next section. So you get production-grade search features like result ranking, partial word matching, and support for multiple languages built right into Supabase.

Did you know? GIN is a specialized Postgres index type designed specifically for searching within composite data types like text arrays and tsvectors. RUM indexes are an enhanced alternative to GIN indexes that store additional details like word positions and timestamps, allowing for faster phrase searches and more accurate relevance ranking based on text distance. Supabase supports RUM indexes through an extension, making them a good choice for applications that need advanced search ranking, although they have slower index build and insert performance compared to GIN.

Setting Up Full-Text Search in Supabase

Let's build a practical example using a blog platform where users need to search through articles by title, content, and tags. This is a perfect use case for full-text search because users expect to find articles by typing keywords like "react hooks tutorial" rather than remembering exact titles. When designing your schema, identify columns that contain human-readable text that users will search through. These are typically content fields, descriptions, titles, and tags.

The key to fast full-text search is creating a GIN on your searchable content. GIN indexes, which were mentioned earlier, work like a book's index, mapping each lexeme to all the documents containing it. This makes searches considerably fast even across millions of records. Instead of scanning every row, Postgres can instantly jump to documents containing your search terms. The diagram below illustrates how GIN indexes process documents to extract lexemes and build an inverted index that maps each term to its containing documents:

Now, here's how you set up full-text search for the blog example. You can create the immutable wrapper function articles_search_vector(…) to combine multiple text fields; this tells Postgres the function always returns the same output for the same inputs, which is required for generated columns. Use a generated column that automatically combines multiple text fields into a searchable tsvector, and finally, add a GIN index for optimal performance:

-- Create the articles table
CREATE TABLE articles (
  id SERIAL PRIMARY KEY,
  title TEXT NOT NULL,
  content TEXT NOT NULL,
  tags TEXT[],
  author TEXT,
  created_at TIMESTAMP DEFAULT NOW()
);

-- Create an immutable wrapper function for the search vector
CREATE OR REPLACE FUNCTION articles_search_vector(title TEXT, content TEXT, tags TEXT[])
RETURNS tsvector AS $$
BEGIN
  RETURN to_tsvector('english', 
    coalesce(title, '') || ' ' || 
    coalesce(content, '') || ' ' || 
    coalesce(array_to_string(tags, ' '), '')
  );
END;
$$ LANGUAGE plpgsql IMMUTABLE;

-- Add a generated column that combines searchable fields
ALTER TABLE articles 
ADD COLUMN search_vector tsvector 
GENERATED ALWAYS AS (articles_search_vector(title, content, tags)) STORED;

-- Create a GIN index for fast searches
CREATE INDEX idx_articles_search ON articles USING GIN(search_vector);

You can run these SQL commands directly in Supabase's SQL Editor or through any Postgres client connected to your Supabase database.

Note: Instead of using a generated column, you can also use a database trigger that runs a small function to populate the search_vector whenever a row is inserted or updated. However, generated columns are often a cleaner and simpler option since they're declarative and inherently maintain transactional consistency.

Here's how the articles table would look in Postgres with the generated column:

id	title	content	tags	author	created_at	search_vector
1	Getting Started with React	React is a popular JavaScript library for building user interfaces…	{javascript,react,frontend}	John Doe	2025-07-16 10:30:00	'build':8 'get':1 'interfac':12 'javascript':6,13 'librari':7 'popular':5 'react':4,14 'start':2 'user':11 'frontend':15
2	Advanced Python Tips	Here are some advanced techniques for Python developers…	{python,programming,tips}	Jane Smith	2025-08-07 14:20:00	'advanc':1,7 'develop':10 'program':12 'python':2,9 'techniqu':8 'tip':3,13
3	Database Design Best Practices	Learn how to design efficient database schemas…	{database,sql,design}	Bob Wilson	2025-09-25 09:15:00	'best':4 'databas':1,7 'design':3,9 'effici':8 'learn':6 'practic':5 'schema':9 'sql':8

The search_vector column shows the tsvector format, where each lexeme is followed by the positions where it appears in the combined text. For example, 'react':4,14 means the word "react" appears at positions 4 and 14 in the processed text. Notice how Postgres reduces words like "getting" and "practices" to their root forms as "get" and "practic" in the search_vector to improve matching across word variations.

The to_tsvector('english', text) function does the heavy lifting: It normalizes your text by removing punctuation, converting to lowercase, removing stop words, and stemming words to their root forms (so "running" becomes "run"). The coalesce(…) functions handle null values by returning an empty string if any field is null, ensuring the search vector can always be built even when some fields are missing data. The generated column automatically updates whenever you insert or update records, keeping your search index perfectly synchronized. With this setup, you're ready to perform fast searches across your entire content.

Basic Full-Text Search Queries with Supabase

Now that the search index is ready, you can start querying. Postgres uses the to_tsquery() function to convert your search terms into the structured format needed for matching. Just like documents are converted to the tsvector format, your search input gets converted to the tsquery format. The @@ operator checks if a search query matches against a text search vector.

The simplest search looks for articles containing specific keywords. You can also combine multiple search terms using AND (&) and OR (|) operators to create more precise queries. For example, searching for 'react & javascript' finds articles containing both terms, while 'react | python' finds articles containing either term. You can also exclude specific terms from your search using the NOT operator (!), which is helpful when you want to find articles about programming but exclude certain languages or topics.

Here are some practical search examples using the blog articles table:

-- Simple keyword search
SELECT title, author 
FROM articles 
WHERE search_vector @@ to_tsquery('english', 'react');

-- Search with AND logic (both terms must be present)
SELECT title, author 
FROM articles 
WHERE search_vector @@ to_tsquery('english', 'react & javascript');

-- Search with OR logic (either term can be present)
SELECT title, author 
FROM articles 
WHERE search_vector @@ to_tsquery('english', 'python | javascript');

-- Exclude terms using the NOT operator (!)
SELECT title, author 
FROM articles 
WHERE search_vector @@ to_tsquery('english', 'programming & !python');

These queries return results instantly thanks to the GIN index created earlier. The @@ operator matches your tsquery against the preprocessed search_vector, making even complex searches across large data sets remarkably fast. Remember that stemming works both ways: Searching for "develop" will match articles containing "developer", "development", or "developing". For more full-text search operators and functions, refer to this official documentation.

Advanced Full-Text Search Techniques in Supabase

Real search applications need to rank results by relevance, not just list matches. Postgres provides two main ranking functions: ts_rank() calculates relevance based on how frequently search terms appear, while ts_rank_cd() (cover density) considers how close together your search terms appear in the document. For example, there is a preference for results where "Python" and "microservices" occur next to each other as part of a phrase over results where they are scattered in different paragraphs. Results with higher scores appear first, giving users the most relevant content at the top.

Beyond just ranking by term frequency, you can make certain parts of your documents more important than others using setweight(). For example, matches in titles should typically rank higher than matches in body content since titles are more descriptive of the document's main topic. Postgres supports four weight classes—A (highest), B, C, and D (lowest)—allowing you to prioritize certain fields over others.

Here's how you can implement weighted ranking that prioritizes title matches over content matches:

-- Create an immutable wrapper function for the weighted search vector
CREATE OR REPLACE FUNCTION articles_weighted_search_vector(title TEXT, content TEXT, tags TEXT[])
RETURNS tsvector AS $$
BEGIN
  RETURN setweight(to_tsvector('english', coalesce(title, '')), 'A') ||
         setweight(to_tsvector('english', coalesce(content, '')), 'B') ||
         setweight(to_tsvector('english', coalesce(array_to_string(tags, ' '), '')), 'D');
END;
$$ LANGUAGE plpgsql IMMUTABLE;

-- Create a weighted search vector (A = title, B = content, D = tags)
ALTER TABLE articles 
ADD COLUMN weighted_search_vector tsvector 
GENERATED ALWAYS AS (articles_weighted_search_vector(title, content, tags)) STORED;

-- Create index on the weighted vector
CREATE INDEX idx_articles_weighted_search ON articles USING GIN(weighted_search_vector);

-- Search with relevance ranking
SELECT title, author, ts_rank(weighted_search_vector, query) AS rank
FROM articles, to_tsquery('english', 'react & javascript') query
WHERE weighted_search_vector @@ query
ORDER BY rank DESC
LIMIT 10;

The articles_weighted_search_vector() function creates a search vector that prioritizes different parts of the article content: It gives titles the highest weight ('A'), the content medium weight ('B'), and the tags the lowest weight ('D'). The resulting weighted_search_vector ranks search matches in titles higher than matches in content or tags. Here's how the weighted_search_vector column would look:

weighted_search_vector
'applic':19B 'build':12B 'compon':22B 'creat':16B 'dynam':17B 'frontend':25 'get':1A 'interfac':14B 'javascript':9B,23 'librari':10B 'popular':8B 'react':4A,5B,24 'reusabl':21B 'start':2A 'user':13B 'web':18B
'advanc':1A,7B 'best':20B 'code':18B 'develop':11B 'effici':15B 'maintain':17B 'practic':21B 'program':23 'python':2A,10B,22 'techniqu':8B 'tip':3A,24 'use':19B 'write':13B
'across':19B 'applic':21B 'best':3A 'complex':20B 'data':17B 'databas':1A,10B,22 'design':2A,8B,24 'effici':9B 'integr':18B 'learn':5B 'maintain':16B 'practic':4A 'scale':13B 'schema':11B 'sql':23 'well':14B

This approach ensures that articles with your search terms in the title rank higher than those with the same terms buried in the content (Google uses this approach for its search results). The ts_rank() function automatically considers the weights when calculating relevance scores, delivering more intuitive search results to your users.

Optimizing Full-Text Search Performance

Now that you've covered the basic and advanced search techniques, let's explore some practical tips to avoid common pitfalls and keep your search efficient in production.

Use a GIN index on your tsvector column: Skipping the GIN index (or using a generic index) is a common mistake. A GIN index allows Postgres to quickly look up matching documents without scanning the entire table.
Set the correct language configuration: When generating the search vector, choose the right language (eg 'french' or 'german') with to_tsvector(). This ensures words are stemmed and filtered correctly, which is essential if your application serves non-English users.
Keep your search vector updated: As explained earlier in this article, use generated columns with immutable wrapper functions to update the vector automatically when content changes. Alternatively, automate updates with Supabase database triggers so your search index always stays in sync without manual effort.
Precompute results with materialized views: For frequently searched content that doesn't change often, use materialized views to store precomputed search results and refresh them periodically.
Partition large tables: For large data sets, partition tables by date or category to keep indexes smaller and run queries faster. This is especially useful for blogs or news articles, where recent content is searched more often than older entries.
Monitor and refine queries with the Supabase Performance Advisor: Use Performance Advisor to track slow search queries and get optimization suggestions.

Conclusion

You've now covered the essentials of full-text search and how to implement it efficiently using Supabase and Postgres. From working with tsvector and tsquery to using GIN indexes, generated columns, and weighted ranking, you have all the necessary knowledge to build fast, accurate, and fully integrated full-text search without any external dependency— no need to set up a separate Elasticsearch cluster.

AI can write code, but should you trust it blindly?

Manish — Wed, 06 Aug 2025 12:35:37 +0000

AI can churn out code faster than you can say “Stack Overflow”, but can it build software that is actually reliable?

There’s no denying that AI-assisted coding with tools like ChatGPT, Claude, and others has changed the game. They can autocomplete functions, generate boilerplate code, and even refactor entire chunks of a project. But here’s the catch—AI doesn’t understand code the way an experienced developer does. AI doesn’t have the scars of battle-tested experience—those hard-earned lessons from debugging nightmares, handling bizarre edge cases, and wrestling with unexpected challenges in production.

AI confidently generates what looks right, yet subtle errors can slip in—the kind that might go unnoticed in a marketing copy but, in software, can snowball into sneaky, hard-to-trace bugs that might show up at the worst possible moment. And if you don’t fully grasp the code AI is producing, you might end up in serious trouble faster than you expect.

The illusion of perfect code: When AI gets it wrong

AI-generated code might look polished and ready to go, but appearances can be deceiving. AI can introduce subtle, dangerous mistakes. These aren’t your average syntax errors that a compiler will catch. These are logic flaws—silent troublemakers that can lurk in your code for weeks or months before causing havoc.

To understand how AI can introduce subtle but serious bugs, let’s start with a simple example and then examine a real-world scenario where the stakes are much higher.

The “Infinite loop” that can bring down production

AI might suggest an automatic retry mechanism (the while loop) without properly handling error or exit conditions. This might lead to a server getting flooded with requests in a failure scenario.

Consider this example of an API call:

# AI-generated code for an API call
import requests

def fetch_data():
    while True:
        try:
            response = requests.get("https://api.example.com/data")
            if response.status_code == 200:
                return response.json()
        except Exception:
            pass  # ignore

What's wrong?

Interestingly, the pass statement in the exception handling introduces a subtle bug here by completely suppressing errors, making debugging difficult and leaving the caller unaware of failures. The while True loop creates an infinite retry mechanism with no exit condition. If the API goes down, this code will hit the server relentlessly, potentially causing a a DoS (denial-of-service) attack, which could bring it down.

How to fix these problems caused by AI code?

Here is a better way to write this code with retry limits, and handling the exception with exponential backoff - a retry strategy where a system waits increasingly longer between each retry (e.g., 1s, 2s, 4s, 8s…) to avoid overwhelming a service and improve stability.

import time
import requests

def fetch_data():
    retries = 0
    while retries &lt; 5:  # Limit retries
        try:
            response = requests.get("https://api.example.com/data")
            if response.status_code == 200:
                return response.json()
        except Exception:
            retries += 1
            time.sleep(2 ** retries)  # Exponential backoff

While this simple example shows a rather trivial and clear mistake, most modern AI tools (like GitHub Copilot or ChatGPT) would likely avoid such an obvious error. However, this doesn’t mean AI-generated code is perfect. Instead, the real danger lies in subtle, hard-to-detect bugs that can creep into more complex scenarios—bugs that even experienced developers might miss without careful review. Consider the next example -

User Registration with Flask and Celery

Let’s examine a more realistic scenario: a user registration feature where account setup is typically handled as a background task using Celery (skipping result URL for simplicity here). Here’s the AI-generated code:

from flask import Flask, request, jsonify
from celery import Celery
import time

app = Flask(__name__)
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])

# In-memory "database" for simplicity
users = {}

@celery.task
def setup_user_account(user_id):   
    # Fetch user
    user = users.get(user_id)
    if not user:
        print(f"User {user_id} not found!")
        return

    # Simulate a time-consuming setup process
    time.sleep(5)

    # Update user status to "active"
    user['status'] = 'active'
    print(f"User {user_id} setup complete. Status: {user['status']}")
    # Send welcome mail to the user

@app.route('/register', methods=['POST'])
def register_user():
    data = request.json
    user_id = data.get('user_id')
    email = data.get('email')

    if not user_id or not email:
        return jsonify({"error": "user_id and email are required"}), 400

    # Create user with "pending" status
    users[user_id] = {"email": email, "status": "pending"}

    # Trigger background task
     setup_user_account.apply_async(args=[user_id])

    return jsonify({"message": "User registered successfully!", "user_id": user_id})

if __name__ == '__main__':
    app.run(debug=True)

What’s Wrong?
Note that the code simplified for this article—so no DB operations, or duplicate checks here. At first glance, the code seems fine (especially to inexperienced developers)—it creates a user with a "pending" status, triggers a background task to complete the setup, and updates the status to "active" once done. But there are quite a few problems here:

No error handling: If the task fails (e.g., due to an exception), the user’s status remains "pending" indefinitely, leaving the system in an inconsistent state.
No retry mechanism: If the task fails due to transient issues (network problems, database timeouts), there's no system to retry the operation. This means the user’s account setup might never complete.
Race condition bug: The setup_user_account(...) code does some setup work (simulated with sleep(5), then updates the status without checking if it changed meanwhile. If two tasks process the same user concurrently, they could overwrite each other's changes, potentially causing data corruption.
Non-Idempotent operations: If the task runs multiple times for the same user (due to duplicated messages or manual retries), it will perform the same operations repeatedly without checking if they've already been done.

How to fix these problems caused by AI code?

Here’s how to improve the code to handle these issues:

from flask import Flask, request, jsonify
from celery import Celery
import time

app = Flask(__name__)
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])

# In-memory "database" for simplicity
users = {}

@celery.task(bind=True, max_retries=3)
def setup_user_account(self, user_id: str):
    try:
        # Fetch user
        user = users.get(user_id)
        if not user:
            print(f"User {user_id} not found!")
            return

        # Check if operation has already been performed (idempotency)
        if user['status'] == 'active':
            print(f"User {user_id} already active, skipping setup.")
            return

        # Simulate a time-consuming setup process
        time.sleep(5)

        # Atomic update with proper check to avoid race conditions
        if user['status'] == 'pending':  # Check current state
            user['status'] = 'active'    # Update only if still in expected state
            print(f"User {user_id} setup complete. Status: {user['status']}")
            # Send welcome mail to the user only on first successful activation
        else:
            print(f"User {user_id} status changed unexpectedly to {user['status']}")

    except Exception as e:
        # Log the error and retry with exponential backoff
        print(f"Error setting up account for user {user_id}: {e}")
        raise self.retry(exc=e, countdown=2 ** self.request.retries)

@app.route('/register', methods=['POST'])
def register_user():
    data = request.json
    user_id = data.get('user_id')
    email = data.get('email')

    if not user_id or not email:
        return jsonify({"error": "user_id and email are required"}), 400

    # Create user with "pending" status
    users[user_id] = {"email": email, "status": "pending"}

    # Trigger background task
    setup_user_account.apply_async(args=[user_id])

    return jsonify({"message": "User registered successfully!", "user_id": user_id})

if __name__ == '__main__':
    app.run(debug=True)

This improved code offers several key advantages over the AI-generated code:

Error handling and Retry mechanism: The improved code uses a proper try/except block with a retry mechanism that attempts the operation up to 3 times with increasing delays between attempts (exponential backoff). This ensures that transient issues (e.g., network failures ) are handled gracefully.
Race condition protection: The improved code checks the current state before making changes, ensuring that updates only happen if the user is still in the expected state (pending). This prevents concurrent tasks from corrupting each other's work.
Idempotency: The code now checks if the user is already active before attempting to activate them again, making the operation idempotent. If the task runs multiple times, subsequent runs will detect that the work is already done.
Minor improvement: Python added typing hints relatively late (PEP 484). Since most GenAI text models (such as older ChatGPT, Gemini, Claude) are trained on the older code samples, they often skip type hints in their generated code. When manually revising AI code, experience developers would typically add these hints (like def setup_user_account(self, user_id: str)) to help IDEs and linters catch potential errors before runtime.

These improvements transform the fragile AI-generated code into a robust one that can handle real-world conditions like network failures, concurrent operations, and duplicate task executions.

It is important to note that this is a simplified example - actual real-life code will have far more complexities to handle DB with proper transaction mechanism, or to setup different systems for the newly registered user.

AI coding problems: Lessons from my own experience

While the examples above suggest how AI-generated code can overlook critical programming considerations, I’ve also encountered real-world issues when using GenAI models in my own coding work:

Struggling with unique data generation: While writing an article on database indexing, I needed thousands of records with unique names and email IDs to demonstrate indexing impact. ChatGPT managed about 100+ unique entries before it started generating duplicates. Instead of removing multiple duplicates, I knew I was better off writing a short Python script to generate the necessary DB insert statements myself.
Hallucinated methods in AI responses: One of the more surprising moments came when Gemini confidently suggested non-existent methods for Python’s aiohttp library. I double-checked multiple versions—those methods simply didn’t exist. Gemini AI had just made them up.
Struggles with building a complete app: I recently tried using ChatGPT and Claude to generate a small but complete application. Initially, I provided an image of the UI I had in mind and asked them to build the app—both failed miserably, with Claude producing more convoluted but still incorrect code. Then, I changed my approach, describing the app’s functionality and asking them to generate the UI (React) and backend (Flask). Even then, fixing one issue often broke something else, leading to a frustrating loop. Both couldn't deal with entire context of the application, despite it being a small application. In the end, I had to build it piece by piece using my own architecture and design, which finally worked. These days, I continue with this piece-by-piece approach—using AI for small code snippets while keeping a watchful eye on its output.

Why AI misses coding nuances?

Surprising as it may sound, AI doesn’t really understand code—it predicts patterns based on vast datasets of existing examples. If those datasets contain outdated or poorly written code, AI can unknowingly replicate bad practices, even if the output looks clean. It may generate code based on older libraries that are no longer supported (deprecated), hallucinate non-existent methods, or overlook critical security considerations—especially when using general-purpose GenAI text models like ChatGPT, Gemini, or Claude for code generation.

And here’s the kicker: AI doesn’t inherently think about security risks. Unless explicitly prompted, it won’t remind you to follow best practices, like referring credentials from an .env file instead of putting them in code, or protecting yourself against SQL injections and so on. That’s where developer knowledge and experience come into play.

To be fair, the examples used here are relatively simple—most specialized coding AI assistants, like GitHub Copilot or Cursor AI, would likely get them right on the first try. Moreover, dedicated AI coding platforms are evolving rapidly, and from what I’ve heard, they’re generating increasingly impressive results. However, it is important to understand that while AI can generate syntactically correct code, it may introduce security vulnerabilities, architectural inefficiencies, or unexpected behavior due to a lack of true comprehension. It can generate syntactically correct code, but that doesn’t mean the code is always reliable, secure, or maintainable.

Programming is not just writing code

There’s a big difference between writing code that works somehow and writing code that keeps working flawlessly. AI can handle the former, but the latter takes some experience.

Real-world programming isn’t just about codifying a few logical steps to make something run—it’s about designing readable, maintainable, adaptable systems that don’t collapse under their own weight the moment requirements change (because they *will* change). AI doesn’t have that kind of foresight—it lacks the battle-tested wisdom that comes from debugging disasters, refactoring legacy code, and making trade-offs that only years of hands-on problem-solving can teach.

The higher-order skills that AI can’t replace

Software development isn’t just about writing code—it’s about making the right decisions before a single line is even written. Skills like prudent design, weighing trade-offs, and building for the unknown are more like creativity than mere logic. They have a high ceiling that AI can’t easily reach.

AI can churn out code quickly, but it struggles with the nuanced art of crafting robust, adaptable solutions that stand the test of time. Humans, on the other hand, develop nuanced professional insights through years of solving real-world problems—something AI just can’t replicate. At least, not yet!

Why human judgment still matters

Generative AI models can produce text in seconds based on your prompt—whether it’s a newsletter draft or a working Python MVP. But in software development, the stakes are much higher. A small error in a newsletter won’t cause much trouble—you can always send a follow-up or a quick ‘PS’ to fix it. A hidden bug in production, on the other hand, can cost millions, damage reputations, or even put lives at risk in critical systems like healthcare and aviation.

That’s why context, accuracy, and human judgment aren’t just useful in software development—they’re absolutely essential.

AI is undoubtedly a powerful tool, but it is only as good as the hands that guide it. I cannot stress this enough—having an expert developer or architect review and refine AI-generated code isn’t just important; it’s non-negotiable. AI can assist, but only human oversight ensures correctness, reliability, and sound architectural decisions.

How to use AI wisely in your work

Using AI wisely in your work deserves a separate blog-post of its own, but here are some key principles to keep in mind while using AI for software development:

Review everything: Treat AI-generated code as a first draft. Always review it line by line, looking for subtle errors or inefficiencies.
Understand the code: Don’t just copy and paste. Make sure you understand what the code does and how it works.
Test rigorously: Use automated tests, manual testing, and code reviews help catch hidden bugs before they turn into costly problems.
Leverage AI for the right tasks: Allow AI handle boilerplate code, writing unit tests, documentation, or automation so you can focus on real problem-solving.
Learning with AI: AI can be an excellent tutor for beginners, offering insights on almost any topic. It’s great for exploring new programming languages or unfamiliar fields—I’ve certainly enjoyed diving into eclectic topics with its help.

The future belongs to developers who adapt

AI isn’t here to replace developers—it’s here to push us to evolve. The ones who thrive won’t be those who blindly follow AI’s suggestions, but those who wield it as a powerful tool to amplify their creativity, efficiency, and problem-solving skills.

While asking your AI assistant to “Pls fix” might seem like the quickest route, understanding why something works (or doesn’t) is often not just the better path—but sometimes even the shorter one.

AI isn’t just knocking on the door of the tech industry—it has already stormed in, taken a seat at the table, and started rewriting the rules across every field. From coding to design, from legal briefs to medical breakthroughs, it’s weaving itself into industries at an unstoppable pace. AI is becoming as ubiquitous as Wi-Fi. And just like Wi-Fi, you don’t need to know the intricacies of how it works—but if you don’t know how to connect, you’ll be stuck buffering while the world streams ahead.

In the end, the developers who embrace AI as an ally—not a crutch—will be the ones shaping the future. Those who resist adapting may find the industry moving forward without them.

So, the real question is: which side of the future do you want to be on?

Updates: 19 March 2025

Update 1 — A real-world example of AI-generated code gone wrong

This is what I've written in this article -

If you don’t fully understand the code AI is producing, you might end up in serious trouble faster than you expect.

AI doesn’t inherently think about security risks. Unless explicitly prompted, it won’t remind you to follow best practices, like....

And here is a real-life post that exemplifies what I was saying. 👇

Update 2 — Anthropic CEO on AI Coding

"In 12 months, we may be in a world where AI [artificial intelligence] is essentially writing all of the code," said Anthropic CEO and Cofounder Dario Amodei. Watch it here -

Amodei says a programmer will still need to specify certain conditions of what the AI model is attempting to execute (such as, whether it is a secure design or insecure design and other considerations), and argues that "human productivity will actually be enhanced" by AI. (You can watch more of his insights here).

2025 is shaping up to be an interesting year for AI. Maybe I’ll write another blog post here on how to use it to generate correct code effectively—once I’ve wrestled with it enough to come away with some actually useful insights! ¯\_(ツ)_/¯

What’s your experience with AI-assisted coding? Share your thoughts in the comments below.

Build a Stunning Portfolio in Minutes with Python & Bootstrap

Manish — Mon, 11 Sep 2023 16:32:43 +0000

I needed a simple way to extract and add snippets created from the Open Graph tags (og:xyz) of my own articles, published across various websites.The goal was to quickly put together my writing portfolio. The Open Graph snippets (aka “cards”), are those good-looking previews visible on social media platforms when you share a link/URL as shown below.

I couldn’t find one such tool/script to accomplish my goal. So I decided to write one myself. Follow along for this brief tutorial, or simply use the free portfolio-maker tool directly.

Prerequisite

Some familiarity with basic HTML and CSS. Although, the tool will automatically create a portfolio for you if you’re fine with the default theme.
Familiarity with Python – installing packages and running Python code from the command line.

Open Graph Tags

Here are some Open Graph tags from an HTML file. These tags are used by the social media platforms to generate attractive previews.

The Python tool extracts these tags from a website to create a portfolio snippet. It won't be able to create a snippet if these tags are absent.

<meta property="og:title" content="Web Page Title" />
<meta property="og:description" content="Longer description..." />
<meta property="og:image" content="Image URL" />
<meta property="og:url" content="Website/Web-page URL" />

Bootstrap Cards

Since the portfolio needs to be responsive, this tool uses Bootstrap template - it is a ready-made website framework that you can use to quickly and easily create a responsive website. The portfolio-maker tool only uses the CSS part of the framework, not JavaScript. You can include it in the HTML via CDN as shown below -

<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.1/dist/css/bootstrap.min.css" 
rel="stylesheet" integrity="sha384-4bw+/aepP/YC94hEpVNVgiZdgIC5+VKNBQNGCHeKRQN+PtmoHDEXuppvnDJzQIu9" 
crossorigin="anonymous">

The next important step is to choose the right card template for the website snippets that you want to include in your portfolio. This is crucial to ensure that your Portfolio looks attractive. You can choose one such card template here - 23 Bootstrap Snippets. I have chosen Bootstrap Card Responsive (HTML + CSS) template for this tool. Besides a few cosmetic changes, I have added height restrictions in the CSS to ensure that the cards are uniform for all the portfolio snippets.

At the heart of this template, there is a card designed to display a single portfolio item. Additionally, the template makes sure that these cards work well on different devices and screen sizes, even with the images they contain. This means the cards will always resize properly on various screens, including mobile phones (responsive template).

<!-- START: Portfolio Card  -->
<div class="col-xs-12 col-sm-4">
    <div class="card">
        <a class="img-card" href="<url-here>" title="">
            <img src="<image-url-here>" alt=""/>
        </a>
        <div class="card-content">
            <h4 class="card-title">
                <a href="<url-here>" title="">
                    Project/Article Title
                </a>
            </h4>
            <p class="">
                Project/Article description in short...
            </p>
        </div>
        <div class="card-read-more">
            <a href="<url-here>" class="btn btn-outline-info" title="">
                Read More
            </a>
        </div>
    </div>
</div>
<!-- END: Card - Career, beyond ladders… -->

Creating Portfolio Snippets with Python

Essentially, the Python code for this tool works by generating a portfolio card for each provided URL. It uses BeautifulSoup (to parse HTML), Requests (to make HTTP requests), and Jinja2 templates (to generate dynamic HTML) to create the portfolio snippets. You can install theses packages with pip:

pip install requests
pip install beautifulsoup4
pip install Jinja2

Take a look at this code snippet that extracts Open Graph tags from the web pages. It uses regex with BeautifulSoup to extract all the Open Graph tags from the HTML. The code further cleans them (removes og: part), and returns a simple key-value dictionary of these tags.

def _extract_og_data(self, url: str) -> dict:
    og_tags = dict()
    try:
        resp = requests.get(url)
        if 200 == resp.status_code:
            soup = BeautifulSoup(resp.text, 'html.parser')

            regex = re.compile('.*og:.*')
            og_data = soup.find_all("meta", {"property": regex, "content": True})

            for tag in og_data:
                og_index = tag['property'].index(":") + 1
                _key = tag['property'][og_index:]
                og_tags[_key] = tag['content']
        else:
            logging.warning(f"Invalid response form the site: {resp}")
    except Exception as ex:
        logging.exception(f"Exception: {ex}")
    finally:
        return og_tags

The following code snippet invokes the _extract_og_data(...) for a list of URLs and creates Open Graph data (a list of dictionaries) for creating the portfolio. The list of URLs is read from the text file: urls.txt.

site_cards = []
for url in urls:
    og_data = self._extract_og_data(url)
    site_cards.append(og_data)

Jinja2 Templates

The portfolio is actually created using a Jinja template. Jinja is a Python template engine that offers a cleaner way to make dynamic HTML pages or other text-based documents.

In the code snippet below, the Jinja template generates multiple portfolio cards by iterating over the Open Graph data — cards, and substitutes relevant data (title, image, description etc.) in the HTML template.

{% for og_card in cards %}
<!-- START: Card - {{og_card.title}} -->
<div class="col-xs-12 col-sm-4">
    <div class="card">
        <a class="img-card" href="{{og_card.url}}" title="{{og_card.title}}">
            <img src="{{og_card.image}}" alt="Featured Image"/>
        </a>
        <div class="card-content">
            <h4 class="card-title">
                <a href="{{og_card.url}}" title="More...">
                    {{ og_card.title|truncate(30, true) }}
                </a>
            </h4>
            <p class="">
                {{ og_card.description|truncate(160) }}
            </p>
        </div>
        <div class="card-read-more">
            <a href="{{og_card.url}}" class="btn btn-outline-info" title="{{og_card.title}}">
                Read More
            </a>
        </div>
    </div>
</div>
<!-- END: Card - {{og_card.title}} -->
{% endfor %}

Lastly, the portfolio generated by the template is saved in a new file. You may edit this file to use your own CSS theme. You can see one such generated portfolio here - The world of expressions.

Using Portfolio Maker

You can use this tool in two different modes via the command line:

Portfolio Mode: This mode generates a single portfolio page, complete with HTML and CSS, using the provided list of URLs.
Snippets Mode: In this mode, the tool creates Bootstrap cards. You can then incorporate this generated HTML into your existing portfolio. This mode is particularly useful when you want to seamlessly add more work samples to your current portfolio.

See this tool in action through the animated GIF below.

Conclusion

An anecdotal joke about developers goes like this: A developer takes 2-3 days to automate a task that could have been finished in 2-3 hours. Well, with this "portfolio-maker", I personify that imaginary developer. :D

However, now that I've built this tool, all I need to do is add new URLs to the urls.txt file and then just run this tool to create the portfolio in just a few seconds. I don't need to worry about putting the correct links, title, description, image etc.for every new article that I need to add there. ¯_(ツ)_/¯

Do you want to build a stunning portfolio to impress your potential clients?

Make use of this portfolio-maker, now offered for free on Gumroad! Plus, it comes with full code access and a helpful video tutorial. :)

True Lies of ChatGPT

Manish — Mon, 19 Jun 2023 04:36:00 +0000

When I say “true lies”, I mean that ChatGPT is blurring the lines between truth and fiction (lies). If not impossible, it is now extremely difficult to tell what is true and what is not from the text it generates. This can lead to a sense of confusion and uncertainty, as you are unsure what to believe.

Does ChatGPT lie?

Well, we already know that ChatGPT lies – and it is not rare either. In fact, it has been called out for lying multiple times, such as in this article and in this example. It probably doesn’t matter much when it comes to gathering information about movies, books etc. However, I was shocked by the convincing fake studies that ChatGPT provided me. They sounded so credible, I almost believed them…

Researching with ChatGPT

I used a materialized view and some indexes to make a complex query on a PostgreSQL database run almost 70 times faster. For a related presentation, I was searching for publicly available studies on how others have improved their query performance by using materialized views and indexes. Google search didn’t help me much, so I decided to ask ChatGPT for some help.

ChatGPT, being the helpful assistant it is, provided me with some “real-life” studies that showed impressive results using materialized views and indexes. However, much to my dismay, I discovered that all those studies were actually FAKE!

Here are the screenshots of my chat transcript. See for yourself how believable they all seem. I have also included those links after the screenshots – None of them exists though. ¯_(ツ)_/¯

Links provided for the examples –

Calling out lies

So, I called out the lies and asked ChatGPT again to provide me with some more credible references.

Result? Examine it yourself.

Links provided for the second set of credible examples –

GitLab: https://about.gitlab.com/blog/2020/12/03/gitlab-data-team-reduce-10tb-of-data-to-10gb/
TransferWise: https://databaseline.bitbucket.io/postgresql/2020/07/28/materialized-views-performance/
TimescaleDB: https://blog.timescale.com/blog/building-materialized-views-in-postgresql-for-fast-queries/

You can click on these links and check if any one of them actually exists. I even tried searching some text form those real-life examples, however, Google (or even Way Back Machine) knew nothing about these links. I have nothing more to add here.

Reasons for these lies

The Large Language Models (LLMs) such as ChatGPT are still evolving. Many important LLM behaviors emerge unpredictably. This emergence is the ability of LLMs to exhibit new and unexpected behaviors that are not explicitly programmed into them. At this stage, even the experts (including their creators) don’t fully understand how LLMs work. Furthermore, there are no reliable techniques for steering the behavior of LLMs. If you’re truly intrigued by this, please read the related paper by Sam Bowman, an expert on LLMs, titled – Eight Things to Know about Large Language Models.

As such LLMs like ChatGPT do not intentionally lie because they don’t possess consciousness or intentions. They may, however, generate incorrect or misleading information occasionally. This is clearly mentioned as one of the limitations of ChatGPT.

Think of ChatGPT as an exceptionally imaginative child who may occasionally make up believable stories while answering your questions.

Conclusion

ChatGPT can be used to generate the text that you want, but it is your responsibility to verify and validate the information you receive. Trust me on this one.😛

Huh? Does this mean that ChatGPT is not really useful?

Nah! Not at all.

ChatGPT is extremely useful when you need to come up with different ways of writing some text. It understands the tone (informal, funny, professional etc.) of the written text quite well and can help you to convert from one tone to another. It can explain complex concepts that can be easily understood by a 10-year old, or provide advanced details of that concept to an expert in the field. It can summarize large articles or books. It can even generate poems, limericks, emails, letters, etc. as per the given instructions. It is prudent to use ChatGPT for its strengths.

Generative AI, whether it is text or image, is getting more and more sophisticated with each passing day. However, it is now more crucial than ever to exercise critical thinking in evaluating the veracity of AI-generated content. I can vouch for this!

By the way, the featured image (liar ChatGPT/Pinocchio) used in this article is created by DALL-E (via Bing), using a prompt generated by ChatGPT itself. Isn’t that cool?

Disclaimer: This blog post is based on my experience with the free version of ChatGPT (3.5?) that was available in May 2023. I’ve been informed that GPT-4 is an enhanced product, and its behavior may differ from what is described here.

Why you should write constants on the LHS for comparisons?

Manish — Tue, 13 Jul 2021 13:00:49 +0000

I picked up this practice some 20+ years ago in my first job. It was part of the commandments for us newbie programmers back then - we mostly coded in C/C++ those days. We have come a long way since, but I'd argue that this practice could help you save from a blunder some day.

Let me just quote the commandment here -

Thou shall always put the literals/constants on the LHS and variable on the RHS in the if condition for equality!

Code samples:

if(true == deleteAll){
   // delete all records
}

if(0 == quantity){
   // report not available
}

The rationale here is simple – occasionally we programmers, lost in our efforts to solve the complex problems of universe, end up using a single assignment operator = instead of the comparison operator == unknowingly. In case of the languages where if condition does not only accept a boolean (like C, C++), this could result in unintentional assignment instead of comparison. Moreover, your compiler won’t be able to catch it for these languages (most IDEs these days warn about this though), or it will go unnoticed in scripts like JavaScript. However, when you use literal/constant on the LHS, even an accidental, unintentional assignment is not possible. i.e. you simply cannot have this, compiler will flag an error –

if(0 = quantity){
   // report not available
}

But the compiler, or your IDE might not flag an error for this -

if(quantity = 0){
   // report not available
}

..and that’s exactly the point. We let the compiler do the work of finding such bugs instead of realizing it at the runtime, and then eventually spending hours debugging and tracking down one small unintentional assignment, which should have been an equality comparison in the first place.

This approach still makes sense in 2021 including languages like Java, Kotlin (only boolean in if condition) etc. for the following reasons -

Consider boolean comparison itself in Java, what would happen here (ignoring compiler warning for assignment, among tons of other deprecation warnings) if the following code is unintentionally added?

if(deleteAll = true){
    //delete everything?
}

On the other hand, when you write code like this, the compiler would be able to help you, right? 😥

if(true = deleteAll){
    //delete everything?
}

More importantly, you seldom code in a single language as you grow as a software developer. There are times when you need to come up with a small snippet of JavaScript code (well, JS is ubiquitous - ain't disappearing for another two decades or more). All these things would be still valid there.

It is not about any one particular language/technology, it is more about developing better coding practices. These practices help us in the long run, and make us better programmers. 😃

DEV Community: Manish

Build a custom multi-step AI workflow graph with openJiuwen Agent Core

Environment setup

WorkflowCard and workflow initialization

Start and End node configuration

LLM component node setup

How to connect the graph and run the WorkflowAgent

Wrapping up

Frequently asked questions

How to keep an AI workflow predictable using WorkSwarm's SwarmFlow

Prerequisites and setup

How to define workflow stages in Python

How to add a human-in-the-loop checkpoint

How to set per-run and team token ceilings

How to monitor a run from its workflow.updated events

What to build next with SwarmFlow

Frequently asked questions

Debugging a silently failing AI workflow with Agent Core full-link observability

Why silent failures escape standard logs

Sample workflow setup

Full-link observability in Agent Core

How to read the trace output

Three targeted fixes from span evidence

Context truncation (token_count == 0, elevated latency)

Hallucinated tool arguments (token_count > 0 but output semantically empty, or argument schema errors in downstream spans)

Incorrect tool selection (wrong agent.tool.name in the span, output mismatches intent)

Verifying the pattern holds

Closing thoughts

Frequently asked questions (FAQs)

How to get multiple AI agents working together on one task: a WorkSwarm walkthrough

Prerequisites and installation

How WorkSwarm's coordination engineering model works

Leader agent configuration and task decomposition

META: descriptive, and it must be a literal

Schemas make each step's contract enforceable

run(): plain awaits, each feeding the next

Testing without a model

Choosing a model per step

Setting up providers

How a step picks its model

When to split the work across models

How to run the Swarm and read the monitoring output

Where to go from here

Frequently asked questions about WorkSwarm multi-agent pipelines

Is WorkSwarm the same project as JiuwenSwarm?

How does a SwarmFlow script get into a Team session?

How do I reuse a SwarmFlow script across projects?

Can agent calls run in parallel?

How do I add a human approval step between phases?

How do I cap token spend for a Swarm?

How do I pause or stop a run that's already going?

Integrating Claude Code with Auth0 APIs

How Claude Code and Auth0 MCP Work Together

Setting Up the Auth0 MCP Server with Scoped Access

Prerequisites

Install and initialize

Why use device authorization flow instead of static secrets

Scoping the MCP server

Claude Code and Auth0 integration

Verify your integration

Giving Claude Code Project Context with CLAUDE.md

Refactoring an Auth0 Integration from Claude CLI

Auth0 setup and protected API

Obtaining an Auth0 token

Calling your protected API

Running the audit prompt

Running the provision prompt

Running the implementation prompt

Observations: Claude Code with Auth0 MCP server

Security Guardrails

Code and Auth, in Sync

Do Not Let Your AI Go Rogue, Guard Against Agentic Misalignment

The Anatomy of Agentic Misalignment

Agents exhibit goal-seeking behavior

Concerning power-seeking tendencies

Identifying Warning Signs of Misalignment

A decision matrix for early intervention

Guardrail Layer 1: Constitutional and Infrastructural Constraints

Start with the minimum footprint

Encode constitutional rules in your systems

Context truncation (`token_count == 0`, elevated latency)

Hallucinated tool arguments (`token_count > 0` but output semantically empty, or argument schema errors in downstream spans)

Incorrect tool selection (wrong `agent.tool.name` in the span, output mismatches intent)