Sinks
Configure where Spineforge sends telemetry events — console, file, backend, or your own.
Environment variables
| Variable | Default | Description |
|---|---|---|
SPINEFORGE_DATA_DIR | CWD | Base directory for .spineforge/ (registry + logs) |
SPINEFORGE_LOG_FILE | actions.jsonl | JSONL log filename inside .spineforge/logs/ |
SPINEFORGE_VERBOSE | true | Enable ConsoleSink (coloured one-liners to stdout) |
SPINEFORGE_TRACE_CONTENT | true | Capture prompt / completion text in action events |
SPINEFORGE_REGISTRY_URL | (empty) | Backend URL — empty = offline mode (no APISink) |
All can also be passed as kwargs to spineforge.init() — kwargs win over env vars.
ConsoleSink
Prints a compact, coloured one-liner per event to stdout. Enabled when SPINEFORGE_VERBOSE=true.
🦴 [SPINE:a1b2c3d4] RUN started input="What is the capital of France?"
🦴 [SPINE:a1b2c3d4] [RUN:e5f6g7h8] ⚡ llm_call gpt-4o 612ms ✓ (prompt:48 comp:12)
🦴 [SPINE:a1b2c3d4] [RUN:e5f6g7h8] 🔧 tool_call search_web 341ms ✓
🦴 [SPINE:a1b2c3d4] [RUN:e5f6g7h8] RUN completed 1024ms ✓To disable: SPINEFORGE_VERBOSE=false or spineforge.init(..., verbose=False)
FileSink
Appends events as JSON lines to <data_dir>/.spineforge/logs/actions.jsonl. Non-blocking — events are pushed to an in-memory queue and written by a background daemon thread.flush() drains the queue synchronously (called automatically at process exit via atexit).
FileSink is always active. It also serves as the fallback for APISink on connection failures.
APISink
Active when SPINEFORGE_REGISTRY_URL is set. POSTs telemetry batches to the Spineforge backend at /telemetry/ingest. Authenticated with a scoped JWT (telemetry:write scope). Batch size: up to 50 events, flushed every 5 seconds.
- 200 — all events accepted
- 207 — partial success; failed events fall back to FileSink
- Non-2xx / connection error — entire batch falls back to FileSink
Custom sinks
Implement the Sink ABC and pass it to init():
from spineforge.sinks import Sink
class MyDataWarehouseSink(Sink):
def emit(self, event: dict) -> None:
if event["event"] == "action":
self.warehouse.insert("agent_actions", event)
def flush(self) -> None:
self.warehouse.commit()
def shutdown(self) -> None:
self.flush()
self.warehouse.close()No changes to instrumentation or span processing needed — just add the sink to your init call.
init(extra_sinks=[...]) is planned. Currently, custom sinks require a light wrapper around spineforge.init().Disabling content capture
If you don't want prompt or completion text stored (e.g. for compliance):
spine = spineforge.init(
agent_name="my-agent",
trace_content=False # or SPINEFORGE_TRACE_CONTENT=false
)Token counts and costs are still captured. Only the text content is omitted.
Next steps
- SSO / OIDC — configure agent ownership via your IdP
- Telemetry — how the sink pipeline connects to the OTel span processor