tvastar 0.16.2

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

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_...",
)

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

Problem	How Tvastar handles it
Code execution without Docker	In-memory sandbox, zero setup
Agent claims success but fails	Built-in silent failure detection
Crash at step 47 of 50	Step-level checkpoint and resume
Deploy to Lambda, GitHub Actions, web	Single agent definition, any target
Agent loops on the same tool	Built-in loop detection
Context grows past model limit	Automatic compaction and summarisation
Audit what the agent actually did	Full transcript stored every run
Inspect runs visually	Built-in trace viewer UI (tvastar ui)
Flaky network tools fail mid-run	Per-tool retry with exponential backoff
Run 100 prompts at once	Built-in parallel fan-out
Stream tokens to the browser	SSE endpoint out of the box
Tool called in wrong execution phase	GovernancePolicy — tamper-proof phase enforcement
Filesystem changes need atomic rollback	harness.transaction() + sandbox snapshot/restore
Agent needs memory across sessions	tvastar.contrib.ltm — post-session LTM consolidation
Session messages balloon past 50 MB	memory_cap_mb — hard cap with auto-compaction
Regulator asks "what did your AI do?"	ExecutionReceipt — cryptographic proof, court-ready
PHI / PII ends up in your audit log	SanitizationPolicy — redact before hashing, chain stays intact
Agent runs a tool, no record of the output	tool_calls[].output — every input AND output captured
No proof a human approved the action	approvals[] in receipt — who approved, when, what
Anyone on the team can read the audit log	TrustLog(can_read=...) — role-based access, PermissionError if denied
Audit log tampered silently	on_breach callback — fires immediately, per-receipt
7-year SOX retention, data must not grow forever	RetentionPolicy — archive old entries, legal hold freezes the log

---

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[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.

---

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.

---

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)

---

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)

Set meta_model on any LoopConfig and the loop will rewrite 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"))

# After each FAIL, meta_model rewrites the worker's instructions and the next
# retry uses the improved version. Every run is recorded as a LoopGeneration.
run = await loop.trigger()

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

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().

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)

---

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()

---

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.

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])

---

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)

---

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.

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")

# 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}")

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
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 — 25 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