Feat v0.5.1 chat core library + unicode security gate + cluster registry design
All checks were successful
CI/CD / detect-changes (pull_request) Successful in 4s
CI/CD / test-frontend (pull_request) Successful in 5s
CI/CD / test-go-pg (pull_request) Successful in 2m31s
CI/CD / test-sqlite (pull_request) Successful in 2m45s
CI/CD / build-and-deploy (pull_request) Successful in 1m34s

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-30 12:26:16 +00:00
parent 2abf406db8
commit 5d58a1cd4c
14 changed files with 1883 additions and 29 deletions

View File

@@ -2,6 +2,46 @@
All notable changes to Switchboard Core are documented here.
## v0.5.1 — Chat Core Library
### Added
- **`chat-core` library package**: Conversations, messages, participants, and
read cursors as ext_data tables. Permissions: `db.write`, `realtime.publish`.
Consumable by other packages via `lib.require("chat-core")`.
- **4 ext_data tables**: `conversations` (title, type, created_by, updated_at),
`participants` (conversation_id, participant_id, type, display_name, role,
joined_at), `messages` (conversation_id, participant_id, content, content_type,
edited_at), `read_cursors` (conversation_id, participant_id,
last_read_message_id).
- **6 exported Starlark functions**: `create()`, `send()`, `history()`,
`add_participant()`, `remove_participant()`, `mark_read()`.
- **15 REST endpoints**: Full CRUD for conversations, messages, participants.
Cursor-based paginated message history. Per-conversation unread counts.
- **Realtime events**: `message`, `message.edited`, `message.deleted`,
`participant.added`, `participant.removed` published to
`conversation:{id}` channels.
- **`db.query()` range parameters**: New `before` and `after` kwargs accept
dicts of `{col: val}` generating `col < ?` / `col > ?` WHERE clauses.
Enables cursor-based pagination and efficient unread counting for all
extension packages. 7 new Go tests.
### Security
- **Invisible Unicode scanning gate** (GlassWorm / Trojan Source defense):
`ScanSource()` detects variation selectors, bidi overrides, zero-width chars,
tag characters, and other invisible Unicode across 10 ranges. `Verdict()`
blocks GlassWorm payloads (>=10 variation selector / tag chars), Trojan Source
(any bidi override), and excessive zero-width chars (>=6). Two enforcement
points: extension install (422 rejection before files written) and Starlark
execution (belt-and-suspenders). 23 new tests including GlassWorm payload
simulation.
### Changed
- **Admin packages type filter**: Added `Library` option alongside Surface/Extension/Full/Workflow. Filter dropdown replaced button group with themed `Dropdown` primitive (keyboard nav, CSS-var theming, consistent with rest of admin UI).
- SDK version bumped from 0.5.0 to 0.5.1.
## v0.5.0 — Realtime Primitive + Dialog Audit + Permissions UI
### Added

View File

@@ -1,6 +1,6 @@
# Switchboard Core — Roadmap
## Current: v0.5.0Realtime Primitive + Dialog Audit + Permissions UI
## Current: v0.5.1Chat Core Library
Fork of chat-switchboard, gutted to a pure extension platform. All AI/chat
features removed from the kernel. What remains is the minimum viable
@@ -341,18 +341,22 @@ participation is post-MVP.
| Native dialog audit | ✅ | 5 bare `confirm()` calls migrated to `sw.confirm()` in tasks, schedules, notes (×2), editor packages. |
| Admin permissions UI | ✅ | Permissions drawer in admin Packages page. Grant/revoke per permission, grant-all bulk action. `pending_review`/`suspended` status badges. SDK API client methods added. Closes gap identified in v0.4.7. |
### v0.5.1 — Chat Core Library (planned)
### v0.5.1 — Chat Core Library
| Step | Status | Description |
|------|--------|-------------|
| Library package | | `chat-core` library with `permissions: ["db.write"]`. Private ext_data tables, exported Starlark functions, REST endpoints. |
| Conversations table | | `conversations` — title, type (direct/group), created_by, updated_at. |
| Participants table | | `participants` — conversation_id, participant_id, participant_type (user/bot), display_name, role (member/admin), joined_at. |
| Messages table | | `messages` — conversation_id, participant_id, content, content_type (text/system/file), edited_at. |
| Read cursors table | | `read_cursors` — conversation_id, participant_id, last_read_message_id. |
| Exported Starlark API | | `chat.create()`, `chat.send()`, `chat.history()`, `chat.add_participant()`, `chat.remove_participant()`, `chat.mark_read()`. |
| REST endpoints | | Full CRUD for conversations, messages, participants. Paginated message history. Unread counts. |
| Realtime integration | | On `chat.send()`: insert message, emit `chat.message.created` event, call `realtime.publish("conversation:{id}", "message", data)`. |
| Library package | | `chat-core` library with `permissions: ["db.write", "realtime.publish"]`. Private ext_data tables, exported Starlark functions, REST endpoints. |
| Conversations table | | `conversations` — title, type (direct/group), created_by, updated_at. |
| Participants table | | `participants` — conversation_id, participant_id, participant_type (user/bot), display_name, role (member/admin), joined_at. |
| Messages table | | `messages` — conversation_id, participant_id, content, content_type (text/system/file), edited_at. |
| Read cursors table | | `read_cursors` — conversation_id, participant_id, last_read_message_id. |
| Exported Starlark API | | `create()`, `send()`, `history()`, `add_participant()`, `remove_participant()`, `mark_read()` — available via `lib.require("chat-core")`. |
| REST endpoints | ✅ | 15 routes: full CRUD for conversations, messages, participants. Cursor-based paginated message history. Per-conversation unread counts. |
| Realtime integration | | On `send()`: insert message, publish `realtime.publish("conversation:{id}", "message", data)`. Also emits `participant.added`, `participant.removed`, `message.edited`, `message.deleted` events. |
| db.query() range params | ✅ | Kernel enhancement: `before`/`after` kwargs for `db.query()` enabling cursor-based pagination and efficient unread counting. 7 new tests. |
| Admin Library filter | ✅ | Added `Library` type filter to admin Packages page alongside Surface/Extension/Full/Workflow. |
| Admin filter Dropdown | ✅ | Replaced button group with themed `Dropdown` primitive for type filter. Keyboard nav, consistent theming via CSS vars. |
| Unicode security gate | ✅ | `ScanSource()` detects variation selectors, bidi overrides, zero-width chars, tag characters. `Verdict()` blocks GlassWorm payloads and Trojan Source (CVE-2021-42574). Two enforcement points: install (422) + Starlark exec. 23 new tests. |
### v0.5.2 — Chat Surface (planned)
@@ -377,6 +381,12 @@ participation is post-MVP.
| Multi-user E2E | | Docker compose test: multiple browser sessions, verify real-time message delivery, presence indicators, cross-replica broadcast. |
| Workflow integration | | Verify: workflow stage with `audience: team` can create a conversation via `chat.create()`, add assigned members as participants. Conversation scoped to instance. |
### Backlog — Admin UI Polish
| Item | Description |
|------|-------------|
| "Requires Review" filter | Dropdown filter option or dedicated button for `pending_review` packages. Useful when the package list grows long. |
### v0.5.4 — Package Updates (planned)
| Step | Status | Description |
@@ -403,9 +413,33 @@ participation is post-MVP.
Extension, communication, and operations tracks converge. First
externally usable release.
- Health monitoring (extension package wrapping kernel `/health`)
- Backup/restore tooling (ext_data export/import, admin surface)
- Documentation site (extension authoring guide, API reference, deployment guide)
Design docs: `docs/DESIGN-cluster-registry.md` — PG-backed cluster registry and self-assembling mesh.
### v0.6.0 — Cluster Registry + HA (planned)
PG is the consensus layer. Zero new infrastructure. `UNLOGGED` table + `LISTEN/NOTIFY` replaces etcd/Consul/Redis for homelab-to-small-team scale.
| Step | Status | Description |
|------|--------|-------------|
| `node_registry` table | | `UNLOGGED TABLE` — node_id, endpoint, seq, registered_at, heartbeat, stats JSONB. Folds into existing schema init. |
| Node registration | | Self-registration on startup: `INSERT ... ON CONFLICT DO UPDATE`. `node_id` = `hostname-PID` or `CLUSTER_NODE_ID` env override. |
| Heartbeat tick | | Every 10s: update own heartbeat + collect runtime stats (goroutines, heap, GC, uptime, ws_clients, extensions). |
| Stale sweep | | Every heartbeat tick: `DELETE WHERE heartbeat < now() - 30s`. All nodes run it — idempotent, no ring topology. |
| Self-eviction | | If heartbeat UPDATE returns 0 rows: node was swept by peer → log error + `os.Exit(1)`. K8s restarts → re-register. |
| LISTEN/NOTIFY routing | | Durable events (messages, state changes) fan-out via `pg_notify`. All replicas receive, push to local WS subscribers, drop if irrelevant. Phase 1: ephemeral events (typing, presence) also via NOTIFY (8KB limit, ~60 bytes each). |
| Cluster API | | `GET /api/admin/cluster``SELECT node_id, endpoint, registered_at, heartbeat, stats FROM node_registry ORDER BY seq`. Returns `{data: [...]}` envelope. |
| Admin cluster dashboard | | Extension renders one card per node: node_id, endpoint, uptime, ws_clients, heap, GC pause. JSONB stats — future keys render automatically, no schema migration. |
| Health endpoint | | `GET /health` includes `cluster: {size, peers, heartbeat_age_ms}`. Satisfies MVP health monitoring requirement. |
| Config | | `CLUSTER_NODE_ID`, `CLUSTER_HEARTBEAT_INTERVAL` (default 10s), `CLUSTER_STALE_THRESHOLD` (default 30s), `CLUSTER_ENDPOINT` (Phase 2 mesh, auto-detect). |
| Single-node regression | | One-node behavior identical to pre-cluster: one registry row, NOTIFY delivers back to same instance. No special-casing. |
| Multi-node integration test | | Docker Compose: 3 instances, shared PG. Verify registration, heartbeat, peer discovery, stale sweep on shutdown, WS fan-out. |
### v0.6.1 — Backup/Restore + Documentation (planned)
| Step | Status | Description |
|------|--------|-------------|
| Backup tooling | | ext_data export/import, admin surface for backup management. |
| Documentation site | | Extension authoring guide, API reference, deployment guide. |
## Post-MVP
@@ -434,5 +468,6 @@ externally usable release.
| Notes over Editor | First surface is Obsidian-style notes (rich text, folders, backlinks) instead of a code editor. Notes is a stronger E2E proof — it exercises ext_data, storage, and the SDK more fully than a pure-browser CM6 editor. |
| No built-in auto-install | Extensions ship in the repo but are not auto-installed. Distribution model TBD — explicit install only. Keeps the kernel clean and avoids opinionated defaults. |
| Chat as extension, not kernel | Human-to-human chat built entirely as library + surface packages. Zero kernel awareness of conversations, messages, or participants. Kernel gains one generic `realtime` module. Proves near-infinite extensibility. LLM participation layers on top via a separate bridge extension — the chat system doesn't know or care whether a participant is human or AI. |
| PG as consensus layer | Horizontal scaling uses PG as the sole coordinator (UNLOGGED node_registry + LISTEN/NOTIFY). No etcd, Consul, Redis, or Raft. Rationale: system is already tightly coupled to PG; adding a second consensus layer doubles operational complexity for zero benefit at homelab-to-small-team scale. UNLOGGED table is visible to all connections, survives session disconnect, and cleans up automatically on PG crash — correct behavior since all nodes are dead anyway. Sweep-all for health (every node deletes stale rows) is simpler than ring topology and has no edge cases. |
| Two trigger tiers | Event + webhook triggers are extension-declared (manifest contract, full sandbox). Scheduled tasks are user-created ad-hoc (restricted sandbox — no raw HTTP, no DB table creation, connections-only outbound). Separation keeps extension contracts static and user automation safe. |
| Scheduled task identity | Tasks run as their creator (RBAC-scoped). Admin-created tasks can opt into system context. Creator deactivation pauses the schedule. Ensures audit trail and permission boundaries. |

View File

@@ -1 +1 @@
0.5.0
0.5.1

View File

@@ -0,0 +1,290 @@
# Design: PG-Backed Cluster Registry & Self-Assembling Mesh
**Status:** Proposed — Pre-MVP / MVP wrap-up
**Author:** Jeff
**Date:** 2026-03-30
**Scope:** Kernel primitive — zero domain awareness
---
## Problem
Switchboard-core is tightly coupled to PostgreSQL. Scaling horizontally requires instance coordination: peer discovery, health monitoring, ephemeral event routing, and (optionally) leader election. Traditional HA solutions (Raft, etcd, Consul) introduce a second consensus layer on top of PG — doubling operational complexity for a system that already has serializable transactions and LISTEN/NOTIFY.
## Principle
PG is already the consensus system. Use it as the sole coordinator for service discovery, health, pub/sub, and cluster state. Zero new infrastructure dependencies.
---
## Design
### 1. Node Registry (Unlogged Table)
```sql
CREATE UNLOGGED TABLE IF NOT EXISTS node_registry (
node_id TEXT PRIMARY KEY,
endpoint TEXT NOT NULL,
seq SERIAL,
registered_at TIMESTAMPTZ DEFAULT now(),
heartbeat TIMESTAMPTZ DEFAULT now(),
stats JSONB DEFAULT '{}'
);
```
**Why UNLOGGED:** No WAL overhead, visible to all connections, survives session disconnect. Contents do NOT survive a PG crash — but a PG crash means all nodes are dead anyway. Clean slate on recovery is the correct behavior; all nodes re-register on restart.
**Why not TEMPORARY:** PG temporary tables are session-scoped and invisible to other connections. Useless for cross-instance coordination.
### 2. Node Lifecycle
#### Startup
```
boot → connect PG → register self → read peer list →
subscribe LISTEN/NOTIFY channels → open WS to clients → ready
```
```sql
-- Self-registration (atomic, idempotent)
INSERT INTO node_registry (node_id, endpoint)
VALUES ($1, $2)
ON CONFLICT (node_id) DO UPDATE
SET endpoint = EXCLUDED.endpoint,
registered_at = now(),
heartbeat = now(),
stats = '{}';
```
`node_id` is generated at startup: `hostname + PID` or a UUID. Deterministic IDs enable restart-in-place without orphaned rows.
#### Heartbeat Tick (every N seconds, e.g., 10s)
Each node performs two operations per tick:
```go
// 1. Update own heartbeat + stats
stats := collectStats() // see §5
db.Exec(`UPDATE node_registry SET heartbeat = now(), stats = $2 WHERE node_id = $1`,
nodeID, stats)
// 2. Sweep stale entries (idempotent — safe for concurrent execution)
db.Exec(`DELETE FROM node_registry WHERE heartbeat < now() - interval '30 seconds'`)
```
The sweep is the distributed health check. Every node runs it. Multiple nodes deleting the same stale row is a no-op. No ring topology required — the sweep is the safety net that guarantees eventual cleanup regardless of which nodes are alive.
#### Self-Eviction
If a node's own heartbeat UPDATE returns 0 rows affected, it has been swept by a peer:
```go
result := db.Exec(`UPDATE node_registry SET heartbeat = now() WHERE node_id = $1`, nodeID)
if result.RowsAffected() == 0 {
log.Error("evicted from registry, shutting down")
os.Exit(1)
}
```
Kubernetes restarts the process. On boot it re-registers as a fresh node. The restart loop IS the recovery mechanism.
### 3. Leaderless by Default
No `role` column. No primary/replica distinction. All nodes are equal peers. PG is the sole coordinator.
**If a leader is ever needed** (future use case), it's a small lift:
```sql
ALTER TABLE node_registry ADD COLUMN role TEXT DEFAULT 'peer';
-- Atomic leader claim: first writer wins
UPDATE node_registry SET role = 'primary'
WHERE node_id = $1
AND NOT EXISTS (SELECT 1 FROM node_registry WHERE role = 'primary');
```
No quorum, no voting, no term numbers. PG serialization handles it. But the current design intentionally avoids this — consistent with the existing pattern where scheduled tasks use first-to-mark-wins execution, not leader assignment.
### 4. Realtime Event Routing (Two Tiers)
#### Tier 1: Durable Events (messages, state changes)
These are PG writes. Fan-out via `LISTEN/NOTIFY`:
```
User A sends message → Instance 1 writes to PG →
pg_notify('channel:xyz', '{"type":"message","id":"..."}') →
All instances receive → push to local WS clients subscribed to channel:xyz
```
This maps directly to the v0.5.0 `realtime.publish()` Starlark primitive. Kernel implementation is `pg_notify(channel, payload)`.
#### Tier 2: Ephemeral Events (typing indicators, presence, cursor position)
These NEVER hit durable storage.
**Phase 1 (MVP):** Use PG LISTEN/NOTIFY for ephemeral events too.
```
User A types → Instance 1 WS handler →
pg_notify('ephemeral:xyz', '{"type":"typing","user":"A"}') →
Instance 2 receives → pushes to User B's WS
Instance 3 receives → pushes to User C's WS
```
PG NOTIFY payload limit is 8KB. A typing indicator is ~60 bytes. At homelab scale (3 nodes, dozens of users) this is a non-issue. One codepath, one transport, zero new infrastructure.
**Phase 2 (Post-MVP, if needed):** Direct instance-to-instance mesh for ephemeral events.
- Peer endpoints are already in the registry
- Each node dials peers discovered from the registry on startup and on peer list changes
- Ephemeral events route over the mesh; durable events stay on PG NOTIFY
- The `realtime.publish()` Starlark API does not change — kernel swaps the underlying transport
```
┌─────────────────────────────────┐
│ PG (source of truth) │
│ ┌───────────┐ ┌──────────────┐ │
│ │ node_ │ │ LISTEN/ │ │
│ │ registry │ │ NOTIFY │ │
│ └───────────┘ └──────────────┘ │
└──────┬──────────────┬────────────┘
│ durable │
┌────────────┼──────────────┼────────────┐
│ │ │ │
┌────┴────┐ ┌───┴─────┐ ┌────┴─────┐
│ Node 1 │◄─┤ Node 2 ├─►│ Node 3 │
│ User A │ │ User B │ │ User C │
└─────────┘ └─────────┘ └──────────┘
◄──── ephemeral mesh ────►
(Phase 2, if needed)
```
**Channel subscription model (Phase 1):** Broadcast and discard. Every node receives every event, checks local WebSocket subscriptions, drops if irrelevant. At 3 nodes this is the correct answer — trivial, no subscription state to sync, fully leaderless.
### 5. Stats Collection & Admin Dashboard
Each heartbeat tick publishes runtime stats in the `stats` JSONB column:
```go
var m runtime.MemStats
runtime.ReadMemStats(&m)
stats := map[string]any{
"goroutines": runtime.NumGoroutine(),
"heap_alloc": m.HeapAlloc,
"heap_sys": m.HeapSys,
"gc_cycles": m.NumGC,
"gc_pause_ns": m.PauseNs[(m.NumGC+255)%256],
"uptime_sec": time.Since(startTime).Seconds(),
"ws_clients": wsHub.Count(),
"extensions": sandbox.LoadedCount(),
}
```
Admin dashboard API endpoint:
```sql
SELECT node_id, endpoint, registered_at, heartbeat, stats
FROM node_registry
ORDER BY seq;
```
**ICD envelope:** `{"data": [...]}` — consistent with list endpoint convention.
The admin dashboard extension renders one card per node. One node = one card. Three nodes = three cards. The extension has zero awareness of clustering — it reads a list and renders it. This is the extensibility proof working as designed.
Since `stats` is JSONB, future additions (mesh peer latency, extension execution counts, queue depths) require no schema migration. The dashboard extension renders whatever keys are present.
### 6. Self-Assembling Mesh Properties
The mesh builds itself from the registry with zero configuration:
| Event | Behavior |
|---|---|
| Node starts | Inserts into registry → peers discover it on next heartbeat tick |
| Node dies | Heartbeat goes stale → swept by peers → mesh shrinks |
| Node restarts | Re-registers → peers discover it → mesh restores |
| Scale up (K8s) | New replicas register → mesh grows automatically |
| Scale down (K8s) | Terminated nodes go stale → swept → mesh shrinks |
| PG crash | All nodes lose connection → all exit → K8s restarts all → clean re-registration |
No config files listing peers. No seed nodes. No service discovery sidecar. No bootstrap ceremony. The registry table IS the service mesh control plane.
---
## What This Enables (Future)
These are NOT in scope for this work. Listed to validate the design supports them without rearchitecting.
- **Work distribution:** Primary (if elected) assigns extension execution shards via registry metadata. Replicas poll for assignments. No message queue.
- **Realtime routing (v0.5.0):** When a node owns a pub/sub channel, it registers ownership in PG. Other nodes route publishes to the owner. Node dies → sweep clears routes → subscribers reconnect.
- **Rolling deploys:** New version registers, old version's heartbeat goes stale, clean handoff with zero coordination protocol.
- **Session affinity:** Registry tracks which node holds a user's WS connection. Targeted event delivery instead of broadcast (optimization for scale).
---
## Implementation Notes
### Migration
Pre-MVP: fold `CREATE UNLOGGED TABLE` into existing schema initialization. No new migration file needed. DB rebuild required (acceptable — no install base).
### Configuration
| Setting | Default | Notes |
|---|---|---|
| `CLUSTER_NODE_ID` | `hostname-PID` | Override for deterministic identity across restarts |
| `CLUSTER_HEARTBEAT_INTERVAL` | `10s` | Tick frequency |
| `CLUSTER_STALE_THRESHOLD` | `30s` | 3x heartbeat = stale |
| `CLUSTER_ENDPOINT` | auto-detect | Advertised address for peer mesh (Phase 2) |
### Single-Node Behavior
When only one node is registered, the system behaves identically to pre-cluster switchboard-core. The registry has one row. LISTEN/NOTIFY delivers events back to the same instance. No special-casing required.
### Health Endpoint Integration
`GET /health` includes cluster status:
```json
{
"status": "healthy",
"node_id": "node-abc-1234",
"cluster": {
"size": 3,
"peers": ["node-def-5678", "node-ghi-9012"],
"heartbeat_age_ms": 2340
}
}
```
Satisfies the v0.6.0 MVP health monitoring requirement.
---
## Rejected Alternatives
| Alternative | Why rejected |
|---|---|
| etcd / Consul | Second consensus layer on top of PG. Doubles operational complexity for a system already tightly coupled to PG. Violates KISS. |
| Redis pub/sub for ephemeral events | Second stateful service. Two failure domains, two connection pools, two things to monitor. PG NOTIFY handles the same job at homelab scale. |
| Raft implementation in Go | Massive complexity for a problem PG serializable transactions already solve. |
| Ring topology monitoring | Each node watches the one above it. Adjacent failures create blind spots. Sweep-all is simpler and has no edge cases. |
| Leader election at startup | Unnecessary. Leaderless design with first-to-mark-wins task execution already works. Leader election is a future option, not a requirement. |
---
## Checklist
- [ ] `node_registry` unlogged table in schema init
- [ ] Node registration on startup
- [ ] Heartbeat tick with stats collection and stale sweep
- [ ] Self-eviction on 0-rows-affected heartbeat
- [ ] PG LISTEN/NOTIFY integration for realtime events
- [ ] Admin API endpoint: `GET /api/admin/cluster`
- [ ] Admin dashboard extension: cluster node cards
- [ ] Health endpoint cluster status
- [ ] Configuration via environment variables
- [ ] Single-node regression test (no behavioral change)
- [ ] Multi-node integration test (Docker Compose, 3 instances)

View File

@@ -0,0 +1,80 @@
{
"id": "chat-core",
"title": "Chat Core",
"type": "library",
"tier": "starlark",
"version": "0.1.0",
"description": "Core chat library — conversations, messages, participants, read cursors. Usable by other packages via lib.require().",
"author": "switchboard",
"permissions": ["db.write", "realtime.publish"],
"exports": [
"create", "send", "history",
"add_participant", "remove_participant", "mark_read"
],
"api_routes": [
{"method": "GET", "path": "/conversations"},
{"method": "POST", "path": "/conversations"},
{"method": "GET", "path": "/conversations/*"},
{"method": "PUT", "path": "/conversations/*"},
{"method": "DELETE", "path": "/conversations/*"},
{"method": "GET", "path": "/messages/*"},
{"method": "POST", "path": "/messages/*"},
{"method": "PUT", "path": "/messages/*"},
{"method": "DELETE", "path": "/messages/*"},
{"method": "GET", "path": "/participants/*"},
{"method": "POST", "path": "/participants/*"},
{"method": "DELETE", "path": "/participants/*"},
{"method": "POST", "path": "/read/*"},
{"method": "GET", "path": "/unread"}
],
"db_tables": {
"conversations": {
"columns": {
"title": "text",
"type": "text",
"created_by": "text",
"updated_at": "text"
},
"indexes": [["created_by"], ["updated_at"]]
},
"participants": {
"columns": {
"conversation_id": "text",
"participant_id": "text",
"participant_type": "text",
"display_name": "text",
"role": "text",
"joined_at": "text"
},
"indexes": [
["conversation_id"],
["participant_id"],
["conversation_id", "participant_id"]
]
},
"messages": {
"columns": {
"conversation_id": "text",
"participant_id": "text",
"content": "text",
"content_type": "text",
"edited_at": "text"
},
"indexes": [["conversation_id"]]
},
"read_cursors": {
"columns": {
"conversation_id": "text",
"participant_id": "text",
"last_read_message_id": "text"
},
"indexes": [["conversation_id", "participant_id"]]
}
},
"schema_version": 1
}

View File

@@ -0,0 +1,691 @@
# Chat Core — Starlark Backend (v0.1.0)
#
# Library package providing conversations, messages, participants,
# and read cursors. Consumable via lib.require("chat-core").
#
# Entry points:
# on_request(req) → REST API routes
# Exported globals → create, send, history, add_participant,
# remove_participant, mark_read
#
# Modules: db, json, realtime
# ═══════════════════════════════════════════════
# Helpers
# ═══════════════════════════════════════════════
def _resp(status, data):
return {"status": status, "body": json.encode(data), "headers": {"Content-Type": "application/json"}}
def _str(v):
if v == None:
return ""
return str(v)
def _int(v):
if v == None:
return 0
s = str(v)
if not s:
return 0
return int(s)
def _is_participant(conversation_id, user_id):
"""Check if user is a participant in the conversation."""
rows = db.query("participants", filters={"conversation_id": conversation_id, "participant_id": user_id}, limit=1)
return len(rows or []) > 0
def _get_participant(conversation_id, user_id):
"""Get participant record or None."""
rows = db.query("participants", filters={"conversation_id": conversation_id, "participant_id": user_id}, limit=1)
if rows and len(rows) > 0:
return rows[0]
return None
def _is_admin(conversation_id, user_id):
"""Check if user has admin role in the conversation."""
p = _get_participant(conversation_id, user_id)
if p:
return _str(p.get("role", "")) == "admin"
return False
def _get_conversation(conversation_id):
"""Get conversation by ID or None."""
rows = db.query("conversations", filters={"id": conversation_id}, limit=1)
if rows and len(rows) > 0:
return rows[0]
return None
def _now():
"""Current timestamp placeholder — db auto-populates created_at."""
return ""
# ═══════════════════════════════════════════════
# Exported API (lib.require("chat-core"))
# ═══════════════════════════════════════════════
def create(title, type="group", participants=None, creator_id=""):
"""Create a conversation and add initial participants.
Args:
title: conversation title
type: "direct" or "group" (default "group")
participants: list of dicts with {id, type?, display_name?, role?}
creator_id: user ID of the creator (added as admin)
Returns:
dict with conversation data
"""
if type not in ("direct", "group"):
type = "group"
conv = db.insert("conversations", {
"title": _str(title),
"type": type,
"created_by": _str(creator_id),
"updated_at": "",
})
cid = conv["id"]
# Add creator as admin participant
if creator_id:
db.insert("participants", {
"conversation_id": cid,
"participant_id": _str(creator_id),
"participant_type": "user",
"display_name": "",
"role": "admin",
"joined_at": "",
})
# Add initial participants
for p in (participants or []):
pid = _str(p.get("id", ""))
if not pid or pid == _str(creator_id):
continue
db.insert("participants", {
"conversation_id": cid,
"participant_id": pid,
"participant_type": _str(p.get("type", "user")),
"display_name": _str(p.get("display_name", "")),
"role": _str(p.get("role", "member")),
"joined_at": "",
})
return conv
def send(conversation_id, participant_id, content, content_type="text"):
"""Send a message to a conversation.
Args:
conversation_id: target conversation
participant_id: sender ID
content: message content
content_type: "text", "system", or "file" (default "text")
Returns:
dict with message data
"""
if content_type not in ("text", "system", "file"):
content_type = "text"
msg = db.insert("messages", {
"conversation_id": _str(conversation_id),
"participant_id": _str(participant_id),
"content": _str(content),
"content_type": content_type,
"edited_at": "",
})
# Update conversation timestamp
db.update("conversations", _str(conversation_id), {"updated_at": msg.get("created_at", "")})
# Publish realtime event
realtime.publish(
"conversation:" + _str(conversation_id),
"message",
{
"id": msg.get("id", ""),
"conversation_id": _str(conversation_id),
"participant_id": _str(participant_id),
"content": _str(content),
"content_type": content_type,
"created_at": msg.get("created_at", ""),
},
)
return msg
def history(conversation_id, limit=50, cursor=""):
"""Get paginated message history for a conversation.
Args:
conversation_id: target conversation
limit: max messages to return (1-100, default 50)
cursor: created_at value of last message from previous page
Returns:
dict with {messages, has_more, next_cursor}
"""
lim = _int(limit)
if lim < 1 or lim > 100:
lim = 50
filters = {"conversation_id": _str(conversation_id)}
before = {}
if cursor:
before = {"created_at": _str(cursor)}
rows = db.query("messages", filters=filters, order="-created_at", limit=lim + 1, before=before)
rows = rows or []
has_more = len(rows) > lim
if has_more:
rows = rows[:lim]
next_cursor = ""
if has_more and rows:
next_cursor = _str(rows[-1].get("created_at", ""))
return {"messages": rows, "has_more": has_more, "next_cursor": next_cursor}
def add_participant(conversation_id, participant_id, participant_type="user", display_name="", role="member"):
"""Add a participant to a conversation.
Returns:
dict with participant data
"""
cid = _str(conversation_id)
pid = _str(participant_id)
# Check if already a participant
existing = db.query("participants", filters={"conversation_id": cid, "participant_id": pid}, limit=1)
if existing and len(existing) > 0:
return existing[0]
p = db.insert("participants", {
"conversation_id": cid,
"participant_id": pid,
"participant_type": _str(participant_type),
"display_name": _str(display_name),
"role": _str(role),
"joined_at": "",
})
# Publish realtime event
realtime.publish(
"conversation:" + cid,
"participant.added",
{
"conversation_id": cid,
"participant_id": pid,
"participant_type": _str(participant_type),
"display_name": _str(display_name),
"role": _str(role),
},
)
# Insert system message
send(cid, pid, "joined the conversation", "system")
return p
def remove_participant(conversation_id, participant_id):
"""Remove a participant from a conversation.
Returns:
True on success
"""
cid = _str(conversation_id)
pid = _str(participant_id)
# Delete participant record
rows = db.query("participants", filters={"conversation_id": cid, "participant_id": pid}, limit=1)
for r in (rows or []):
db.delete("participants", r["id"])
# Delete read cursor
cursors = db.query("read_cursors", filters={"conversation_id": cid, "participant_id": pid}, limit=1)
for c in (cursors or []):
db.delete("read_cursors", c["id"])
# Publish realtime event
realtime.publish(
"conversation:" + cid,
"participant.removed",
{"conversation_id": cid, "participant_id": pid},
)
return True
def mark_read(conversation_id, participant_id, last_read_message_id):
"""Update the read cursor for a participant.
Returns:
True on success
"""
cid = _str(conversation_id)
pid = _str(participant_id)
mid = _str(last_read_message_id)
# Delete existing cursor (upsert via delete+insert)
existing = db.query("read_cursors", filters={"conversation_id": cid, "participant_id": pid}, limit=1)
for r in (existing or []):
db.delete("read_cursors", r["id"])
db.insert("read_cursors", {
"conversation_id": cid,
"participant_id": pid,
"last_read_message_id": mid,
})
return True
# ═══════════════════════════════════════════════
# REST API dispatcher
# ═══════════════════════════════════════════════
def on_request(req):
path = req["path"]
method = req["method"]
user_id = req.get("user_id", "")
# ── Conversations ──────────────────────────
# GET /conversations — list user's conversations
if method == "GET" and path == "/conversations":
return _handle_list_conversations(req, user_id)
# POST /conversations — create
if method == "POST" and path == "/conversations":
return _handle_create_conversation(req, user_id)
# GET /unread — unread counts
if method == "GET" and path == "/unread":
return _handle_unread(user_id)
# GET /conversations/:id
if method == "GET" and path.startswith("/conversations/"):
cid = path[len("/conversations/"):]
return _handle_get_conversation(cid, user_id)
# PUT /conversations/:id
if method == "PUT" and path.startswith("/conversations/"):
cid = path[len("/conversations/"):]
return _handle_update_conversation(cid, req, user_id)
# DELETE /conversations/:id
if method == "DELETE" and path.startswith("/conversations/"):
cid = path[len("/conversations/"):]
return _handle_delete_conversation(cid, user_id)
# ── Messages ───────────────────────────────
# Routes: /messages/:conversation_id or /messages/:conversation_id/:message_id
if path.startswith("/messages/"):
remainder = path[len("/messages/"):]
parts = remainder.split("/")
cid = parts[0]
mid = parts[1] if len(parts) > 1 else ""
if method == "GET" and not mid:
return _handle_list_messages(cid, req, user_id)
if method == "POST" and not mid:
return _handle_send_message(cid, req, user_id)
if method == "PUT" and mid:
return _handle_edit_message(cid, mid, req, user_id)
if method == "DELETE" and mid:
return _handle_delete_message(cid, mid, user_id)
# ── Participants ───────────────────────────
if path.startswith("/participants/"):
remainder = path[len("/participants/"):]
parts = remainder.split("/")
cid = parts[0]
pid = parts[1] if len(parts) > 1 else ""
if method == "GET" and not pid:
return _handle_list_participants(cid, user_id)
if method == "POST" and not pid:
return _handle_add_participant(cid, req, user_id)
if method == "DELETE" and pid:
return _handle_remove_participant(cid, pid, user_id)
# ── Read cursors ──────────────────────────
if method == "POST" and path.startswith("/read/"):
cid = path[len("/read/"):]
return _handle_mark_read(cid, req, user_id)
return _resp(404, {"error": "not found"})
# ═══════════════════════════════════════════════
# Conversation handlers
# ═══════════════════════════════════════════════
def _handle_list_conversations(req, user_id):
"""List conversations the user is a participant in."""
# Get all conversation IDs for this user
my_parts = db.query("participants", filters={"participant_id": user_id}, limit=500)
if not my_parts:
return _resp(200, {"data": []})
cids = []
for p in my_parts:
cids.append(_str(p.get("conversation_id", "")))
# Fetch conversations and enrich with last message
items = []
for cid in cids:
conv = _get_conversation(cid)
if not conv:
continue
# Get last message
last_msgs = db.query("messages", filters={"conversation_id": cid}, order="-created_at", limit=1)
last_msg = None
if last_msgs and len(last_msgs) > 0:
last_msg = {
"id": last_msgs[0].get("id", ""),
"content": _str(last_msgs[0].get("content", "")),
"participant_id": _str(last_msgs[0].get("participant_id", "")),
"content_type": _str(last_msgs[0].get("content_type", "")),
"created_at": _str(last_msgs[0].get("created_at", "")),
}
# Get participant count
parts = db.query("participants", filters={"conversation_id": cid}, limit=500)
part_count = len(parts or [])
items.append({
"id": conv.get("id", ""),
"title": conv.get("title", ""),
"type": conv.get("type", ""),
"created_by": conv.get("created_by", ""),
"updated_at": conv.get("updated_at", ""),
"created_at": conv.get("created_at", ""),
"last_message": last_msg,
"participant_count": part_count,
})
# Sort by updated_at descending (most recent first)
items = sorted(items, key=lambda x: x.get("updated_at", "") or x.get("created_at", ""), reverse=True)
return _resp(200, {"data": items})
def _handle_create_conversation(req, user_id):
"""Create a new conversation."""
body = json.decode(req.get("body", "{}"))
title = _str(body.get("title", ""))
conv_type = _str(body.get("type", "group"))
participants_list = body.get("participants", [])
conv = create(title, conv_type, participants_list, user_id)
return _resp(201, conv)
def _handle_get_conversation(cid, user_id):
"""Get conversation detail with participants."""
if not _is_participant(cid, user_id):
return _resp(403, {"error": "not a participant"})
conv = _get_conversation(cid)
if not conv:
return _resp(404, {"error": "conversation not found"})
parts = db.query("participants", filters={"conversation_id": cid}, limit=500)
return _resp(200, {
"id": conv.get("id", ""),
"title": conv.get("title", ""),
"type": conv.get("type", ""),
"created_by": conv.get("created_by", ""),
"updated_at": conv.get("updated_at", ""),
"created_at": conv.get("created_at", ""),
"participants": parts or [],
})
def _handle_update_conversation(cid, req, user_id):
"""Update conversation (admin only)."""
if not _is_admin(cid, user_id):
return _resp(403, {"error": "admin only"})
body = json.decode(req.get("body", "{}"))
patch = {}
title = body.get("title", None)
if title != None:
patch["title"] = _str(title)
if patch:
db.update("conversations", cid, patch)
conv = _get_conversation(cid)
return _resp(200, conv)
def _handle_delete_conversation(cid, user_id):
"""Delete conversation and all associated data (admin only)."""
if not _is_admin(cid, user_id):
return _resp(403, {"error": "admin only"})
# Cascade delete: messages, participants, read_cursors, then conversation
msgs = db.query("messages", filters={"conversation_id": cid}, limit=5000)
for m in (msgs or []):
db.delete("messages", m["id"])
parts = db.query("participants", filters={"conversation_id": cid}, limit=500)
for p in (parts or []):
db.delete("participants", p["id"])
cursors = db.query("read_cursors", filters={"conversation_id": cid}, limit=500)
for c in (cursors or []):
db.delete("read_cursors", c["id"])
db.delete("conversations", cid)
return _resp(200, {"ok": True})
# ═══════════════════════════════════════════════
# Message handlers
# ═══════════════════════════════════════════════
def _handle_list_messages(cid, req, user_id):
"""Paginated message history."""
if not _is_participant(cid, user_id):
return _resp(403, {"error": "not a participant"})
q = req.get("query", {})
limit_val = _int(q.get("limit", "50"))
cursor = _str(q.get("cursor", ""))
result = history(cid, limit_val, cursor)
return _resp(200, result)
def _handle_send_message(cid, req, user_id):
"""Send a message."""
if not _is_participant(cid, user_id):
return _resp(403, {"error": "not a participant"})
body = json.decode(req.get("body", "{}"))
content = _str(body.get("content", ""))
content_type = _str(body.get("content_type", "text"))
if not content:
return _resp(400, {"error": "content required"})
msg = send(cid, user_id, content, content_type)
return _resp(201, msg)
def _handle_edit_message(cid, mid, req, user_id):
"""Edit a message (author only)."""
if not _is_participant(cid, user_id):
return _resp(403, {"error": "not a participant"})
# Verify ownership
msgs = db.query("messages", filters={"id": mid}, limit=1)
if not msgs:
return _resp(404, {"error": "message not found"})
msg = msgs[0]
if _str(msg.get("participant_id", "")) != user_id:
return _resp(403, {"error": "can only edit own messages"})
body = json.decode(req.get("body", "{}"))
content = _str(body.get("content", ""))
if not content:
return _resp(400, {"error": "content required"})
# Use created_at of the message as the edited_at marker
db.update("messages", mid, {"content": content, "edited_at": msg.get("created_at", "")})
# Publish edit event
realtime.publish(
"conversation:" + cid,
"message.edited",
{"id": mid, "conversation_id": cid, "content": content, "edited_by": user_id},
)
updated = db.query("messages", filters={"id": mid}, limit=1)
return _resp(200, updated[0] if updated else {})
def _handle_delete_message(cid, mid, user_id):
"""Delete a message (author or admin)."""
if not _is_participant(cid, user_id):
return _resp(403, {"error": "not a participant"})
msgs = db.query("messages", filters={"id": mid}, limit=1)
if not msgs:
return _resp(404, {"error": "message not found"})
msg = msgs[0]
is_author = _str(msg.get("participant_id", "")) == user_id
is_conv_admin = _is_admin(cid, user_id)
if not is_author and not is_conv_admin:
return _resp(403, {"error": "can only delete own messages or be admin"})
db.delete("messages", mid)
# Publish delete event
realtime.publish(
"conversation:" + cid,
"message.deleted",
{"id": mid, "conversation_id": cid, "deleted_by": user_id},
)
return _resp(200, {"ok": True})
# ═══════════════════════════════════════════════
# Participant handlers
# ═══════════════════════════════════════════════
def _handle_list_participants(cid, user_id):
"""List participants in a conversation."""
if not _is_participant(cid, user_id):
return _resp(403, {"error": "not a participant"})
parts = db.query("participants", filters={"conversation_id": cid}, limit=500)
return _resp(200, {"data": parts or []})
def _handle_add_participant(cid, req, user_id):
"""Add a participant (admin only)."""
if not _is_admin(cid, user_id):
return _resp(403, {"error": "admin only"})
body = json.decode(req.get("body", "{}"))
pid = _str(body.get("participant_id", ""))
if not pid:
return _resp(400, {"error": "participant_id required"})
ptype = _str(body.get("participant_type", "user"))
display_name = _str(body.get("display_name", ""))
role = _str(body.get("role", "member"))
p = add_participant(cid, pid, ptype, display_name, role)
return _resp(201, p)
def _handle_remove_participant(cid, pid, user_id):
"""Remove a participant (admin only, or self-remove)."""
is_self = pid == user_id
if not is_self and not _is_admin(cid, user_id):
return _resp(403, {"error": "admin only (or remove yourself)"})
remove_participant(cid, pid)
return _resp(200, {"ok": True})
# ═══════════════════════════════════════════════
# Read cursor handlers
# ═══════════════════════════════════════════════
def _handle_mark_read(cid, req, user_id):
"""Mark conversation as read up to a message."""
if not _is_participant(cid, user_id):
return _resp(403, {"error": "not a participant"})
body = json.decode(req.get("body", "{}"))
mid = _str(body.get("last_read_message_id", ""))
if not mid:
return _resp(400, {"error": "last_read_message_id required"})
mark_read(cid, user_id, mid)
return _resp(200, {"ok": True})
def _handle_unread(user_id):
"""Get unread counts for all user's conversations."""
my_parts = db.query("participants", filters={"participant_id": user_id}, limit=500)
if not my_parts:
return _resp(200, {"data": {}})
counts = {}
for p in my_parts:
cid = _str(p.get("conversation_id", ""))
if not cid:
continue
# Get read cursor
cursors = db.query("read_cursors", filters={"conversation_id": cid, "participant_id": user_id}, limit=1)
if cursors and len(cursors) > 0:
last_read_id = _str(cursors[0].get("last_read_message_id", ""))
if last_read_id:
# Get the created_at of the last read message
last_read_msgs = db.query("messages", filters={"id": last_read_id}, limit=1)
if last_read_msgs and len(last_read_msgs) > 0:
last_read_at = _str(last_read_msgs[0].get("created_at", ""))
# Count messages after the read cursor
unread = db.query("messages", filters={"conversation_id": cid}, after={"created_at": last_read_at}, limit=1000)
counts[cid] = len(unread or [])
else:
# Last read message was deleted — count all
all_msgs = db.query("messages", filters={"conversation_id": cid}, limit=1000)
counts[cid] = len(all_msgs or [])
else:
# No cursor value — all messages are unread
all_msgs = db.query("messages", filters={"conversation_id": cid}, limit=1000)
counts[cid] = len(all_msgs or [])
else:
# No cursor at all — all messages are unread
all_msgs = db.query("messages", filters={"conversation_id": cid}, limit=1000)
counts[cid] = len(all_msgs or [])
return _resp(200, {"data": counts})

View File

@@ -299,6 +299,38 @@ func (h *PackageHandler) InstallPackage(c *gin.Context) {
}
}
// Unicode security gate — scan all scannable files for invisible/deceptive
// characters before writing anything to the extension store.
scanExts := map[string]bool{".star": true, ".json": true, ".js": true, ".html": true}
for _, f := range zr.File {
if f.FileInfo().IsDir() {
continue
}
ext := strings.ToLower(filepath.Ext(f.Name))
if !scanExts[ext] {
continue
}
rc, err := f.Open()
if err != nil {
continue
}
data, err := io.ReadAll(rc)
rc.Close()
if err != nil {
continue
}
findings := sandbox.ScanSource(string(data), f.Name)
if len(findings) > 0 {
if blocked, reason := sandbox.Verdict(findings); blocked {
c.JSON(http.StatusUnprocessableEntity, gin.H{
"error": "extension_blocked",
"reason": fmt.Sprintf("%s (file: %s)", reason, f.Name),
})
return
}
}
}
// Validate required fields
pkgID, _ := manifest["id"].(string)
title, _ := manifest["title"].(string)

View File

@@ -8,7 +8,7 @@
//
// Starlark API:
//
// rows = db.query("logs", filters={"user_id": "abc"}, order="created_at", limit=50)
// rows = db.query("logs", filters={"user_id": "abc"}, order="created_at", limit=50, before={"created_at": "2026-01-01"}, after={"count": 5})
// row = db.insert("logs", {"message": "hello"})
// ok = db.update("logs", row_id, {"message": "updated"})
// ok = db.delete("logs", row_id)
@@ -221,19 +221,64 @@ func goToStarlark(v any) (starlark.Value, error) {
// ── Builtins ─────────────────────────────────
// dbQuery implements db.query(table, filters=None, order=None, limit=100).
// starlarkRangeToSQL converts a Starlark dict of {col: val} into range
// comparison clauses (e.g. col < $N or col > $N). op must be "<" or ">".
func (cfg DBModuleConfig) starlarkRangeToSQL(rangeVal starlark.Value, op string, startIdx int) ([]string, []any, error) {
if rangeVal == starlark.None || rangeVal == nil {
return nil, nil, nil
}
d, ok := rangeVal.(*starlark.Dict)
if !ok {
return nil, nil, fmt.Errorf("db: range param must be a dict, got %s", rangeVal.Type())
}
if d.Len() == 0 {
return nil, nil, nil
}
var parts []string
var args []any
idx := startIdx
for _, item := range d.Items() {
col, ok := item[0].(starlark.String)
if !ok {
return nil, nil, fmt.Errorf("db: range key must be a string, got %s", item[0].Type())
}
colStr := string(col)
if strings.ContainsAny(colStr, " \t\n\"';-") {
return nil, nil, fmt.Errorf("db: invalid column name %q", colStr)
}
val, err := starlarkToGoValue(item[1])
if err != nil {
return nil, nil, fmt.Errorf("db: range value for %q: %w", colStr, err)
}
parts = append(parts, fmt.Sprintf("%s %s %s", colStr, op, cfg.ph(idx)))
args = append(args, val)
idx++
}
return parts, args, nil
}
// dbQuery implements db.query(table, filters=None, order=None, limit=100, before=None, after=None).
func dbQuery(ctx context.Context, cfg DBModuleConfig) func(*starlark.Thread, *starlark.Builtin, starlark.Tuple, []starlark.Tuple) (starlark.Value, error) {
return func(thread *starlark.Thread, b *starlark.Builtin, args starlark.Tuple, kwargs []starlark.Tuple) (starlark.Value, error) {
var table string
var filters starlark.Value = starlark.None
var order starlark.Value = starlark.None
var limit starlark.Int = starlark.MakeInt(100)
var before starlark.Value = starlark.None
var after starlark.Value = starlark.None
if err := starlark.UnpackArgs(b.Name(), args, kwargs,
"table", &table,
"filters?", &filters,
"order?", &order,
"limit?", &limit,
"before?", &before,
"after?", &after,
); err != nil {
return nil, err
}
@@ -248,14 +293,41 @@ func dbQuery(ctx context.Context, cfg DBModuleConfig) func(*starlark.Thread, *st
return nil, err
}
// Build range clauses (before → <, after → >)
beforeParts, beforeArgs, err := cfg.starlarkRangeToSQL(before, "<", len(whereArgs)+1)
if err != nil {
return nil, err
}
afterParts, afterArgs, err := cfg.starlarkRangeToSQL(after, ">", len(whereArgs)+len(beforeArgs)+1)
if err != nil {
return nil, err
}
// Merge all WHERE conditions
var allParts []string
var allArgs []any
// Extract equality parts from whereClause
if whereClause != "" {
// whereClause is "WHERE x = $1 AND y = $2"; strip the "WHERE " prefix
allParts = append(allParts, strings.TrimPrefix(whereClause, "WHERE "))
allArgs = append(allArgs, whereArgs...)
} else {
allArgs = append(allArgs, whereArgs...)
}
allParts = append(allParts, beforeParts...)
allArgs = append(allArgs, beforeArgs...)
allParts = append(allParts, afterParts...)
allArgs = append(allArgs, afterArgs...)
lim, ok := limit.Int64()
if !ok || lim < 1 || lim > 1000 {
lim = 100
}
query := fmt.Sprintf("SELECT * FROM %s", physTable)
if whereClause != "" {
query += " " + whereClause
if len(allParts) > 0 {
query += " WHERE " + strings.Join(allParts, " AND ")
}
if order != starlark.None {
@@ -275,11 +347,11 @@ func dbQuery(ctx context.Context, cfg DBModuleConfig) func(*starlark.Thread, *st
query += fmt.Sprintf(" ORDER BY %s %s", colStr, dir)
}
limitPH := cfg.ph(len(whereArgs) + 1)
limitPH := cfg.ph(len(allArgs) + 1)
query += " LIMIT " + limitPH
whereArgs = append(whereArgs, lim)
allArgs = append(allArgs, lim)
rows, err := cfg.DB.QueryContext(ctx, query, whereArgs...)
rows, err := cfg.DB.QueryContext(ctx, query, allArgs...)
if err != nil {
return nil, fmt.Errorf("db.query: %w", err)
}

View File

@@ -226,6 +226,104 @@ func TestDBViewRejectsUnknownView(t *testing.T) {
}
}
// ── before / after range parameters ──────────
func TestDBQueryBefore(t *testing.T) {
db, cfg := newTestDB(t)
db.Exec(`INSERT INTO ext_test_ext_logs (id, message, created_at) VALUES ('a', 'old', '2026-01-01T00:00:00Z')`)
db.Exec(`INSERT INTO ext_test_ext_logs (id, message, created_at) VALUES ('b', 'mid', '2026-02-01T00:00:00Z')`)
db.Exec(`INSERT INTO ext_test_ext_logs (id, message, created_at) VALUES ('c', 'new', '2026-03-01T00:00:00Z')`)
result, err := execScript(t, cfg, `rows = db.query("logs", before={"created_at": "2026-02-15T00:00:00Z"}, order="created_at")`)
if err != nil {
t.Fatalf("script error: %v", err)
}
rowsStr := result.Globals["rows"].String()
if !strings.Contains(rowsStr, "old") || !strings.Contains(rowsStr, "mid") {
t.Errorf("expected old and mid rows, got: %s", rowsStr)
}
if strings.Contains(rowsStr, "new") {
t.Errorf("unexpected 'new' row in before results: %s", rowsStr)
}
}
func TestDBQueryAfter(t *testing.T) {
db, cfg := newTestDB(t)
db.Exec(`INSERT INTO ext_test_ext_logs (id, message, created_at) VALUES ('a', 'old', '2026-01-01T00:00:00Z')`)
db.Exec(`INSERT INTO ext_test_ext_logs (id, message, created_at) VALUES ('b', 'mid', '2026-02-01T00:00:00Z')`)
db.Exec(`INSERT INTO ext_test_ext_logs (id, message, created_at) VALUES ('c', 'new', '2026-03-01T00:00:00Z')`)
result, err := execScript(t, cfg, `rows = db.query("logs", after={"created_at": "2026-02-01T00:00:00Z"}, order="created_at")`)
if err != nil {
t.Fatalf("script error: %v", err)
}
rowsStr := result.Globals["rows"].String()
if !strings.Contains(rowsStr, "new") {
t.Errorf("expected 'new' row, got: %s", rowsStr)
}
if strings.Contains(rowsStr, "old") || strings.Contains(rowsStr, "mid") {
t.Errorf("unexpected old/mid rows in after results: %s", rowsStr)
}
}
func TestDBQueryBeforeWithFilters(t *testing.T) {
db, cfg := newTestDB(t)
db.Exec(`INSERT INTO ext_test_ext_logs (id, message, user_id, created_at) VALUES ('a', 'u1-old', 'u1', '2026-01-01T00:00:00Z')`)
db.Exec(`INSERT INTO ext_test_ext_logs (id, message, user_id, created_at) VALUES ('b', 'u1-new', 'u1', '2026-03-01T00:00:00Z')`)
db.Exec(`INSERT INTO ext_test_ext_logs (id, message, user_id, created_at) VALUES ('c', 'u2-old', 'u2', '2026-01-01T00:00:00Z')`)
result, err := execScript(t, cfg, `rows = db.query("logs", filters={"user_id": "u1"}, before={"created_at": "2026-02-01T00:00:00Z"})`)
if err != nil {
t.Fatalf("script error: %v", err)
}
rowsStr := result.Globals["rows"].String()
if !strings.Contains(rowsStr, "u1-old") {
t.Errorf("expected u1-old, got: %s", rowsStr)
}
if strings.Contains(rowsStr, "u1-new") || strings.Contains(rowsStr, "u2-old") {
t.Errorf("unexpected rows: %s", rowsStr)
}
}
func TestDBQueryBeforeAfterCombined(t *testing.T) {
db, cfg := newTestDB(t)
db.Exec(`INSERT INTO ext_test_ext_logs (id, message, created_at) VALUES ('a', 'old', '2026-01-01T00:00:00Z')`)
db.Exec(`INSERT INTO ext_test_ext_logs (id, message, created_at) VALUES ('b', 'mid', '2026-02-01T00:00:00Z')`)
db.Exec(`INSERT INTO ext_test_ext_logs (id, message, created_at) VALUES ('c', 'new', '2026-03-01T00:00:00Z')`)
result, err := execScript(t, cfg, `rows = db.query("logs", after={"created_at": "2026-01-15T00:00:00Z"}, before={"created_at": "2026-02-15T00:00:00Z"})`)
if err != nil {
t.Fatalf("script error: %v", err)
}
rowsStr := result.Globals["rows"].String()
if !strings.Contains(rowsStr, "mid") {
t.Errorf("expected mid row, got: %s", rowsStr)
}
if strings.Contains(rowsStr, "old") || strings.Contains(rowsStr, "new") {
t.Errorf("unexpected old/new rows: %s", rowsStr)
}
}
func TestDBQueryBeforeEmptyDict(t *testing.T) {
_, cfg := newTestDB(t)
// Empty before dict should be a no-op (same as no before)
_, err := execScript(t, cfg, `rows = db.query("logs", before={})`)
if err != nil {
t.Fatalf("empty before dict should not error: %v", err)
}
}
func TestDBQueryBeforeInvalidColumn(t *testing.T) {
_, cfg := newTestDB(t)
_, err := execScript(t, cfg, `rows = db.query("logs", before={"bad name": "x"})`)
if err == nil {
t.Fatal("expected error for invalid column name in before")
}
if !strings.Contains(err.Error(), "invalid column name") {
t.Errorf("unexpected error: %v", err)
}
}
// ── namespace isolation ──────────────────────
func TestDBQueryRejectsOtherExtensionTable(t *testing.T) {

View File

@@ -145,6 +145,14 @@ func (s *Sandbox) ExecWithLoader(ctx context.Context, filename, source string, m
}
}()
// Unicode security gate — scan source for invisible/deceptive characters
// before execution (defense in depth; install gate is primary).
if findings := ScanSource(source, filename); len(findings) > 0 {
if blocked, reason := Verdict(findings); blocked {
return nil, fmt.Errorf("unicode security gate: %s (file: %s)", reason, filename)
}
}
globals, err := starlark.ExecFile(thread, filename, source, predeclared)
if err != nil {
return nil, wrapError(err)

View File

@@ -0,0 +1,154 @@
// Package sandbox — unicode_scan.go
//
// v0.5.1: Invisible Unicode scanning gate.
//
// Defends against GlassWorm (variation selector payloads), Trojan Source
// (bidi override attacks / CVE-2021-42574), and other invisible Unicode
// smuggling techniques.
//
// Two public functions:
// ScanSource(source, filename) → []Finding
// Verdict(findings) → (blocked bool, reason string)
//
// No dependencies beyond stdlib (unicode, fmt, strings).
// Single-pass rune iteration. No regex.
package sandbox
import (
"fmt"
"strings"
"unicode"
)
// invisibleRanges defines the Unicode ranges we scan for, grouped by category.
// Uses unicode.RangeTable with R16/R32 slices for efficient lookup.
var invisibleRanges = map[string]*unicode.RangeTable{
"zero_width": {
R16: []unicode.Range16{
{Lo: 0x200B, Hi: 0x200F, Stride: 1}, // zero-width space, joiners, bidi marks
{Lo: 0x2060, Hi: 0x2064, Stride: 1}, // word joiner, invisible operators
{Lo: 0xFEFF, Hi: 0xFEFF, Stride: 1}, // BOM / zero-width no-break space
},
},
"bidi_override": {
R16: []unicode.Range16{
{Lo: 0x202A, Hi: 0x202E, Stride: 1}, // bidi overrides (Trojan Source)
{Lo: 0x2066, Hi: 0x2069, Stride: 1}, // bidi isolates
},
},
"variation_selector": {
R16: []unicode.Range16{
{Lo: 0xFE00, Hi: 0xFE0F, Stride: 1}, // variation selectors (GlassWorm primary)
},
R32: []unicode.Range32{
{Lo: 0xE0100, Hi: 0xE01EF, Stride: 1}, // variation selector supplement (GlassWorm secondary)
},
},
"tag_character": {
R32: []unicode.Range32{
{Lo: 0xE0001, Hi: 0xE007F, Stride: 1}, // tag characters
},
},
"other_invisible": {
R16: []unicode.Range16{
{Lo: 0x3164, Hi: 0x3164, Stride: 1}, // Hangul filler
{Lo: 0xFFF9, Hi: 0xFFFB, Stride: 1}, // interlinear annotations
},
},
}
// Finding records a single invisible/deceptive Unicode character found in source.
type Finding struct {
Offset int // byte offset in source
Line int // 1-based line number
Rune rune // the character found
Category string // one of: variation_selector, bidi_override, zero_width, tag_character, other_invisible
}
// ScanSource scans source code for invisible/deceptive Unicode characters.
// Returns a slice of findings (empty if clean). Single-pass rune iteration.
func ScanSource(source string, filename string) []Finding {
var findings []Finding
line := 1
offset := 0
for _, r := range source {
for cat, rt := range invisibleRanges {
if unicode.Is(rt, r) {
findings = append(findings, Finding{
Offset: offset,
Line: line,
Rune: r,
Category: cat,
})
break
}
}
if r == '\n' {
line++
}
offset += len(string(r))
}
return findings
}
// CategoryCounts returns a map of category → count from a slice of findings.
func CategoryCounts(findings []Finding) map[string]int {
counts := make(map[string]int)
for _, f := range findings {
counts[f.Category]++
}
return counts
}
// Verdict evaluates findings and returns whether the source should be blocked.
//
// Blocking rules:
// - Any bidi_override → block (Trojan Source, zero legitimate use in Starlark)
// - variation_selector + tag_character >= 10 → block (GlassWorm signature)
// - zero_width >= 6 → block
//
// Returns blocked=false, reason="" if clean or below threshold.
func Verdict(findings []Finding) (blocked bool, reason string) {
if len(findings) == 0 {
return false, ""
}
counts := CategoryCounts(findings)
// Rule 1: Any bidi override → always block (Trojan Source / CVE-2021-42574)
if counts["bidi_override"] > 0 {
return true, fmt.Sprintf(
"blocked: %d bidirectional override character(s) detected (Trojan Source / CVE-2021-42574). "+
"These characters can reverse the visual display order of code, hiding malicious logic.",
counts["bidi_override"],
)
}
// Rule 2: GlassWorm signature — variation selectors + tag characters >= 10
glasswormCount := counts["variation_selector"] + counts["tag_character"]
if glasswormCount >= 10 {
return true, fmt.Sprintf(
"blocked: %d variation selector / tag characters detected (GlassWorm pattern). "+
"Legitimate emoji use requires 1-2; %d suggests encoded hidden payload.",
glasswormCount, glasswormCount,
)
}
// Rule 3: Excessive zero-width characters
if counts["zero_width"] >= 6 {
return true, fmt.Sprintf(
"blocked: %d zero-width characters detected. "+
"This exceeds the safe threshold and may indicate hidden content.",
counts["zero_width"],
)
}
// Below threshold — not blocked
var parts []string
for cat, n := range counts {
parts = append(parts, fmt.Sprintf("%d %s", n, cat))
}
return false, strings.Join(parts, ", ")
}

View File

@@ -0,0 +1,354 @@
package sandbox
import (
"context"
"strings"
"testing"
)
// ── ScanSource ──────────────────────────────
func TestScanSource_CleanStarlark(t *testing.T) {
source := `
def on_request(req):
path = req["path"]
method = req["method"]
return {"status": 200, "body": "ok"}
`
findings := ScanSource(source, "clean.star")
if len(findings) != 0 {
t.Errorf("expected 0 findings for clean source, got %d", len(findings))
}
}
func TestScanSource_FewVariationSelectors(t *testing.T) {
// 1-2 variation selectors — legitimate emoji use (e.g. text vs emoji presentation)
source := "emoji = \"\u2764\uFE0F\" # heart with variation selector"
findings := ScanSource(source, "emoji.star")
if len(findings) != 1 {
t.Fatalf("expected 1 finding for single VS, got %d", len(findings))
}
if findings[0].Category != "variation_selector" {
t.Errorf("expected category variation_selector, got %q", findings[0].Category)
}
if findings[0].Rune != '\uFE0F' {
t.Errorf("expected rune U+FE0F, got U+%04X", findings[0].Rune)
}
}
func TestScanSource_GlassWormPattern(t *testing.T) {
// 50+ variation selectors — GlassWorm payload pattern
var sb strings.Builder
sb.WriteString("data = \"")
for i := 0; i < 50; i++ {
sb.WriteRune(rune(0xFE00 + (i % 16))) // cycle through VS1-VS16
}
sb.WriteString("\"")
source := sb.String()
findings := ScanSource(source, "glassworm.star")
count := 0
for _, f := range findings {
if f.Category == "variation_selector" {
count++
}
}
if count != 50 {
t.Errorf("expected 50 variation_selector findings, got %d", count)
}
}
func TestScanSource_BidiOverride(t *testing.T) {
// RLO (U+202E) — Trojan Source attack
source := "name = \"\u202Emalicious\""
findings := ScanSource(source, "bidi.star")
if len(findings) != 1 {
t.Fatalf("expected 1 finding, got %d", len(findings))
}
if findings[0].Category != "bidi_override" {
t.Errorf("expected bidi_override, got %q", findings[0].Category)
}
}
func TestScanSource_BidiIsolate(t *testing.T) {
// LRI (U+2066), PDI (U+2069)
source := "text = \"\u2066hidden\u2069\""
findings := ScanSource(source, "bidi_isolate.star")
bidiCount := 0
for _, f := range findings {
if f.Category == "bidi_override" {
bidiCount++
}
}
if bidiCount != 2 {
t.Errorf("expected 2 bidi_override findings, got %d", bidiCount)
}
}
func TestScanSource_ZeroWidthBelow(t *testing.T) {
// 3 zero-width chars — below threshold
source := "a = \"\u200B\u200B\u200B\""
findings := ScanSource(source, "zw.star")
zwCount := 0
for _, f := range findings {
if f.Category == "zero_width" {
zwCount++
}
}
if zwCount != 3 {
t.Errorf("expected 3 zero_width findings, got %d", zwCount)
}
}
func TestScanSource_ZeroWidthAbove(t *testing.T) {
// 8 zero-width chars — above threshold
source := "a = \"\u200B\u200B\u200B\u200B\u200B\u200B\u200B\u200B\""
findings := ScanSource(source, "zw_many.star")
zwCount := 0
for _, f := range findings {
if f.Category == "zero_width" {
zwCount++
}
}
if zwCount != 8 {
t.Errorf("expected 8 zero_width findings, got %d", zwCount)
}
}
func TestScanSource_TagCharacters(t *testing.T) {
// Tag characters (U+E0001-U+E007F)
var sb strings.Builder
sb.WriteString("tag = \"")
for i := 0; i < 15; i++ {
sb.WriteRune(rune(0xE0001 + i))
}
sb.WriteString("\"")
source := sb.String()
findings := ScanSource(source, "tags.star")
tagCount := 0
for _, f := range findings {
if f.Category == "tag_character" {
tagCount++
}
}
if tagCount != 15 {
t.Errorf("expected 15 tag_character findings, got %d", tagCount)
}
}
func TestScanSource_HangulFiller(t *testing.T) {
source := "x = \"\u3164\"" // Hangul filler
findings := ScanSource(source, "hangul.star")
if len(findings) != 1 {
t.Fatalf("expected 1 finding, got %d", len(findings))
}
if findings[0].Category != "other_invisible" {
t.Errorf("expected other_invisible, got %q", findings[0].Category)
}
}
func TestScanSource_LineTracking(t *testing.T) {
source := "line1\nline2\nline3 = \"\u202E\""
findings := ScanSource(source, "lines.star")
if len(findings) != 1 {
t.Fatalf("expected 1 finding, got %d", len(findings))
}
if findings[0].Line != 3 {
t.Errorf("expected line 3, got %d", findings[0].Line)
}
}
func TestScanSource_VariationSelectorSupplement(t *testing.T) {
// U+E0100 — GlassWorm secondary range (R32)
var sb strings.Builder
sb.WriteString("x = \"")
for i := 0; i < 12; i++ {
sb.WriteRune(rune(0xE0100 + i))
}
sb.WriteString("\"")
source := sb.String()
findings := ScanSource(source, "vs_supp.star")
vsCount := 0
for _, f := range findings {
if f.Category == "variation_selector" {
vsCount++
}
}
if vsCount != 12 {
t.Errorf("expected 12 variation_selector findings from supplement range, got %d", vsCount)
}
}
// ── Verdict ─────────────────────────────────
func TestVerdict_NoFindings(t *testing.T) {
blocked, reason := Verdict(nil)
if blocked {
t.Error("expected not blocked for nil findings")
}
if reason != "" {
t.Errorf("expected empty reason, got %q", reason)
}
}
func TestVerdict_BidiAlwaysBlocks(t *testing.T) {
findings := []Finding{{Rune: 0x202E, Category: "bidi_override"}}
blocked, reason := Verdict(findings)
if !blocked {
t.Error("expected blocked for bidi_override")
}
if !strings.Contains(reason, "bidirectional override") {
t.Errorf("expected bidi reason, got %q", reason)
}
}
func TestVerdict_GlassWormBlocks(t *testing.T) {
var findings []Finding
for i := 0; i < 10; i++ {
findings = append(findings, Finding{Rune: rune(0xFE00 + i), Category: "variation_selector"})
}
blocked, reason := Verdict(findings)
if !blocked {
t.Error("expected blocked for 10 variation selectors")
}
if !strings.Contains(reason, "GlassWorm") {
t.Errorf("expected GlassWorm reason, got %q", reason)
}
}
func TestVerdict_GlassWormMixedVSAndTag(t *testing.T) {
var findings []Finding
for i := 0; i < 5; i++ {
findings = append(findings, Finding{Rune: rune(0xFE00 + i), Category: "variation_selector"})
}
for i := 0; i < 5; i++ {
findings = append(findings, Finding{Rune: rune(0xE0001 + i), Category: "tag_character"})
}
blocked, _ := Verdict(findings)
if !blocked {
t.Error("expected blocked for 5 VS + 5 tag = 10 total")
}
}
func TestVerdict_FewVariationSelectorsPass(t *testing.T) {
findings := []Finding{
{Rune: 0xFE0F, Category: "variation_selector"},
{Rune: 0xFE0E, Category: "variation_selector"},
}
blocked, _ := Verdict(findings)
if blocked {
t.Error("expected not blocked for 2 variation selectors (emoji use)")
}
}
func TestVerdict_ZeroWidthBelowThresholdPasses(t *testing.T) {
var findings []Finding
for i := 0; i < 5; i++ {
findings = append(findings, Finding{Rune: 0x200B, Category: "zero_width"})
}
blocked, _ := Verdict(findings)
if blocked {
t.Error("expected not blocked for 5 zero_width chars")
}
}
func TestVerdict_ZeroWidthAboveThresholdBlocks(t *testing.T) {
var findings []Finding
for i := 0; i < 6; i++ {
findings = append(findings, Finding{Rune: 0x200B, Category: "zero_width"})
}
blocked, reason := Verdict(findings)
if !blocked {
t.Error("expected blocked for 6 zero_width chars")
}
if !strings.Contains(reason, "zero-width") {
t.Errorf("expected zero-width reason, got %q", reason)
}
}
func TestVerdict_OtherInvisibleDoesNotBlock(t *testing.T) {
findings := []Finding{
{Rune: 0x3164, Category: "other_invisible"},
{Rune: 0xFFF9, Category: "other_invisible"},
}
blocked, _ := Verdict(findings)
if blocked {
t.Error("expected not blocked for 2 other_invisible chars")
}
}
// ── GlassWorm payload simulation ────────────
func TestGlassWormPayloadDetection(t *testing.T) {
// Simulate a GlassWorm-style payload: encode each byte of a secret string
// as a variation selector offset (U+FE00 + low nibble, U+E0100 + high nibble).
// This is the core technique: invisible characters carrying data.
secret := "eval(malicious_code)"
var sb strings.Builder
sb.WriteString("innocent_looking_var = \"")
for _, b := range []byte(secret) {
// Encode low nibble as VS1-VS16 (U+FE00-U+FE0F)
sb.WriteRune(rune(0xFE00 + (int(b) & 0x0F)))
// Encode high nibble as VS supplement (U+E0100-U+E010F)
sb.WriteRune(rune(0xE0100 + ((int(b) >> 4) & 0x0F)))
}
sb.WriteString("\"")
source := sb.String()
findings := ScanSource(source, "glassworm_payload.star")
// Should have 2 findings per character (low + high nibble)
expectedFindings := len(secret) * 2
vsCount := 0
for _, f := range findings {
if f.Category == "variation_selector" {
vsCount++
}
}
if vsCount != expectedFindings {
t.Errorf("expected %d variation_selector findings, got %d", expectedFindings, vsCount)
}
// Verdict should block
blocked, reason := Verdict(findings)
if !blocked {
t.Errorf("GlassWorm payload should be blocked, got: %s", reason)
}
if !strings.Contains(reason, "GlassWorm") {
t.Errorf("expected GlassWorm in reason, got: %s", reason)
}
}
// ── Execution gate integration ──────────────
func TestExecGateBlocksBidiSource(t *testing.T) {
sb := New(DefaultConfig())
// Source with a bidi override character
source := "x = \"\u202E\""
_, err := sb.Exec(context.Background(), "bidi.star", source, nil)
if err == nil {
t.Fatal("expected error from unicode gate for bidi override")
}
if !strings.Contains(err.Error(), "unicode security gate") {
t.Errorf("expected 'unicode security gate' error, got: %v", err)
}
}
func TestExecGateAllowsCleanSource(t *testing.T) {
sb := New(DefaultConfig())
_, err := sb.Exec(context.Background(), "clean.star", "x = 42", nil)
if err != nil {
t.Fatalf("clean source should not be blocked: %v", err)
}
}
func TestExecGateAllowsFewVS(t *testing.T) {
sb := New(DefaultConfig())
source := "emoji = \"\u2764\uFE0F\""
_, err := sb.Exec(context.Background(), "emoji.star", source, nil)
if err != nil {
t.Fatalf("2 VS should not be blocked: %v", err)
}
}

View File

@@ -167,7 +167,7 @@ export async function boot() {
};
// Marker for idempotency
sw._sdk = '0.5.0';
sw._sdk = '0.5.1';
// 8. Expose globally
window.sw = sw;

View File

@@ -6,8 +6,9 @@
*/
const { html } = window;
const { useState, useEffect, useCallback } = hooks;
import { Dropdown } from '../../primitives/dropdown.js';
const TYPE_OPTIONS = ['all', 'surface', 'extension', 'full', 'workflow'];
const TYPE_OPTIONS = ['all', 'surface', 'extension', 'full', 'workflow', 'library'];
const CORE_IDS = new Set(['admin']);
function typeBadge(type) {
@@ -224,12 +225,11 @@ export default function PackagesSection() {
${/* ── Action bar ─────────────── */``}
<div style="display:flex;gap:6px;align-items:center;flex-wrap:wrap;margin-bottom:14px;">
${TYPE_OPTIONS.map(t => html`
<button key=${t}
class="btn-small ${typeFilter === t ? 'btn-primary' : ''}"
onClick=${() => setTypeFilter(t)}
style="text-transform:capitalize;">${t}</button>
`)}
<${Dropdown}
value=${typeFilter}
onChange=${setTypeFilter}
options=${TYPE_OPTIONS.map(t => ({ label: t.charAt(0).toUpperCase() + t.slice(1), value: t }))}
/>
<span style="flex:1;" />
<button class="btn-small" onClick=${uploadPackage} disabled=${installing}>
${installing ? 'Installing\u2026' : 'Install Package'}