Managed expansion API

On this page

• Controller
• Requests and providers
• Configuration and operations

Managed expansion keeps application-owned source data separate from cached records and the displayed graph. Start with the branch guide.

Controller

st_graph_workbench.component.expansion.ExpansionController

ExpansionController(elements: Elements, *, source_id: str = 'graph', source_version: str = '1', config: ExpansionConfig | None = None)

Maintain reversible graph views, independent of a Streamlit runtime.

Initialize with the starting graph. Store one controller per application session/component. Call request then the provider, then apply_response; or use expand for a synchronous batch. view is safe to render and loaded includes collapsed records. Neither returns mutable internal data.

Source code in st_graph_workbench/component/expansion.py

def __init__(
    self,
    elements: Elements,
    *,
    source_id: str = "graph",
    source_version: str = "1",
    config: ExpansionConfig | None = None,
) -> None:
    elements = _validate_loaded(elements)
    self.source_id = str(source_id)
    self._instance_id = uuid4().hex
    self.source_version = str(source_version)
    self.limits = config or ExpansionConfig()
    self._base = deepcopy(elements)
    self._cache = deepcopy(elements)
    self._record_index: _RecordIndex | None = None
    self._branches: dict[str, _Branch] = {}
    self._requests: dict[str, ExpansionRequest] = {}
    self._protected: set[str] = set()
    self._hidden: set[str] = set()
    self._filtered: set[str] = set()
    self._deleted: set[str] = set()
    self._handled: set[str] = set()
    self._last_view = deepcopy(elements)
    self.changes: list[dict[str, Any]] = []
    self.revision = 0
    self.viewport: dict[str, Any] = {}
    self.bulk: dict[str, Any] | None = None
    self.search_result: dict[str, Any] | None = None
    self.analysis_result: dict[str, Any] | None = None
    self.last_error: str | None = None

branch_id

branch_id(node_id: ElementId, query: dict[str, Any] | None = None) -> str

Return a stable identity for a starting node and normalized filters.

Source code in st_graph_workbench/component/expansion.py

def branch_id(self, node_id: ElementId, query: dict[str, Any] | None = None) -> str:
    """Return a stable identity for a starting node and normalized filters."""
    payload = [normalize_id(node_id, option_name="node_id"), _query(query)]
    return hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()[
        :24
    ]

view

view() -> Elements

Return the displayed graph, excluding collapsed, hidden and filtered data.

Source code in st_graph_workbench/component/expansion.py

def view(self) -> Elements:
    """Return the displayed graph, excluding collapsed, hidden and filtered data."""
    return deepcopy(self._projection()[0])

loaded

loaded() -> Elements

Return all cached records, including collapsed branches, without fetching.

Source code in st_graph_workbench/component/expansion.py

def loaded(self) -> Elements:
    """Return all cached records, including collapsed branches, without fetching."""
    return deepcopy(self._cache)

request

request(node_id: ElementId, *, query: dict[str, Any] | None = None, more: bool = False) -> ExpansionRequest | None

Start a batch, or restore a cached branch without fetching again.

Source code in st_graph_workbench/component/expansion.py

def request(
    self,
    node_id: ElementId,
    *,
    query: dict[str, Any] | None = None,
    more: bool = False,
) -> ExpansionRequest | None:
    """Start a batch, or restore a cached branch without fetching again."""
    if str(node_id) not in {str(n["data"]["id"]) for n in self.view()["nodes"]}:
        raise ValueError("Expansion requires a displayed starting node.")
    key, branch = self._branch(node_id, query)
    if branch.pending:
        return None
    if branch.fetched and not more:
        was_open = branch.opened
        branch.opened = True
        if self._over_bulk_limit():
            branch.opened = was_open
            branch.error = "Expansion stopped at the configured record limit."
            return None
        branch.error = None
        self._changed()
        return None
    if not branch.has_more:
        return None
    request = ExpansionRequest(
        uuid4().hex,
        key,
        str(node_id),
        deepcopy(branch.query),
        deepcopy(branch.cursor),
        self.limits.page_size,
        self.source_id,
        self.source_version,
        branch.revision,
    )
    self._requests[request.request_id] = request
    branch.pending = request.request_id
    branch.error = None
    return request

apply_response

apply_response(response: ExpansionResponse) -> bool

Atomically accept a current response; return false for stale/duplicate data.

Source code in st_graph_workbench/component/expansion.py

def apply_response(self, response: ExpansionResponse) -> bool:
    """Atomically accept a current response; return false for stale/duplicate data."""
    request = self._requests.get(response.request_id)
    if request is None:
        return False
    branch = self._branches[request.branch_id]
    displayed_before = self._ids(self.view())
    if (
        request.source_version != self.source_version
        or request.source_id != self.source_id
        or request.revision != branch.revision
        or branch.pending != response.request_id
        or request.node_id not in self._ids(self.view())
    ):
        self._requests.pop(response.request_id, None)
        if branch.pending == response.request_id:
            branch.pending = None
        return False
    try:
        for count in (response.total_nodes, response.total_edges):
            if count is not None and (
                isinstance(count, bool) or not isinstance(count, int) or count < 0
            ):
                raise ValueError(
                    "Expansion totals must be nonnegative integers or None."
                )
        if response.has_more and response.cursor == request.cursor:
            raise ValueError("A continuing response must advance its cursor.")
        json.dumps(response.cursor, allow_nan=False)
        incoming = _validate_loaded(response.elements, strict=False)
        if not isinstance(response.has_more, bool):
            raise ValueError("has_more must be a boolean.")
        for group, other in (("nodes", "edges"), ("edges", "nodes")):
            other_records = (
                self._index().nodes if other == "nodes" else self._index().edges
            )
            if any(e["data"]["id"] in other_records for e in incoming[group]):
                raise ValueError(
                    "A response cannot change a record's node/edge identity."
                )
        incoming["nodes"] = [
            n
            for n in incoming["nodes"]
            if str(n["data"]["id"]) not in self._deleted
        ]
        incoming["edges"] = [
            e
            for e in incoming["edges"]
            if not (
                {str(e["data"][x]) for x in ("id", "source", "target")}
                & self._deleted
            )
        ]
        known = self._index().order
        additions = {
            group: [e for e in incoming[group] if str(e["data"]["id"]) not in known]
            for group in ("nodes", "edges")
        }
        candidate = (
            upsert_elements(
                self._cache, nodes=additions["nodes"], edges=additions["edges"]
            )
            if additions["nodes"] or additions["edges"]
            else self._cache
        )
        node_ids = {str(n["data"]["id"]) for n in incoming["nodes"]} - {
            request.node_id
        }
        cached_edges = (
            self._index().edges
            if candidate is self._cache
            else {e["data"]["id"]: e for e in candidate["edges"]}
        )
        for edge in incoming["edges"]:
            data = cached_edges[str(edge["data"]["id"])]["data"]
            node_ids.update(
                str(data[x])
                for x in ("source", "target")
                if str(data[x]) != request.node_id
            )
        old_cache, old_branch = self._cache, deepcopy(branch)
        self._cache = candidate
        branch.node_ids.update(node_ids)
        branch.edge_ids.update(str(e["data"]["id"]) for e in incoming["edges"])
        branch.opened = True
        if self._over_bulk_limit():
            self._cache = old_cache
            self._branches[request.branch_id] = old_branch
            self.fail(
                response.request_id,
                "Expansion stopped at the configured record limit.",
            )
            return False
        branch.fetched = True
        branch.has_more = response.has_more
        branch.cursor = deepcopy(response.cursor)
        branch.total_nodes, branch.total_edges = (
            response.total_nodes,
            response.total_edges,
        )
        branch.pending = None
        self._requests.pop(response.request_id)
        self._place_new_nodes(request.node_id, displayed_before)
        self._changed()
        return True
    except (TypeError, ValueError, KeyError) as error:
        self.fail(response.request_id, str(error))
        raise

fail

fail(request_id: str, message: str) -> None

Finish a failed request without advancing its cursor or changing its view.

Source code in st_graph_workbench/component/expansion.py

def fail(self, request_id: str, message: str) -> None:
    """Finish a failed request without advancing its cursor or changing its view."""
    request = self._requests.pop(request_id, None)
    if request:
        branch = self._branches[request.branch_id]
        if branch.pending == request_id:
            branch.pending = None
            branch.error = str(message)

expand

expand(node_id: ElementId, provider: ExpansionProvider, *, query: dict[str, Any] | None = None, more: bool = False) -> bool

Accept one synchronous batch, preserving the last valid view on failure.

Return true for an accepted provider response. Restoring cached branches, already-pending requests, and failures return false; inspect describe for pending/error status and view for the resulting graph.

Source code in st_graph_workbench/component/expansion.py

def expand(
    self,
    node_id: ElementId,
    provider: ExpansionProvider,
    *,
    query: dict[str, Any] | None = None,
    more: bool = False,
) -> bool:
    """Accept one synchronous batch, preserving the last valid view on failure.

    Return true for an accepted provider response. Restoring cached branches,
    already-pending requests, and failures return false; inspect ``describe``
    for pending/error status and ``view`` for the resulting graph.
    """
    request = self.request(node_id, query=query, more=more)
    if request is None:
        return False
    try:
        response = provider.expand(request)
        if response.request_id != request.request_id:
            raise ValueError(
                "Provider response request_id does not match its request."
            )
        return self.apply_response(response)
    except Exception as error:
        self.fail(request.request_id, str(error))
        return False

collapse

collapse(node_ids: list[ElementId] | None = None, *, branch_id: str | None = None) -> None

Close branches, retaining anchors; an explicit branch_id takes precedence.

Source code in st_graph_workbench/component/expansion.py

def collapse(
    self, node_ids: list[ElementId] | None = None, *, branch_id: str | None = None
) -> None:
    """Close branches, retaining anchors; an explicit branch_id takes precedence."""
    ids = set(normalize_id_list(node_ids or [], option_name="node_ids"))
    for key, branch in self._branches.items():
        matches = (
            key == branch_id if branch_id is not None else branch.node_id in ids
        )
        if matches:
            branch.opened = False
            branch.revision += 1
            if branch.pending:
                self._requests.pop(branch.pending, None)
                branch.pending = None
    self._invalidate_inactive_requests()
    if self.bulk and self.bulk["status"] == "running":
        self.cancel()
    self._changed()

collapse_all

collapse_all() -> None

Close all branches, preserving starting and explicitly protected records.

Source code in st_graph_workbench/component/expansion.py

def collapse_all(self) -> None:
    """Close all branches, preserving starting and explicitly protected records."""
    self.collapse([b.node_id for b in self._branches.values()])

protect

protect(element_ids: list[ElementId], *, enabled: bool = True) -> None

Keep records visible independently of branch state; not a position lock.

Source code in st_graph_workbench/component/expansion.py

def protect(self, element_ids: list[ElementId], *, enabled: bool = True) -> None:
    """Keep records visible independently of branch state; not a position lock."""
    ids = set(normalize_id_list(element_ids, option_name="element_ids"))
    if not ids <= self._ids(self._cache):
        raise ValueError("Only loaded records can be protected.")
    self._protected = self._protected | ids if enabled else self._protected - ids
    self._changed()

set_visibility

set_visibility(*, hidden: list[str] | None = None, filtered: list[str] | None = None) -> None

Set independent hide/filter masks; an empty list restores only that mask.

Source code in st_graph_workbench/component/expansion.py

def set_visibility(
    self, *, hidden: list[str] | None = None, filtered: list[str] | None = None
) -> None:
    """Set independent hide/filter masks; an empty list restores only that mask."""
    masks = {
        name: set(normalize_id_list(values, option_name=name))
        for name, values in (("hidden", hidden), ("filtered", filtered))
        if values is not None
    }
    for name, values in masks.items():
        setattr(self, "_" + name, values)
    self._invalidate_inactive_requests()
    self._changed()

update_records

update_records(elements: Elements) -> None

Apply explicit application-owned edits to the cache, including positions.

Source code in st_graph_workbench/component/expansion.py

def update_records(self, elements: Elements) -> None:
    """Apply explicit application-owned edits to the cache, including positions."""
    candidate = upsert_elements(self._cache, **elements, replace=False)
    self._cache = _validate_loaded(candidate)
    edge_index = {
        str(edge["data"]["id"]): edge["data"] for edge in candidate["edges"]
    }
    for branch in self._branches.values():
        for edge_id in branch.edge_ids & edge_index.keys():
            branch.node_ids.update(
                str(edge_index[edge_id][endpoint])
                for endpoint in ("source", "target")
                if str(edge_index[edge_id][endpoint]) != branch.node_id
            )
    self._changed()

delete_records

delete_records(element_ids: list[ElementId]) -> None

Apply actual source deletion, unlike collapse; suppress stale resurrection.

Source code in st_graph_workbench/component/expansion.py

def delete_records(self, element_ids: list[ElementId]) -> None:
    """Apply actual source deletion, unlike collapse; suppress stale resurrection."""
    ids = set(normalize_id_list(element_ids, option_name="element_ids"))
    nodes = {str(n["data"]["id"]) for n in self._cache["nodes"]} & ids
    updated = delete_elements(
        self._cache, node_ids=sorted(nodes), edge_ids=sorted(ids - nodes)
    )
    self._deleted.update(self._ids(self._cache) - self._ids(updated))
    self._cache = updated
    self._base = {
        group: [
            e
            for e in self._base[group]
            if str(e["data"]["id"]) not in self._deleted
        ]
        for group in ("nodes", "edges")
    }
    self._protected -= self._deleted
    self.cancel()
    self._changed()

invalidate

invalidate(elements: Elements, *, source_version: str) -> None

Replace the source checkpoint explicitly, clearing incompatible exploration.

Source code in st_graph_workbench/component/expansion.py

def invalidate(self, elements: Elements, *, source_version: str) -> None:
    """Replace the source checkpoint explicitly, clearing incompatible exploration."""
    replacement = ExpansionController(
        elements,
        source_id=self.source_id,
        source_version=source_version,
        config=self.limits,
    )
    self.__dict__.update(replacement.__dict__)

commands

commands(previous: Elements, *, restore_positions: bool = False) -> list[GraphCommand]

Reconcile the view; restore_positions explicitly reapplies saved positions.

Source code in st_graph_workbench/component/expansion.py

def commands(
    self, previous: Elements, *, restore_positions: bool = False
) -> list[GraphCommand]:
    """Reconcile the view; restore_positions explicitly reapplies saved positions."""
    current = self.view()
    known = self._ids(previous)
    removed = known - self._ids(current)
    nodes = {str(n["data"]["id"]) for n in previous.get("nodes", [])}
    commands: list[GraphCommand] = []
    if removed:
        commands.append(
            delete_elements_command(
                uuid4().hex,
                node_ids=sorted(removed & nodes),
                edge_ids=sorted(removed - nodes),
            )
        )
    updates: Elements = {"nodes": [], "edges": []}
    for group in ("nodes", "edges"):
        previous_records = {
            str(e["data"]["id"]): e for e in previous.get(group, [])
        }
        for record in current[group]:
            old = previous_records.get(str(record["data"]["id"]), {})
            if group == "nodes" and old and not restore_positions:
                record = {k: v for k, v in record.items() if k != "position"}
                old = {k: v for k, v in old.items() if k != "position"}
            if record != old:
                updates[group].append(record)
    if updates["nodes"] or updates["edges"]:
        commands.append(
            upsert_elements_command(uuid4().hex, **updates, replace=False)
        )
    return commands

start_bulk

start_bulk(node_ids: list[ElementId], *, query: dict[str, Any] | None = None) -> None

Queue breadth-first bounded exploration; call step once per application tick.

Source code in st_graph_workbench/component/expansion.py

def start_bulk(
    self, node_ids: list[ElementId], *, query: dict[str, Any] | None = None
) -> None:
    """Queue breadth-first bounded exploration; call step once per application tick."""
    ids = normalize_id_list(
        node_ids or [n["data"]["id"] for n in self._base["nodes"]],
        option_name="node_ids",
    )
    view = self.view()
    if not set(ids) <= {n["data"]["id"] for n in view["nodes"]}:
        raise ValueError("Bulk expansion requires displayed starting nodes.")
    filters = _query(query)
    self.cancel()
    self.bulk = {
        "id": uuid4().hex,
        "status": "running",
        "queue": [[i, 0] for i in sorted(set(ids))],
        "visited": [],
        "query": filters,
        "batches": 0,
        "baseline": {
            g: [str(e["data"]["id"]) for e in view[g]] for g in ("nodes", "edges")
        },
        "limits": {"nodes": self.limits.max_nodes, "edges": self.limits.max_edges},
    }

step

step(provider: ExpansionProvider) -> bool

Process at most one batch. Return whether more bulk work remains.

Source code in st_graph_workbench/component/expansion.py

def step(self, provider: ExpansionProvider) -> bool:
    """Process at most one batch. Return whether more bulk work remains."""
    if not self.bulk or self.bulk["status"] != "running":
        return False
    bulk = self.bulk
    while bulk["queue"]:
        node_id, depth = bulk["queue"].pop(0)
        if depth >= self.limits.max_depth or node_id not in self._ids(self.view()):
            continue
        key, branch = self._branch(node_id, bulk["query"])
        if key in bulk["visited"]:
            continue
        if not branch.fetched or branch.has_more:
            self.expand(node_id, provider, query=bulk["query"], more=branch.fetched)
            branch = self._branches[key]
            bulk["batches"] += 1
            if branch.error:
                if bulk["status"] == "running":
                    bulk["status"] = "failed"
                bulk["queue"].insert(0, [node_id, depth])
                return False
        else:
            self.request(node_id, query=bulk["query"])
            if bulk["status"] != "running":
                bulk["queue"].insert(0, [node_id, depth])
                return False
        if branch.has_more:
            bulk["queue"].insert(0, [node_id, depth])
        else:
            bulk["visited"].append(key)
            bulk["queue"].extend(
                [[child, depth + 1] for child in sorted(branch.node_ids)]
            )
        if not bulk["queue"]:
            bulk["status"] = "complete"
        return bool(bulk["status"] == "running")
    bulk["status"] = "complete"
    return False

cancel

cancel() -> None

Invalidate outstanding responses and stop scheduling new bulk batches.

Source code in st_graph_workbench/component/expansion.py

def cancel(self) -> None:
    """Invalidate outstanding responses and stop scheduling new bulk batches."""
    for branch in self._branches.values():
        if branch.pending:
            branch.revision += 1
            branch.pending = None
    self._requests.clear()
    if self.bulk and self.bulk["status"] in {"running", "failed"}:
        self.bulk["status"] = "cancelled"

retry_bulk

retry_bulk() -> None

Resume a failed bulk request at its unchanged cursor.

Source code in st_graph_workbench/component/expansion.py

def retry_bulk(self) -> None:
    """Resume a failed bulk request at its unchanged cursor."""
    if self.bulk and self.bulk["status"] == "failed":
        self.bulk["status"] = "running"

reveal

reveal(element_id: ElementId) -> None

Restore a cached result through recorded branches, including real context.

Source code in st_graph_workbench/component/expansion.py

def reveal(self, element_id: ElementId) -> None:
    """Restore a cached result through recorded branches, including real context."""
    target = str(element_id)
    routes: dict[str, list[str]] = {key: [] for key in self._anchors()}
    pending = deque(routes)
    by_node: dict[str, list[tuple[str, _Branch]]] = defaultdict(list)
    for key, branch in self._branches.items():
        if branch.fetched:
            by_node[branch.node_id].append((key, branch))
    while pending and target not in routes:
        anchor = pending.popleft()
        for key, branch in by_node[anchor]:
            for child in sorted(branch.node_ids | branch.edge_ids):
                if child not in routes:
                    routes[child] = [*routes[anchor], key]
                    pending.append(child)
    if target not in routes or target in self._deleted:
        raise ValueError("No cached exploration path reaches this record.")
    restore = {target}
    for edge in self._cache["edges"]:
        if str(edge["data"]["id"]) == target:
            restore.update(
                str(edge["data"][endpoint]) for endpoint in ("source", "target")
            )
    for key in routes[target]:
        branch = self._branches[key]
        branch.opened = True
        restore.update(branch.node_ids | branch.edge_ids | {branch.node_id})
    self._hidden -= restore
    self._filtered -= restore
    self._changed()

reveal_search_result

reveal_search_result(result: dict[str, Any], node_id: str) -> None

Validate and reveal a provider-supplied real path from an existing anchor.

Source code in st_graph_workbench/component/expansion.py

def reveal_search_result(self, result: dict[str, Any], node_id: str) -> None:
    """Validate and reveal a provider-supplied real path from an existing anchor."""
    for name, expected in (
        ("source_version", self.source_version),
        ("source_dataset_id", self.source_id),
    ):
        if name in result and result[name] != expected:
            raise ValueError(
                "Source search result is obsolete; refresh the source search."
            )
    path = result.get("paths", {}).get(node_id)
    if not path or path[-1] != node_id or path[0] not in self._anchors():
        raise ValueError(
            "Source search must provide a path from a starting/protected record."
        )
    incoming = _validate_loaded(result["elements"])
    if set(path) & self._deleted:
        raise ValueError(
            "Search result contains deleted records; refresh the source."
        )
    known = self._ids(self._cache)
    candidate = upsert_elements(
        self._cache,
        nodes=[
            n
            for n in incoming["nodes"]
            if n["data"]["id"] not in known | self._deleted
        ],
        edges=[
            e
            for e in incoming["edges"]
            if e["data"]["id"] not in known
            and not {e["data"][name] for name in ("id", "source", "target")}
            & self._deleted
        ],
    )
    if not set(path) <= {n["data"]["id"] for n in candidate["nodes"]}:
        raise ValueError("Source search path has a missing node.")
    contributions = []
    supplied_edges = {e["data"]["id"] for e in incoming["edges"]}
    for source, target in zip(path, path[1:]):
        edges = {
            str(e["data"]["id"])
            for e in candidate["edges"]
            if e["data"]["id"] in supplied_edges
            and {str(e["data"]["source"]), str(e["data"]["target"])}
            == {source, target}
        }
        if not edges:
            raise ValueError("Source search path has a missing relationship.")
        contributions.append((source, target, edges))
    self._cache = candidate
    for source, target, edges in contributions:
        _, branch = self._branch(source)
        branch.node_ids.add(target)
        branch.edge_ids.update(edges)
        branch.fetched = True
        branch.opened = True
    self.reveal(node_id)

snapshot

snapshot() -> dict[str, Any]

Export state without provider objects; source properties may be sensitive.

Source code in st_graph_workbench/component/expansion.py

def snapshot(self) -> dict[str, Any]:
    """Export state without provider objects; source properties may be sensitive."""
    branches = {
        key: {
            **asdict(b),
            "node_ids": sorted(b.node_ids),
            "edge_ids": sorted(b.edge_ids),
            "pending": None,
        }
        for key, b in self._branches.items()
    }
    return deepcopy(
        {
            "schema_version": 1,
            "source_id": self.source_id,
            "source_version": self.source_version,
            "config": asdict(self.limits),
            "base": self._starting_graph(),
            "loaded": self._cache,
            "branches": branches,
            "protected": sorted(self._protected),
            "hidden": sorted(self._hidden),
            "filtered": sorted(self._filtered),
            "deleted": sorted(self._deleted),
            "viewport": self.viewport,
        }
    )

restore classmethod

restore(snapshot: dict[str, Any], *, source_version: str, source_id: str | None = None) -> ExpansionController

Restore a compatible snapshot; require explicit invalidation for new sources.

Source code in st_graph_workbench/component/expansion.py

@classmethod
def restore(
    cls,
    snapshot: dict[str, Any],
    *,
    source_version: str,
    source_id: str | None = None,
) -> ExpansionController:
    """Restore a compatible snapshot; require explicit invalidation for new sources."""
    if snapshot.get("schema_version") != 1 or snapshot.get("source_version") != str(
        source_version
    ):
        raise ValueError(
            "Snapshot schema/source version differs; invalidate explicitly."
        )
    if source_id is not None and snapshot.get("source_id") != str(source_id):
        raise ValueError("Snapshot source identity differs; invalidate explicitly.")
    obj = cls(
        snapshot["base"],
        source_id=snapshot["source_id"],
        source_version=source_version,
        config=ExpansionConfig(**snapshot["config"]),
    )
    obj._cache = _validate_loaded(snapshot["loaded"])
    nodes = {str(n["data"]["id"]) for n in obj._cache["nodes"]}
    edges = {str(e["data"]["id"]) for e in obj._cache["edges"]}
    if not obj._ids(obj._base) <= nodes | edges:
        raise ValueError("Snapshot starting records must exist in its cache.")
    for key, value in snapshot["branches"].items():
        data = deepcopy(value)
        data["node_ids"], data["edge_ids"] = (
            set(data["node_ids"]),
            set(data["edge_ids"]),
        )
        data["query"] = _query(data["query"])
        data["pending"] = None
        if (
            isinstance(data["revision"], bool)
            or not isinstance(data["revision"], int)
            or data["revision"] < 0
            or any(
                not isinstance(data[field], bool)
                for field in ("opened", "fetched", "has_more")
            )
        ):
            raise ValueError("Invalid snapshot branch state.")
        if obj.branch_id(data["node_id"], data["query"]) != key:
            raise ValueError("Invalid snapshot branch identity.")
        deleted = set(snapshot["deleted"])
        if (
            data["node_id"] not in nodes | deleted
            or not data["node_ids"] <= nodes | deleted
            or not data["edge_ids"] <= edges | deleted
        ):
            raise ValueError("Snapshot branch references unknown records.")
        obj._branches[key] = _Branch(**data)
    for name in ("protected", "hidden", "filtered", "deleted"):
        setattr(obj, "_" + name, set(snapshot[name]))
    obj.viewport = deepcopy(snapshot["viewport"])
    if not isinstance(obj.viewport, dict):
        raise ValueError("Snapshot viewport must be a dictionary.")
    if obj.viewport:
        from .commands import viewport_command

        viewport_command("snapshot-validation", "set_viewport", **obj.viewport)
    if not obj._protected <= nodes | edges:
        raise ValueError("Snapshot protected records must exist in its cache.")
    json.dumps(obj.snapshot(), allow_nan=False)
    obj._changed()
    return obj

describe

describe(provider: ExpansionProvider | None = None) -> dict[str, Any]

Return JSON configuration for graph_workbench(expansion=...).

Includes independent expand/collapse availability and exact next-change counts when a local preview exists. Unknown remote counts remain None.

Source code in st_graph_workbench/component/expansion.py

def describe(self, provider: ExpansionProvider | None = None) -> dict[str, Any]:
    """Return JSON configuration for graph_workbench(expansion=...).

    Includes independent expand/collapse availability and exact next-change
    counts when a local preview exists. Unknown remote counts remain None.
    """
    view = self.view()
    visible = self._ids(view)
    node_ids = {n["data"]["id"] for n in view["nodes"]}
    rows: dict[str, Any] = {}
    for node in view["nodes"]:
        node_id = str(node["data"]["id"])
        key, branch = self._branch(node_id)
        closed = {k for k, b in self._branches.items() if b.node_id == node_id}
        collapsed, _ = self._projection(closed)
        hidden = visible - self._ids(collapsed)
        next_nodes: int | None = None
        next_edges: int | None = None
        available: bool | None = None
        preview = getattr(provider, "preview", None)
        if branch.fetched and not branch.opened:
            restored = copy(self)
            restored._branches = {**self._branches, key: deepcopy(branch)}
            restored._branches[key].opened = True
            restored_view = restored.view()
            next_nodes = len(
                {str(n["data"]["id"]) for n in restored_view["nodes"]} - visible
            )
            next_edges = len(
                {str(e["data"]["id"]) for e in restored_view["edges"]} - visible
            )
        elif preview and branch.has_more:
            request = ExpansionRequest(
                "preview",
                key,
                node_id,
                deepcopy(branch.query),
                deepcopy(branch.cursor),
                self.limits.page_size,
                self.source_id,
                self.source_version,
                branch.revision,
            )
            try:
                batch = preview(request)
                available = batch.has_more
                trial = copy(self)
                trial._branches = {**self._branches, key: deepcopy(branch)}
                trial._requests = {request.request_id: request}
                trial._branches[key].pending = request.request_id
                trial.bulk = None
                trial.apply_response(batch)
                trial_view = trial.view()
                next_nodes = len(
                    {str(n["data"]["id"]) for n in trial_view["nodes"]} - visible
                )
                next_edges = len(
                    {str(e["data"]["id"]) for e in trial_view["edges"]} - visible
                )
            except Exception:
                # A preview is optional; an unavailable source must not blank the view.
                next_nodes = next_edges = None
        elif not branch.has_more:
            next_nodes, next_edges = 0, 0
        rows[node_id] = {
            "branch_id": key,
            "opened": branch.opened,
            "fetched": branch.fetched,
            "has_more": branch.has_more,
            "can_expand": not branch.opened
            and (
                branch.fetched
                or next_nodes is None
                or available
                or bool(next_nodes or next_edges)
            ),
            "can_load_more": branch.opened and branch.has_more,
            "can_collapse": any(self._branches[k].opened for k in closed),
            "next_nodes": next_nodes,
            "next_edges": next_edges,
            "collapse_nodes": len(hidden & node_ids),
            "collapse_edges": len(hidden - node_ids),
            "protected": node_id in self._protected,
            "pending": bool(branch.pending),
            "error": branch.error
            or next(
                (
                    self._branches[k].error
                    for k in sorted(closed)
                    if self._branches[k].error
                ),
                None,
            ),
        }
    return {
        "nodes": rows,
        "controller_id": self._instance_id,
        "branches": [
            {
                "branch_id": k,
                "node_id": b.node_id,
                "query": deepcopy(b.query),
                "opened": b.opened,
                "active": b.opened and b.node_id in visible,
                "has_more": b.has_more,
                "error": b.error,
            }
            for k, b in self._branches.items()
            if b.fetched or b.pending or b.error
        ],
        "limits": asdict(self.limits),
        "bulk": deepcopy(self.bulk),
        "source_id": self.source_id,
        "source_version": self.source_version,
        "view_revision": self.revision,
        "loaded": self.loaded(),
        "hidden_count": len(self._hidden | self._filtered),
        "source_search": callable(getattr(provider, "search", None)),
        "source_analysis": callable(getattr(provider, "analyze", None)),
        "search_result": deepcopy(self.search_result),
        "analysis_result": deepcopy(self.analysis_result),
        "error": self.last_error,
    }

handle_event

handle_event(event: GraphEvent | None, provider: ExpansionProvider) -> None

Handle explicit expansion intents; applications schedule bulk step separately.

Source code in st_graph_workbench/component/expansion.py

def handle_event(
    self, event: GraphEvent | None, provider: ExpansionProvider
) -> None:
    """Handle explicit expansion intents; applications schedule bulk step separately."""
    self.last_error = None
    try:
        self._handle_event(event, provider)
    except Exception as error:
        self.last_error = str(error)

Requests and providers

st_graph_workbench.component.expansion.ExpansionRequest dataclass

ExpansionRequest(request_id: str, branch_id: str, node_id: str, query: dict[str, Any], cursor: Any, limit: int, source_id: str, source_version: str, revision: int)

One versioned, repeatable provider query, with an opaque JSON cursor.

st_graph_workbench.component.expansion.ExpansionResponse dataclass

ExpansionResponse(request_id: str, elements: Elements, has_more: bool = False, cursor: Any = None, total_nodes: int | None = None, total_edges: int | None = None)

One complete batch; edges may reference records in the loaded cache.

Totals count distinct nodes and edges separately, not domain rows. None means unknown. A continuing response must advance its cursor.

st_graph_workbench.component.expansion.ExpansionProvider

Bases: Protocol

Application-owned loader. Optional search/analyze stay explicit.

Implement expand(request) to return a batch or raise an actionable error. Optional preview(request) returns the same batch without I/O, enabling exact next-change counts. It must not fetch from a remote service on render.

expand

expand(request: ExpansionRequest) -> ExpansionResponse

Retrieve one authorized batch without modifying controller state.

Source code in st_graph_workbench/component/expansion.py

def expand(self, request: ExpansionRequest) -> ExpansionResponse:
    """Retrieve one authorized batch without modifying controller state."""
    ...

st_graph_workbench.component.expansion_provider.InMemoryExpansionProvider

InMemoryExpansionProvider(elements: Elements)

Expand dataframe-derived dictionaries without modifying the source graph.

Pages contain distinct neighboring nodes and all matching edges to them. Relationship labels and direction filter edges; attributes match the merged edge/neighbor data (neighbor values win). Time filters use ISO-8601 values from the configured edge field, falling back to the neighboring node field. search returns verified paths from supplied roots, not isolated matches.

Source code in st_graph_workbench/component/expansion_provider.py

def __init__(self, elements: Elements) -> None:
    self._elements = normalize_elements(elements)
    self._nodes = {str(n["data"]["id"]): n for n in self._elements["nodes"]}
    self._adjacency: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for edge in self._elements["edges"]:
        for node_id in {str(edge["data"]["source"]), str(edge["data"]["target"])}:
            self._adjacency[node_id].append(edge)

expand

expand(request: ExpansionRequest) -> ExpansionResponse

Return an immutable source page, with separate exact node and edge totals.

Source code in st_graph_workbench/component/expansion_provider.py

def expand(self, request: ExpansionRequest) -> ExpansionResponse:
    """Return an immutable source page, with separate exact node and edge totals."""
    positive_limit(request.limit, "Expansion limit")
    if request.node_id not in self._nodes:
        raise ValueError(f"Unknown source node: {request.node_id}")
    neighbors = self._neighbors(request.node_id, request.query)
    keys = sorted(neighbors)
    cursor = request.cursor if request.cursor is not None else 0
    if isinstance(cursor, bool) or not isinstance(cursor, int) or cursor < 0:
        raise ValueError("The in-memory cursor must be a nonnegative integer.")
    selected = keys[cursor : cursor + request.limit]
    edges = [edge for node_id in selected for edge in neighbors[node_id]]
    return ExpansionResponse(
        request.request_id,
        {
            "nodes": self._context_nodes(set(selected), exclude=request.node_id),
            "edges": deepcopy(edges),
        },
        has_more=cursor + len(selected) < len(keys),
        cursor=cursor + len(selected),
        total_nodes=len(set(keys) - {request.node_id}),
        total_edges=sum(len(value) for value in neighbors.values()),
    )

preview

preview(request: ExpansionRequest) -> ExpansionResponse

Preview the next page using the local index, without remote I/O.

Source code in st_graph_workbench/component/expansion_provider.py

def preview(self, request: ExpansionRequest) -> ExpansionResponse:
    """Preview the next page using the local index, without remote I/O."""
    return self.expand(request)

search

search(query: str, *, roots: list[str], limit: int = 50) -> dict[str, Any]

Search source node properties and return real connecting paths from roots.

Disconnected matches are reported separately and are not silently added as new exploration anchors. Applications may explicitly protect/load them.

Source code in st_graph_workbench/component/expansion_provider.py

def search(
    self, query: str, *, roots: list[str], limit: int = 50
) -> dict[str, Any]:
    """Search source node properties and return real connecting paths from roots.

    Disconnected matches are reported separately and are not silently added
    as new exploration anchors. Applications may explicitly protect/load them.
    """
    positive_limit(limit, "Search limit")
    all_matches = [
        key
        for key, node in self._nodes.items()
        if query.casefold() in str(node["data"]).casefold()
    ]
    matches = all_matches[:limit]
    parents: dict[str, tuple[str, dict[str, Any]] | None] = {
        key: None for key in roots if key in self._nodes
    }
    pending = deque(parents)
    while pending:
        node_id = pending.popleft()
        for edge in self._adjacency[node_id]:
            source, target = (str(edge["data"][x]) for x in ("source", "target"))
            other = target if source == node_id else source
            if other not in parents:
                parents[other] = (node_id, edge)
                pending.append(other)
    paths: dict[str, list[str]] = {}
    nodes: set[str] = set()
    edges: dict[str, dict[str, Any]] = {}
    for match in matches:
        if match not in parents:
            continue
        path = [match]
        while parents[path[-1]] is not None:
            entry = parents[path[-1]]
            assert entry is not None
            parent, edge = entry
            path.append(parent)
            edges[str(edge["data"]["id"])] = edge
        paths[match] = list(reversed(path))
        nodes.update(path)
    return {
        "node_ids": matches,
        "node_count": len(self._elements["nodes"]),
        "edge_count": len(self._elements["edges"]),
        "total_matches": len(all_matches),
        "complete": len(all_matches) <= limit,
        "paths": paths,
        "unreachable_node_ids": [key for key in matches if key not in paths],
        "elements": {
            "nodes": self._context_nodes(nodes),
            "edges": deepcopy(list(edges.values())),
        },
    }

Configuration and operations

st_graph_workbench.component.expansion.ExpansionConfig dataclass

ExpansionConfig(page_size: int = 50, max_depth: int = 3, max_nodes: int = 500, max_edges: int = 2000)

Finite exploration limits; bulk limits count additional displayed records.

st_graph_workbench.component.expansion.ExpansionOperation module-attribute

ExpansionOperation = Literal['expand', 'load_more', 'collapse', 'collapse_all', 'expand_all', 'protect', 'unprotect', 'cancel', 'retry', 'reveal', 'search', 'analyze', 'continue']

Explicit operations emitted by opt-in expansion controls.

Events use action="expansion". The operation is one of expand, load_more, collapse, collapse_all, expand_all, protect, unprotect, cancel, retry, reveal, search, analyze, or continue. The exploration dialog is browser-local; continue schedules a single bulk batch. Applications normally pass returned events to controller.handle_event(event, provider).