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 (OMEGA droplet) 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, cost_usd=0.0, 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 the OMEGA token_usage / subscription_usage TimescaleDB hypertables (see hedge-fund-mcp database/init_timescale.sql). 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 = 0
cost_usd: float = 0.0
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, cost_usd=0.0, duration_s=0.0, cached=False, error_class=None, extra=None)[source]

Construct an AttributionEvent with timestamp=now.

Parameters:
Return type:

AttributionEvent