blockether.vis.activity

Immutable Activity receipts, counts, grouping and evidence.

For reading event data, see the Sessions API guide. For authoring tool presentations, see the Activity presentation guide.

@dataclass(frozen=True, slots=True)

class ActivityResource:

A typed reference to a resource involved in an invocation, identified by id.

ActivityResource(type: str, id: str)

type: str

id: str

@dataclass(frozen=True, slots=True)

class ActivityDiffLine:

One classified diff line. is_redacted marks content withheld from the receipt.

ActivityDiffLine(kind: str, text: str, is_redacted: bool | None = None)

kind: str

text: str

is_redacted: bool | None

@dataclass(frozen=True, slots=True)

class ActivityEvidence:

Text or structured change evidence attached to an invocation.

Optional counts describe the observed changes. Truncation and redaction flags explain why the displayed evidence can be incomplete.

kind: str

text: str

lines: tuple[ActivityDiffLine, ...] | None

additions: int | None

deletions: int | None

modifications: int | None

is_truncated: bool | None

is_redacted: bool | None

@dataclass(frozen=True, slots=True)

class ActivityRow:

One invocation's lifecycle, outcome and human-readable evidence.

id and sequence keep identity and order. state describes the outcome. summary is display text and does not replace the status. Linked receipts appear as children with their shared handle_id. presentation contains the content that the extension selected. evidence keeps what the engine observed.

ActivityRow( id: str, sequence: int, operation: str, presenter: str, signal: str, state: str, summary: str, resources: tuple[ActivityResource, ...], evidence: tuple[ActivityEvidence, ...], argument_key: str | None = None, read_key: str | None = None, handle_id: str | None = None, group_token: str | None = None, duration_ms: int | None = None, result_summary: str | None = None, error_summary: str | None = None, summary_format: str | None = None, result_format: str | None = None, is_truncated: bool | None = None, children: tuple[ActivityRow, ...] | None = None, presentation: Mapping[str, typing.Any] | None = None)

id: str

sequence: int

operation: str

presenter: str

signal: str

state: str

summary: str

resources: tuple[ActivityResource, ...]

evidence: tuple[ActivityEvidence, ...]

argument_key: str | None

read_key: str | None

handle_id: str | None

group_token: str | None

duration_ms: int | None

result_summary: str | None

error_summary: str | None

summary_format: str | None

result_format: str | None

is_truncated: bool | None

children: tuple[ActivityRow, ...] | None

presentation: Mapping[str, typing.Any] | None

@dataclass(frozen=True, slots=True)

class ActivityArgumentGroup:

One operation with identical complete arguments. All invocation evidence is kept.

ActivityArgumentGroup(id: str, rows: tuple[ActivityRow, ...])

id: str

rows: tuple[ActivityRow, ...]

@dataclass(frozen=True, slots=True)

class ActivityGroup:

One exact operation across the block, ordered by first invocation.

Shell polling remains child evidence. Unknown operations use their exact names. Disclosure state belongs to the reader, never to this receipt.

ActivityGroup( id: str, label: str, rows: tuple[ActivityRow, ...])

id: str

label: str

rows: tuple[ActivityRow, ...]

argument_groups: tuple[ActivityArgumentGroup, ...]

Repeated arguments in first-entry order. Unknown argument keys stay separate.

@dataclass(frozen=True, slots=True)

class ActivityCounts:

Invocation counts by lifecycle outcome, distinct from the visible row count.

ActivityCounts(running: int, succeeded: int, failed: int, cancelled: int)

running: int

succeeded: int

failed: int

cancelled: int

@dataclass(frozen=True, slots=True)

class ActivityProjection:

One form's Activity receipt, optionally a page of its durable history.

For external data, use from_wire instead of direct construction. rows and nested values are immutable. to_wire makes an independent, mutable, JSON-compatible copy. Grouping properties are computed, not stored.

ActivityProjection( state: str, counts: ActivityCounts, rows: tuple[ActivityRow, ...], history: Mapping[str, typing.Any] | None = None)

state: str

counts: ActivityCounts

rows: tuple[ActivityRow, ...]

history: Mapping[str, typing.Any] | None

groups: tuple[ActivityGroup, ...]

The same per-operation groups that Companion and TUI use.

They are not serialized.

argument_groups: tuple[ActivityArgumentGroup, ...]

Exact operation/argument pairs within this block. They are not serialized.

@classmethod

def from_wire(cls, value: Any) -> ActivityProjection:

Validate a receipt and construct immutable rows and nested records.

Raises:

ValueError: The schema, identity uniqueness, presentation bounds or history page limits are invalid.

def to_wire(self) -> dict[str, typing.Any]:

Return a fresh portable receipt without computed reader grouping fields.