shipit-watcher 1.0.0

Observability for LLM applications — tracing, prompt governance, cost allocation and PII masking.

Observability for LLM applications.

Tracing · Agent graphs · Prompt governance · Cost allocation · PII masking

Python 3.11+ · zero required dependencies · framework-agnostic · 246 tests

pip install shipit-watcher

---

One coherent record of what an AI system did: which prompt ran, what it cost, which tenant it belonged to, which tools it called, what it retrieved — and why it chose what it chose.

---

Contents

• The problem
• Install
• Quick start
• Core concepts
• Prompt identity
• Prompt registry
• Agent graphs
• PII masking
• The event model
• Sinks
• The local ledger
• Reporting
• Configuration
• Architecture
• Design rules
• Testing
• Integration guide

---

The problem

A typical LLM stack records LLM calls and nothing else. Ours recorded them three times:

litellm-acompletion   407 → 476   $0.001312   {}
litellm-completion    407 → 476   $0.001312   {}     ← same call
OpenAI-generation                             {}     ← same call again

Same tokens, same cost, three entries, no metadata, no tool calls.

The cause is double instrumentation. litellm.success_callback = ["langfuse"] makes LiteLLM open its own trace, the application opens another, and a proxy-side callback opens a third. None of them carries the tenant, the cost centre, or which prompt produced the answer.

Watcher settles the ownership question — exactly one component traces — then adds the dimensions that let a trace answer a business question.

---

Install

pip install shipit-watcher[all]     # langfuse + litellm + django
pip install shipit-watcher          # core only, no dependencies

Every integration degrades to a no-op when its library is absent, so the core is safe to import anywhere.

---

Quick start

import shipit_watcher as wt

wt.configure(service_name="fleetflow", environment="production")
wt.instrument_litellm()          # one trace per call, not three

with wt.trace("chat.request",
              company_id=str(company.id),
              user_id=str(user.id),
              cost_center="fleet-ops"):

    with wt.get_tracer().tool("df_list_cars") as tool:
        tool.output = list_cars(company)

    wt.get_tracer().decision(
        "route.expert",
        chosen="fuel-analysis",
        options=["fuel-analysis", "driver-communication"],
        rationale="query mentions fuel consumption",
        confidence=0.87,
    )

Everything inside the block attaches to the trace automatically. Nothing needs a trace_id parameter threaded through it.

Decorators

@wt.observe_agent("planner")       # opens a root trace
async def run_agent(query: str): ...

@wt.observe_tool("search_fleet")   # records a tool invocation
def search_fleet(q: str): ...

@wt.observe("summarise")           # a plain span
def summarise(text: str): ...

Sync, async, generators and async generators are all handled. Generators are consumed inside the span — a decorator that returned the generator object unconsumed would record a 0 ms span and never see the real work or its errors.

---

Core concepts

Concept	What it is
Trace	One unit of work — typically one user request
Event	One observation inside a trace: a span, decision, tool call, retrieval
Context	Ambient tenant / user / cost-centre, propagated via contextvars
Sink	A destination: Langfuse, your database, the console
Prompt identity	Name + version + content fingerprint, carried on every call

Context propagates through contextvars, so it follows the logical flow of execution across await boundaries and into tasks — without being global state shared between concurrent requests.

with wt.bind(company_id="acme", cost_center="fleet-ops"):
    ...    # every event here carries both

---

Prompt identity

"Prompt identity carried in every call — precondition for enforce, nothing else works without it." — Shipit Watcher requirements

Without it a trace can say which model ran and what it cost, but not which prompt version produced this answer — so governance, replay and "why this answer" have nothing to hang off.

prompt = wt.identify_prompt(
    agent.system_prompt,
    name=f"agent:{agent.slug}",
    version="v3",
    registered=True,          # came from a managed registry
)

Field	Meaning
name	Where the prompt came from — agent:inbox-manager
version	Registry version, when one exists
fingerprint	SHA-256 over normalised text — changes exactly when the prompt does
registered	Defaults to False — the gap report is only useful if its default is honest

The fingerprint makes a compliance gap report possible on day one, with no registry: group by fingerprint, and any prompt with no registry entry is by definition unregistered.

---

Prompt registry

Fingerprints tell you which prompt ran. The registry decides what runs — so a prompt change becomes a release someone can review, label and roll back, instead of a diff buried in a Python string.

The whole turn, in one call

answer = wt.run_prompt(
    "fleet-assistant",
    variables={"company": "VivaDrive", "vehicles": 142, "drivers": 100},
    user_message="How many cars do we have?",
)

answer.text            # the answer
answer.total_cost      # what it cost
answer.total_tokens    # 467 → 121

run_prompt resolves the live prompt, compiles it, takes model, temperature and max_tokens from the prompt's own config, calls the model, links the generation to that exact version, and records all of it. Credentials default to LITELLM_API_BASE / LITELLM_API_KEY, so the common case passes neither.

Everything below is that call taken apart, for when you need the pieces.

One prompt per agent

An application with several agents should not share one prompt namespace. Prompts are keyed agent/<slug>, and the slug is derived from the agent so "FleetFlow Assistant" and "fleetflow-assistant" can never resolve to two different prompts:

prompt = wt.get_agent_prompt(agent, fallback=agent.system_prompt)
system = prompt.compile(company=company.name, vehicles=142)

Passing the agent's own system_prompt as fallback makes adoption incremental — agents with a registry entry are managed from Langfuse, agents without one keep working exactly as before, and prompt.registered tells the compliance report which is which.

wt.agent_prompt_name("Fuel Expert")       # → "agent/fuel-expert"

Attributing calls you cannot reach

A litellm.completion three frames deep inside a tool has no way to pass a prompt identity. Bind it to the scope instead:

with wt.use_prompt(prompt):
    ...                       # every LLM call in here is attributed

Without this, prompt attribution only covers the call sites you remembered to annotate — which is exactly the gap governance cannot have. Works for LLMClient and for direct litellm calls under instrument_litellm().

Fetch the latest stable prompt

This is the call an agent makes on every turn:

import shipit_watcher as wt

prompt = wt.get_prompt("fleet-assistant", fallback=LOCAL_DEFAULT)

system = prompt.compile(
    company="VivaDrive",
    vehicles=142,
    drivers=100,
    language="English",
)

get_prompt returns the version currently labelled production — not the newest version. Publishing and releasing are separate acts: a new version goes live only when the production label moves onto it, which you do from the Langfuse UI or from code, with no deploy.

Argument	Default	What it selects
label	"production"	The deployment channel. "staging" to try one first.
version	–	An exact version. Pins a run; overrides label.
fallback	–	Template used if the prompt cannot be resolved at all.

Why it does not fail your request

Prompts sit on the critical path of every turn, so resolution degrades in steps rather than raising:

fresh cache  →  registry  →  stale cache  →  fallback
   0.01ms        ~30ms       last good      your string

The stale rung is the one that earns its keep. If Langfuse is unreachable the agent keeps answering with the last prompt it successfully fetched, marked prompt.stale is True so a dashboard can show the degradation instead of the outage hiding.

Fallbacks are cached too. A prompt missing from the registry is missing on every request, so without caching each one pays a round-trip and a 404 to learn the same thing.

Wire it into an agent

The identity travels with the call, so every generation in Langfuse says which prompt version produced it — and config lets the prompt carry its own model settings, so tuning temperature is also a registry change, not a deploy:

prompt = wt.get_prompt("fleet-assistant", fallback=LOCAL_DEFAULT)

client = wt.LLMClient(
    model=prompt.config.get("model", "gemini-2.5-pro"),
    api_base=os.environ["LITELLM_API_BASE"],
    api_key=os.environ["LITELLM_API_KEY"],
    temperature=prompt.config.get("temperature", 0.2),
)

with wt.trace("agent.turn", user_id=user.email, session_id=session.id) as ctx:
    answer = client.complete(
        [{"role": "system", "content": prompt.compile(**variables)},
         {"role": "user", "content": question}],
        prompt=prompt.identity,        # ← links the generation to the version
    )
    ctx.set_output({"answer": answer.text})

With WATCHER_GOVERNANCE=enforce, a prompt that did not come from the registry is refused before the call is made — see Configuration.

Publish a version

wt.create_prompt(
    "fleet-assistant",
    "You are {{company}}'s fleet assistant. Fleet: {{vehicles}} vehicles.",
    labels=["production"],                 # omit to stage without releasing
    tags=["fleetflow", "agent"],
    config={"model": "gemini-2.5-pro", "temperature": 0.2, "max_tokens": 2000},
    commit_message="Tighten citation rule",
)

Langfuse versions by name — this never overwrites, it appends version n+1. Publishing with labels=[] stages the prompt for review; moving the production label is the release.

Unlike get_prompt, this raises on failure. A write that did not land is a release that did not happen, and a release script must not report success having changed nothing.

Chat prompts

Pass a message list and the type is inferred:

wt.create_prompt("triage", [
    {"role": "system", "content": "You triage fleet incidents."},
    {"role": "user", "content": "{{incident}}"},
])

They fingerprint like text prompts — the messages are joined before hashing — so a chat prompt is just as traceable as a string one.

---

Agent graphs

Langfuse draws a graph of a trace only when its observations carry a semantic type — agent, tool, retriever, embedding, guardrail, chain, evaluator. A trace of undifferentiated spans renders as a list, because nothing in it says which box is an agent and which is a tool it called.

Turn it on:

export WATCHER_LANGFUSE_TRANSPORT=otlp

Then write the turn as you would anyway — the types come from which helper you reach for, not from extra arguments:

with wt.trace("fleetflow.agent.turn", user_id=user.email, session_id=sid) as ctx:
    with wt.tool("df_list_cars") as t:          # → tool node
        t.output = runner.run("df_list_cars")

    wt.retrieval("kb.search", query=q, chunks=chunks)   # → retriever node
    answer = client.complete(messages, prompt=prompt.identity)   # → generation
    ctx.set_output({"answer": answer.text})

renders as:

[AGENT]      fleetflow.agent.turn      4.20s
  ├ [TOOL]       df_list_cars            0.55s
  ├ [TOOL]       df_list_drivers         1.40s
  └ [GENERATION] llm.fleet_assistant     1.22s   571 tok   $0.000400

Watcher call	Langfuse node
wt.trace(...)	agent (the graph's entry point)
wt.tool(...)	tool
wt.retrieval(...)	retriever
wt.generation(...) / LLMClient	generation
wt.policy(...)	guardrail
wt.decision(...)	chain
wt.handoff(...)	agent
wt.span(...)	span — no graph node, by design

Why a separate transport

Semantic types cannot be sent over the classic ingestion API. Offered one, a Langfuse v3 server replies:

"Invalid option: expected one of \"GENERATION\"|\"SPAN\"|\"EVENT\""

They exist only over OTLP, as the span attribute langfuse.observation.type. The Langfuse Python SDK exposes this as as_type= from 3.3.1 — but v3 also removed client.trace(), which most existing integrations call. So this sink speaks OTLP directly over plain HTTP, with no OpenTelemetry dependency and no SDK upgrade: only the server has to be v3.

Requirements: a Langfuse server ≥ 3.x. Check yours with curl $LANGFUSE_HOST/api/public/health.

sdk vs otlp

	sdk (default)	otlp
Server needed	any	v3+
Agent graph	✗	✓
Observation types	span / generation	all ten
Delivery	Langfuse client's own batching	one request per finished trace

Both carry user, session, tags, tokens, cost and prompt version. The only difference is the graph — so sdk stays the default, and otlp is a one-variable upgrade when your server supports it.

---

PII masking

Applied before anything is persisted or leaves the process. Masking at display time is theatre once the raw value is on someone else's infrastructure.

wt.mask_text("Driver Jan Kowalski PESEL 44051401359, [e-mail]")
# 'Driver Jan Kowalski PESEL [PESEL], [EMAIL]'

wt.mask_text("Odometer 1234567890 km")
# unchanged — ten digits, but not a valid NIP

Polish identifiers are first-class and checksum-validated:

Detector	Validation
PESEL	weights 1,3,7,9 · complement mod 10
NIP	weights 6,5,7,2,3,4,5,6,7 · mod 11
REGON	9- and 14-digit variants
IBAN	ISO 13616 mod-97
Card	Luhn
Email / Phone / IP	pattern, digit-boundary anchored

Checksums are the point. A fleet database is full of ten-digit numbers that are not tax IDs. Validating the check digit is what stops this redacting the data the traces exist to explain.

wt.configure(mask_pii=True)                       # default
policy = wt.MaskingPolicy(enabled_rules=frozenset({"EMAIL"}))   # relax explicitly

---

The event model

Typed, because a decision path cannot be rendered from span(name="something").

Event	Carries
GenerationEvent	model, provider, tokens, cost, prompt identity
DecisionEvent	chosen, options_considered, rationale, confidence
ToolInvocationEvent	tool name, arguments, success, error
RetrievalEvent	query, knowledge base, chunks with provenance
HandoffEvent	from_agent → to_agent, reason
PolicyEvent	policy name, blocked, reason
HumanReviewEvent	reviewer, verdict, comment

Two fields do the heavy lifting:

options_considered — recorded at the moment of choosing, "why not the other option" is answerable. Reconstructed afterwards, it is a guess.

RetrievedChunk.content_hash — a citation without one cannot prove the source said what the answer claims. The document may have changed since.

tracer.retrieval("kb.policy", query="fuel policy",
    chunks=[wt.RetrievedChunk(source="policy_2026.pdf", score=0.93,
                              version="4", content_hash="9f2a1b")])

---

Sinks

Sink	Purpose
LangfuseSink	analysis surface — supports both v2 and v3 client shapes
DjangoSink	the local ledger — retention, MPK reporting, your boundary
ConsoleSink	development

FanOutSink isolates them: if Langfuse is unreachable the database row is still written, and vice versa. A sink that raises is logged once and skipped, never retried in-line — that would put a failing backend on the user's critical path.

Adding ClickHouse later is a new file, not a change to any call site.

---

The local ledger

Langfuse is where you look at traces. The database is where they are kept.

LLMCallRecord — the cost ledger

One row per generation: tenant, cost centre, model, tokens, cost, prompt identity, latency, and the trace_id that joins back to Langfuse.

Enable it with WATCHER_PERSIST_DB=true (Django apps only — the sink is imported lazily, so nothing here loads in a process without an ORM).

User identity is not assumed to be a primary key. user is a FK and resolves only when the value is a pk; user_ref always holds the raw identifier you passed, whether that is a UUID, an email, or an SSO subject. Both are indexed. Without this split, Django rejects the entire row for a non-UUID user — so tracing by email lost the record altogether, with only a warning in the log:

with wt.trace("turn", user_id="[e-mail]", session_id=session.id):
    ...
# LLMCallRecord(user=None, user_ref="[e-mail]", session_id=…)

session_id is stored verbatim, so a conversation in your database and a session in Langfuse are the same string and join without a mapping table.

TraceEventRecord — the full tree (opt-in)

Every event including parent/child edges, so the decision path is reconstructable from your own database:

wt.configure(persist_to_database=True, persist_all_events=True)

├─ span         planning
│  ├─ decision      route.expert   → fuel-analysis  ✗['driver-communication']
├─ span         execution
│  ├─ tool          df_list_cars
│  ├─ retrieval     kb.policy      → policy.pdf #9f2a1b
│  ├─ generation    llm.completion → 883 tok $0.001312
├─ span         validation
│  ├─ policy        pii_masking

Children finish before their parents, so the parent FK is usually null at insert time. parent_event_id always records the edge; resolve the relations once the trace completes:

DjangoSink.stitch_parents(trace_id)

Volume is real — one agent turn emits a dozen events — so persist_all_events is off by default. Enable it for the systems under audit.

---

Reporting

Because generations land in your own table, the reports are SQL:

# Spend per MPK cost centre
LLMCallRecord.objects.values('cost_center').annotate(
    calls=Count('id'), tokens=Sum('total_tokens'), cost=Sum('total_cost'))

# Spend per tenant
LLMCallRecord.objects.values('company__name').annotate(cost=Sum('total_cost'))

# Prompt compliance gap
LLMCallRecord.objects.filter(prompt_registered=False).values(
    'prompt_name', 'prompt_fingerprint').annotate(n=Count('id'))

# Every decision this tenant made
TraceEventRecord.objects.filter(company=company, event_type='decision')

---

Configuration

Environment first, configure() overrides, safe defaults throughout.

The three you actually have to set

export LANGFUSE_PUBLIC_KEY=pk-lf-…
export LANGFUSE_SECRET_KEY=sk-lf-…
export LANGFUSE_HOST=https://langfuse.digitalfleet.eu   # your instance

LANGFUSE_HOST is the base URL, and it is not optional for self-hosting: it defaults to Langfuse Cloud, so leaving it unset silently ships your traces to cloud.langfuse.com — where the keys do not work and nothing appears. The keys are two rather than one because Langfuse itself issues them that way: the public key identifies the project, the secret key authorises the write, and they are sent as an HTTP Basic pair.

Everything below has a working default. Nothing else is required to start.

Variable	Default	Meaning
WATCHER_LANGFUSE_TRANSPORT	sdk	otlp for the agent graph
WATCHER_SERVICE	unknown-service	tags every trace
WATCHER_ENV	development	environment
WATCHER_ENABLED	true	master switch
WATCHER_MASK_PII	true	redact before persistence
WATCHER_CAPTURE_CONTENT	true	false = metrics-only traces
WATCHER_SAMPLE_RATE	1.0	fraction of traces kept
WATCHER_PERSIST_DB	false	enable the local ledger
WATCHER_PERSIST_ALL_EVENTS	false	persist the whole tree
WATCHER_GOVERNANCE	audit	audit / warn / enforce
LANGFUSE_PUBLIC_KEY LANGFUSE_SECRET_KEY LANGFUSE_HOST	—	Langfuse

---

Architecture

        your application
               │
               ▼
    ┌──────────────────────┐
    │       Tracer         │  ← contextvars: tenant, user, MPK
    └──────────┬───────────┘
               │  typed events
               ▼
    ┌──────────────────────┐
    │   masking (PII)      │  ← before anything leaves the process
    └──────────┬───────────┘
               ▼
    ┌──────────────────────┐
    │     FanOutSink       │  ← failures isolated per sink
    └───┬──────────┬───────┘
        ▼          ▼
   Langfuse    your database
   (analysis)  (system of record)

LiteLLM calls report into the ambient trace rather than opening their own, which is what removes the duplicates.

---

Design rules

1. Observability never breaks the request it observes. Every entry point swallows its own failures. Failing to record is a monitoring incident; failing a user's request because recording broke is a worse one.
2. Exactly one component owns tracing. instrument_litellm() removes LiteLLM's Langfuse callback. Keep both and the duplicates return.
3. PII is masked before persistence, never at display time.
4. Fail closed on privacy, open on availability. A masking error redacts the whole value; a sink error is dropped and logged.
5. Exceptions propagate unchanged. The tracer marks the span and re-raises.
6. Sampling never drops errors. A 10% sample that also discards 90% of failures is useless precisely when it is needed.

---

Testing

pytest shipit_watcher/tests -q --cov=shipit_watcher --cov-report=term-missing

148 tests · 90% coverage.

The negative cases carry as much weight as the positive ones. Over-masking destroys the data traces exist to explain, so "an odometer reading survives untouched" is as much a requirement as "a PESEL is redacted".

Bugs the suite caught during development, rather than assumptions that shipped:

• a PHONE pattern matching inside a ten-digit number
• REGON alternation precedence
• a double-yield in trace() that replaced the caller's exception with "generator didn't stop after throw()"
• a public tracer() helper shadowing the shipit_watcher.tracer submodule
• typed events not inheriting their parent, flattening the decision path

---

Integration guide

# settings.py or AppConfig.ready()
import shipit_watcher as wt

wt.configure(
    service_name="fleetflow",
    environment=os.getenv("ENV", "development"),
    persist_to_database=True,
)
wt.instrument_litellm()

Then wrap your entry point:

with wt.trace("chat.request",
              company_id=str(company.id),
              user_id=str(user.id),
              session_id=str(session.id),
              cost_center=company.cost_center):
    ...

⚠️ Do not run both. instrument_litellm() replaces LiteLLM's native Langfuse callback. To keep LiteLLM's own traces instead, pass replace_langfuse_callback=False — but then do not create application traces as well, or you are back to duplicates.

Requirements coverage

Module	Capability	Status
B	Filter by system, model, user, prompt, date, MPK	✅
C	Prompt identity in every call	✅
C	Compliance gap report	✅
C	Block unregistered prompts (enforce)	⚠️ flag present, blocking not implemented
D	Event model + retrieval provenance	✅
D	Decision path, why this / why not	✅ data model — UI still to build
E	MPK tagging and allocation	✅ tagging — hierarchy and rules still custom
E	Budgets and alert ladder	❌ not started
G	PII masking before persistence	✅

---

_{Built for FleetFlow · reusable anywhere}