Attribution — swarph_mesh.attribution

Per-call attribution events plus pluggable AttributionWriter Protocol.

Tier 2 stability. AttributionWriter Protocol + default writers (FileAttributionWriter, NullAttributionWriter) are consumer-facing.

Attribution event + writer protocol.

Per PLAN.md §8, every SwarphCall writes one row of attribution data after a successful adapter call. The package ships writer implementations for common cases (no-op, JSONL file); production consumers wire up a TSDB writer at module load.

The writer is a Protocol — consumers can plug in any callable matching the shape. Default in v0.1.0: FileAttributionWriter appending to ~/.swarph/attribution.jsonl so cost data isn’t silently dropped on first run, but no DB driver is forced.

class swarph_mesh.attribution.AttributionEvent(timestamp, provider, model, role, caller, mesh_peer, input_tokens, output_tokens, cached_tokens=0, thinking_tokens=None, cache_read_tokens=None, cache_creation_1h=None, cache_creation_5m=None, cost_usd=0.0, cost_basis='unknown', duration_s=0.0, cached=False, error_class=None, extra=<factory>)[source]

Bases: object

One row of cost / latency / token attribution for an LLM call.

Fields match the shape consumed by a token_usage / subscription_usage TimescaleDB hypertables (see a time-series store). External consumers can map to their own schema in their writer impl.

Parameters:

  • timestamp (float)

  • provider (str)

  • model (str)

  • role (str)

  • caller (str)

  • mesh_peer (str | None)

  • input_tokens (int)

  • output_tokens (int)

  • cached_tokens (int)

  • thinking_tokens (int | None)

  • cache_read_tokens (int | None)

  • cache_creation_1h (int | None)

  • cache_creation_5m (int | None)

  • cost_usd (float)

  • cost_basis (str)

  • duration_s (float)

  • cached (bool)

  • error_class (str | None)

  • extra (dict[str, Any])

timestamp: float
provider: str
model: str
role: str
caller: str
mesh_peer: str | None
input_tokens: int
output_tokens: int
cached_tokens: int = 0
thinking_tokens: int | None = None
cache_read_tokens: int | None = None
cache_creation_1h: int | None = None
cache_creation_5m: int | None = None
cost_usd: float = 0.0
cost_basis: str = 'unknown'
duration_s: float = 0.0
cached: bool = False
error_class: str | None = None
extra: dict[str, Any]
to_dict()[source]
Return type:

dict[str, Any]

class swarph_mesh.attribution.AttributionWriter(*args, **kwargs)[source]

Bases: Protocol

Protocol for sink-side attribution writers. Implementations can write to TSDB, JSONL files, OTLP, stdout, or be no-op.

async write(event)[source]
Parameters:

event (AttributionEvent)

Return type:

None

class swarph_mesh.attribution.NullAttributionWriter[source]

Bases: object

No-op. Useful in test fixtures where attribution noise is distracting and the call’s LLMResponse already carries the cost/token data inline.

async write(event)[source]
Parameters:

event (AttributionEvent)

Return type:

None

class swarph_mesh.attribution.FileAttributionWriter(path=None)[source]

Bases: object

Append-only JSONL writer at ~/.swarph/attribution.jsonl.

Default writer in v0.1.0 — every call lands a parseable line, no DB driver required. Single-process safe via an asyncio.Lock. Multi-process scenarios (which v0.1.0 doesn’t target) would need a file-lock; document and defer.

Parameters:

path (Optional[Path])

async write(event)[source]
Parameters:

event (AttributionEvent)

Return type:

None

swarph_mesh.attribution.get_default_writer()[source]

Return the module-level default writer. SwarphCall uses this when no per-call writer is provided.

Return type:

AttributionWriter

swarph_mesh.attribution.set_default_writer(writer)[source]

Override the module-level default writer (e.g., production consumers call this at startup with a TSDB-backed writer).

Parameters:

writer (AttributionWriter)

Return type:

None

swarph_mesh.attribution.make_event(*, provider, model, role, caller, mesh_peer, input_tokens, output_tokens, cached_tokens=0, thinking_tokens=None, cache_read_tokens=None, cache_creation_1h=None, cache_creation_5m=None, cost_usd=0.0, cost_basis='unknown', duration_s=0.0, cached=False, error_class=None, extra=None)[source]

Construct an AttributionEvent with timestamp=now.

Parameters:

  • provider (str)

  • model (str)

  • role (str)

  • caller (str)

  • mesh_peer (str | None)

  • input_tokens (int)

  • output_tokens (int)

  • cached_tokens (int)

  • thinking_tokens (int | None)

  • cache_read_tokens (int | None)

  • cache_creation_1h (int | None)

  • cache_creation_5m (int | None)

  • cost_usd (float)

  • cost_basis (str)

  • duration_s (float)

  • cached (bool)

  • error_class (str | None)

  • extra (dict[str, Any] | None)

Return type:

AttributionEvent