forge.telemetry

Python telemetry package; imports are explicit from this package.

Package contract

Field	Value
Language	python
Source version	0.1.0
Manifest	`forge-py/pyproject.toml`
Source files	5
Evidence	Source reference; registry publication and runtime conformance are separate checks

Import boundary

import forge.telemetry

Use a source checkout or your verified private registry. Manifest coordinates identify the package; they do not establish that a public registry release exists.

Download package reference JSON. Each original source file and generated declaration artifact has its own SHA-256 digest. Function bodies and constant values are omitted from downloads. These are source declaration inventories, not compiler-resolved rustdoc, TypeDoc, DocC, or Dokka output. Private modules can contain public declarations that are not reachable through the package boundary; consult the entry point before importing.

audit.py

Read declaration text · 9 declaration entries

class AuditEntry()

class AuditTrail()

def __init__(self, agent_did: str) -> None

def agent_did(self) -> str

def append(self, event: AuditEvent) -> None

def entries(self) -> list[AuditEntry]

def is_empty(self) -> bool

def last(self) -> AuditEntry | None

def filter_by_kind(self, kind: AuditEventKind) -> list[AuditEntry]

collector.py

Read declaration text · 28 declaration entries

class CompletedSpan()

class SpanCollector(Protocol)

def record_span(self, span: CompletedSpan) -> None

def spans(self) -> list[CompletedSpan]

def clear(self) -> None

def is_empty(self) -> bool

class NoopCollector()

def record_span(self, span: CompletedSpan) -> None

def spans(self) -> list[CompletedSpan]

def clear(self) -> None

def is_empty(self) -> bool

class InMemoryCollector()

def __init__(self) -> None

def record_span(self, span: CompletedSpan) -> None

def spans(self) -> list[CompletedSpan]

def clear(self) -> None

def is_empty(self) -> bool

class InMemoryTelemetry()

def __init__(self, agent_did: str | None=None) -> None

def start_span(self, name: str, attributes: list[tuple[str, str]] | None=None) -> SpanId

def end_span(self, span_id: SpanId) -> None

def emit_event(self, event: AuditEvent) -> None

def flush(self) -> None

def completed_spans(self) -> list[CompletedSpan]

def events(self) -> list[AuditEvent]

def clear(self) -> None

def span_count(self) -> int

def event_count(self) -> int

contract.py

Read declaration text · 18 declaration entries

class SpanId()

def __init__(self, value: str) -> None

def generate(cls) -> SpanId

def from_string(cls, value: str) -> SpanId

class AuditEventKind(Enum)

class AuditEvent()

def new(cls, kind: AuditEventKind, agent_did: str, details: Any) -> AuditEvent

def with_signature(self, signature: str) -> AuditEvent

class TelemetryContract(Protocol)

def start_span(self, name: str, attributes: list[tuple[str, str]] | None=None) -> SpanId

def end_span(self, span_id: SpanId) -> None

def emit_event(self, event: AuditEvent) -> None

def flush(self) -> None

class NoopTelemetry()

def start_span(self, name: str, attributes: list[tuple[str, str]] | None=None) -> SpanId

def end_span(self, span_id: SpanId) -> None

def emit_event(self, event: AuditEvent) -> None

def flush(self) -> None

errors.py

Read declaration text · 15 declaration entries

class TelemetryError(ForgeError)

class SpanNotFoundError(TelemetryError)

def __init__(self, span_id: str) -> None

class SpanAlreadyClosedError(TelemetryError)

def __init__(self, span_id: str) -> None

class AuditEntryInvalidError(TelemetryError)

def __init__(self, reason: str) -> None

class AuditTrailCorruptedError(TelemetryError)

def __init__(self, index: int, reason: str) -> None

class FlushFailedError(TelemetryError)

def __init__(self, reason: str) -> None

class CollectorFullError(TelemetryError)

def __init__(self, capacity: int) -> None

class ExportFailedError(TelemetryError)

def __init__(self, reason: str) -> None