> ## Documentation Index
> Fetch the complete documentation index at: https://docs.blobhub.io/llms.txt
> Use this file to discover all available pages before exploring further.

# Code Component (logic.code)

> Execute custom Python code within the workflow environment.

The `logic.code` component executes custom Python code safely within the workflow execution environment. It provides a sandboxed space equipped with helpers for input/output data handling and flow control.

## Runtime Environment

The execution occurs inside a hardened **Python 3** environment.

### Available Modules

You can leverage several standard libraries and helpful built-ins:

**Standard Libraries:**

* `json`, `uuid`, `time`, `datetime`, `zoneinfo`
* `functools`, `traceback`, `copy`, `decimal`
* `textwrap`, `math`, `re`, `abc`
* `typing`, `Union`, `Literal`

**Exceptions:**

* `Exception`
* `ValueError`
* **`Exit`**: A special exception intentionally designed to exit the active code block immediately.

**Session exceptions**: raised by the Session Management functions below, and catchable in your code so a
workflow can react — most importantly, retry an optimistic-concurrency clash on a later run.

* **`InvalidSessionObjectAlias`**: An alias violates the namespace grammar.
* **`InvalidThreadEnvelope`** / **`InvalidGraphEnvelope`**: The target object is missing, or is not a thread / graph.
* **`GraphMutationConflict`**: An `if_rev` optimistic-lock check failed — the element changed since it was read.
* **`GraphTraversalLimitExceeded`**: A `query_session_graph` traversal exceeded a hard bound.
* **`InvalidGraphQuery`**: A malformed traversal query.

## Input & Output

### Reading Inputs

You can access data arriving at the component's input ports. Inputs are received as explicitly typed **Value Objects**.

* **`data_inputs`**: A list of all received input port data.
* **`get_data_input(route)`**: A helper function to grab the Value Object of a specific input port by its exact route name.

```python theme={null}
# Returns a dict structurally conforming to a Value Object.
input_val = get_data_input("user_query")

if input_val and input_val.get("type") == "message":
    pass
```

### Writing Outputs

You can emit data out of the component by pushing **Value Objects** to output routes.

* **`set_data_output(route, value, persist=False)`**: Assigns a value to a targeted output port.
  * `route`: Output port name (e.g., "result").
  * `value`: The Value Object payload.
  * `persist`: If `True`, stores the output specifically to processor data tracking logic for external visibility.

```python theme={null}
output_val = create_message(role="assistant", text="Processed Data")
set_data_output("result", output_val)
```

## Flow Control

You determine which logic route the workflow should navigate following code execution.

* **`set_result(route, terminal=False)`**: Declares the destination route.
  * `route`: Connection edge to traverse (`"success"`, `"failure"`, etc).
  * `terminal`: Set to `True` if this intentionally ends the entire workflow execution.
* **`exit()`**: Aborts the execution of the remainder of the active code script.

```python theme={null}
if user_exists:
    set_result("success")
else:
    set_result("failure")
```

## Helper Functions

### Data Helpers

* **`create_message(role, text=None, attribute_id=None)`**: Generates a standardized "message" Value Object dynamically.
* **`create_value(data_type, data)`**: Quickly wrap arbitrary matching data into a recognized Value structure.

### Session Management

Every function below is bound to the execution's **own session** — none of them takes a `session_id`, so your code
can only ever reach its own session's objects. Names and behavior mirror the platform's REST session commands
one-to-one, and the same validation applies here as on the REST API (alias grammar, thread/graph envelope checks,
size limits, graph mutation/traversal caps, and `if_rev` optimistic concurrency). Writes are attributed to the user
the execution runs as.

The request/response shapes for each function are documented in full on the linked `operations` reference pages
below — this section covers the signature, behavior, and return shape only.

#### Objects

A session object is a typed document addressed by an `alias` — a path such as `missions/a1/graph`.

* **`download_session_object(alias)`**: Reads the value of a session object. Returns the typed payload, or `None`
  if it doesn't exist. See [Download Session Object](/blob-types/workflow/operations/download-session-object).
* **`upload_session_object(alias, value)`**: Writes/creates a session object, validating the alias. See
  [Upload Session Object](/blob-types/workflow/operations/upload-session-object).
* **`delete_session_object(alias)`**: Deletes a session object. If the object is a `thread` or `graph` envelope,
  this also cascades — deleting all of its child thread items / graph elements. See
  [Session Objects](/blob-types/workflow/session-objects/introduction) for the envelope model (there's no
  dedicated delete reference page).
* **`list_session_objects(prefix="", delimiter=None, cursor=None, limit=None)`**: Lists session objects under an
  optional alias `prefix`, with optional `delimiter`-based folder rollup and cursor pagination. Returns
  `{"objects": [...], "cursor": ..., "common_prefixes": [...]}` (the last key only when `delimiter` is given). See
  [List Session Objects](/blob-types/workflow/operations/list-session-objects).

```python theme={null}
config = download_session_object("my_config")                     # read
upload_session_object("counter", {"count": 1})                     # write
listing = list_session_objects(prefix="missions/", delimiter="/")  # one-level "folder" listing
```

#### Threads

A `thread` object holds an append-only list of items. Create the envelope first with
`upload_session_object(alias, {"type": "thread", ...})` before posting to it.

* **`post_session_thread_item(alias, content, parent_id=None, metadata=None)`**: Appends an item. `content` is a
  list of typed parts (e.g. `[{"type": "text", "text": "..."}]`); `metadata` is a free-form tag dict. Returns
  `{"item": <item>}`. See [Post Session Thread Item](/blob-types/workflow/operations/post-session-thread-item).
* **`list_session_thread_items(alias, ascending=False, created_since=None, created_before=None, cursor=None,
  limit=None)`**: Reads items, newest-first by default; `created_since`/`created_before` bound the range (use
  `created_since` as a delta cursor). Returns `{"items": [...], "cursor": ...}`. See
  [List Session Thread Items](/blob-types/workflow/operations/list-session-thread-items).
* **`get_session_thread_item(alias, item_id)`**: Fetches a single item. Returns `{"item": <item>}`. See
  [Get Session Thread Item](/blob-types/workflow/operations/get-session-thread-item).

```python theme={null}
posted = post_session_thread_item("missions/a1/cells/imp-7f/meta",
                                   [{"type": "text", "text": "Plan ready."}],
                                   metadata={"type": "summary"})
recent = list_session_thread_items("missions/a1/cells/imp-7f/meta", created_since=cursor)
```

#### Graphs

A `graph` object holds vertices and edges. Create the envelope first with
`upload_session_object(alias, {"type": "graph", ...})` — vertices and edges are then changed only via mutations.

* **`apply_session_graph_mutations(alias, operations)`**: Applies a batch of mutations (`add_vertex`, `add_edge`,
  `set_vertex_props`, `set_edge_props`, `remove_*_props`, `delete_vertex`, `delete_edge`), each optionally
  carrying `if_rev` for optimistic concurrency. Returns `{"elements": [...], "changes": [...]}`. See
  [Apply Session Graph Mutations](/blob-types/workflow/operations/apply-session-graph-mutations).
* **`query_session_graph(alias, query)`**: Runs a bounded traversal against a query AST. Returns
  `{"result": ...}`. See [Query Session Graph](/blob-types/workflow/operations/query-session-graph).
* **`list_session_graph_elements(alias, type=None, ascending=False, updated_since=None, ids_only=False,
  cursor=None, limit=None)`**: Pages elements, optionally filtered to `type` `"vertex"`/`"edge"`. Returns
  `{"elements": [...], "cursor": ...}`, or `{"element_ids": [...], "cursor": ...}` when `ids_only=True`. See
  [List Session Graph Elements](/blob-types/workflow/operations/list-session-graph-elements).
* **`get_session_graph_element(alias, element_id)`**: Fetches one element. Returns `{"element": ...}`. See
  [Get Session Graph Element](/blob-types/workflow/operations/get-session-graph-element).
* **`get_session_graph_elements(alias, element_ids)`**: Batch-fetches elements. Returns `{"elements": [...]}`. See
  [Get Session Graph Elements](/blob-types/workflow/operations/get-session-graph-elements).
* **`list_session_graph_neighbors(alias, from_ids, direction, label=None, neighbor_label=None, cursor=None,
  limit=None)`**: Expands adjacency from `from_ids` (`direction` is `"out"`/`"in"`/`"both"`, optionally filtered
  by edge `label` / `neighbor_label`). Returns `{"edges": [...], "vertices": [...], "cursor": ...}`. See
  [List Session Graph Neighbors](/blob-types/workflow/operations/list-session-graph-neighbors).

```python theme={null}
mutated = apply_session_graph_mutations("missions/a1/graph", [
    {"op": "add_vertex", "type": "cell", "props": {"status": "planned"}},
])
open_cells = query_session_graph("missions/a1/graph", {
    "start": ["V"], "steps": [["hasLabel", "cell"], ["has", "status", "neq", "completed"]],
})
```

#### Advanced: optimistic concurrency

Mutations can carry `if_rev` to guard against a concurrent write. If another writer already advanced the element
since it was read, the call raises `GraphMutationConflict` instead of silently overwriting it — catch the
exception and reconcile on a later run:

```python theme={null}
try:
    apply_session_graph_mutations("missions/a1/graph", [
        {"op": "set_vertex_props", "element_id": "imp-7f", "props": {"status": "working"}, "if_rev": 4},
    ])
except GraphMutationConflict:
    pass  # another writer advanced it; reconcile on the next run
```

### Logging & Events

Produce informational, warning, or failure events logged to the execution stream:

* **`post_output_event(message)`**
* **`post_warning_event(message)`**
* **`post_error_event(message)`**

### System Metadata

* **`current_time_utc_iso()`**: Gets UTC time dynamically.
* **`result`**: The immediate runtime representation of the result dictionary.

## Full Example

```python theme={null}
request_val = get_data_input("request")

if request_val and request_val.get("type") == "message":
    
    response_text = "Message received and acknowledged."
    
    msg = create_message("assistant", response_text)
    set_data_output("response", msg)
    
    post_output_event("Ack sent")
    set_result("success")
else:
    post_error_event("Request generic or missing type string")
    set_result("failure")
```