InputLayer
StarFork

WebSocket API

The WebSocket API provides real-time, bidirectional communication with InputLayer. It supports authentication, query execution, streaming results, session rules, and push notifications.

Connection

Connect to the global WebSocket endpoint:

ws://host:port/ws

With TLS (via reverse proxy):

wss://host:port/ws

Optionally specify a knowledge graph via query parameter:

ws://host:port/ws?kg=mydb

Authentication

The first message after connection must authenticate. There are two methods:

Username/Password

{
  "type": "login",
  "username": "admin",
  "password": "admin"
}

API Key

{
  "type": "authenticate",
  "api_key": "your-api-key"
}

Auth Response

On success:

{
  "type": "authenticated",
  "session_id": "a1b2c3d4",
  "knowledge_graph": "default",
  "version": "0.1.0",
  "role": "admin"
}

On failure:

{
  "type": "auth_error",
  "message": "Invalid credentials"
}

The server allows 30 seconds for authentication before closing the connection.

Executing Statements

Send any InputLayer statement or meta command:

{
  "type": "execute",
  "program": "?edge(X, Y)"
}

The program field accepts any valid InputLayer statement: queries, inserts, rule definitions, meta commands, etc.

Result Response

{
  "type": "result",
  "columns": ["X", "Y"],
  "rows": [[1, 2], [2, 3], [3, 4]],
  "row_count": 3,
  "total_count": 3,
  "truncated": false,
  "execution_time_ms": 2,
  "errors": []
}

Error Response

{
  "type": "error",
  "message": "String value too long: 70000 bytes (max 65536)",
  "code": "validation"
}

A program of one statement that fails gets an error. code says why: validation, not_found, conflict, unsupported or internal; it is absent for connection, auth and engine failures.

Statement Failures

A multi-statement program keeps running past a failed statement and replies with a result (or result_start) whose errors lists each failure by 0-based statement index:

{
  "type": "result",
  "columns": ["message"],
  "rows": [["Inserted 1 fact(s) into 'a'."], ["Relation 'x' not found."]],
  "errors": [{"index": 1, "code": "not_found", "message": "Relation 'x' not found."}]
}

An empty errors means every statement succeeded. Do not match message text to detect failures.

Knowledge Graph Operations

Use meta commands via execute:

{"type": "execute", "program": ".kg create mydb"}
{"type": "execute", "program": ".kg use mydb"}
{"type": "execute", "program": ".kg list"}
{"type": "execute", "program": ".kg drop mydb"}

When .kg use switches knowledge graphs, the result includes a switched_kg field.

Streaming Results

For large result sets (> 1MB serialized), the server automatically streams results in chunks:

Stream Start

{
  "type": "result_start",
  "columns": ["X", "Y"],
  "total_count": 50000,
  "truncated": false,
  "execution_time_ms": 245
}

Stream Chunk

{
  "type": "result_chunk",
  "rows": [[1, 2], [2, 3]],
  "chunk_index": 0
}

Each chunk contains up to 500 rows. Chunks are sent sequentially.

Stream End

{
  "type": "result_end",
  "row_count": 50000,
  "chunk_count": 100
}

Small results (< 1MB) use the single result message, maintaining backward compatibility.

Session Rules

Session rules are ephemeral rules scoped to the current WebSocket connection. They are automatically cleared when the connection closes.

Define Session Rule

Session rules omit the + prefix:

{
  "type": "execute",
  "program": "temp(X, Y) <- edge(X, Y), X < Y"
}

Session Facts

Insert transient facts scoped to the session:

{
  "type": "execute",
  "program": "context(\"user\", \"alice\")"
}

Query with Session Rules

Session rules are automatically included when querying:

{
  "type": "execute",
  "program": "?temp(X, Y)"
}

Clear Session

{
  "type": "execute",
  "program": ".session clear"
}

Notifications

The server pushes notifications when persistent data changes in the session's knowledge graph.

Persistent Update

Sent when base facts in a relation change:

{
  "type": "persistent_update",
  "knowledge_graph": "default",
  "relation": "edge",
  "operation": "insert",
  "count": 5,
  "timestamp_ms": 1700000000000,
  "seq": 42
}

Rule Change

Sent when a rule is registered, removed, or dropped:

{
  "type": "rule_change",
  "knowledge_graph": "default",
  "rule_name": "reachable",
  "operation": "registered",
  "timestamp_ms": 1700000000000,
  "seq": 43
}

KG Change

Sent when a knowledge graph is created or dropped:

{
  "type": "kg_change",
  "knowledge_graph": "mydb",
  "operation": "created",
  "timestamp_ms": 1700000000000,
  "seq": 44
}

Schema Change

Sent when a relation is dropped:

{
  "type": "schema_change",
  "knowledge_graph": "default",
  "entity": "edge",
  "operation": "dropped",
  "timestamp_ms": 1700000000000,
  "seq": 45
}

The seq field is a monotonic sequence number for deduplication on reconnect.

Standing Queries

A standing query keeps a query's result current: after each committed change that can affect it, the server pushes the rows that entered and left the result. Use it to react to derived conclusions without polling.

Subscribe

{"type": "execute", "program": ".subscribe late_orders ?late(Order, Customer)"}

The reply is a normal result holding the initial snapshot. The subscription is bound to the connection and to its current knowledge graph.

<id> is client-chosen: 1-128 characters from [A-Za-z0-9_.:-], unique per connection.

Delta

{
  "type": "subscription_delta",
  "subscription": "late_orders",
  "knowledge_graph": "default",
  "seq": 1,
  "columns": ["Order", "Customer"],
  "inserted": [[1042, "acme"]],
  "retracted": [[1007, "globex"]]
}
  • Set semantics: inserted and retracted are the difference between consecutive result sets, each sorted. Applying every delta to the snapshot reproduces the current result.
  • seq starts at 1 per subscription and increases by one per delta. Unchanged results push nothing.
  • Triggers: inserts, deletes, rule add/drop/remove, and relation drop on any relation the query depends on (its body plus, transitively, the bodies of the persistent rules defining them). Writes to other relations cause no evaluation.
  • Coalescing: at most one evaluation runs per subscription. Commits that land meanwhile trigger one follow-up evaluation, so a burst produces few deltas and the last one always reflects the final state.
  • Evaluation runs off the connection's read loop and off the write path.

Errors

If re-evaluation fails (for example read access was revoked), the server pushes:

{"type": "subscription_error", "subscription": "late_orders", "message": "Access denied"}

The subscription stays registered; the next successful evaluation's delta is relative to the last good result.

.subscribe replies with an error when the id is taken, the per-connection limit is reached, the query uses limit/offset, or the user lacks read access to the knowledge graph (the same check as a query).

Unsubscribe

{"type": "execute", "program": ".unsubscribe late_orders"}

Subscriptions are also removed when the connection closes and when it switches knowledge graph (.kg use, .kg create). The switch removes them silently: re-subscribe in the new knowledge graph.

Scope and limits

  • Subscriptions see persistent data only: session facts and session rules are excluded, as with .why.
  • At most http.rate_limit.ws_max_subscriptions per connection (default 64, 0 = unlimited).
  • A delta larger than the maximum message size is replaced by a subscription_error.

Keep-Alive

Send a ping to keep the connection alive:

{"type": "ping"}

Response:

{"type": "pong"}

The server also sends WebSocket-level pings every 30 seconds to detect dead connections.

Connection Lifecycle

  1. Connect to ws://host:port/ws
  2. Authenticate with login or authenticate
  3. Execute statements and queries
  4. Receive notifications for data changes and subscription deltas
  5. Close gracefully by sending a WebSocket Close frame

Graceful Close

Always send a Close frame before disconnecting to avoid server warnings:

# Python
await websocket.close()
// JavaScript
ws.close(1000, "Normal closure");

Reconnection

If the connection drops:

  1. Wait with exponential backoff (1s, 2s, 4s, 8s, max 30s)
  2. Reconnect and re-authenticate
  3. Session rules, facts, and subscriptions are lost - re-create them after reconnecting
  4. Persistent data (facts, rules, indexes) is unaffected
  5. Use the seq field from notifications to detect missed events

Rate Limiting

The server applies per-connection rate limiting (default: 1000 messages/sec, configurable). Exceeding the limit returns:

{
  "type": "error",
  "message": "Rate limit exceeded (1000 msgs/sec)"
}

Example: Python Client





async def main():
    async with websockets.connect("ws://localhost:8080/ws") as ws:
        # Authenticate
        await ws.send(json.dumps({
            "type": "login",
            "username": "admin",
            "password": "admin"
        }))
        auth_resp = json.loads(await ws.recv())
        assert auth_resp["type"] == "authenticated"
        print(f"Session: {auth_resp['session_id']}")

        # Insert data
        await ws.send(json.dumps({
            "type": "execute",
            "program": "+edge[(1, 2), (2, 3), (3, 4)]"
        }))
        result = json.loads(await ws.recv())

        # Query
        await ws.send(json.dumps({
            "type": "execute",
            "program": "?edge(X, Y)"
        }))
        result = json.loads(await ws.recv())
        print(f"Got {result['row_count']} rows")

        # Close gracefully
        await ws.close()

asyncio.run(main())

Example: JavaScript Client

const ws = new WebSocket("ws://localhost:8080/ws");

ws.onopen = () => {
  ws.send(JSON.stringify({
    type: "login",
    username: "admin",
    password: "admin"
  }));
};

ws.onmessage = (event) => {
  const msg = JSON.parse(event.data);

  if (msg.type === "authenticated") {
    // Authenticated - start querying
    ws.send(JSON.stringify({
      type: "execute",
      program: "?edge(X, Y)"
    }));
  }

  if (msg.type === "result") {
    console.log(`Query returned ${msg.row_count} rows`);
  }

  // Handle streaming results
  if (msg.type === "result_start") {
    console.log(`Streaming ${msg.total_count} rows...`);
  }
  if (msg.type === "result_chunk") {
    console.log(`Received chunk ${msg.chunk_index}`);
  }
  if (msg.type === "result_end") {
    console.log(`Stream complete: ${msg.row_count} rows`);
  }

  // Handle notifications
  if (msg.type === "persistent_update") {
    console.log(`${msg.relation}: ${msg.operation} ${msg.count} rows`);
  }
};