TraceForge

Framework-Agnostic Execution Replay & Analysis Platform

---

⚡ Overview

TraceForge is a complete, developer-first execution tracing, replay, and analysis platform for Python software. It helps software teams capture the exact execution graph of their applications—nested spans, parent-child relationships, durations, structured events, exceptions, state snapshots, and context propagation—across synchronous and asynchronous workflows.

TraceForge provides an end-to-end ecosystem comprising:

• Python SDK: Zero-dependency hot-path tracing with thread & task isolation (contextvars).
• Storage & Query Engine: High-performance SQLite batch writer & append-only repositories.
• Replay & Execution Diff Engine: Historical execution tree reconstruction and deep diff comparisons.
• FastAPI HTTP Gateway & Web Dashboard: Production-grade REST API, JWT authentication, Prometheus /metrics, and embedded Web UI.
• Command-Line Interface (traceforge): Modular CLI for server lifecycle, replay analysis, visualization export, and project bootstrap.
• Plugin Subsystem: Dynamic plugin loading, registration, and isolated lifecycle execution.

---

💡 Why TraceForge?

Most Python observability tools are either heavy APM platforms (requiring external SaaS collectors, agents, and complex cloud infrastructure) or simple logging libraries that lack execution graph awareness. TraceForge bridges this gap as a lightweight, SQLite-native platform with zero external infrastructure dependencies. Unlike standard OpenTelemetry setups, TraceForge includes a built-in execution replay engine to reconstruct historical function trees offline and a structural diff engine to compare execution paths and latency regressions across runs. It installs in seconds via pip and runs entirely within your local or self-hosted environment.

---

🌟 Key Features

Feature	Description
Async-First Tracing	Native support for with and async with context managers, functions, and decorators.
Execution Replay	Reconstruct full historical execution graphs, session trees, timeline events, and state snapshots.
Execution Diff Engine	Compare two sessions/traces to detect timing regressions, structural execution branch shifts, and exceptions.
High-Performance Storage	Lock-free lockless ring-buffer async writer with SQLite persistence and zero-overhead hot-path emission.
Security Layer	JWT authentication, role-based access control (RBAC), and token rate-limiting middleware.
Web Dashboard & REST API	Embedded static dashboard (/dashboard), OpenAPI interactive docs (/docs), and /metrics.
Modular CLI	Comprehensive CLI tool (traceforge) to manage server instances, replay traces, export data, and inspect graphs.
Extensible Plugin System	Modular plugin registry and manager supporting lifecycle hooks (initialize, enable, shutdown).

---

🏗️ Architecture Flow

External Clients / CLI / Web Dashboard
                 │
                 ▼
  Authentication & Security (JWT, RBAC, Rate-Limiter)
                 │
                 ▼
      FastAPI Gateway / REST API Engine
                 │
                 ▼
      TraceForgeApiService Facade
        ┌────────┴────────┬─────────────────┐
        ▼                 ▼                 ▼
   Query Engine     Replay Engine   Execution Diff Engine
        │                 │                 │
        └────────┬────────┴─────────────────┘
                 ▼
     Storage Engine (SQLite Batch Writer)
                 ▲
                 │
           Recorder Engine
                 ▲
                 │
      Tracer / SDK Core (ContextVars Propagation)

---

📦 Installation

# Basic installation from source
pip install .

# Installation with development dependencies (pytest, mypy, ruff, httpx)
pip install ".[dev]"

# Installation with optional features
pip install ".[websocket,yaml]"

Requires Python 3.12+.

---

🚀 Quickstart

1. Python SDK Usage (Span Tracing & Storage)

import asyncio
import traceforge

# Initialize storage, exporters, and recorder
storage = traceforge.SQLiteStorage("traces.db")
recorder = traceforge.Recorder(
    storage=storage,
    exporters=[traceforge.ConsoleExporter()],
).start()

# Setup Tracer
tracer = traceforge.Tracer("order-service")
tracer.add_hook(recorder)

# Track execution with nested spans
with tracer.start_span("process-order") as span:
    span.set_attribute("customer.id", "cust_99812")

    with tracer.start_span("validate-inventory") as inv_span:
        inv_span.add_event("inventory-checked", payload={"sku": "ITEM-42", "qty": 1})

    with tracer.start_span("charge-card") as pay_span:
        pay_span.set_attribute("payment.method", "credit_card")

# Stop recorder to flush all spans to storage
recorder.stop()

# Query recorded spans back from storage
spans = asyncio.run(storage.query_spans(limit=10))
print(f"Recorded Spans: {len(spans)}")

2. Async Context & Decorators

import asyncio
from traceforge import traced, Tracer, configure

tracer = Tracer("async-service")
configure(tracer)

@traced(name="fetch-external-api")
async def fetch_user_data(user_id: str):
    # Async span automatically created and context-propagated
    return {"user_id": user_id, "status": "active"}

async def main():
    async with tracer.start_span("async-pipeline") as span:
        data = await fetch_user_data("usr_100")
        span.set_attribute("pipeline.complete", True)

asyncio.run(main())

3. Platform Query Engine, Execution Replay & Diff Analysis

from traceforge.service.service import TraceForgeApiService
from traceforge.storage.drivers.sqlite import SQLiteStorageDriver

# Connect to TraceForge relational database (initialized via SQLiteStorageDriver or CLI)
driver = SQLiteStorageDriver("traceforge.db")
conn = driver.connection_manager.get_connection()

# Query sessions via QueryEngine
service = TraceForgeApiService(conn)
sessions = service.query_engine.sessions.list()
print(f"Total Sessions: {len(sessions)}")

# Replay session execution graph
if sessions:
    session_replay = service.replay_session(sessions[0].session_id)
    print(f"Session Status: {session_replay.session.status}")
    print(f"Recorded Nodes: {len(session_replay.nodes)}")

---

💻 Command-Line Interface (CLI)

TraceForge comes with a modular CLI tool:

# Initialize a new TraceForge project workspace
traceforge init --name my-project

# Launch the FastAPI Gateway server & Web Dashboard
traceforge server --host 0.0.0.0 --port 8000 --db traceforge.db

# Replay an execution session directly from the CLI
traceforge replay sess_12345 --db traceforge.db

# Generate visualization models (graph, timeline, flamegraph)
traceforge visualize sess_12345 --type flamegraph --db traceforge.db

# Export session traces to JSON or Markdown
traceforge export sess_12345 --format markdown --output session_report.md

---

🌐 Web Dashboard & REST API

Launch the HTTP Gateway:

python -m traceforge.gateway.server

• Web Dashboard: http://localhost:8000/dashboard
• Interactive API Documentation (Swagger): http://localhost:8000/docs
• Prometheus Metrics: http://localhost:8000/metrics
• Health Check: http://localhost:8000/health

Key API Endpoints

• POST /api/v1/auth/token: Acquire JWT Bearer tokens
• GET /api/v1/sessions: List recorded sessions with filters
• GET /api/v1/sessions/{id}/replay: Retrieve historical session replay graph
• GET /api/v1/visualize/timeline/{id}: Fetch timeline visualization model
• POST /api/v1/diff/sessions: Compare two sessions for performance regressions

---

⚙️ Configuration System

TraceForge uses a hierarchical, immutable configuration loader:

1. CLI Arguments (highest priority)
2. Environment Variables (e.g. TRACEFORGE_STORAGE__DATABASE_URI=app.db)
3. Configuration Files (traceforge.yaml, traceforge.toml, traceforge.json)
4. Default Settings (lowest priority)

---

🧪 Testing & Verification

Run the test suite with coverage:

pytest --cov=traceforge

Run code formatting and type checks:

ruff check traceforge/
ruff format --check traceforge/
mypy traceforge/

---

📄 License

This project is licensed under the MIT License — see the LICENSE file for details.