One request_id.
Three streams. Four pillars.

Get started in 30 seconds

One env var picks the engine.
Same code, SQLite to Postgres.

fasten is a library — runs in your process, Cloud optional.
Audit rows go to durable storage.

# .env
FASTEN_SERVICE_ID=my-service
FASTEN_NODE_ID=host-01
FASTEN_AUDIT_DSN=sqlite:///./audit.db

# app.py
import fasten
fasten.init()  # reads env

# .env
FASTEN_SERVICE_ID=my-service
FASTEN_NODE_ID=host-01
FASTEN_AUDIT_DSN=postgres://user:pw
  @db:5432/app?table=audit_log

# app.py — identical
import fasten
fasten.init()  # reads env

The emit call is identical in both modes. Storage swapped at the env-var boundary.

audit model

5 Ws + H — enforced at the type level.

The classical audit vocabulary. No string-soup attributes.

same shape, six languages

init → emit + log.
That's the entire API surface.

Each language's signature matches its dominant structured-logging idiom (structlog / slog / tracing / pino / SLF4J).

# .env — secrets never appear in code
FASTEN_SERVICE_ID=api-server
FASTEN_NODE_ID=host-01
FASTEN_AUDIT_DSN=postgres://user:pw@db/app?table=audit_log

# app.py
import fasten
from fasten.shim.http import RequestIDMiddleware, APILogger

fasten.init()                                      # reads env
app.add_middleware(RequestIDMiddleware)            # mints / honours X-Request-ID
app.add_middleware(APILogger, skip={"/health"}) # one api row per request

@app.post("/users")
async def create(u, actor):
    fasten.emit(code="USER_CREATED", target=u.id,
               actor=actor.id, detail={"email": u.email})
    fasten.log.info("user_created", user_id=u.id)

// .env
FASTEN_SERVICE_ID=checkout-svc
FASTEN_AUDIT_DSN=postgres://user:${DB_PW}@db/shop?table=audit_log

// main.go
import (
    "github.com/nerdapplabs/fasten-go/fasten"
    httpshim "github.com/nerdapplabs/fasten-go/fasten/shim/http"
)

fasten.Init(fasten.Config{})                        // reads env

r := chi.NewRouter()
r.Use(httpshim.RequestID)                           // mint/honour X-Request-ID
r.Use(fasten.APILogger("/health"))                 // one api row per request

r.Post("/orders", func(w http.ResponseWriter, req *http.Request) {
    ctx := req.Context()
    fasten.Emit(ctx, audit.CodeOrderPlaced,
        fasten.Target(orderID),
        fasten.Detail(map[string]any{"amount": amount}),
    )
    fasten.Log(ctx).Info("order_placed", "order_id", orderID)
})

// .env
FASTEN_SERVICE_ID=notification-svc
FASTEN_AUDIT_DSN=sqlite:///data/app.db?table=audit_log

// server.ts
import fasten, { EmitInput, withRequestID, mintID } from '@nerdapplabs/fasten';

fasten.init();                                     // reads env

app.post('/sensors/:id', async (req, res) => {
    // honour or mint X-Request-ID; everything emit/log inside inherits it
    const rid = req.headers['x-request-id'] ?? mintID();
    await withRequestID(rid, async () => {
        fasten.emit({
            code: 'SENSOR_UPDATED',
            target: req.params.id, actor: 'system',
            detail: { reading: req.body.value },
        } satisfies EmitInput);
        fasten.log.info('sensor_updated', { id: req.params.id });
    });
    res.sendStatus(200);
});

// .env
FASTEN_SERVICE_ID=batch-worker
FASTEN_AUDIT_DSN=sqlite:///var/lib/worker/audit.db?table=audit_log

// main.rs
use fasten::{EmitBuilder, Config};

let cfg = fasten::config_from_env()?;

// Fluent Emit — anchors auto-filled from ctx + config
fasten::EmitBuilder::new("BATCH_PROCESSED", "batch-4211")
    .actor("worker-01", "service")
    .detail(detail_map)
    .emit()
    .await?;

// tracing-style log
tracing::info!(batch_id = "4211", "batch_completed");

AI agent fit

Agents call tools. Tools touch resources.
Who audits that today? Nobody.

LangSmith, OTel, OpenAI Assistants — LLM-ops or timing tools, not typed audit. fasten fits natively: every tool call is an audit row — actor / target / method / correlation.

advanced · optional

Beyond HTTP — cross-transport correlation.

HTTP is table stakes. fasten's wedge is the transports HTTP-only tools skip:

• MQTT — industrial / IoT / pub-sub.
• scheduler — autonomous, cron-fired jobs.

Each is an opt-in shim. Most adopters never need these.

# publisher (HTTP handler — user sets a new setpoint)
await fasten.shim.mqtt.publish(
    client, f"devices/{device_id}/setpoint",
    {"target_temp_c": 22.5},
)
# shim injects _req = current_request_id() into payload

# subscriber (device firmware, any language)
@mqtt_shim.handler("devices/+/setpoint")
async def on_setpoint(msg, ctx):
    # ctx.request_id == original HTTP caller's id
    fasten.log.info("setpoint_received")
    fasten.emit(code="DEVICE_SETPOINT_UPDATED", ...)

# operator — one query stitches both services
$ fasten tail --stream audit --request-id abc
SETPOINT_REQUESTED       api-server  actor=admin   target=dev-42
DEVICE_SETPOINT_UPDATED  device-42   actor=system  # same id ✓

@scheduler.cron("0 */6 * * *")
@fasten.shim.scheduler.instrument
async def nightly_cleanup():
    # request_id = "scheduler-a1b2c3d4"
    fasten.emit(code="RETENTION_SWEEP_RAN", ...)
    fasten.log.info("cleanup_done")
# all rows for this run share the same id

Custom transport (gRPC, NATS, AMQP, Kafka)? Write a 10-line shim following the same read-id-from-wire / set-ctx / call-downstream pattern.

what it replaces

Honest split. No overclaiming.

faq

Common questions.