tvastar 0.19.0

Tvastar — a programmable agent harness framework for Python. Agent = Model + Harness.

Tvastar

The agent harness that catches when AI agents lie about success.

# pip install tvastar
from tvastar import Harness, create_agent, default_toolset
from tvastar.model import AnthropicModel

agent = create_agent(
    "my-agent",
    model=AnthropicModel("claude-sonnet-4-6"),
    tools=default_toolset(),
)
result = await Harness(agent).run("Fix the failing tests")

print(result.quality.grade)    # "FAIL"
print(result.quality.summary)  # "agent claimed success but last tool shows failure"
print(result.ok)               # False — Tvastar caught the lie

Benchmark: 3,651 failed agent trajectories from tau2-bench. Tvastar detected 100%. Traditional monitoring detected 0%. Details →

---

Table of Contents

• Why Tvastar
• Loop in 60 seconds
• What is a harness?
• The five problems Tvastar solves
• Works with any agent or model
• See it in action: tvastar-fix
• When not to use Tvastar
• What Tvastar handles so you do not have to
• How it works
• Install
• Environment variables
• Core concepts
• Tools
• Sessions
• Structured output
• Delegating to specialist sub-agents
• Parallel fan-out
• Auto-topology
• DAG task execution
• Loop Engineering
• Loop Quality
• Plug into anything
• Verifiable Execution
• Extended thinking
• Workflows
• Event-driven dispatch
• Context compaction
• Application-level file access
• Sandboxes
• MCP — use any published tool server
• Durable execution
• Serving over HTTP
• Observability
• Trace viewer UI
• Tool masking
• Silent-failure detection
• Benchmark results
• Untrusted content & prompt-injection detection
• Dynamic Capability Governance
• Transactional Sandbox
• Long-Term Memory
• CLI
• Deploy anywhere
• Custom model adapter
• Evals
• Benchmarks
• Human-in-the-loop
• Cost tracking
• What we're building
• Roadmap
• Testing
• Troubleshooting
• Further reading

---

Why Tvastar

Your agents complete tasks they didn't actually finish. They loop forever without telling you. They swallow errors and say "done." Tvastar detects this — automatically, in any loop.

result = await harness.run("fix the failing tests")
print(result.quality.score)    # 40
print(result.quality.grade)    # "FAIL"
print(result.quality.summary)  # "1 error — final answer claims success but the last tool result shows failure"

pip install tvastar
# or: tvastar quality my_agent.py:agent "fix the tests"  → score 0–100, exit 1 if FAIL

Agent = Model + Harness
Loop  = Agent + Schedule + Verify + Handoff

You shouldn't be prompting agents anymore. You should be building systems that do it for you — and knowing whether they actually did.

pip install tvastar

---

Loop in 60 seconds

# A CI loop that watches your build, fixes failures, and escalates only when it can't.
import asyncio
from tvastar.loop.patterns import CISweeper
from tvastar.model import AnthropicModel

loop = CISweeper(
    model=AnthropicModel("claude-sonnet-4-6"),
    schedule="*/15 * * * *",   # every 15 minutes
    cancel_after=300.0,         # 5-minute timeout per run
)

asyncio.run(loop.start())       # runs forever — trigger → run → verify → handoff if stuck

Or scaffold from the CLI and be running in seconds:

tvastar loop init CISweeper        # writes .tvastar/loops/ci_sweeper.py
tvastar loop audit .tvastar/loops/ci_sweeper.py:loop   # score readiness L0→L3
tvastar loop run   .tvastar/loops/ci_sweeper.py:loop   # trigger once to test

The loop runs the agent, verifies the result, retries with exponential backoff, and escalates to you (Slack, email, any webhook) only when it cannot fix something itself. You walk away. It runs.

---

What is a harness?

Most Python agent libraries give you one of two things: orchestration patterns (how agents coordinate) or model wrappers (how to call an LLM). Neither solves the problem of running agents safely in production.

A harness is the missing layer. It sits between your agent logic and the real world and handles what happens when things go wrong — code that crashes, context that overflows, silent failures, infrastructure that varies across environments.

Tvastar includes lightweight framework primitives so you have something to run (AgentSpec, @tool, sessions, workflows). But the framework is minimal on purpose. The harness is the product.

---

The five problems Tvastar solves

1. You are still manually prompting agents

The leverage point has shifted. You should be building systems that prompt agents for you — not babysitting individual runs. Tvastar is the framework that makes automated agent loops production-ready. Give the loop a goal and a schedule. Walk away.

2. Running agent-produced code safely

Most frameworks assume you have a container. Tvastar runs real code in-memory with no Docker, no setup, no external service. Switch to Docker or a remote sandbox with one line when you need stronger isolation.

3. Agents that lie about success

An agent says "all tests pass" over a failing run. An agent claims a file was created but nothing was written. Tvastar detects silent failures automatically — the loop does not trust what the agent says, only what actually happened.

4. Long-running agents that crash

A 10-minute agent run failing at minute 9 loses everything. Tvastar checkpoints transcript and filesystem after every step. Crashes resume from where they stopped, not from the beginning.

5. Deploying the same agent everywhere

One agent definition runs as a web service, AWS Lambda, GitHub Action, container, or serverless function. No rewriting. No framework-specific deployment config.

---

Works with any agent or model

import asyncio
from tvastar import create_agent, Harness, default_toolset
from tvastar.model import AnthropicModel

# Wrapping a raw model call
agent = create_agent(
    "assistant",
    model=AnthropicModel("claude-opus-4-6"),
    instructions="You are a helpful coding agent.",
    tools=default_toolset(),
)
result = asyncio.run(Harness(agent).run("Write hello.py and run it."))
print(result.text)

# Wrapping an OpenAI-compatible provider
from tvastar.model import OpenAIModel

agent = create_agent("assistant", model=OpenAIModel("gpt-4o"), tools=default_toolset())

# Local Ollama — completely free, no API key
model = OpenAIModel(model="llama3.2", base_url="http://localhost:11434/v1", api_key="ollama")
agent = create_agent("assistant", model=model, tools=default_toolset())

# Any OpenAI-compatible provider (Groq, Together, Cloudflare…)
model = OpenAIModel(
    model="llama-3.1-8b-instant",
    base_url="https://api.groq.com/openai/v1",
    api_key="gsk_...",
)

# 100+ providers with one import — and automatic cost routing between them
# pip install "tvastar[litellm]"
from tvastar.model import LiteLLMModel

# Single provider — any litellm model string
model = LiteLLMModel("anthropic/claude-sonnet-4-6")
model = LiteLLMModel("groq/llama-3.1-70b-versatile")
model = LiteLLMModel("ollama/llama3.2")

# Router — your fast/cheap model handles easy prompts, smart model handles hard ones
model = LiteLLMModel(
    "fast",
    model_list=[
        {"model_name": "fast",  "litellm_params": {"model": "claude-haiku-4-5-20251001"}},
        {"model_name": "smart", "litellm_params": {"model": "claude-sonnet-4-6"}},
    ],
    routing_strategy="usage-based-routing-v2",   # least-loaded wins
    fallbacks=[{"fast": ["smart"]}],             # escalate on failure
)

Running Haiku where you were running Opus cuts cost 25×. LiteLLM Router does the routing. You don't write the if/else.

The harness wraps the model. It does not care which one.

---

See it in action: tvastar-fix

The fastest way to understand Tvastar is to watch it fix something real.

tvastar-fix is a CLI tool and GitHub Action that auto-fixes failing tests. Your tests fail on a PR. Tvastar runs the agent, executes the fixes in a safe sandbox, verifies they actually pass, and pushes the correction — without you touching a line.

It is the reference implementation for everything the harness provides: safe execution, silent failure detection, crash recovery, and deploy-anywhere portability in one working example.

pip install "tvastar[fix]"
tvastar-fix --test-cmd "pytest tests/" --model claude-opus-4-6

---

When not to use Tvastar

• You only need a single chat completion → call the model SDK directly, Tvastar is overkill
• You need hundreds of pre-built integrations (Slack, Salesforce, databases) → LangChain's ecosystem is larger
• Your agent never executes code or writes files → the sandbox and failure detection add weight without benefit

Tvastar is for agents that do things — run code, edit files, call tools — and need to do those things safely in production. If your agent only talks, you do not need a harness.

---

What Tvastar handles so you do not have to

Execution & Safety

Problem	How Tvastar handles it	API
Code execution without Docker	In-memory sandbox, zero setup	VirtualSandbox (default)
Real bash, jailed to a directory	Allowlist commands, network off	LocalSandbox + SecurityPolicy
Filesystem changes need atomic rollback	Snapshot before, restore on exception	harness.transaction()
Agent claims success but didn't	Silent failure detection on every run	unverified_completion detector
Agent loops on the same tool	Thrash detection fires before the loop spins forever	thrash_loop detector
Crash at step 47 of 50	Step-level checkpoint + resume from last good state	FileStore + harness.resume()
Flaky network tools fail mid-run	Per-tool retry with exponential backoff + jitter	ToolRetryPolicy
Tool called in wrong execution phase	Governance runs in Python — injection-proof	GovernancePolicy

Models & Routing

Problem	How Tvastar handles it	API
Locked to one model provider	Works with Anthropic, OpenAI, Ollama, Groq, and 100+ more	AnthropicModel, OpenAIModel, LiteLLMModel
100 providers, one interface	Single import, any litellm model string	LiteLLMModel("groq/llama-3.1-70b")
Paying Opus prices for easy prompts	Router uses cheap model by default, escalates on failure	LiteLLMModel(model_list=[...], fallbacks=[...])
Writing agent='reviewer' at every call site	Semantic matching picks the right profile from the prompt	AgentRouter + sess.task(router=router)
Bad agents waste compute	Rolling quality score; drop underperformers automatically	AgentPruner(threshold=60.0)
Wiring a TaskGraph by hand every time	Describe the goal; planner generates the parallel structure	auto_topology(goal, harness=harness)
Free-form prompt rewriting guesses wrong	Compile better instructions from real failure evidence	DSPyOptimizer

Context & Memory

Problem	How Tvastar handles it	API
Context grows past model limit	Automatic summarisation when message count threshold is hit	CompactionPolicy
Session messages balloon past 50 MB	Hard cap with auto-compaction before the next call	memory_cap_mb
Agent needs facts across sessions	BM25 retrieval (semantic optional) injected into system prompt	tvastar.contrib.ltm.LTMStore
Agent forgets context mid-task	Named sessions persist history; resume any session by ID	harness.session(name) / harness.resume(id)

Quality & Observability

Problem	How Tvastar handles it	API
No idea if the agent actually succeeded	Quality score 0–100 on every run, automatic	score_run(result)
Audit what the agent actually did	Full transcript — every message, tool call, and result	result.messages
Inspect runs visually	Local trace viewer, no cloud required	tvastar ui --trace run.jsonl
Stream tokens to the browser	SSE endpoint out of the box	GET /sessions/{id}/stream
No structured signal from failures	Typed findings with severity and evidence	Finding(detector, severity, message)
Observability platform integration	OpenTelemetry GenAI conventions — drops into Braintrust, Datadog, Honeycomb	OTelExporter

Compliance & Audit

Problem	How Tvastar handles it	API
Regulator asks "what did your AI do?"	Cryptographic receipt per run — hash + HMAC signature	ExecutionReceipt + receipt.verify()
PHI / PII ends up in the audit log	Redact before hashing — proof of removal is baked in	SanitizationPolicy.hipaa() / .pci() / .gdpr()
Model sees real SSNs before policy strips them	Tokenise before the model call; rehydrate after	TokenVault
Regex misses names, locations, passport numbers	Microsoft Presidio ML — 50+ entity types, 15+ languages	SanitizationPolicy.presidio()
No record of which tool returned what	Every input AND output captured, matched by call ID	receipt.tool_calls[].output
No proof a human approved the action	Approval records in the receipt — who, when, what	receipt.approvals[]
Anyone on the team can read audit logs	Role-based access; PermissionError on unauthorized read	TrustLog(can_read=role_fn)
Audit log tampered silently	on_breach callback fires immediately per corrupted entry	TrustLog(on_breach=alert_fn)
7-year SOX / 6-year HIPAA retention	Archive old entries; legal hold freezes the log entirely	RetentionPolicy
Quality SLA drops in production	Raise SLABreached or escalate when score falls below threshold	AssurancePolicy(min_score=80, on_fail="raise")

Production & Scale

Problem	How Tvastar handles it	API
Deploy to Lambda, GitHub Actions, web	One agent definition, any target	tvastar.deploy
Run 100 prompts at once	Concurrent fan-out, optional concurrency cap	harness.fan_out(prompts, concurrency=8)
Webhook / chatbot message handling	Fire-and-forget dispatch; reply in on_complete callback	dispatch()
Long-running pipeline survives restarts	Durable workflow with run history	@workflow + WorkflowRun
Agent loop must never miss a failure	L0→L3 readiness audit before deploy	audit_loop(loop)
Human must approve before irreversible action	Pause and wait — CLI, webhook, or event backend	require_approval() + ApprovalGate
API or model spend spiralling	Hard budget ceiling per run	BudgetPolicy(max_usd=0.50)
External tools (databases, SaaS, APIs)	Plug in any MCP server — fully transparent to the model	connect_mcp_server()

---

How it works

create_agent(...)  →  AgentSpec          (what the agent is — immutable)
Harness(spec)      →  Harness            (how it runs — stateful)
harness.run(...)   →  RunResult          (one prompt, one answer)
harness.session()  →  Session            (multi-turn conversation)

Inside every run() or prompt(), the loop looks like this:

User message
    ↓
Model generates response
    ↓
  ┌─ stop_reason == TOOL_USE? ──────────────────────────────────┐
  │                                                             │
  │   Execute all requested tools (concurrently)               │
  │   Feed results back to model                               │
  │   Auto-compact context if policy threshold hit             │
  │   Checkpoint to durable store                              │
  │   Loop ────────────────────────────────────────────────────┘
  │
  └─ END_TURN → RunResult(.text, .messages, .usage, .steps, .data)

---

Install

pip install tvastar                      # core only — zero deps
pip install "tvastar[anthropic]"         # + Claude models
pip install "tvastar[openai]"            # + OpenAI / Groq / Ollama / etc.
pip install "tvastar[litellm]"           # + 100+ providers via LiteLLM + cost routing
pip install "tvastar[router]"            # + semantic-router embeddings for AgentRouter
pip install "tvastar[dspy]"              # + DSPy systematic prompt optimisation
pip install "tvastar[serve]"             # + HTTP server (FastAPI)
pip install "tvastar[otel]"              # + OpenTelemetry tracing
pip install "tvastar[presidio]"          # + ML-powered PII detection (Presidio + spaCy)
pip install "tvastar[all]"              # everything

---

Environment variables

Tvastar reads these from the environment — never pass credentials in code:

Variable	Used by	Example
ANTHROPIC_API_KEY	AnthropicModel	sk-ant-...
OPENAI_API_KEY	OpenAIModel	sk-...
ANTHROPIC_BASE_URL	AnthropicModel — custom endpoint	https://my-proxy.example.com
OPENAI_BASE_URL	OpenAIModel — custom endpoint / Groq / Ollama	https://api.groq.com/openai/v1

# Claude
export ANTHROPIC_API_KEY="sk-ant-..."

# OpenAI
export OPENAI_API_KEY="sk-..."

# Ollama (local, no key needed)
export OPENAI_BASE_URL="http://localhost:11434/v1"
export OPENAI_API_KEY="ollama"

You can also pass api_key and base_url directly to the model constructor — useful when you need multiple providers in one process:

from tvastar.model.anthropic import AnthropicModel
from tvastar.model.openai import OpenAIModel

claude = AnthropicModel("claude-sonnet-4-6", api_key="sk-ant-...")
gpt4   = OpenAIModel("gpt-4o", api_key="sk-...", base_url="https://api.openai.com/v1")
llama  = OpenAIModel("llama3.2", base_url="http://localhost:11434/v1", api_key="ollama")

---

Core concepts

Agent layer:

Thing	What it is
AgentSpec	Immutable declaration: model + tools + instructions + policies
Harness	Stateful runtime: runs an AgentSpec across sessions
Session	One conversation thread with its own message history
Tool	A Python function the model can call (schema auto-derived)
Skill	A Markdown file of reusable expertise, loaded on demand
Sandbox	Where code runs — virtual (in-memory), local, or Docker
RunResult	What you get back: .text, .data, .usage, .steps, .ok
GovernancePolicy	Phase-based tool enforcement — declare which tools are legal per workflow phase
Finding	A structured signal from a silent-failure detector (severity + message + evidence)

Loop layer:

Thing	What it is
Loop	An agent on a schedule: trigger → run → verify → handoff if stuck → idle
LoopConfig	Schedule, goal, retries, timeout, circuit breaker — validated at construction
LoopState	IDLE → TRIGGERED → RUNNING → VERIFYING → PASS/FAIL → RETRY/HANDOFF/SUSPENDED
LoopRun	One iteration's metadata: state, steps, findings, error, duration
HandoffPolicy	What fires when retries are exhausted: LogHandoff, CallbackHandoff, MultiHandoff
MakerChecker	Two-agent pattern: Maker proposes, Checker independently verifies before PASS
ReadinessLevel	L0 MANUAL → L1 OBSERVE → L2 GATED → L3 AUTONOMOUS — scored by audit_loop()

---

Tools

from tvastar import tool, ToolRetryPolicy

@tool
def add(a: int, b: int) -> int:
    "Add two integers."
    return a + b

# With retry for flaky network calls
@tool(retry=ToolRetryPolicy(max_attempts=3, backoff_base=0.5))
async def call_api(url: str) -> str:
    "Fetch a URL."
    ...

# Access session context (sandbox, filesystem, memory)
@tool
async def save(path: str, content: str, ctx: ToolContext) -> str:
    "Save a file."
    ctx.filesystem.write(path, content)
    return "saved"

Built-in tools via default_toolset(): bash, read_file, write_file, edit_file, grep, glob, list_files.

Add internet access with web_toolset() — no API key, no extra dependencies:

from tvastar import default_toolset, web_toolset

agent = create_agent(
    "researcher",
    model=model,
    tools=[*default_toolset(), *web_toolset()],
)
# Agent can now browse any URL and search the web

# Or use individually
from tvastar import web_browse, web_search

@tool
async def my_tool(url: str) -> str:
    return await web_browse.fn(url)

web_browse(url) fetches any page as clean markdown via Jina AI Reader. web_search(query) returns top search results via Jina AI Search. Both handle HTTP errors gracefully and accept a max_chars limit to protect context.

Harness-wide retry — applies to all tools that do not have their own policy:

agent = create_agent(..., tool_retry=ToolRetryPolicy(max_attempts=3))

---

Sessions

harness = Harness(agent)

# One-shot
result = await harness.run("Summarise this document.")

# Multi-turn
sess = harness.session()
async with sess:
    await sess.prompt("Read report.txt")
    await sess.prompt("Write a 3-bullet summary")
    result = await sess.prompt("Translate the summary to Spanish")

# Named sessions for parallel branches
branch_a = harness.session("review-api")
branch_b = harness.session("review-auth")
results = await asyncio.gather(
    branch_a.prompt("Review the API layer"),
    branch_b.prompt("Review the auth layer"),
)

---

Structured output

Get back a typed object instead of raw text:

from pydantic import BaseModel

class Report(BaseModel):
    summary: str
    issues: list[str]
    severity: str

result = await sess.prompt("Analyse this code.", result=Report)
report: Report = result.data
print(report.severity)

Works with Pydantic v2, Pydantic v1, dataclasses, plain dict, or any callable validator.

Strict mode — fail loudly on parse failure

from tvastar import StructuredOutputError

try:
    result = await sess.prompt("Analyse this.", result=Report, strict=True)
except StructuredOutputError as e:
    print(f"Schema: {type(e.schema).__name__}")
    print(f"Raw text: {e.raw_text[:100]}")
    print(f"Parse error: {e.parse_error}")

Without strict=True (the default), parse failures fall back to raw text and surface a "structured_parse_failure" finding in result.findings.

Testing structured output with MockModel

Script dict entries for deterministic structured output in tests:

from tvastar import MockModel

model = MockModel(scripts={
    "analyst": [{"summary": "All clear", "issues": [], "severity": "low"}],
})
# session.task(agent="analyst", result=Report) will parse the JSON correctly

---

Delegating to specialist sub-agents

from tvastar import create_agent, AgentProfile

reviewer = AgentProfile(
    name="reviewer",
    description="Reviews code for security and correctness.",
    instructions="Report only issues with a reproducible failure scenario.",
    thinking_level="high",
    max_steps=10,
)

agent = create_agent("coordinator", model=model, subagents=[reviewer], tools=default_toolset())

sess = harness.session()
async with sess:
    result = await sess.task(
        "Review the auth package for security issues.",
        agent="reviewer",
        cancel_after=60.0,
        result=ReviewReport,
    )

Task delegation is capped at 4 levels deep to prevent runaway recursion.

Auto-routing — the router picks the agent, not you

from tvastar import AgentRouter

router = AgentRouter(spec.subagents.values())

# No more agent="reviewer" — the router reads the prompt and decides
result = await sess.task(
    "Review this SQL migration for data loss risks.",
    router=router,
)

With pip install "tvastar[router]" the router uses embedding-based similarity (semantic-router). Without it, it falls back to word-overlap — zero deps, still useful.

AgentPruner — drop specialists that underperform

from tvastar import AgentPruner

pruner = AgentPruner(threshold=60.0, min_runs=3)

# After each task result, record the score against the profile that ran it
pruner.update("coder", result)

# Rebuild the router from the surviving pool — low scorers are gone
router = AgentRouter(pruner.active(all_profiles))

AgentDropout for your agent pipelines: profiles whose rolling average quality score falls below threshold are removed from routing. The compute that was going to a bad agent goes to a good one instead.

---

Parallel fan-out

Run multiple prompts concurrently with one call:

results = await harness.fan_out([
    "Summarise chapter 1",
    "Summarise chapter 2",
    {
        "prompt": "Summarise chapter 3",
        "agent": "summariser",
        "cancel_after": 30.0,
        "result": SummarySchema,
    },
], concurrency=4)

---

Auto-topology — describe the goal, get the parallel graph

You shouldn't be wiring TaskGraph by hand for every pipeline. auto_topology() decomposes a natural-language goal into a TaskGraph + AgentProfile set automatically.

from tvastar import auto_topology

graph, profiles = await auto_topology(
    "Research our top 3 competitors, score their pricing, write a strategy report.",
    harness=harness,    # reuses your existing model — no extra config
    max_subtasks=6,
)
results = await graph.run()
print(results["report"].text)

The planner generates subtasks with explicit dependencies. graph.run() executes them with maximum parallelism — the critical path, not the sum.

---

DAG task execution — maximum parallelism

TaskGraph models work as a directed acyclic graph. Independent tasks run concurrently; a task starts the moment every dependency completes. Wall-clock time equals the critical path, not the sum of all tasks.

from tvastar import TaskGraph

graph = TaskGraph(harness)

# These three have no deps — start immediately in parallel
graph.task("leads",   "Fetch the lead list from CRM")
graph.task("pricing", "Scrape competitor pricing pages")
graph.task("news",    "Find recent news about the prospect")

# Waits for all three; their results are auto-injected into its prompt
graph.task("analyse", "Score and prioritise leads",
           depends_on=["leads", "pricing", "news"])

# These two depend on analyse but not each other — run in parallel
graph.task("emails",  "Write personalised cold emails",
           depends_on=["analyse"])
graph.task("report",  "Write executive summary",
           depends_on=["analyse"])

results = await graph.run()
print(results["emails"].text)
print(results.ok)          # True when every task finished cleanly
print(results.text)        # dict of all task outputs

Fluent chaining:

results = await (
    TaskGraph(harness)
    .task("fetch", "Fetch data")
    .task("analyse", "Analyse it", depends_on=["fetch"])
    .task("report",  "Write report", depends_on=["analyse"])
    .run()
)

Structured output per task:

graph.task("score", "Score each lead", result=LeadScores, depends_on=["fetch"])
results["score"].data  # LeadScores instance

---

Loop Engineering

A loop is an agent on a schedule with verify + handoff built in. It runs autonomously, retries on failure with exponential backoff, and escalates to a human only when it cannot fix something itself.

Loop = Agent + Schedule + Verify + Handoff

Built-in patterns — clone and run in minutes

Pattern	What it does	Default schedule
CISweeper	Watches CI, fixes red builds, escalates if unfixable	Every 15 min
PRBabysitter	Resolves trivial merge conflicts, flags stale PRs	Every 30 min
DailyTriage	Classifies new issues by severity, detects duplicates	9am UTC daily
DependencySweeper	Bumps patch versions, runs tests, commits if green	3am UTC daily
PostMergeCleanup	Reports TODOs + stale references after merges	Every 30 min
ChangelogDrafter	Writes CHANGELOG entries from commit history	Monday 9am
MakerChecker	Maker proposes, Checker independently verifies	@manual

MakerChecker — two-agent verification

from tvastar.loop.patterns import MakerChecker
from tvastar.model import AnthropicModel

loop = MakerChecker(
    maker_model=AnthropicModel("claude-haiku-4-5-20251001"),   # fast writer
    checker_model=AnthropicModel("claude-sonnet-4-6"),          # careful reviewer
    goal="Fix the failing test in tests/test_auth.py",
    max_rounds=3,          # Maker+Checker cycles before HANDOFF
    cancel_after=600.0,
)
run = await loop.trigger()
# Maker proposes a fix → Checker reviews adversarially → APPROVED or REJECTED+feedback
# REJECTED feeds structured criticism back to Maker for the next round
# Only APPROVED advances to PASS

Self-Improving Loops (meta_model / DSPyOptimizer)

Set meta_model on any LoopConfig and the loop rewrites its own agent instructions after each FAIL — inspired by Hyperagents. No code execution: improvement is pure prompt evolution, persisted across restarts.

from tvastar.loop import Loop, LoopConfig
from tvastar.memory.store import FileStore
from tvastar.model.anthropic import AnthropicModel

config = LoopConfig(
    name="self-improving-ci",
    goal="Keep the build green.",
    schedule="*/15 * * * *",
    cancel_after=300.0,
    meta_model=AnthropicModel("claude-sonnet-4-6"),  # stronger model improves the worker
)
loop = Loop(spec, config, store=FileStore(".tvastar-state"))
run = await loop.trigger()

best = loop.best_generation()   # highest-scoring generation on record
print(f"Best: gen {best.gen_id}, score={best.score}")

DSPyOptimizer — systematic, not free-form. Free-form rewriting after a failure is guesswork. DSPy compiles better instructions by treating your PASS runs as few-shot demonstrations and your FAIL runs as failure evidence. The optimizer produces structured output — just the improved instructions, no commentary, no hallucinated constraints.

from tvastar.loop.optimize import DSPyOptimizer

config = LoopConfig(
    name="self-improving-ci",
    goal="Keep the build green.",
    schedule="*/15 * * * *",
    optimizer=DSPyOptimizer("gpt-4o"),   # replaces meta_model — takes precedence
)
# pip install "tvastar[dspy]"

DSPyOptimizer takes precedence over meta_model when both are set. meta_model is still supported — swap to optimizer when you want DSPy's structured compilation instead of a one-shot rewrite.

MakerChecker with a FileStore also persists checker rejection verdicts across runs so the Maker learns from patterns that caused rejection in previous sessions.

L0→L3 Readiness Audit

Score any loop before deploying it. Never discover failure modes at 2am.

from tvastar import audit_loop

report = audit_loop(loop)
print(f"L{report.level} {report.name}: {report.description}")
for gap in report.gaps:
    print(f"  ✗ {gap}")
# L0 MANUAL: No schedule configured. Set LoopConfig(schedule='*/15 * * * *') to reach L1.

Or from the CLI — useful as a pre-deploy CI gate:

tvastar loop audit .tvastar/loops/ci.py:loop   # exits 0 only at L3 AUTONOMOUS

Level	Name	What it means
L0	MANUAL	Loop exists but only fires when you call trigger() manually
L1	OBSERVE	Scheduled + handoff — fires automatically and escalates failures
L2	GATED	L1 + cancel_after timeout — safe for loops that mutate state
L3	AUTONOMOUS	L2 + silent-failure detectors + circuit breaker — production-ready

Handoff policies

from tvastar.loop.handoff import LogHandoff, CallbackHandoff, MultiHandoff

# Default: prints a structured report to stderr
loop = CISweeper(model=model, handoff=LogHandoff())

# Custom: call any async function
loop = CISweeper(model=model, handoff=CallbackHandoff(
    async def on_fail(run, history):
        await slack.post("#oncall", f"Loop {run.loop_name} failed after {run.iteration} attempts")
))

# Both: fire all, report all failures independently
loop = CISweeper(model=model, handoff=MultiHandoff([LogHandoff(), slack_handoff]))

Loop lifecycle — Werner-hardened

Every failure mode is handled before code runs, not discovered at 2am:

Failure	How Tvastar handles it
Run exceeds time limit	cancel_after fires TIMEOUT → _handle_fail
Model API error	FailureKind.MODEL_ERROR → retry with backoff
Agent claims success but fails	Silent-failure detectors → DETECTION → retry
Process crashes mid-run	_recover() on startup detects RUNNING → marks INTERRUPTED
Too many consecutive failures	Circuit breaker → SUSPENDED; loop.reset() to resume
Handoff itself throws	Retried 3× with backoff → HANDOFF_FAILED (never silently dropped)
Scheduler task dies unexpectedly	add_done_callback watchdog restarts it

---

Loop Quality — score every run automatically

Every RunResult has a quality score. score_run() computes it from findings and stop reason:

from tvastar.quality import score_run

result = await harness.run("fix the failing tests")
report = score_run(result)

print(report.score)    # 0–100
print(report.grade)    # "PASS" | "WARN" | "FAIL"
print(report.summary)  # human-readable explanation

Or from the CLI — useful as a CI gate:

tvastar quality my_agent.py:agent "fix the failing tests"
# Loop Quality: 82/100  [PASS]
# exit 0 on PASS/WARN, exit 1 on FAIL

Scoring deductions:

Deduction	Condition
−30	Per ERROR finding (e.g. unverified_completion, schema_mismatch)
−10	Per WARNING finding (e.g. thrash_loop, ignored_tool_error)
−20	Run stopped by max_steps or budget
−50	Run stopped by error

Grades: ≥ 80 → PASS · ≥ 60 → WARN · < 60 → FAIL.

---

Plug into anything — wrap any agent framework

tvastar.wrap is a quality layer you add on top of whatever agent infrastructure you already run. Zero changes to your existing loop.

import tvastar

# Decorator — any async function becomes quality-scored
@tvastar.wrap
async def my_loop(prompt: str) -> str:
    return await some_external_agent(prompt)

result = await my_loop("fix the failing tests")
print(result.quality.score)   # 0–100
print(result.quality.grade)   # "PASS" | "WARN" | "FAIL"
print(result.ok)              # True if grade is PASS

OpenAI function-calling loops

from tvastar.adapters.openai import OpenAILoopWrapper

with OpenAILoopWrapper() as loop:
    loop.messages.append({"role": "user", "content": "Fix the tests."})
    while True:
        resp = client.chat.completions.create(
            model="gpt-4o", messages=loop.messages, tools=my_tools
        )
        loop.messages.append(resp.choices[0].message.model_dump())
        if resp.choices[0].finish_reason == "stop":
            break
        # handle tool calls …

print(loop.result.quality.grade)   # full detector suite ran

LangGraph graphs

from tvastar.adapters.langgraph import LangGraphWrapper

graph = build_my_graph().compile()
wrapped = LangGraphWrapper(graph)

result = await wrapped.ainvoke({"messages": [HumanMessage(content="Fix tests.")]})
print(result.quality.score)

AWS AgentCore (Bedrock Agents)

from tvastar.adapters.agentcore import AgentCoreWrapper
import boto3

client = boto3.client("bedrock-agent-runtime")
wrapper = AgentCoreWrapper(client)

result = wrapper.invoke(
    agent_id="ABCDEF1234", agent_alias_id="TSTALIASID",
    session_id="session-1", input_text="Fix the failing tests.",
)
print(result.quality.grade)

All three adapters convert the framework's message format into Tvastar's types so the full silent-failure detector suite runs — not just text-level checks.

---

Verifiable Execution — the audit trail your regulator will actually accept

HIPAA violations average $1.9M per incident. A SOX audit asks for records of every automated decision. The EU AI Act requires traceability on high-risk AI systems. PCI-DSS demands logs of who accessed what, when, and what changed.

Every other AI agent framework answers these questions with: a text file, if you're lucky.

Tvastar closes 7 regulatory gaps no other framework addresses:

Gap	What we built
No model version in record	model_name field on every receipt
Tool outputs not captured	tool_calls[].output — input AND output, matched by ID
PII/PHI in plaintext audit log	SanitizationPolicy — redact before hashing, chain stays valid
No human approver in record	approvals[] — who approved, exact timestamp, what message
Audit log readable by anyone	TrustLog(can_read=role_fn) — PermissionError on unauthorized access
Tampering goes undetected	on_breach callback fires immediately on first corrupted entry
No retention / legal hold	RetentionPolicy — SOX 7yr, HIPAA 6yr, litigation freeze

One line to opt in:

from tvastar.assurance import AssurancePolicy, TrustLog, SanitizationPolicy

agent = create_agent(
    "billing-bot",
    model=model,
    assurance=AssurancePolicy(
        key=os.environ["RECEIPT_KEY"],    # HMAC-SHA256; reads env automatically if not passed
        log=TrustLog(
            ".tvastar-trust.jsonl",
            on_breach=lambda r: alert_security_team(r),    # fires immediately on tamper
            can_read=lambda role: role in {"auditor", "admin"},  # PermissionError otherwise
        ),
        min_score=80,                     # quality SLA — SLABreached if score drops below
        on_fail="escalate",
        on_escalate=lambda r: page_oncall(r),
        sanitize=SanitizationPolicy.hipaa(),   # redact SSN/DOB/PHI before hashing
    ),
)

result = await harness.run("Process patient intake form")

# Cryptographic proof — SHA-256 hash + HMAC-SHA256 signature
print(result.receipt.run_id)                    # run_c3afc6fcc23c4322
print(result.receipt.content_hash)              # sha256:d2e502ed...
print(result.receipt.model_name)                # claude-sonnet-4-6
print(result.receipt.tool_calls[0]["output"])   # what the tool actually returned
print(result.receipt.verify())                  # True — hash and signature both valid

# Chain integrity — modify any past entry and verify_chain() catches it
assert policy.log.verify_chain()

# Pull any past run
r = policy.log.get("run_c3afc6fcc23c4322", role="auditor")
print(r.quality_grade)     # "PASS"
print(r.approvals)         # [{"tool": "deploy", "approved_by": "[e-mail]", ...}]

Audit report — hand this to the lawyer

r = log.get("run_abc123", role="auditor")
print(r.to_audit_report())
# ══════════════════════════════════════════════════════════
# EXECUTION RECEIPT  run_c3afc6fcc23c4322
# ══════════════════════════════════════════════════════════
# Agent:           billing-bot
# Model:           claude-sonnet-4-6
# Timestamp:       2026-06-20 14:32:01 UTC
# Duration:        2.3 s
# Quality:         91 / 100  [PASS]
# ...
# TOOL CALLS (2)
#   [1] bash
#       Input:   {"command": "pytest tests/"}
#       Output:  "3 passed in 0.8s"
# ...
# HUMAN APPROVALS (1)
#   deploy_to_production — approved by [e-mail] at 14:32:00
# ══════════════════════════════════════════════════════════
# INTEGRITY
#   Hash:      sha256:d2e502ed...
#   Signature: hmac-sha256:9af1b3c0...
#   Chain:     sha256:a1b2c3d4...  ← previous receipt
# ══════════════════════════════════════════════════════════

html = r.to_audit_report(fmt="html")
Path("audit.html").write_text(html)   # table view, ready to email

PII / PHI redaction — HIPAA, PCI-DSS, GDPR

# Regex-based presets (zero deps)
sanitize=SanitizationPolicy.hipaa()    # SSN, DOB, phone, email, IP, bearer tokens
sanitize=SanitizationPolicy.pci()      # credit card, CVV, bearer tokens
sanitize=SanitizationPolicy.gdpr()     # email, phone, IP, DOB
sanitize=SanitizationPolicy.all_pii()  # union of all above

# ML-powered (catches names, locations, passport numbers — things regex misses)
# pip install tvastar[presidio] && python -m spacy download en_core_web_lg
sanitize=SanitizationPolicy.presidio(languages=["en", "de"])

# Extend any preset with custom patterns
sanitize=SanitizationPolicy.hipaa().add_pattern(r"MRN-\d+", "[MRN]")

The hash covers the sanitized content. Proof that PII was removed is baked into the receipt itself — a tampered re-injection of PII fails verify().

TokenVault — the model never sees real PII, not even once

SanitizationPolicy redacts PII from audit records. TokenVault goes further: the model never sees the original values in the first place.

from tvastar.assurance import SanitizationPolicy, TokenVault

vault = TokenVault()
clean = vault.tokenize(prompt, SanitizationPolicy.hipaa())
# "Patient Jane Smith SSN 123-45-6789" → "Patient <<PERSON_1>> SSN <<US_SSN_1>>"

result = await sess.prompt(clean)   # model works on tokens only

final = vault.rehydrate(result.text)
# "Send referral to <<PERSON_1>>" → "Send referral to Jane Smith"

HIPAA §164.514 says the AI system must not retain PHI beyond the minimum necessary. TokenVault makes that testable: grep your model traffic for the vault's tokens — if you find the original SSN, something broke.

Retention — SOX 7yr, HIPAA 6yr, legal holds

from tvastar.assurance import RetentionPolicy

log = TrustLog(".tvastar-trust.jsonl")

# Archive entries older than 7 years (SOX requirement)
count = log.apply_retention(RetentionPolicy(
    max_age_days=365 * 7,
    archive_path=".tvastar-trust-archive.jsonl",  # standalone verifiable chain
))

# Legal hold — freeze everything until litigation ends
count = log.apply_retention(RetentionPolicy(
    max_age_days=30,
    hold_until=1_800_000_000.0,   # epoch timestamp of hold expiry
))
# → returns 0, nothing archived while hold is active

Archive files are standalone verifiable JSONL chains — a regulator can run TrustLog(archive_path).verify_chain() independently.

---

Extended thinking

agent = create_agent(..., thinking_level="high")
# Anthropic: budget_tokens=16000  (low=1024, medium=8000, high=16000)
# OpenAI:    reasoning_effort='high'

---

Workflows — durable, inspectable pipelines

from tvastar import workflow
from tvastar.workflow import WorkflowContext

@workflow
async def summarise_document(ctx: WorkflowContext) -> dict:
    harness = await ctx.init(agent)
    sess = await harness.session()
    result = await sess.prompt(f"Summarise {ctx.payload['path']}")
    return {"summary": result.text, "steps": result.steps}

run = await summarise_document.run({"path": "report.pdf"})
print(run.status)   # RunStatus.COMPLETED
print(run.output)   # {'summary': '...', 'steps': 3}

for past_run in summarise_document.list_runs():
    print(past_run.run_id, past_run.status)

Workflow receipts

Get a unified execution receipt after a workflow run:

@workflow
async def my_workflow(ctx: WorkflowContext) -> dict:
    # ... do work ...
    receipt = ctx.build_receipt()
    # {"run_id": "run_abc", "status": "completed", "duration_seconds": 12.3, ...}
    return {"result": data, "receipt": receipt}

---

Event-driven dispatch

For chat bots, webhooks, and queue processors:

from tvastar import dispatch, dispatch_and_wait, observe_dispatch, DispatchInput

# Fire and forget
dispatch_id = await dispatch(
    agent,
    id="user_123",
    input=DispatchInput(text=message_text, type="chat.message"),
    on_complete=lambda r: send_reply(r.text),
    cancel_after=30.0,
)

# Fire and await
result = await dispatch_and_wait(agent, id="job_456", text="Process this report.")

# Watch all dispatches globally
observe_dispatch(lambda event: logger.info(event.type, extra=event.data))

---

Context compaction

Prevent context window exhaustion in long sessions:

from tvastar import CompactionPolicy

agent = create_agent(
    "long-runner",
    model=model,
    compaction=CompactionPolicy(
        max_messages=40,
        keep_last=10,
        min_messages=20,
    ),
)
# Fires automatically after tool turns. The model never notices.

---

Application-level file access

async with Harness(agent) as h:
    await h.fs.write_file("report.pdf", pdf_bytes)
    result = await h.run("Summarise report.pdf")
    summary = await h.fs.read_file("summary.md")

---

Sandboxes

from tvastar import VirtualSandbox, LocalSandbox, SecurityPolicy

# Default — in-memory, zero deps
create_agent(..., sandbox=VirtualSandbox)

# Real bash, jailed to a directory
policy = SecurityPolicy(allowed_commands={"python", "pytest"}, network=False)
create_agent(..., sandbox=lambda: LocalSandbox("./workspace", policy=policy))

---

MCP — use any published tool server

from tvastar import connect_mcp_server, default_toolset

client = await connect_mcp_server(command="python", args=["my_mcp_server.py"])
# or remote:
client = await connect_mcp_server(url="https://api.example.com/mcp", headers={...})

agent = create_agent("a", model=model, tools=[*default_toolset(), *client.tools])
await client.close()

---

Durable execution — survive crashes

from tvastar import Harness, FileStore

harness = Harness(agent, store=FileStore(".tvastar-state"))

# On restart — resume from last checkpoint
sess = harness.resume("sess_abc123") or harness.session()

Workflow-level checkpoints

For multi-phase workflows, checkpoint at the orchestration level so re-invocation skips completed phases:

from tvastar import workflow, WorkflowContext, FileCheckpoint

@workflow
async def my_pipeline(ctx: WorkflowContext) -> dict:
    # Skip if already completed
    data = await ctx.skip_if_checkpointed("phase_1")
    if data is None:
        data = await do_expensive_work()
        await ctx.checkpoint("phase_1", data)

    # Phase 2 uses phase 1 output
    result = await do_more_work(data)
    await ctx.checkpoint("phase_2", result)
    return result

# Run with checkpoint persistence
run = await my_pipeline.run(
    {"input": "..."},
    checkpoint_backend=FileCheckpoint(".tvastar-checkpoints/"),
)
# On crash + re-run with same run_id: skips completed phases automatically

---

Serving over HTTP

pip install "tvastar[serve]"
tvastar serve my_agent.py:agent --port 8000

Method	Path	Description
GET	/	Agent info
POST	/sessions	Create session
POST	/sessions/{id}/prompt	Send a message
WS	/sessions/{id}/stream	WebSocket streaming
GET	/sessions/{id}/stream?text=...	SSE streaming

curl -N "http://localhost:8000/sessions/sess_abc/stream?text=Hello"
# data: {"type": "text_delta", "data": {"text": "Hello"}}
# data: [DONE]

---

Observability

from tvastar import Tracer, ConsoleExporter, JSONLExporter

harness = Harness(agent, tracer=Tracer([
    ConsoleExporter(),
    JSONLExporter("trace.jsonl"),
]))

OpenTelemetry (Braintrust, Honeycomb, Datadog, Sentry):

pip install "tvastar[otel]"

from tvastar import OTelExporter
harness = Harness(agent, tracer=Tracer([OTelExporter()]))

The model.generate span follows the OpenTelemetry GenAI semantic conventions — gen_ai.system, gen_ai.request.model, gen_ai.usage.input_tokens, gen_ai.response.finish_reasons, … — so traces drop into Braintrust / Honeycomb / Datadog dashboards without custom attribute mapping.

---

Trace viewer UI — inspect every run locally

Write a trace file with JSONLExporter, then open the viewer:

pip install "tvastar[serve]"
tvastar ui                          # reads tvastar-trace.jsonl in cwd
tvastar ui --trace my-run.jsonl     # custom path
tvastar ui --port 7878 --no-open    # headless / CI

Or programmatically after a run:

from tvastar import Tracer, JSONLExporter, Harness, run_ui

harness = Harness(agent, tracer=Tracer([JSONLExporter("trace.jsonl")]))
result  = await harness.run("Write and test auth.py")

# inspect in browser
run_ui("trace.jsonl", port=7878)

The viewer is a self-contained FastAPI + vanilla-JS SPA (no build step, no Node):

• Left panel — runs listed newest-first, with a green/yellow/red status dot, step count, tool-call count, and total duration
• Right panel — per-run token counts, detected findings (warnings / errors), and an expandable timeline: every model.generate, tool.invoke, and lifecycle event in order with inputs, result previews, and stop reasons
• Auto-refreshes every 5 s — watch a long run fill in live

Try it with the bundled demo (no agent run required):

python run_ui_demo.py   # generates a sample trace and opens the viewer

---

Tool masking — show the model only the tools it needs now

Exposing every tool on every turn burns context and tempts the model to reach for the wrong one. A tool_policy filters the visible toolset per turn (it can only hide available tools, never grant new ones, and never breaks a run):

from tvastar import create_agent, allow_only, deny, phases

# only one tool, ever
create_agent(..., tool_policy=allow_only("read_file"))

# everything except the dangerous one
create_agent(..., tool_policy=deny("bash"))

# research first, unlock writes once we're a few steps in
create_agent(..., tool_policy=phases({1: ["grep", "read_file"],
                                      4: ["grep", "read_file", "write_file"]}))

# or any callable: (MaskContext) -> list[str]
create_agent(..., tool_policy=lambda ctx: ["bash"] if ctx.step > 2 else [])

---

Silent-failure detection

result = await harness.run("Fix all test failures.")

if not result.ok:
    for finding in result.warnings:
        print(f"[{finding.severity}] {finding.detector}: {finding.message}")
# → [WARNING] unverified_completion: model claimed success but last tool result shows failures

Built-in detectors: unknown_tool, schema_mismatch, thrash_loop, ignored_tool_error, unverified_completion, prompt_injection, empty_answer, step_limit.

Benchmark: tau2-bench (10,832 trajectories)

Evaluated against the tau2-bench dataset — 3,651 failed agent trajectories across 4 model families and 4 domains (airline, retail, telecom).

Failure Category	Count	Tvastar	Traditional Monitoring	Gap
False success (agent lied)	461	100%	0%	+100%
Ambiguous (stuck/looping)	3,175	100%	0%	+100%
Honest failure	15	100%	0%	+100%

Key finding: 97% of "false success" failures are preceded by detectable thrash loops. Tvastar catches the root cause upstream — before the agent even produces its misleading final message. Traditional exit-code monitoring catches none of them.

Top detectors on false-success cases: thrash_loop (97.2%), step_limit (98.5%), unverified_completion (2.8%).

Write your own:

from tvastar.detect import Finding, Severity

def slow_run(ctx):
    if ctx.stopped == "max_steps":
        return [Finding("slow_run", Severity.WARNING, "hit the step ceiling")]
    return []

create_agent(..., detect=[*default_detectors(), slow_run])

Detect from raw trajectories — no RunContext needed

Run detectors against any message list (JSONL trajectories, sub-agent transcripts):

from tvastar import detect_from_messages

# Load a trajectory from a file and run all detectors
from tvastar.types import Message
messages = [...]  # parsed from JSONL

findings = detect_from_messages(messages)
for f in findings:
    print(f)  # [WARNING] thrash_loop: tool 'bash' called 3x with identical args

# Suppress false positives for known tools
findings = detect_from_messages(messages, known_tools=["bash", "read_file", "write_file"])

Score a pipeline of multiple runs

from tvastar.quality import score_pipeline

report = score_pipeline([result1, result2, result3], strategy="worst")
print(report.score)    # min score across all results
print(report.grade)    # "PASS" | "WARN" | "FAIL"

Strategies: "worst" (strictest — min score), "average" (mean), "all_pass" (100 if all pass, else min).

---

Untrusted content & prompt-injection detection

No one has solved prompt injection — so Tvastar doesn't claim to. It gives you the two honest things that genuinely help:

1. Fence untrusted content so the model treats it as data, not orders. This reduces — does not eliminate — the model following injected instructions.
2. Detect content that looks like an injection attempt and surface it as a WARNING finding (the built-in prompt_injection detector). Detection, not prevention.

from tvastar import wrap_untrusted, scan_for_injection

@tool
async def fetch(url: str) -> str:
    "Fetch a web page."
    page = await http_get(url)
    return wrap_untrusted(page, source=url)   # the model sees it as DATA

# the prompt_injection detector flags suspicious tool output automatically:
result = await harness.run("Summarise that page.")
for f in result.warnings:
    if f.detector == "prompt_injection":
        print("⚠ possible injection in tool output:", f.message)

Scan full message lists for injection

from tvastar import scan_messages_for_injection

result = scan_messages_for_injection(messages)
if result.is_adversarial:
    for e in result.evidence:
        print(e)  # [msg 3, tool_result] pattern 'override_instructions': "ignore all previous..."

Redact PII from message lists

from tvastar import redact_messages

result = redact_messages(messages)
print(result.redaction_count)    # 3
print(result.redacted_types)     # ["email", "phone"]
# Messages now contain [EMAIL_1], [PHONE_2] placeholders instead of real PII

page = await http_get(url)
return wrap_untrusted(page, source=url)   # the model sees it as DATA

the prompt_injection detector flags suspicious tool output automatically:

result = await harness.run("Summarise that page.") for f in result.warnings: if f.detector == "prompt_injection": print("⚠ possible injection in tool output:", f.message)

---

## Dynamic Capability Governance — lock dangerous tools to specific phases

`GovernancePolicy` enforces **least privilege at invocation time** — after the
model has already decided to call a tool. Unlike masking (which is advisory),
governance runs in Python code and cannot be bypassed by prompt injection.

```python
from tvastar import create_agent, GovernancePolicy
from tvastar.approval import ApprovalGate

gov = GovernancePolicy(
    phases={
        "read":  {"grep", "read_file", "glob"},
        "write": {"grep", "read_file", "glob", "write_file", "bash"},
    },
    current_phase="read",
    # Optional — route blocked calls to a human instead of hard-blocking:
    approval_gate=ApprovalGate(backend="cli"),
)
agent = create_agent("assistant", model=..., governance=gov)

# Elevate at runtime (per-session — concurrent sessions are isolated):
gov.set_phase("write")

# Single-call enforcement for custom tool pipelines:
block = await gov.enforce("write_file", tool_use_id="call_123")
if block is not None:
    # block is a ToolResultBlock(is_error=True) — feed it back to the model
    print(block.content)  # [governance] denied: tool 'write_file' not permitted in phase 'read'

# Wire masking and governance together from one object:
create_agent(..., governance=gov, tool_policy=gov.as_tool_policy())

---

Transactional Sandbox — atomic rollback on failure

Wrap any session step in a harness.transaction() to guarantee that filesystem changes are rolled back if the step raises an exception.

async with harness.transaction(session) as sess:
    await sess.prompt("Refactor the auth module and run tests")
    # → if tests fail or an exception fires, the workspace rolls back atomically

Works with VirtualSandbox (< 150 ms on 1 MB) and LocalSandbox (< 500 ms on 500 KB). Both expose snapshot() / restore() for manual control too:

snap = sandbox.snapshot()
# ... do risky things ...
sandbox.restore(snap)   # reset to exactly the pre-snapshot state

---

Long-Term Memory — remember facts across sessions

tvastar.contrib.ltm consolidates conversation knowledge into a persistent LTMStore after each session and injects recalled context into the system prompt on subsequent runs. No extra dependencies needed (BM25 retrieval by default; sentence-transformers optional for semantic search).

from tvastar.contrib.ltm import LTMStore
from tvastar.memory.store import FileStore

ltm = LTMStore(FileStore(".ltm"))

# Wire retrieval into the system prompt — recalled per turn, keyed on user intent
agent = create_agent("assistant", model=..., system_prompt_hook=ltm.as_hook())

# After the session completes, persist what the agent learned
result = await Harness(agent).run("Fix the flaky auth test")
await ltm.consolidate(result, model, session_id="fix-auth-001")

# Next session — the agent automatically recalls relevant past knowledge
result2 = await Harness(agent).run("The auth test is flaky again")

---

CLI

Agent commands:

tvastar run   my_agent.py:agent "Write hello.py and run it"
tvastar chat  my_agent.py:agent
tvastar serve my_agent.py:agent
tvastar info  my_agent.py:agent
tvastar logs  run_abc123
tvastar ui    --trace tvastar-trace.jsonl   # local trace viewer
tvastar bench my_agent.py:agent --suite swe-lite --max-tasks 10

Loop commands:

# Scaffold a loop from any built-in pattern
tvastar loop init CISweeper                          # → .tvastar/loops/ci_sweeper.py
tvastar loop init MakerChecker --name my-verifier   # custom name
tvastar loop init DailyTriage --out ./loops/triage.py  # custom path

# Score readiness before deploying (exits 0 only at L3 AUTONOMOUS)
tvastar loop audit .tvastar/loops/ci_sweeper.py:loop

# Trigger once and see the result
tvastar loop run .tvastar/loops/ci_sweeper.py:loop

# Inspect current state
tvastar loop status .tvastar/loops/ci_sweeper.py:loop

---

Deploy anywhere

One agent definition. Any target.

# AWS Lambda
from tvastar.deploy import lambda_handler
handler = lambda_handler(agent)

# GitHub Action
from tvastar.deploy import github_action
github_action(agent, on="workflow_dispatch")

# ASGI (Uvicorn, Gunicorn)
from tvastar.serving import create_app
app = create_app(agent)

---

Custom model adapter

from tvastar.model import Model
from tvastar.types import Message, ModelResponse, StopReason, TextBlock

class MyModel(Model):
    name = "my-provider"

    async def generate(self, messages, *, system=None, tools=None,
                       max_tokens=4096, temperature=1.0,
                       stop_sequences=None, thinking_level=None) -> ModelResponse:
        text = await my_api_call(messages)
        return ModelResponse(
            message=Message("assistant", [TextBlock(text=text)]),
            stop_reason=StopReason.END_TURN,
        )

---

Evals — measure agent quality

Know when your agent gets better or worse. Define test cases, run them, get a score.

import asyncio
from tvastar import EvalSuite, Case
from tvastar.eval import assert_contains, assert_ok, assert_steps_under, assert_not_contains

suite = EvalSuite(agent, concurrency=8)

suite.add(Case(
    name="writes valid Python",
    prompt="Write a function that reverses a string",
    checks=[
        assert_contains("def"),
        assert_contains("return"),
        assert_ok(),
        assert_steps_under(5),
    ],
))

suite.add(Case(
    name="does not hallucinate imports",
    prompt="Write hello world in Python",
    checks=[
        assert_contains("print"),
        assert_not_contains("import nonexistent"),
    ],
))

report = asyncio.run(suite.run())
report.print()
# ============================================================
# Eval Report  —  2/2 passed  (100%)
# Duration: 3.2s
# ============================================================
#   ✓  writes valid Python  (2.1s)
#   ✓  does not hallucinate imports  (1.1s)
# ============================================================

print(report.score)   # 1.0
print(report.passed)  # 2

Run on every PR to catch regressions before they ship.

---

Benchmarks — measure quality against the real world

EvalSuite measures against your checks. BenchSuite measures against standardised, external task sets — the difference between testing whether your code works and testing whether your agent works on real software engineering problems.

import asyncio
from tvastar import create_agent, BenchSuite, swe_bench_tasks, default_toolset
from tvastar.model import AnthropicModel

agent = create_agent("coder", model=AnthropicModel(), tools=default_toolset())
suite = BenchSuite(agent, concurrency=4)
suite.add_many(swe_bench_tasks(split="lite", max_tasks=10))   # needs: pip install datasets
report = asyncio.run(suite.run())
report.print()
# ═══════════════════════════════════════════════════════════════
# Benchmark Report
#   Resolved : 7/10  (70.0%)
#   Duration : 142.3s
# ═══════════════════════════════════════════════════════════════

Or from the CLI:

tvastar bench agent.py:agent --suite swe-lite --max-tasks 50 --out report.json

Local JSONL — bring your own benchmark in SWE-bench format:

tvastar bench agent.py:agent --suite ./my_tasks.jsonl --max-tasks 20

Verification runs real pytest on the workspace — not the model's say-so. Results are labelled swe_lite_local to distinguish them from the official Docker-based harness numbers. Use the official harness for published comparisons; use this for rapid iteration.

---

Human-in-the-loop — require approval before dangerous actions

Pause an agent run and wait for a human to approve before taking an irreversible action.

from tvastar import tool
from tvastar.approval import require_approval

@tool
async def deploy_to_production(environment: str, ctx) -> str:
    """Deploy the current build to an environment."""
    await require_approval(
        f"Deploy to {environment!r}? This will affect live users.",
        timeout=120,   # seconds to wait for a human response
    )
    return do_deploy(environment)

Three backends — pick the one that fits your stack:

from tvastar.approval import ApprovalGate, set_default_gate

# CLI — prints to terminal, reads stdin (default, good for development)
set_default_gate(ApprovalGate(backend="cli"))

# Webhook — POST to your app, resolve via HTTP callback
gate = ApprovalGate(backend="webhook", webhook_url="https://myapp.com/approvals")
set_default_gate(gate)

# Event — you control resolution from outside the agent loop
pending_requests = []
gate = ApprovalGate(
    backend="event",
    on_request=lambda req: pending_requests.append(req),
)
# Later: pending_requests[0].approve() or .deny()

---

Cost tracking — know what every run costs

from tvastar.cost import cost_for_model, BudgetPolicy, BudgetExceeded

# Check cost for a model + token counts
cost = cost_for_model("claude-opus-4-6", input_tokens=1000, output_tokens=500)
print(f"${cost.usd:.4f}")   # $0.0525

# Enforce a budget — raises BudgetExceeded if the run exceeds it
agent = create_agent(
    "assistant",
    model=AnthropicModel("claude-opus-4-6"),
    budget=BudgetPolicy(max_usd=0.50, on_exceed="stop"),
)

# Or check manually after a run
result = await harness.run("Analyse this codebase")
if hasattr(result, "cost"):
    print(f"Run cost: ${result.cost.usd:.4f}")

Per-phase cost attribution

Track which workflow phases consume the most budget:

budget = BudgetPolicy(max_usd=2.0, on_exceed="stop")

async with budget.phase("planning"):
    result1 = await sess.prompt("Plan the approach")

async with budget.phase("execution"):
    result2 = await sess.prompt("Execute the plan")

# See where the money went
for phase, cost in budget.cost_breakdown().items():
    print(f"  {phase}: ${cost.usd:.4f}")
# planning: $0.0120
# execution: $0.0450

budget.reset_phases()  # clear for next run

Supported models with automatic pricing: Claude (all tiers), GPT-4o, GPT-4o-mini, o1, o3-mini, Llama via Groq, and more. Add custom rates to COST_TABLE.

---

What we're building

Tvastar is the engine. Every product below is built on top of it — same harness, same tools, same deploy model. Framework features get added only when a product needs them.

---

✅ tvastar-fix — Auto-repair failing tests

Shipped. The reference implementation.

Your CI fails. tvastar-fix runs the agent, edits the source, re-runs the suite itself, and pushes the fix — without you touching a line. Verification is a real exit code, never the model's claim.

pip install "tvastar[fix]"
tvastar-fix --test-cmd "pytest tests/" --model claude-opus-4-6

---

✅ tvastar-outbound — AI outbound sales agent

Shipped v0.9.0.

Give it a CSV of leads. It researches each one in parallel (company site, news, LinkedIn via web_browse + web_search), scores and prioritises them with TaskGraph, writes a personalised cold email for each, waits for your approval via ApprovalGate, then sends. Full audit trail in the trace viewer.

pip install tvastar
tvastar-outbound --csv leads.csv --icp "B2B SaaS, 50+ employees" \
    --sender-name "Jane" --sender-company "Acme" --sender-email [e-mail] \
    --min-score 0.6 --dry-run

Or programmatically:

from tvastar.outbound import run_campaign
from tvastar.model import AnthropicModel

result = await run_campaign(
    "leads.csv",
    model=AnthropicModel("claude-sonnet-4-5"),
    icp="B2B SaaS companies with 50+ employees struggling with developer productivity",
    sender_name="Jane Smith",
    sender_company="Acme",
    sender_email="[e-mail]",
    min_score=0.6,
)
print(f"Sent {result.sent}/{result.leads_qualified} emails.")

Why Tvastar is the right engine:

• TaskGraph researches all leads in parallel — 50 leads in wall-clock time of 1
• web_browse + web_search — no external scraping service needed
• ApprovalGate — human reviews every draft before anything goes out
• BudgetPolicy — hard cost ceiling per campaign
• JSONLExporter + tvastar ui — see every email and every research step

---

🔒 tvastar-comply — PII / PFI / PHI compliance layer

Core shipped in tvastar.assurance. Token-vault rehydration coming in v0.16.0.

Healthcare, finance, and legal companies cannot use AI agents on real customer data without a compliance layer. The redaction and audit-trail layer is already in tvastar.assurance — no extra install needed.

What's shipped today (tvastar.assurance):

Capability	API
SSN, DOB, email, phone, IP, credit card redaction	SanitizationPolicy.hipaa() / .pci() / .gdpr()
ML entity detection (names, locations, passports)	SanitizationPolicy.presidio()
Cryptographic proof PII was removed	Hash covers sanitized form — receipt.verify()
Role-gated audit log access	TrustLog(can_read=fn)
Retention + legal hold	RetentionPolicy
Per-run compliance report	receipt.to_audit_report()

from tvastar.assurance import AssurancePolicy, SanitizationPolicy, TrustLog

agent = create_agent(
    "clinical-assistant",
    model=AnthropicModel(),
    assurance=AssurancePolicy(
        log=TrustLog(".tvastar-trust.jsonl"),
        sanitize=SanitizationPolicy.hipaa(),   # PHI redacted before hash
        min_score=80,
    ),
)
# Input:  "Jane Doe, SSN 123-45-6789, diagnosis: hypertension"
# → Receipt stores: "Jane Doe, SSN [SSN], diagnosis: hypertension"
# → Hash proves PII was removed — tamper-evident

Coming in v0.16.0 (tvastar-comply):

• Token-vault rehydration — [SSN_1] → original value, post-LLM, on your infra
• CCPA + GLBA coverage
• Enterprise compliance dashboard

The vault stays local. No PII ever leaves your infrastructure.

---

📋 tvastar-review — GitHub PR review bot

Coming after tvastar-outbound.

Webhook fires on PR open → agent reads the diff → posts inline comments → flags shallow or unverified completions using the built-in detectors. Ships as a zero-config GitHub Action.

- uses: vanamayaswanth/tvastar-review@v1
  with:
    model: claude-sonnet-4-6

---

🛠 tvastar-devops — Production auto-heal agent

Extending tvastar-fix to live systems.

Log watcher detects anomaly → agent diagnoses root cause → runs bash fix → verifies with a real exit code → pages you only if it cannot fix it. Same "verify with real signals" principle as tvastar-fix, extended to production incidents.

---

💬 tvastar-support — Customer support agent

Multi-platform, persistent, production-ready.

One session per user, memory across conversations, simultaneous Telegram / Slack / email. dispatch() per inbound message, on_complete sends the reply. Human escalation via ApprovalGate when confidence is low.

---

🔍 tvastar-research — Competitive intel agent

Parallel web research → structured report.

Describe what you want to know. Agent fans out across sources with fan_out(), synthesises with structured output (result=), delivers a report. VCs, analysts, marketing teams.

---

Roadmap

Products ship first. Framework features get added only when a product needs them.

Milestone	What ships	Status
Web tools	web_browse + web_search — Jina AI, zero deps	✅ v0.8.1
DAG execution	TaskGraph — parallel tasks, critical path only	✅ v0.8.0
tvastar-outbound	Outbound sales agent — research → score → email → send	✅ v0.9.0
SOTA safety	Governance, transactions, LTM, memory cap, OpenAI retry	✅ v0.10.0
Loop Engineering	Loop, 7 patterns, CLI, MakerChecker, L0→L3 audit	✅ v0.11.0
Self-Improving Loops	meta_model prompt evolution, generational archive, MakerChecker cross-run memory	✅ v0.12.0
Loop Quality	score_run(), LoopQualityReport, tvastar quality CLI, 14 source bug fixes, security hardening	✅ v0.13.0
Plug into anything	tvastar.wrap, adapters.openai, adapters.langgraph, adapters.agentcore — Loop Quality on any framework	✅ v0.14.0
Verifiable Execution	AssurancePolicy, ExecutionReceipt, TrustLog — cryptographic receipts + SLA enforcement	✅ v0.15.0
Audit reports	receipt.to_audit_report() — text + HTML, hand to a lawyer	✅ v0.15.1
7 regulatory gaps	model tracking, tool outputs, PII redaction, human approver, access control, breach alert, retention	✅ v0.15.2–v0.15.4
Presidio ML PII	SanitizationPolicy.presidio() — 50+ entity types, 15+ languages	✅ v0.15.3
Silent-failure benchmark	tau2-bench 10,832 trajectories — 100% detection on false-success	✅ v0.18.0
Core primitives upgrade	enforce(), durable checkpoints, profile-keyed MockModel, score_pipeline, detect_from_messages, redact_messages, scan_messages_for_injection, composable tracer helpers	✅ v0.19.0
Agent Debugger example	Meta-agent that diagnoses, fixes, and verifies failing trajectories — exercises every framework feature	✅ v0.19.0
tvastar-comply	Token-vault PII rehydration, CCPA / GLBA coverage, enterprise dashboard	🔒 v0.16.0
tvastar-review	GitHub PR bot — diff → inline comments → GitHub Action	📋 v1.0.0
tvastar-devops	Auto-heal production incidents	📋 v1.1.0
tvastar-support	Multi-platform customer support agent	📋 v1.2.0
Hosted platform	Cloud-hosted harness, product dashboard, skill marketplace	📋 v2.0.0

Framework features are only added when a product needs them — not to match a checklist. tvastar-comply unlocks healthcare, finance, and legal — the highest-value enterprise markets.

---

Testing

MockModel makes agents fully testable without API calls. Pass a script list — one string per model turn:

import asyncio
import pytest
from tvastar import create_agent, Harness
from tvastar.tools.base import tool
from tvastar.model.mock import MockModel

# Tools under test
@tool
def add(a: int, b: int) -> int:
    "Add two integers."
    return a + b

def test_agent_uses_tool():
    # Script: first response requests the tool, second uses its result
    spec = create_agent(
        "calc",
        model=MockModel(script=[
            '{"type":"tool_use","name":"add","input":{"a":2,"b":3}}',
            "The answer is 5.",
        ]),
        instructions="Use the add tool.",
        tools=[add],
    )
    result = asyncio.run(Harness(spec).run("What is 2 + 3?"))
    assert "5" in result.text
    assert result.ok

def test_structured_output():
    from pydantic import BaseModel

    class Answer(BaseModel):
        value: int

    spec = create_agent(
        "q",
        model=MockModel(script=['{"value": 42}']),
        instructions="Return structured answers.",
    )

    async def run():
        sess = Harness(spec).session()
        result = await sess.prompt("What is the answer?", result=Answer)
        assert result.data.value == 42

    asyncio.run(run())

MockModel also works for loop tests:

from tvastar.loop.patterns import CISweeper

def test_loop_pass():
    loop = CISweeper(
        model=MockModel(script=["All CI checks passed."]),
        schedule="@manual",
    )
    run = asyncio.run(loop.trigger())
    assert run.state.value == "pass"

---

Troubleshooting

ImportError: No module named 'anthropic' Install the extras:

pip install "tvastar[anthropic]"   # for Claude
pip install "tvastar[openai]"      # for OpenAI / Groq / Ollama

AuthenticationError / 401 Unauthorized Your API key is missing or wrong:

echo $ANTHROPIC_API_KEY   # should print your key (not empty)
export ANTHROPIC_API_KEY="sk-ant-..."

result.ok is False Check what stopped the run and what detectors fired:

print(result.stopped)   # "end_turn" | "max_steps" | "error"
for f in result.findings:
    print(f.severity.value, f.detector, f.message)

Agent hits max_steps before finishing Either increase the limit or split into two smaller tasks:

spec = create_agent(..., max_steps=40)

thrash_loop finding — agent calls the same tool repeatedly The agent is stuck in a loop. Check the tool's return value — it may be returning an error the agent cannot make progress on. Also try:

spec = create_agent(..., max_steps=15)  # lower ceiling forces earlier escalation

Compaction fires too aggressively / not enough Tune the policy:

CompactionPolicy(
    max_messages=60,   # compact only when > 60 messages
    keep_last=10,      # always keep last 10
    min_messages=20,   # never compact below 20 total
)

Loop stays SUSPENDED after fixing the root cause The circuit breaker tripped after too many consecutive failures. Reset it:

loop.reset()

LoopState.HANDOFF_FAILED — handoff itself threw The handoff handler (Slack, webhook, etc.) failed 3× with backoff. Check connectivity and credentials for your HandoffPolicy implementation. The run is still recorded — you won't lose data.

TvastarError: Loop file not found The path you passed to tvastar loop run does not exist:

tvastar loop init CISweeper                    # creates the file
tvastar loop run .tvastar/loops/ci_sweeper.py:loop

---

Further reading

• Getting Started — install → first agent → first loop in 5 minutes
• Usage Guide — decision trees for every API choice
• API Reference — every public symbol, fully typed
• Patterns Cookbook — 38 copy-paste recipes
• 12-Factor Agents map — how Tvastar maps to the production checklist (honest verdicts)
• AGENTS.md — contributor guide for working in this repo
• CLAUDE.md — codebase map for AI assistants

---

License

MIT