# Starlark Reference

Armature extensions can include Starlark scripts for server-side logic.
Starlark is a Python-like language designed for configuration and
embedding — see the [official spec](https://github.com/google/starlark-go).

## Sandbox constraints

- **No `while` loops** — use `for` with bounded ranges.
- **No `load()`** — use `lib.require()` for library dependencies.
- **Max steps:** 1,000,000 bytecode operations per execution.
- **No filesystem or OS access** — all I/O goes through gated modules.
- **Deterministic** — same input produces same output (no `random`, no `time`).

## Always-available modules

These modules are injected into every script with no permission required.

### json

Standard Starlark JSON module.

```python
data = json.decode('{"key": "value"}')
text = json.encode({"key": "value"})
```

### settings

Read resolved package settings (global → team → user cascade).

```python
val = settings.get("theme", "light")
# Returns the resolved value, or the default if unset.
```

The cascade respects the `user_overridable` flag from the package manifest.
See [Permissions & Groups](PERMISSIONS-AND-GROUPS) for details.

```python
# Check if a runtime capability is available
if settings.has_capability("pgvector"):
    # Use native vector search
    ...
```

`has_capability(name)` returns `True` if the named environment capability
is detected by the kernel. Detected capabilities: `pgvector`, `workspace`,
`object_storage`, `s3`, `postgres`.

### lib

Load exported functions from other packages.

```python
helpers = lib.require("my-utils")
result = helpers.format_date("2026-01-15")
```

Requirements:
- The target package must be declared in your manifest's `depends` array.
- The target package must declare `exports` in its manifest.
- The target package must be status `active`, tier `starlark`.
- Any package type (`library`, `extension`, `full`) can be called as long
  as it declares exports. This enables full packages to expose callable
  functions alongside their UI and API routes.
- Circular dependencies are detected and rejected.
- Results are cached per execution (calling `require` twice returns the
  same object).
- The called function runs with the *target* package's permissions, not
  the caller's.

### permissions

Check whether a user has a specific permission.

```python
if permissions.check(user_id, "image-gen.use"):
    # User is authorized
    ...
```

Returns `True` if the user has the permission, `False` otherwise (including
when the user is not found). Resolves the user's groups and merges granted
permissions — works for both kernel and extension-declared permissions.

## Permission-gated modules

These modules are only available if the package has the corresponding
permission granted in **Admin > Packages**.

### secrets

**Permission:** `secrets.read`

Read admin-configured secrets for this package.

```python
api_key = secrets.get("OPENAI_KEY")  # str or None
all_keys = secrets.list()             # list of key names
```

Secrets are set in **Admin > Packages > Secrets** and scoped per package.

### notifications

**Permission:** `notifications.send`

Send in-app notifications to users.

```python
notifications.send(
    user_id,        # str — target user UUID
    title,          # str — notification title
    body="",        # str — optional body text
    type="extension.notify"  # str — notification type
)
```

### db

**Permission:** `db.read` (queries) or `db.write` (mutations)

Read and write extension data tables. All tables are automatically
namespaced as `ext_{package_id}_{table_name}`.

#### Read operations

```python
# Query with filters, ordering, and pagination
rows = db.query(
    "tasks",                    # table name (without prefix)
    filters={"status": "open"}, # equality WHERE clauses
    order="-created_at",        # column name (prefix - for DESC)
    limit=50,                   # max 1000
    before={"created_at": ts},  # range: column < value
    after={"created_at": ts},   # range: column > value
    search_like={"title": "%bug%"}  # LIKE/ILIKE search
)

# Read from system views (read-only)
users = db.view("users", filters={"display_name": "Alice"}, limit=10)
channels = db.view("channels", limit=100)

# List all tables owned by this package
tables = db.list_tables()
```

Available views: `users`, `channels`.

#### Aggregate operations

```python
# Count rows matching filters
count = db.count("tasks", filters={"status": "open"})

# Aggregate a column (sum, avg, min, max, count)
total = db.aggregate("orders", "amount", "sum", filters={"status": "paid"})
# Returns int, float, or None (if no matching rows)

# Batch multiple queries in a single call
results = db.query_batch([
    {"table": "tasks", "filters": {"status": "open"}, "limit": 10},
    {"table": "logs", "order": "-created_at", "limit": 5},
])
# Returns list of result lists. Max 10 queries per batch.
# Each query spec supports: table (required), filters, order, limit, before, after, search_like
```

#### Vector similarity search

```python
# Find rows with the most similar embeddings (cosine distance)
rows = db.query_similar(
    "documents",        # table name
    "embedding",        # vector column name
    vector=[0.1, 0.2, ...],  # query vector (list of floats)
    limit=10,           # max results (default 10, max 100)
    filters={"active": True},  # optional equality filters
    metric="cosine",    # only "cosine" supported
)
# Returns rows ordered by ascending _distance (0.0 = identical, 1.0 = orthogonal)
# Each row dict includes an injected "_distance" float key.
```

Vector columns are declared as `"vector(N)"` in the manifest `db_tables` block
(N = dimension, 1–4096). Storage varies by backend:

| Backend | Column type | Search |
|---------|-------------|--------|
| Postgres + pgvector | `vector(N)` with HNSW index | Native `<=>` operator |
| Postgres (no pgvector) | `JSONB` | Go-side cosine computation |
| SQLite | `TEXT` | Go-side cosine computation |

Insert vectors as lists: `db.insert("docs", {"embedding": [0.1, 0.2, 0.3]})`.

#### Write operations

```python
row = db.insert("tasks", {"title": "Fix bug", "status": "open"})
# Returns the inserted row dict (with generated id, created_at)

db.update("tasks", row_id, {"status": "closed"})
# Returns True on success

db.delete("tasks", row_id)
# Returns True on success
```

### http

**Permission:** `api.http`

Make outbound HTTP requests.

```python
resp = http.get("https://api.example.com/data", headers={"Authorization": "Bearer ..."})
resp = http.post(url, body='{"key": "val"}', headers={"Content-Type": "application/json"})
resp = http.put(url, body="...", headers={})
resp = http.delete(url, headers={})
resp = http.request("PATCH", url, body="...", headers={})
```

Response dict:

```python
{
    "status": 200,
    "headers": {"content-type": "application/json"},
    "body": "..."   # capped at 1 MB
}
```

#### Batch requests

```python
responses = http.batch([
    {"method": "GET", "url": "https://api.example.com/a"},
    {"method": "POST", "url": "https://api.example.com/b", "body": "{}", "headers": {"Content-Type": "application/json"}},
])
# Returns list of response dicts (same shape as individual calls).
# Individual failures return {"status": 0, "body": "error: ...", "headers": {}}.
# Max 10 requests per batch. Dispatched concurrently.
```

**Security:**
- Private/loopback IPs are blocked (SSRF protection).
- Packages can declare `network_access.allow` (allowlist) or
  `network_access.block` (blocklist) in their manifest.
- Max 10 redirects. 10-second timeout. 1 MB response body limit.

### realtime

**Permission:** `realtime.publish`

Publish WebSocket events to subscribed clients.

```python
realtime.publish(
    "my-channel",     # channel name
    "item.updated",   # event label
    {"id": "abc"}     # payload dict (max 7 KB)
)
```

The payload is automatically tagged with `_pkg: package_id`.

### connections

**Permission:** `connections.read`

Read external connection configurations (secrets are decrypted).

```python
conn = connections.get("postgres", "main-db")
# Returns dict with id, type, name, scope, plus flattened config fields
# Returns None if not found

all_pg = connections.list("postgres")
# Returns list of connection dicts
```

Connections are resolved via scope chain: personal → team → global.

### workflow

**Permission:** `workflow.access`

Read workflow definitions and instances (read-only from Starlark;
mutations go through the HTTP API).

```python
defn = workflow.get_definition(workflow_id)
# Returns dict: id, name, slug, entry_mode, is_active, version, stages[]

inst = workflow.get_instance(instance_id)
# Returns dict: id, workflow_id, current_stage, status, stage_data, ...

instances = workflow.list_instances(workflow_id, status="active")
# Returns list of instance dicts
```

### batch

**Permission:** `batch.exec`

Run multiple callables concurrently. Each callable gets its own
execution thread with an independent step budget.

```python
jira = lib.require("jira-client")
confluence = lib.require("confluence-client")

results, errors = batch.exec([
    lambda: jira.create_issue(issue_data),
    lambda: confluence.create_page(page_data),
    lambda: send_notification(user_id),
], timeout=15)

# results[i] = return value of callables[i], or None on error
# errors[i]  = None on success, or error string on failure
# All three ran concurrently.
```

**Constraints:**

- Max 8 callables per call. Dispatched concurrently via goroutines.
- `timeout` (optional): 1–30 seconds per branch (default 10).
- `lib.require()` is not available inside branch callables.
  Load libraries before the `batch.exec` call.
- `batch.exec()` cannot be called from within a branch (no nesting).
- `print()` output from branches is discarded.

---

### files

**Permissions:** `files.read`, `files.write`

Store and retrieve files via the kernel ObjectStore (PVC or S3).
All keys are scoped to `ext/{packageID}/` — extensions cannot access
each other's files.

```python
# Store a file with optional metadata
files.put("reports/q1.pdf", pdf_bytes,
          content_type="application/pdf",
          metadata={"quarter": "Q1", "year": 2026})

# Read a file
result = files.get("reports/q1.pdf")
# result = {"content": b"...", "content_type": "application/pdf",
#           "size": 12345, "metadata": {"quarter": "Q1", "year": 2026}}

# Metadata only (no content transfer)
meta = files.meta("reports/q1.pdf")

# List files by prefix
entries = files.list(prefix="reports/", limit=50)
# entries = [{"name": "reports/q1.pdf", "size": 12345, "content_type": "..."}]

# Check existence
if files.exists("reports/q1.pdf"):
    files.delete("reports/q1.pdf")

# Bulk delete
files.delete_prefix("temp/")
```

**Notes:**
- `content` accepts string or bytes; `get()` returns bytes.
- Maximum file size: 50 MB (configurable via `EXT_FILES_MAX_SIZE`).
- Metadata is stored as a companion JSON object, not in the file itself.
- `files.list()` automatically filters out internal metadata companions.

---

### workspace

**Permission:** `workspace.manage`

Managed disk directories for extensions that need a real filesystem
(git clones, compilers, media tools). Each workspace is scoped to
`{WORKSPACE_ROOT}/{packageID}/{name}/`.

```python
# Create a workspace (idempotent)
path = workspace.create("my-repo")
# Returns the absolute path to the directory

# Get the path (None if workspace doesn't exist)
path = workspace.path("my-repo")

# List all workspaces owned by this extension
names = workspace.list()  # ["my-repo", "cache"]

# Delete a workspace and all its contents
workspace.delete("my-repo")

# Get disk usage in bytes (10-second timeout)
size = workspace.usage("my-repo")  # 1048576
```

**Constraints:**
- Names must match `^[a-z][a-z0-9_]{0,62}$` (lowercase, no spaces or separators).
- Path traversal and symlink escape are blocked.
- Quota enforcement via `WORKSPACE_QUOTA_MB` env var (0 = unlimited).
- Module not available if `WORKSPACE_ROOT` is unset or not writable.
  Use `settings.has_capability("workspace")` to check availability.

---

## Example: automated stage hook

A simple hook that reads a setting, queries data, and advances:

```python
def on_run(ctx):
    threshold = settings.get("approval_threshold", 1000)
    amount = ctx["stage_data"].get("amount", 0)

    if amount > threshold:
        notifications.send(
            ctx["started_by"],
            "High-value submission",
            body="Amount %d exceeds threshold." % amount,
        )
        return {"advance": True, "data": {"needs_review": True}}

    return {"advance": True, "data": {"needs_review": False}}
```