Skip to content

blockether.vis.extension

Tool declarations, host APIs and extension lifecycle contracts.

For authoring workflows and examples, use the extension tutorial and extension API guide.

@runtime_checkable
class Host(typing.Protocol):

Operations every injected or outside vis host must implement.

Host(*args, **kwargs)
def state_get(self, key: str) -> Any:

Read one value out of the extension's durable state.

def state_put(self, key: str, value: Any) -> Any:

Write one JSON value into the extension's durable state.

def state_del(self, key: str) -> Any:

Drop one key from the extension's durable state.

def state_keys(self) -> Any:

List every key the extension's durable state holds.

def log(self, level: str, message: str) -> Any:

Emit one engine log line at a level.

def notify(self, text: str, level: str) -> Any:

Show one notification on the user's channel.

def shell(self, options: Mapping[str, typing.Any]) -> Mapping[str, typing.Any]:

Run one canonical shell operation and return its result shape.

def jailed_shell(self, options: Mapping[str, typing.Any]) -> Mapping[str, typing.Any]:

Run one shell op inside the workspace jail.

def jailed_shell_session(self, options: Mapping[str, typing.Any]) -> Mapping[str, typing.Any]:

Run one shell op inside a persistent jailed session.

def request_input( self, request_json: str, validator_arities_json: str, run_validator: Callable[[str, str], str]) -> str:

Ask the human, and block until the answer settles or is cancelled.

def live(self, envelope_json: str) -> str:

Open, patch, read or close a live view.

state accepts timeout_ms (0..86400000) and after_seq (nonnegative). A positive timeout blocks until the sequence differs or the view closes. An unchanged timeout returns is_open=True, timed_out=True, without view. Ordinary state reads and changed waits return the current view with seq.

def activity(self, presentation: dict[str, typing.Any]) -> bool:

Replace the running symbol's headline, summary, content and sections.

def reveal_secret(self, handle: str) -> Any:

Resolve a vis-secret: handle to its plaintext.

def forget_secret(self, handle: str) -> Any:

Drop the plaintext a secret handle stands for.

def declare_env(self, declarations_json: str) -> str:

Resolve the environment variables the extension declared.

@dataclass(frozen=True, slots=True)
class ActivityText(_ActivityBlock):

Plain Activity text, never a model message or a question.

ActivityText(text: str)
text: str
type: ClassVar[str] = 'text'
Inherited methods and attributes
def to_wire(self) -> dict[str, typing.Any]:

Return a fresh canonical Activity content block, with no host lifecycle fields.

@dataclass(frozen=True, slots=True)
class ActivityHeading(ActivityText):

Heading for a section of this invocation's current content.

ActivityHeading(text: str)
type: ClassVar[str] = 'heading'
Inherited methods and attributes

Inherited from ActivityText.text.

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

Return a fresh canonical Activity content block, with no host lifecycle fields.

@dataclass(frozen=True, slots=True)
class ActivityMarkdown(ActivityText):

Markdown content for this invocation.

ActivityMarkdown(text: str)
type: ClassVar[str] = 'markdown'
Inherited methods and attributes

Inherited from ActivityText.text.

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

Return a fresh canonical Activity content block, with no host lifecycle fields.

@dataclass(frozen=True, slots=True)
class ActivityCode(ActivityText):

Code with an optional language hint.

ActivityCode(text: str, language: str | None = None)
language: str | None
type: ClassVar[str] = 'code'
Inherited methods and attributes

Inherited from ActivityText.text.

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

Return a fresh canonical Activity content block, with no host lifecycle fields.

@dataclass(frozen=True, slots=True)
class ActivityDiff(ActivityCode):

A diff, not engine-observed file-change evidence.

ActivityDiff(text: str, language: str | None = None)
type: ClassVar[str] = 'diff'
Inherited methods and attributes

Inherited from ActivityCode.language.

Inherited from ActivityText.text.

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

Return a fresh canonical Activity content block, with no host lifecycle fields.

@dataclass(frozen=True, slots=True)
class ActivityTable(_ActivityBlock):

A rectangular table; input lists are snapshotted as tuples.

ActivityTable(columns: tuple[str, ...], rows: tuple[tuple[str, ...], ...])
columns: tuple[str, ...]
rows: tuple[tuple[str, ...], ...]
type: ClassVar[str] = 'table'
Inherited methods and attributes
def to_wire(self) -> dict[str, typing.Any]:

Return a fresh canonical Activity content block, with no host lifecycle fields.

@dataclass(frozen=True, slots=True)
class ActivityFile(_ActivityBlock):

Reference an existing attachment, never an external media URL.

ActivityFile(attachment_id: str, label: str)
attachment_id: str
label: str
type: ClassVar[str] = 'file'
Inherited methods and attributes
def to_wire(self) -> dict[str, typing.Any]:

Return a fresh canonical Activity content block, with no host lifecycle fields.

@dataclass(frozen=True, slots=True)
class ActivityImage(ActivityFile):

Display an existing image attachment.

ActivityImage(attachment_id: str, label: str)
type: ClassVar[str] = 'image'
Inherited methods and attributes

Inherited from ActivityFile.attachment_id.

Inherited from ActivityFile.label.

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

Return a fresh canonical Activity content block, with no host lifecycle fields.

@dataclass(frozen=True, slots=True)
class ActivityVideo(ActivityFile):

Display an existing video attachment.

ActivityVideo(attachment_id: str, label: str)
type: ClassVar[str] = 'video'
Inherited methods and attributes

Inherited from ActivityFile.attachment_id.

Inherited from ActivityFile.label.

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

Return a fresh canonical Activity content block, with no host lifecycle fields.

@dataclass(frozen=True, slots=True)
class ActivityAudio(ActivityFile):

Display an existing audio attachment.

ActivityAudio(attachment_id: str, label: str)
type: ClassVar[str] = 'audio'
Inherited methods and attributes

Inherited from ActivityFile.attachment_id.

Inherited from ActivityFile.label.

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

Return a fresh canonical Activity content block, with no host lifecycle fields.

@dataclass(frozen=True, slots=True)
class ActivityProgress(_ActivityBlock):

Progress with paired value/total, or an indeterminate indicator when omitted.

ActivityProgress( label: str, value: int | float | None = None, total: int | float | None = None)
label: str
value: int | float | None
total: int | float | None
type: ClassVar[str] = 'progress'
Inherited methods and attributes
def to_wire(self) -> dict[str, typing.Any]:

Return a fresh canonical Activity content block, with no host lifecycle fields.

@dataclass(frozen=True, slots=True)
class ActivitySection:

A visible headline and one-line summary; only content waits behind disclosure.

summary_format="markdown" opts into inline Markdown with HTTP(S) links. Unmarked or "inline" summaries are literal text. The 512-byte, one-line limit includes Markdown source; images, HTML and block layout are not shown.

ActivitySection( headline: str, summary: str, content: tuple[ActivityText | ActivityHeading | ActivityMarkdown | ActivityCode | ActivityDiff | ActivityTable | ActivityFile | ActivityImage | ActivityVideo | ActivityAudio | ActivityProgress, ...] = (), *, summary_format: Optional[Literal['inline', 'markdown']] = None)
headline: str
summary: str
summary_format: Optional[Literal['inline', 'markdown']]
def to_wire(self) -> dict[str, typing.Any]:

Return fresh portable data, without engine-owned lifecycle fields.

@dataclass(frozen=True, slots=True)
class ActivityPresentation(ActivitySection):

One atomic symbol presentation retaining all content and non-nested sections.

ActivityPresentation( headline: str, summary: str, content: tuple[ActivityText | ActivityHeading | ActivityMarkdown | ActivityCode | ActivityDiff | ActivityTable | ActivityFile | ActivityImage | ActivityVideo | ActivityAudio | ActivityProgress, ...] = (), sections: tuple[ActivitySection, ...] = (), *, summary_format: Optional[Literal['inline', 'markdown']] = None)
sections: tuple[ActivitySection, ...]
def to_wire(self) -> dict[str, typing.Any]:

Return fresh portable data, without engine-owned lifecycle fields.

Inherited methods and attributes

Inherited from ActivitySection.headline.

Inherited from ActivitySection.summary.

Inherited from ActivitySection.content.

@dataclass(frozen=True, slots=True)
class Activity:

Human-facing symbol presentation; the engine owns identity, timing and outcome.

Declare Activity on every exported callable, including each object method. Write understandable English for people, not Python identifiers or object reprs. Labels and headlines use sentence case ("Read file", "Run tests"), preserving proper names and acronyms. Summaries explain the target, useful counts or outcome. Use render or publish_activity for selected content; return values stay independent.

show_start=False makes a fast operation end-only: no running row or start callback. Use it for quick reads, patches and local lookups. Keep show_start=True for work people wait for, such as tests, network requests or transfers. Internal start/end tracking still preserves ordering, timing, errors and cancellation. Published content is retained but stays hidden until an end-only invocation settles. The final presentation must stand alone: name the target and outcome, including empty results or failures reported as return values. A normally returned failed workflow is still a failed workflow; "Completed" alone is not enough. Label partial lists and excerpts, and retain useful counts, errors and changes.

render is an optional synchronous callback receiving phase, args, kwargs, result and error as keyword arguments; it returns an ActivityPresentation (or None to keep the current presentation). It runs on success and failure, and on start only when show_start=True. Use publish_activity for intermediate stages of long-running tools. Rendering failures never change returns or errors.

Arguments:
  • presenter: Presentation category; leave "generic" for ordinary tools.
  • label: Optional nonblank, single-line label of at most 96 characters.
  • render: Synchronous callback accepting phase, args, kwargs, result and error keyword arguments. Return an ActivityPresentation or None.
  • show_start: Whether readers see a running row before the call settles.
Raises:
  • TypeError: show_start is not boolean or render is not synchronous/callable.
  • ValueError: The presenter is unknown or the label is invalid.
Activity( presenter: str = 'generic', label: str | None = None, render: Callable[..., ActivityPresentation | None] | None = None, show_start: bool = True)
presenter: str
label: str | None
render: Callable[..., ActivityPresentation | None] | None
show_start: bool
def publish_activity(presentation: ActivityPresentation) -> bool:

Replace headline, summary, content and sections without authoring lifecycle.

Headline and summary stay visible when collapsed. Content uses typed text, Markdown, table, code, diff, attachment and progress blocks. Sections group multiple results with one blank line between them. To clear content, publish the same headline/summary with empty content. None from render keeps the last snapshot. Returns False outside an invocation or when publication is rejected.

@dataclass(frozen=True, slots=True, kw_only=True)
class Extension:

Group tools and host capabilities in one immutable declaration.

Arguments:
  • name: Human-readable, nonblank extension name.
  • description: Nonblank description of what the extension does.
  • alias: Python namespace for exported symbols; required with symbols.
  • symbols: Symbol declarations for functions or object namespaces.
  • version: Optional extension version string.
  • kind: Optional extension kind understood by the host.
  • activation: Optional host activation callback.
  • prompt: Instructions as text or a host callback.
  • slash_commands: SlashCommand declarations.
  • op_hooks: OpHook declarations for host lifecycle operations.
  • ctx: Optional host context callback.
  • providers: Provider declarations.
  • network_filters: NetworkFilter declarations.
  • env: Environment variable names resolved at host registration.

Collection inputs are copied into tuples. Constructing an Extension is pure; it does not run callbacks or register anything. In an application, pass it to blockether.vis.engine.Agent or its register_extension method. In an installed extension entrypoint, call this module's register_extension once instead. The application callback bridge rejects host-only fields before connecting.

Raises:
  • ValueError: A required name, alias, callback or environment name is invalid.
  • TypeError: A field has the wrong type or a collection contains something other than its corresponding SDK declaration.
Extension( *, name: str, description: str, version: str | None = None, kind: str | None = None, alias: str | None = None, activation: Callable[..., typing.Any] | None = None, symbols: Sequence[Symbol] = (), prompt: str | Callable[..., typing.Any] | None = None, slash_commands: Sequence[SlashCommand] = (), op_hooks: Sequence[OpHook] = (), ctx: Callable[..., typing.Any] | None = None, providers: Sequence[Provider] = (), network_filters: Sequence[NetworkFilter] = (), env: Sequence[str] = ())
name: str
description: str
version: str | None
kind: str | None
alias: str | None
activation: Callable[..., typing.Any] | None
symbols: Sequence[Symbol]
prompt: str | Callable[..., typing.Any] | None
slash_commands: Sequence[SlashCommand]
op_hooks: Sequence[OpHook]
ctx: Callable[..., typing.Any] | None
providers: Sequence[Provider]
network_filters: Sequence[NetworkFilter]
env: Sequence[str]
def register_extension(extension: Extension) -> None:

Register one typed declaration and resolve its declared environment in this context.

Construction is pure; registration is the sole host boundary. A failure before completion leaves the context unregistered. No constructor registers itself.

def host_env(name, default=None):
def sequence(*, field: str) -> Callable[[~_SequenceClass], ~_SequenceClass]:

Declare a dataclass's public list or tuple field as its sandbox sequence.

Apply this outside @dataclass. The returned class is unchanged except for metadata: original methods are neither called nor transported. Each subclass must opt in separately. Values are checked when returned by an extension; the backing field must hold a built-in list or tuple, not a lazy iterable.

Sandbox records iterate over the received field and support length, truth testing, integer indices and slices. String keys still read named fields. Only the received items are exposed; iteration never fetches another page.

def method( fn: Optional[~_Method] = None, *, tag: Literal['observation', 'mutation'] = 'observation', is_hidden: bool = False, activity: Activity | None = None) -> Union[~_Method, blockether.vis.extension._MethodDecorator]:

Describe a public method on an object exported through Symbol.

Arguments:
  • fn: Method to annotate; omit it to use @method(...).
  • tag: "observation" for reads or "mutation" for state-changing work.
  • is_hidden: Hide the method from discovery without removing the callable.
  • activity: This method's human-facing presentation. Declare it on each exported method, not on the containing object namespace.
Returns:

The original method, or a decorator returning it. Calling your method directly still uses normal Python behavior; this decorator adds metadata.

Raises:
  • ValueError: The tag is unsupported or the decorated value is not callable.
  • TypeError: The Activity declaration has the wrong type.
from blockether.vis.extension import Activity, Symbol, method


class Greeter:
    @method(activity=Activity(label="Greet person", show_start=False))
    def hello(self, name: str) -> str:
        "Return a greeting for the named person."
        return f"Hello, {name}!"


greeter = Greeter()
tool = Symbol(greeter, name="greeter")
assert greeter.hello("Ada") == "Hello, Ada!"
assert tool.contract["members"][0]["name"] == "greeter.hello"
@dataclass(frozen=True, slots=True)
class Symbol:

Expose a function or an object's public methods as typed tools.

Arguments:
  • fn: Function, or an object whose public methods form a namespace.
  • name: Override the function name; required for objects and must then be a public Python identifier.
  • tag: Default operation classification: "observation" or "mutation".
  • is_hidden: Hide the symbol from discovery without removing the callable.
  • activity: Human-facing presentation for a function; declare one on every exported callable. For an object, leave this unset and put an Activity on every exported method instead.

Construction validates the declaration and derives its contract without calling the function or starting Vis. Function docstrings and type annotations become tool documentation, not runtime argument validation. Export narrow objects: public methods and nested namespace objects are traversed, not just methods bearing the decorator.

Raises:
  • ValueError: A callable lacks a docstring, or the name, operation tag, Activity placement or namespace is invalid, including cycles and repeated object references.
  • TypeError: A declaration field has an unsupported type.
Symbol( fn: Callable[..., typing.Any] | object, name: str | None = None, tag: Literal['observation', 'mutation'] = 'observation', is_hidden: bool = False, activity: Activity | None = None)
fn: Callable[..., typing.Any] | object
name: str | None
tag: Literal['observation', 'mutation']
is_hidden: bool
activity: Activity | None
contract: dict[str, typing.Any]

Fresh portable tool description; no callable, default values or host access.

Namespace members carry their full public names. Strings in Annotated describe meaning; unresolved annotations remain explicit, never evaluated. This is documentation, not runtime argument or result validation.

@dataclass(frozen=True, slots=True)
class TypeSpec:

An inert Python type description; references bound recursive records.

TypeSpec( kind: Literal['any', 'null', 'scalar', 'opaque', 'unresolved', 'union', 'literal', 'generic', 'record', 'reference'], name: str, description: str = '', arguments: tuple[TypeSpec, ...] = (), fields: tuple[FieldSpec, ...] = (), values: tuple[str | int | bool | None, ...] = (), variadic: bool = False, sequence_field: str | None = None)
kind: Literal['any', 'null', 'scalar', 'opaque', 'unresolved', 'union', 'literal', 'generic', 'record', 'reference']
name: str
description: str
arguments: tuple[TypeSpec, ...]
fields: tuple[FieldSpec, ...]
values: tuple[str | int | bool | None, ...]
variadic: bool
sequence_field: str | None
@dataclass(frozen=True, slots=True)
class FieldSpec:

A dataclass field; default values and factories are never exported or run.

FieldSpec( name: str, type: TypeSpec, required: bool, has_default: bool, default_is_none: bool)
name: str
type: TypeSpec
required: bool
has_default: bool
default_is_none: bool
@dataclass(frozen=True, slots=True)
class ParameterSpec(FieldSpec):

A callable parameter, preserving Python binding and safe default metadata.

ParameterSpec( name: str, type: TypeSpec, required: bool, has_default: bool, default_is_none: bool, kind: Literal['positional_only', 'positional_or_keyword', 'var_positional', 'keyword_only', 'var_keyword'])
kind: Literal['positional_only', 'positional_or_keyword', 'var_positional', 'keyword_only', 'var_keyword']
Inherited methods and attributes

Inherited from FieldSpec.name.

Inherited from FieldSpec.type.

Inherited from FieldSpec.required.

Inherited from FieldSpec.has_default.

Inherited from FieldSpec.default_is_none.

@dataclass(frozen=True, slots=True)
class ToolSpec:

Typed view of one public callable's authoritative Symbol contract.

ToolSpec( version: int, name: str, tag: Literal['observation', 'mutation'], description: str, signature: str, parameters: tuple[ParameterSpec, ...], returns: TypeSpec)
version: int
name: str
tag: Literal['observation', 'mutation']
description: str
signature: str
parameters: tuple[ParameterSpec, ...]
returns: TypeSpec
@dataclass(frozen=True, slots=True)
class NamespaceSpec:

A public namespace and its callable descendants, with full dotted names.

NamespaceSpec(name: str, members: tuple[ToolSpec, ...])
name: str
members: tuple[ToolSpec, ...]
@dataclass(frozen=True, slots=True)
class HelpDocument:

Generated reference text, not captured CLI output or an operation result.

HelpDocument(tool: str, text: str)
tool: str
text: str
class Catalog:

Read-only snapshot of public Symbols; construction and lookup perform no IO.

Pass the same symbols to Catalog and Extension. Rebuild after changing declarations. This is an adapter, not a registry, dispatcher or runtime type validator.

Catalog(symbols: Sequence[Symbol])
@method(activity=Activity(label='Inspect tool catalog', show_start=False, render=_catalog_presentation))
def spec( self, name: Annotated[str | None, 'Full public name; None lists top-level tools and namespaces.'] = None) -> ToolSpec | NamespaceSpec | tuple[ToolSpec | NamespaceSpec, ...]:

Inspect declared tools without invoking them. Unknown or hidden names raise ValueError.

@method(activity=Activity(label='Read tool reference', show_start=False, render=_catalog_presentation))
def help( self, name: Annotated[str, 'Full public tool or namespace name.']) -> HelpDocument:

Render the same metadata as doc(). No configuration, authentication or operation runs.

Raises TypeError for a non-string name; ValueError for unknown or hidden names.

@dataclass(frozen=True, slots=True)
class SlashCommand:

A user-facing slash command, not a model-facing tool.

SlashCommand( name: str, run: Callable[..., typing.Any], doc: str | None = None, usage: str | None = None)
name: str
run: Callable[..., typing.Any]
doc: str | None
usage: str | None
GATE_OPS = ('fs_access',)
@dataclass(frozen=True, slots=True)
class OpHook:

An operation observer or a fail-closed gate; never mix both kinds in one hook.

OpHook( ops: Sequence[str], fn: Callable[..., typing.Any], phase: Literal['before', 'after'] = 'before')
ops: Sequence[str]
fn: Callable[..., typing.Any]
phase: Literal['before', 'after']
@dataclass(frozen=True, slots=True)
class NetworkFilter:

Request/response policy at the host's network-filter boundary.

NetworkFilter(fn: Callable[..., typing.Any])
fn: Callable[..., typing.Any]
ProviderJSON: TypeAlias = str | int | float | bool | None | Mapping[str, 'ProviderJSON'] | Sequence['ProviderJSON']
ProviderAPIStyle: TypeAlias = Literal['anthropic', 'anthropic-messages', 'anthropic_messages', 'claude', 'messages', 'openai', 'openai-chat', 'openai_chat', 'openai-compatible', 'openai_compatible', 'openai-compatible-chat', 'openai_compatible_chat', 'chat', 'chat-completions', 'chat_completions', 'openai-responses', 'openai_responses', 'openai-compatible-responses', 'openai_compatible_responses', 'responses', 'gemini', 'google', 'google-gemini', 'google_gemini']
ProviderLimitStatus: TypeAlias = Literal['error', 'ok', 'unauthenticated', 'unknown-provider', 'unsupported']
ProviderLimitScope: TypeAlias = Literal['account', 'model', 'plan', 'workspace']
ProviderLimitKind: TypeAlias = Literal['credits', 'rate', 'requests', 'sessions', 'tokens', 'usd']
ProviderWindowKind: TypeAlias = Literal['calendar', 'lifetime', 'rolling']
ProviderWindowUnit: TypeAlias = Literal['day', 'hour', 'minute', 'month', 'week', 'year']
ProviderLimitPrecision: TypeAlias = Literal['derived', 'estimate', 'exact', 'unknown']
ProviderLimitSource: TypeAlias = Literal['derived', 'local', 'provider-api', 'static']
@dataclass(frozen=True, slots=True, kw_only=True)
class ProviderPreset(_ProviderValue):

Endpoint defaults, not credentials. Opaque API payload keys remain unchanged.

extra carries additional router settings as JSON, never overriding named fields. Use enrich_models_fn for typed model metadata beyond default model names.

ProviderPreset( *, base_url: str | None = None, api_style: Optional[Literal['anthropic', 'anthropic-messages', 'anthropic_messages', 'claude', 'messages', 'openai', 'openai-chat', 'openai_chat', 'openai-compatible', 'openai_compatible', 'openai-compatible-chat', 'openai_compatible_chat', 'chat', 'chat-completions', 'chat_completions', 'openai-responses', 'openai_responses', 'openai-compatible-responses', 'openai_compatible_responses', 'responses', 'gemini', 'google', 'google-gemini', 'google_gemini']] = None, default_models: Sequence[str] = (), responses_path: str | None = None, llm_headers: Mapping[str, str] | None = None, extra_body: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None = None, is_hidden: bool | None = None, extra: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None = None)
base_url: str | None
api_style: Optional[Literal['anthropic', 'anthropic-messages', 'anthropic_messages', 'claude', 'messages', 'openai', 'openai-chat', 'openai_chat', 'openai-compatible', 'openai_compatible', 'openai-compatible-chat', 'openai_compatible_chat', 'chat', 'chat-completions', 'chat_completions', 'openai-responses', 'openai_responses', 'openai-compatible-responses', 'openai_compatible_responses', 'responses', 'gemini', 'google', 'google-gemini', 'google_gemini']]
default_models: Sequence[str]
responses_path: str | None
llm_headers: Mapping[str, str] | None
extra_body: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None
is_hidden: bool | None
extra: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None
Inherited methods and attributes
def to_wire( self) -> dict[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]]:

Return fresh host data; declared optional fields are omitted, not null.

@dataclass(frozen=True, slots=True)
class ProviderCredential(_ProviderValue):

A usable credential; return None when absent. Tokens/headers are excluded from repr.

ProviderCredential( token: str, api_url: str | None = None, api_style: Optional[Literal['anthropic', 'anthropic-messages', 'anthropic_messages', 'claude', 'messages', 'openai', 'openai-chat', 'openai_chat', 'openai-compatible', 'openai_compatible', 'openai-compatible-chat', 'openai_compatible_chat', 'chat', 'chat-completions', 'chat_completions', 'openai-responses', 'openai_responses', 'openai-compatible-responses', 'openai_compatible_responses', 'responses', 'gemini', 'google', 'google-gemini', 'google_gemini']] = None, responses_path: str | None = None, llm_headers: Mapping[str, str] | None = None, source: str | None = None)
token: str
api_url: str | None
api_style: Optional[Literal['anthropic', 'anthropic-messages', 'anthropic_messages', 'claude', 'messages', 'openai', 'openai-chat', 'openai_chat', 'openai-compatible', 'openai_compatible', 'openai-compatible-chat', 'openai_compatible_chat', 'chat', 'chat-completions', 'chat_completions', 'openai-responses', 'openai_responses', 'openai-compatible-responses', 'openai_compatible_responses', 'responses', 'gemini', 'google', 'google-gemini', 'google_gemini']]
responses_path: str | None
llm_headers: Mapping[str, str] | None
source: str | None
Inherited methods and attributes
def to_wire( self) -> dict[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]]:

Return fresh host data; declared optional fields are omitted, not null.

@dataclass(frozen=True, slots=True)
class ProviderStatus(_ProviderValue):

Connection verdict, separate from usage limits. extra is display-only JSON metadata.

ProviderStatus( is_authenticated: bool, error: str | None = None, source: str | None = None, provider_id: str | None = None, status: str | None = None, base_url: str | None = None, label: str | None = None, config_path: str | None = None, extra: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None = None)
is_authenticated: bool
error: str | None
source: str | None
provider_id: str | None
status: str | None
base_url: str | None
label: str | None
config_path: str | None
extra: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None
Inherited methods and attributes
def to_wire( self) -> dict[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]]:

Return fresh host data; declared optional fields are omitted, not null.

@dataclass(frozen=True, slots=True)
class ProviderModel(_ProviderValue):

Model metadata returned by enrichment; extra preserves additional router fields.

ProviderModel( name: str, context: int | None = None, is_tool_call: bool | None = None, is_image_input: bool | None = None, extra: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None = None)
name: str
context: int | None
is_tool_call: bool | None
is_image_input: bool | None
extra: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None
Inherited methods and attributes
def to_wire( self) -> dict[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]]:

Return fresh host data; declared optional fields are omitted, not null.

@dataclass(frozen=True, slots=True)
class ProviderLimitWindow(_ProviderValue):

The canonical calendar, rolling or lifetime window for one limit.

ProviderLimitWindow( kind: Literal['calendar', 'lifetime', 'rolling'], unit: Optional[Literal['day', 'hour', 'minute', 'month', 'week', 'year']] = None, size: int | None = None, resets_at_ms: int | None = None)
kind: Literal['calendar', 'lifetime', 'rolling']
unit: Optional[Literal['day', 'hour', 'minute', 'month', 'week', 'year']]
size: int | None
resets_at_ms: int | None
Inherited methods and attributes
def to_wire( self) -> dict[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]]:

Return fresh host data; declared optional fields are omitted, not null.

@dataclass(frozen=True, slots=True)
class ProviderLimit(_ProviderValue):

One canonical usage row; finite measurements retain their original precision.

ProviderLimit( id: str, label: str, scope: Literal['account', 'model', 'plan', 'workspace'], kind: Literal['credits', 'rate', 'requests', 'sessions', 'tokens', 'usd'], precision: Literal['derived', 'estimate', 'exact', 'unknown'], source: Literal['derived', 'local', 'provider-api', 'static'], is_unlimited: bool = False, used: int | float | None = None, limit: int | float | None = None, remaining: int | float | None = None, window: ProviderLimitWindow | None = None, subject: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None = None, note: str | None = None)
id: str
label: str
scope: Literal['account', 'model', 'plan', 'workspace']
kind: Literal['credits', 'rate', 'requests', 'sessions', 'tokens', 'usd']
precision: Literal['derived', 'estimate', 'exact', 'unknown']
source: Literal['derived', 'local', 'provider-api', 'static']
is_unlimited: bool
used: int | float | None
limit: int | float | None
remaining: int | float | None
window: ProviderLimitWindow | None
subject: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None
note: str | None
Inherited methods and attributes
def to_wire( self) -> dict[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]]:

Return fresh host data; declared optional fields are omitted, not null.

@dataclass(frozen=True, slots=True)
class ProviderError(_ProviderValue):

A limits failure, not an exception carrying a credential.

ProviderError( type: str, message: str, data: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None = None)
type: str
message: str
data: Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]] | None
Inherited methods and attributes
def to_wire( self) -> dict[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]]:

Return fresh host data; declared optional fields are omitted, not null.

@dataclass(frozen=True, slots=True, kw_only=True)
class ProviderLimits(_ProviderValue):

Usage snapshot; the host fills provider_id and fetched_at_ms when omitted.

ProviderLimits( *, status: Literal['error', 'ok', 'unauthenticated', 'unknown-provider', 'unsupported'] = 'ok', limits: Sequence[ProviderLimit] = (), rpm: int | None = None, tpm: int | None = None, note: str | None = None, error: ProviderError | None = None, provider_id: str | None = None, fetched_at_ms: int | None = None)
status: Literal['error', 'ok', 'unauthenticated', 'unknown-provider', 'unsupported']
limits: Sequence[ProviderLimit]
rpm: int | None
tpm: int | None
note: str | None
error: ProviderError | None
provider_id: str | None
fetched_at_ms: int | None
def to_wire( self) -> dict[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]]:

Return fresh host data; declared optional fields are omitted, not null.

@dataclass(frozen=True, slots=True)
class Provider:

Pure provider declaration; register_extension adapts callbacks to the host.

Credential reads are passive; only auth_fn may initiate login. Callbacks are synchronous and may run without a session. Refresh accepts zero arguments or one rejected token (None if unknown), chosen without retrying callback errors. Enrichment/selection inputs remain JSON mappings owned by the router/config, not a second SDK schema for those domains. Callback outputs use typed records.

Provider( id: str, label: str, preset: ProviderPreset | None = None, is_managed: bool = False, get_token_fn: Callable[[], ProviderCredential | None] | None = None, detect_fn: Callable[[], ProviderCredential | None] | None = None, status_fn: Callable[[], ProviderStatus | None] | None = None, logout_fn: Callable[[], None] | None = None, limits_fn: Callable[[], ProviderLimits | None] | None = None, refresh_token_fn: Callable[[str | None], ProviderCredential | None] | Callable[[], ProviderCredential | None] | None = None, auth_fn: Callable[[Callable[[str], None]], str | bool | None] | None = None, auth_prompt_fn: Callable[[], Sequence[str] | str | None] | None = None, enrich_models_fn: Callable[Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]], Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]], Sequence[ProviderModel] | None] | None = None, on_selected_fn: Callable[Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]], None] | None = None)
id: str
label: str
preset: ProviderPreset | None
is_managed: bool
get_token_fn: Callable[[], ProviderCredential | None] | None
detect_fn: Callable[[], ProviderCredential | None] | None
status_fn: Callable[[], ProviderStatus | None] | None
logout_fn: Callable[[], None] | None
limits_fn: Callable[[], ProviderLimits | None] | None
refresh_token_fn: Callable[[str | None], ProviderCredential | None] | Callable[[], ProviderCredential | None] | None
auth_fn: Callable[[Callable[[str], None]], str | bool | None] | None
auth_prompt_fn: Callable[[], Sequence[str] | str | None] | None
enrich_models_fn: Callable[Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]], Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]], Sequence[ProviderModel] | None] | None
on_selected_fn: Callable[Mapping[str, str | int | float | bool | None | Mapping[str, str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]] | Sequence[str | int | float | bool | None | Mapping[str, ForwardRef('ProviderJSON')] | Sequence[ForwardRef('ProviderJSON')]]], None] | None
def ok(title, body=None, data=None):
def err(title, body=None, data=None):
def block(reason):
def strings_of(value):
state = <blockether.vis.extension._State object>
def log(level, msg):
def notify(text, level='info'):
class Shell(builtins.dict):
def logs(self, offset=None, limit=None):
def type(self, text, is_enter=True):
def stop(self):
def wait(self, seconds=120):
def shell(opts):
def jailed_shell(opts):
def jailed_shell_session(opts):
fs = <blockether.vis.extension._Fs object>
class Answer:
Answer(raw)
is_submitted
reason
request_id
values
def get(self, name, default=None):
def reveal(self, name):
def ask(title, fields, **options):
def plaintext(name, **spec):
def password(name, **spec):
def multiline(name, **spec):
def select(name, options, **spec):
def multiselect(name, options, **spec):
def checkbox(name, **spec):
def slider(name, **spec):
def otp(name, **spec):
def option(value, label=None):
def row(*fields):
def column(*fields):
def heading(text, live_text=None, **spec):

Form heading, or live heading(id, text, level=1..6).

def paragraph(text, live_text=None, **spec):

Form prose, or live paragraph(id, text) with inline Markdown.

def reveal(handle):
def forget(handle):
class Interrupted(builtins.Exception):

The live view this handle drives is no longer open.

Raised by the next push after the human stopped watching — Escape in the terminal, Stop in the app — so an unattended loop ends by itself. A loop that would rather finish its own work reads view.is_interrupted instead and decides.

note is the comment the person left with the stop, when they left one: the reason it is being stopped, in their words.

Interrupted(view_id, reason=None, note=None)
view_id
reason
note
class TextNode(_Node):

A paragraph, heading or code block, replaced in place.

def set(self, text, **spec):
Inherited methods and attributes
TextNode(view, node_id, type_name)
node_id
type
class Spinner(_Node):

One node of a live view, addressed by its id.

def set(self, text=None, *, variant=None, is_active=None):
Inherited methods and attributes
Spinner(view, node_id, type_name)
node_id
type
class Button(_Node):

One node of a live view, addressed by its id.

def set(self, label=None, *, is_disabled=None):
Inherited methods and attributes
Button(view, node_id, type_name)
node_id
type
class Status(_Node):

One node of a live view, addressed by its id.

def set(self, text, tone=None, detail=None, label=None):
Inherited methods and attributes
Status(view, node_id, type_name)
node_id
type
class Progress(_Node):

One node of a live view, addressed by its id.

def set(self, value=None, done=None, total=None, label=None):
Inherited methods and attributes
Progress(view, node_id, type_name)
node_id
type
class Stat(_Node):

One node of a live view, addressed by its id.

def set(self, stat_id, value_text, label=None, tone=None):
def clear(self):
def remove(self, *item_ids):
Inherited methods and attributes
Stat(view, node_id, type_name)
node_id
type
class Steps(_Node):

One node of a live view, addressed by its id.

def set(self, step_id, tone=None, label=None, detail=None, value=None):
def clear(self):
def remove(self, *item_ids):
Inherited methods and attributes
Steps(view, node_id, type_name)
node_id
type
LogTone: TypeAlias = Literal['idle', 'running', 'ok', 'warn', 'error']
class Log(_Node):

One node of a live view, addressed by its id.

def write( self, *lines: str | Sequence[str], tone: Optional[Literal['idle', 'running', 'ok', 'warn', 'error']] = None):

Append complete lines, optionally styled by severity. Redact before writing.

Each argument is a retained line, not a raw byte fragment. A tone applies only to this call; omitted tones are plain. Controls are displayed literally.

def clear(self):
Inherited methods and attributes
Log(view, node_id, type_name)
node_id
type
class Table(_KeyedNode):

A node holding items the extension addresses by id: it can drop them.

def upsert(self, row_id, cells, tone=None, branch=None):
def select(self, *item_ids):
Inherited methods and attributes
Table(view, node_id, type_name)
node_id
type
def remove(self, *item_ids):
def clear(self):
class LiveView:

A live view the human WATCHES, driven by the extension that opened it.

vis.live(...) mounts one and answers this handle. Nodes are addressed by id — view['jobs'], or view.node('jobs') — and each answers the typed handle its own type declares. The view-level shortcuts (view.status(...), view.log(...), view.row(...)) resolve to the one node of that type and raise naming the candidate ids when the view holds several, so an ambiguous call fails where it was written instead of quietly patching the wrong table.

Pushes are BATCHED: ops buffer and cross on the next push after flush_ms, when a coalesced push fills, and always before the view is read or closed. with view.batch(): groups one logical picture explicitly, including structural add/drop operations. Repeated writes to the same row or node collapse into the last one, so a per-row progress counter costs one wire row per tick rather than one per write.

Closing is the point: close() answers either the structured verdict or the compact model_result the extension chose. Used as a context manager the view closes itself — completed on the way out, and failed carrying the error when the body raised, because a run that died mid-way still owes the model what happened.

LiveView(request, flush_ms=None)
view_id
@contextmanager
def batch(self):

Send one complete picture for a related group of node changes.

Node handles still coalesce exactly as usual, but neither the leading edge nor structural add/drop operations cross the host seam until the outermost batch ends. Reads and close remain explicit flush points.

def flush(self):

Send everything buffered. Called for you before any read or close.

def state(self):

What the view looks like right now, as the surfaces paint it.

def sleep(self, seconds):

Block until the view changes, ends, or seconds elapse.

One host wait replaces periodic state reads. An unchanged timeout returns no view payload. Returns True on change or close, False on timeout. Nonpositive durations do not flush or call the host.

is_interrupted

True once the human stopped watching.

Asks the engine at most once per flush window, so a compute loop can poll it every iteration and still cost one host call per tick.

reason

Why the view ended, or None while open or after a compact result.

is_from_human

True when a PERSON ended it, rather than the run itself or a deadline.

A view is always stoppable — nothing is asked of the human, so nothing is left unanswered by stopping it — and this is how the run finds out that is what happened.

note

The comment the human left with their stop, or None.

The stop always lands; the note says WHY in their own words, and the same words reach the model in the verdict.

result

The structured verdict or compact model result, once ended.

def node(self, node_id):

The typed handle for one node, by id.

def status(self, text, tone=None, detail=None):
def progress(self, value=None, done=None, total=None):
def stat(self, stat_id, value_text, label=None, tone=None):
def step(self, step_id, tone=None, label=None, detail=None, value=None):
def write( self, *lines: str | Sequence[str], tone: Optional[Literal['idle', 'running', 'ok', 'warn', 'error']] = None):
def row(self, row_id, cells, tone=None, branch=None):
def add(self, node, after=None):

Add a whole node to a running view — a scan that discovers a seventh device should not have to have declared it.

def drop(self, node_id):

Drop a whole node, its items with it.

def close( self, reason=None, summary=None, error=None, artifact_id=None, selection_snapshots=None, model_result=None):

End the view and answer the result the model reads.

model_result is an optional compact string returned instead of the full structured verdict. The finished picture and close metadata remain in the durable artifact and on human-facing close events.

selection_snapshots are finished pictures keyed by a selectable table and its selected rows. They are sealed only into the artifact record, so a reopened run can still switch rows without keeping its extension alive.

Closing twice is a no-op answering the first result: a finally that closes what an interrupt already closed must not overwrite the reason the human chose.

def live(title, nodes, **options):

Open a live view and answer the handle that drives it.

View options: description, source, session_id, channel_ids, plus flush_ms for the batching window. EVERY key is a snake_case string, exactly as vis.ask documents. There is no cancellable flag: a human can always stop watching, and the verdict says they did (is_from_human) and why (note). plus flush_ms for the batching window. EVERY key is a snake_case string, exactly as vis.ask documents.

The view is mounted at once and nothing blocks — use it as a context manager so it closes itself:

with vis.live('Deploy', [vis.steps('plan', steps=[...])]) as view:
    view['plan'].set('build', tone='running')

Closing answers the verdict: is_completed, reason, the finished picture as data, and whatever summary the extension chose to end with.

testing = <blockether.vis.extension._Testing object>
def disclosure(node_id, label, *nodes, default_expanded=False):

A collapsible column. Local choices survive updates; receipts start collapsed.

def divider(node_id):

A static horizontal rule across its live container. Add or drop it by id.

def code(node_id, text, *, language=None, **spec):

Literal code; whitespace is retained and content is never executed.

def spinner( node_id, text='Working', *, variant='braille', is_active=True, **spec):

A braille, dots, line or pulse indicator. Receipts never animate.

def button(node_id, label, *, is_disabled=False, **spec):

An operator action. Accepted presses increment clicks in view.state().

No callback or code crosses the wire. The producer decides how to respond. Disabled buttons and completed receipts cannot be activated.

def status(node_id, text=None, **spec):
def progress(node_id, **spec):
def stat(node_id, stats=None, **spec):
def steps(node_id, steps=None, **spec):
def output(node_id, **spec):

Retained output with an independent disclosure, collapsed by default.

default_expanded=True opens an active log initially. Updates preserve the local choice; completion starts a collapsed receipt. Hiding never clears lines. window_lines bounds only the hot window, not the durable record. write(..., tone="warn") styles complete lines with a LogTone; omitted tones remain plain. Seeded line_tones align one-for-one with lines. Redact before writing. ANSI controls are visible text, never executed.

def table(node_id, columns=None, **spec):
def table_column(column_id, label=None, **spec):
def table_row(row_id, cells, **spec):