Guide

Quick Start

Five steps to attach; one more to gate CI. Heuristics work without login or an API key.

After setup

CLI Commands

All commands available from your terminal after installing ARGUS.

Setup

argus init

Writes .cursor/skills/argus-debug/SKILL.md and .claude/skills/argus-debug/SKILL.md. The skill already contains the setup prompt. Safe to re-run (skips unchanged files; pass --force to overwrite). Commit them so later chats can attach ARGUS and read .argus/runs JSON instead of guessing from logs.

Viewing runs

argus list

argus show last

argus fix <run-id>

argus show <run-id>

argus inspect <run-id> --step <node-name>

CI & testing

argus check last

argus check <run-id>

Exit code 0 when the run is clean; 1 on crash, silent failure, semantic fail, missing fields, or tool failures. Use after python my_agent.py in GitHub Actions or any CI job.

pytest --argus

Auto-wraps StateGraph.compile() for the test session. Clean pipelines stay passing tests; missing fields, tool failures, crashes, and semantic degradation fail that test. Tests that never invoke a graph are unchanged. Heuristics only (judge off).

Dashboard

argus ui

Starts a local server on port 7842 and opens the dashboard in your browser. Press Ctrl+C in the terminal to stop it. If the runs table is empty, ARGUS shows which .argus/runs path it is serving — run the graph first, try argus show last, or start the UI from the project root (cwd vs git / pyproject / $ARGUS_DIR).

Replay & compare

argus replay <run-id> <node-name>

argus replay <run-id> <node-name> --app my_pipeline:build_graph

argus replay <run-id> <node-name> --only

argus diff <run-id-a> <run-id-b>

Account & diagnostics

argus login

argus logout

argus whoami

argus doctor

argus update

AI Integration Prompt

After argus init, the skill already contains this AI setup prompt — asking your editor agent to wire ARGUS is enough. The copy below is just a fallback if you still want to paste a one-shot.

I want to add ARGUS monitoring to my LangGraph pipeline. Attach it with a small diff. Do not rewrite my state type or node signatures.

Heuristics, tool-failure scan, placeholders, empty outputs, and crashes work without TypedDict. Do not convert plain-dict state to TypedDict. Do not change node return shapes — returning {**state, ...} is fine. Type hints can be suggested after the first run; they are not a setup gate.

## STEP 1 — FIND THE GRAPH

Find the file where my StateGraph is defined (or the already-compiled app). Note whether nodes are sync or async. Linear, fan-out/fan-in, and cyclic graphs all persist automatically after the outermost invoke()/batch()/stream() returns — no finalize() call needed.

Print a short summary, then integrate. Do not "fix compatibility" by rewriting types first.

## STEP 2 — INTEGRATE ARGUS

Install: pip install argus-agents
(The PyPI package is argus-agents, not argus. Default install includes the CLI, LangGraph adapter, and UI. LLM judge is off by default — heuristics only. Optional later: argus key set, then pass semantic_judge=True.)

Also run: argus init
This writes project skills for Cursor and Claude:
  .cursor/skills/argus-debug/SKILL.md
  .claude/skills/argus-debug/SKILL.md
Commit them. Later chats will read .argus/runs JSON instead of guessing from logs.

Add ArgusRecorder to the file where the graph is built. Keep my existing state and node functions as-is:

from argus import ArgusRecorder

recorder = ArgusRecorder()
app = recorder.attach(compiled_graph)  # returns the app to invoke — nothing is patched
result = app.invoke(initial_state)     # run persists automatically
print(recorder.run_ids[-1])

ArgusRecorder listens to LangGraph callbacks and keeps the dict each node returned — its update, before LangGraph merges it into shared state. That is what makes a node returning {} visible at all. One attach() serves many runs: each invoke() and each .batch() item gets its own run file.

If node functions are async, use await app.ainvoke().

A recording carries state, not code, so it cannot know which node reads a field written three steps earlier. If a later node depends on a specific field, declare it:

recorder = ArgusRecorder(consumers={"audience": ["write"]})

Do not use ArgusWatcher for new code. It patches compile() and rebinds invoke/stream/batch; it is still supported but is not being extended.

## STEP 3 — OPTIONAL CONFIG

Defaults are heuristics-first: semantic_judge is off. record_http and persist_state are on. Do not enable semantic_judge unless I already ran argus key set.

Only add extra kwargs if needed (redact_keys, validators, strict=True). Do not add a large config block by default.

After running the pipeline:
  argus show last         # first aha is in the terminal if something is wrong
  argus list              # see all recorded runs
  argus show <id>         # inspect a specific run by ID
  argus check last        # CI gate — exit 1 on crash / silent failure / semantic fail
  argus ui                # open the web dashboard (empty table = wrong dir or no runs yet)

For pytest, add --argus so silent failures fail the test (no recorder in the test file required):
  pytest --argus

If a node calls a tool directly (tool.invoke(args, config=config)) rather than through the graph, the callback path can miss it under langgraph 1.x. File it explicitly so the tool scan still sees it:

from argus import report_tool_call
report_tool_call("search", input=q, output=hits)   # or error=exc on failure

After the first run, the dashboard may suggest type hints to catch field-drop bugs. That is optional follow-up, not part of this integration.

Runs List

Your pipeline execution history. Every run with ARGUS attached shows up here automatically. An empty table means this UI is reading a different .argus than the project that just ran, or you have not invoked the graph yet — use argus show lastand check cwd vs project root.

Summary cards

Table columns

The Evaluation panel lets you filter runs by constraints like overall_status == clean.

Run Detail

Full picture of a single pipeline execution — metrics, execution trace, AI analysis, and initial state.

Metrics

Execution timeline

Nodes listed in execution order with name, output type, duration, and status. Failed nodes show a root cause annotation — which field was missing and which upstream node dropped it. Expand any row to see full I/O JSON.

Node statuses

AI Analysis

When a provider key is set (OpenAI, Anthropic, or Google — via argus key set), ARGUS investigates non-clean runs automatically. The analysis panel has three sections:

Compare

Side-by-side diff of two runs. Useful for verifying fixes, catching regressions, or understanding performance differences.

Approvals

Gate deployments on ARGUS results. Runs that meet your criteria get approved; everything else is held for review.

Rerun

Re-execute from a specific node using the frozen input state from a previous run. Test a fix without re-running the full pipeline or making upstream LLM calls. Use record_http=True for fully deterministic reruns from disk.

From the CLI

argus replay <run-id> <node-name>

argus replay <run-id> <node-name> --app my_pipeline:build_graph

argus diff <original-run-id> <rerun-run-id>

Configuration Reference

ArgusRecorder is the path under active development — every parameter is keyword-only. It deliberately takes fewer knobs than the legacy watcher: it listens to callbacks instead of patching the graph, so there is no HTTP cassette and no recompile to configure.

recorder = ArgusRecorder(
    # --- Contract: who reads what ---
    consumers={
        # A trace carries state, not code, so it cannot know "write" needs the
        # field "plan" wrote. Declared here, a field never written, written
        # empty, or dropped in between fails on the node responsible — not on
        # the node that noticed.
        "audience": ["write"],

        # Dotted paths reach nested state. A top-level declaration means the
        # whole value, so {"email": {"subject": "...", "body": ""}} is a
        # non-empty "email" — blanking a leaf needs the leaf named.
        "email.body": ["send_email"],

        # Presence-only: issues: [] IS the LGTM, hits: [] IS a clean screen.
        # Also softens empty retrieval in that node's TOOL responses, not just
        # its own update. A tool that raised or a 4xx body still fails hard;
        # undeclared empty lists stay critical (the RAG default).
        "hits": {"readers": ["decide"], "allow_empty": True},
    },

    # --- Detection strictness ---
    strict=False,           # extra checks: nested error keys, rate-limit responses,
                            # empty lists, type mismatches. recommended for CI/staging.

    # --- Semantic validators ---
    validators={
        "summarize": lambda o: (len(o.get("summary","")) > 10, "Summary too short"),
        "*": lambda o: ("error" not in o, "error key present"),  # runs on every node
    },

    # --- LLM semantic judge ---
    semantic_judge=None,    # None = auto: on when a provider key is available, off
                            # otherwise. False keeps the gate fully deterministic.

    # --- Output control ---
    max_field_size=50_000,  # max chars per field before truncation (default: 50k)
)

app = recorder.attach(compiled_graph)   # returns the app you invoke; nothing is patched
result = app.invoke(initial_state)

Read recorder.run_ids after the run — one attach() serves many runs, so it is a list. recorder.session is the most recent one.

report_tool_call

Under langgraph 1.x, code inside a node function can hold an empty callback manager, so a tool the node invokes directly never fires on_tool_start and vanishes from the step's tool list — the tool-failure scan then has nothing to grade. File it yourself:

from argus import report_tool_call

def fetch(state):
    try:
        hits = search_tool.invoke(state["query"])
    except Exception as exc:
        report_tool_call("search", input=state["query"], error=exc)
        raise
    report_tool_call("search", input=state["query"], output=hits)
    return {"hits": hits}

Files the call in the same shape the callback path uses, onto the node step currently open, so the graders cannot tell the difference. It resolves the active recorder itself; pass recorder= or config= to be explicit. It never raises into your code — a dropped call logs one warning on the argus logger and returns False.

ArgusWatcher (legacy)

The older wrap path: it patches compile() and rebinds the runtime methods. Still supported, not being extended — the parameters below are watcher-only. Graph is the only positional argument.

watcher = ArgusWatcher(
    graph,                  # uncompiled StateGraph — or omit and call attach()

    # --- Output control ---
    max_field_size=50_000,  # max chars per field before truncation (default: 50k)
    redact_keys={"token", "api_key"},  # field names to scrub from stored outputs
    persist_state=True,     # save run records to .argus/runs/ (default: True)

    # --- Detection strictness ---
    strict=True,            # extra checks: nested error keys, rate-limit responses,
                            # empty lists, type mismatches. recommended for CI/staging.

    # --- Semantic validators ---
    validators={
        "summarize": lambda o: (len(o.get("summary","")) > 10, "Summary too short"),
        "*": lambda o: ("error" not in o, "error key present"),  # runs on every node
    },

    # --- LLM investigation ---
    investigate=True,       # LLM root-cause analysis on failures (default: True)
                            # set to "always" for every node, False to disable

    # --- Deterministic rerun ---
    record_http=True,       # saves every outbound API call to disk. (default: True)
                            # reruns replay from disk — zero extra cost.

    # --- LLM semantic judge ---
    semantic_judge=True,    # review soft rule flags (placeholders, refusal-like text).
                            # on when a key is set; False keeps the gate deterministic.
    judge_model="gpt-4o",  # tier hint: capable model. Auto-mapped to your active
                            # provider (Claude/Gemini). "gpt-4o-mini" = cheaper tier.

    # --- Latency thresholds ---
    config=ArgusConfig(
        node_timeout_ms=30_000,  # flag nodes that take >=95% of this (likely truncated)
        min_expected_ms=500,     # flag LLM nodes completing faster (likely cached/stale)
    ),
)

app = watcher.attach(graph)
result = app.invoke(initial_state)

Access watcher.run_id after the run. record_http,investigate, and persist_state default to True.semantic_judge turns on when a provider key is set; pass False for a fully deterministic gate.

record_http

Captures every HTTP request/response during the original run (urllib3 for requests, httpcore for httpx). On rerun, serves recorded responses back — same data, zero cost, fully reproducible. A session that captures nothing logs a warning instead of writing a silent empty cassette.

Enable when nodes call paid APIs and you want cheap, identical reruns. Skip when you want the rerun to hit the real API.

semantic_judge

The rules fail the build. The judge reviews the soft flags they raised (warning-level semantic signals, including shape warnings). It does not walk a clean graph looking for hallucinations, and it cannot clear a hard fail.

Requires a provider key (OpenAI, Anthropic, or Google) — set via argus key set. Enable for complex multi-agent pipelines. Skip for simple pipelines or zero-cost monitoring.

Latency Thresholds

Detects timing-correlated degradation — no LLM calls, purely algorithmic. Pass thresholds via ArgusConfig:

Both thresholds are optional and None by default — latency checks only run when configured.