MEGAINFRA-WEBAPP/scripts/pb-add-ask-collections.py
DEls b211ce1c8b feat(ask): PocketBase schema for threads, messages and log
Additive, dry-run-by-default script creating ask_threads, ask_messages
and ask_log (change/memory/request), following the
pb-add-dn-staged-ingestion.py pattern. --dry-run confirms all three
collections are new; --apply is a human step.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-26 17:51:26 +02:00

256 lines
8.6 KiB
Python

#!/usr/bin/env python3
"""Add the three PocketBase collections behind the Ask agent.
``ask_threads`` (one row per conversation), ``ask_messages`` (append-only:
user/assistant/attachment/image rows) and ``ask_log`` (changes for Undo,
long-term memories, and developer change requests — one collection,
distinguished by ``type``).
Dry run unless ``--apply --confirm-host <exact configured PB host>`` is
given. Additive and idempotent, following the same pattern as
scripts/pb-add-dn-staged-ingestion.py: read the live field/index list before
appending, never delete or replace unrelated schema.
All rules are left null (superuser only) — the Next.js server is the only
client of these three collections; there is no browser-direct access.
"""
import sys
from pb_env import PB, http, superuser_token
DRY_RUN = "--dry-run" in sys.argv or "--apply" not in sys.argv
APPLY = "--apply" in sys.argv
def confirmation_host():
for index, argument in enumerate(sys.argv):
if argument.startswith("--confirm-host="):
return argument.split("=", 1)[1]
if argument == "--confirm-host" and index + 1 < len(sys.argv):
return sys.argv[index + 1]
return None
def get_collection(token, name):
code, response = http("GET", f"/api/collections/{name}", token=token)
return response if code == 200 else None
def require_collection(token, name):
collection = get_collection(token, name)
if collection is None:
sys.exit(f"collection {name} not found")
return collection
def relation(name, collection_id, required=False, cascade_delete=False, max_select=1):
return {
"name": name,
"type": "relation",
"required": required,
"collectionId": collection_id,
"maxSelect": max_select,
"cascadeDelete": cascade_delete,
}
def field(name, field_type="text", **options):
return {"name": name, "type": field_type, "required": False, **options}
def select(name, values, required=False, max_select=1):
return {
"name": name,
"type": "select",
"required": required,
"maxSelect": max_select,
"values": values,
}
def system_fields():
return [
field("created", "date", onCreate=True),
field("updated", "date", onCreate=True, onUpdate=True),
]
def add_fields(token, name, wanted):
collection = require_collection(token, name)
existing = list(collection.get("fields", []))
names = {item.get("name") for item in existing}
additions = [item for item in wanted if item["name"] not in names]
if not additions:
print(f" {name}: fields already present")
return
print(f" {name}: add fields {[item['name'] for item in additions]}")
if not DRY_RUN:
code, response = http(
"PATCH", f"/api/collections/{collection['id']}",
{"fields": existing + additions}, token,
)
if code != 200:
sys.exit(f"PATCH {name} failed: {code} {response}")
def add_indexes(token, name, wanted):
collection = require_collection(token, name)
existing = list(collection.get("indexes", []) or [])
# Simple union by exact string match — these are new indexes on new
# collections, so there is nothing pre-existing to reconcile against.
additions = [i for i in wanted if i not in existing]
if not additions:
print(f" {name}: indexes already present")
return
print(f" {name}: add indexes {additions}")
if not DRY_RUN:
code, response = http(
"PATCH", f"/api/collections/{collection['id']}",
{"indexes": existing + additions}, token,
)
if code != 200:
sys.exit(f"PATCH {name} indexes failed: {code} {response}")
def create_collection(token, name, fields, indexes):
existing = get_collection(token, name)
if existing is not None:
print(f" {name}: already exists")
add_fields(token, name, fields)
add_indexes(token, name, indexes)
return existing["id"]
print(f" {name}: create collection")
if DRY_RUN:
return name
code, response = http(
"POST", "/api/collections",
{"name": name, "type": "base", "fields": fields, "indexes": indexes},
token,
)
if code != 200:
sys.exit(f"CREATE {name} failed: {code} {response}")
return response["id"]
def self_test():
r = relation("owner", "users123", required=True)
assert r == {
"name": "owner", "type": "relation", "required": True,
"collectionId": "users123", "maxSelect": 1, "cascadeDelete": False,
}
f = field("text", "text", max=100000)
assert f == {"name": "text", "type": "text", "required": False, "max": 100000}
s = select("kind", ["user", "assistant", "attachment", "image"])
assert s["values"] == ["user", "assistant", "attachment", "image"]
print("self-test: field builders passed")
return 0
def main():
if "--self-test" in sys.argv:
return self_test()
if APPLY and confirmation_host() != PB:
sys.exit(
"REFUSING --apply: pass --confirm-host with the exact configured "
f"PocketBase host ({PB})"
)
try:
token = superuser_token()
except SystemExit as error:
if not DRY_RUN:
raise
print(f"offline preview: {error}")
print("would create: ask_threads, ask_messages, ask_log")
print("\nDRY RUN. No schema changes written.")
return 0
print(f"target: {PB} mode: {'DRY RUN' if DRY_RUN else 'APPLY'}")
users = require_collection(token, "users")
sites = require_collection(token, "sites")
users_id = users["id"]
sites_id = sites["id"]
threads_id = create_collection(
token, "ask_threads",
system_fields() + [
field("title"),
relation("owner", users_id, required=True),
field("archived", "bool"),
field("last_message_at", "date"),
],
[
"CREATE INDEX `idx_ask_threads_owner_archived_last_message` "
"ON `ask_threads` (`owner`, `archived`, `last_message_at`)",
],
)
create_collection(
token, "ask_messages",
system_fields() + [
relation("thread", threads_id, required=True, cascade_delete=True),
relation("owner", users_id),
select("kind", ["user", "assistant", "attachment", "image"], required=True),
field("text", max=100000),
field("display"),
field("charts", "json", maxSize=2000000),
field("action_ids", "json", maxSize=200000),
field("attachment_ids", "json", maxSize=200000),
field("tools", "json", maxSize=200000),
field(
"file", "file", maxSelect=1, maxSize=26214400,
mimeTypes=[
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
"application/vnd.ms-excel",
"text/csv",
"application/pdf",
"image/png",
"image/jpeg",
"image/webp",
"text/plain",
],
),
field("parsed", "json", maxSize=5000000),
field("parse_error"),
],
[
"CREATE INDEX `idx_ask_messages_thread_created` "
"ON `ask_messages` (`thread`, `created`)",
],
)
create_collection(
token, "ask_log",
system_fields() + [
select("type", ["change", "memory", "request"], required=True),
relation("owner", users_id),
field("shared", "bool"),
relation("thread", threads_id, required=False, cascade_delete=False),
select(
"status",
["pending", "applied", "rejected", "failed", "undone",
"open", "planned", "done", "declined"],
),
field("text", max=2000),
field("detail"),
relation("site", sites_id, required=False),
field("ops", "json", maxSize=2000000),
field("before", "json", maxSize=2000000),
field("after", "json", maxSize=2000000),
field("error"),
field("applied_at", "date"),
field("undone_at", "date"),
],
[
"CREATE INDEX `idx_ask_log_type_status` ON `ask_log` (`type`, `status`)",
"CREATE INDEX `idx_ask_log_owner_type` ON `ask_log` (`owner`, `type`)",
"CREATE INDEX `idx_ask_log_thread` ON `ask_log` (`thread`)",
],
)
print("\nDRY RUN. No schema changes written." if DRY_RUN else "\nSchema applied.")
return 0
if __name__ == "__main__":
sys.exit(main())