Finalize research API validation and handoff notes [skip ci]

Reviewed-on: #16

.env.example

@@ -16,3 +16,10 @@ SMTP_FROM=noreply@notice.entroia.com
 # Optional Sign in with Slack.
 SLACK_CLIENT_ID=replace-me
 SLACK_CLIENT_SECRET=replace-me
+
+# Optional shared paper corpus; enable only after PostgreSQL publication.
+# Store the real read-only connection string in the protected server research.env.
+FOUNDRY_RESEARCH_DATABASE_URL=
+FOUNDRY_RESEARCH_SCHEMA=research
+# Deployment migration container; independent of database/schema names.
+FOUNDRY_POSTGRES_CONTAINER=foundry-postgres

.gitea/workflows/check.yml

@@ -74,6 +74,7 @@ jobs:
         env:
           PUBLIC_ORIGIN: https://api.example.com
           DATABASE_URL: postgres://foundry:fixture@127.0.0.1:5432/foundry
+          FOUNDRY_RESEARCH_DATABASE_URL: ''
           GOOGLE_CLIENT_ID: fixture
           GOOGLE_CLIENT_SECRET: fixture
           FOUNDRY_PROJECTS: '{"notes":"http://localhost:3000/auth/callback"}'

README.md

@@ -10,6 +10,10 @@
 4. 由 Gitea Actions 在 `ces-control` 上使用 Go 1.25 构建并检查。部署工作流仅处理 `main` 上影响后端或部署的改动，将二进制启动在服务器 Docker 私网，由 Traefik 提供 HTTPS。通过 `GET /healthz` 检查进程，通过 `GET /readyz` 检查认证库。
 5. 代理避免记录 Google 回调的查询参数及认证请求头。设置入口限流，并定时删除三张临时表（`oauth_attempts`、`login_grants`、`project_sessions`）中 `expires_at <= now()` 的记录。
 
+## 共享论文库（Thesisly）
+
+可选只读接入见[共享论文 API](docs/research-api.md)。Thesisly 使用同一后端的 `/projects/{project}/research/papers` 列表、搜索和详情；只有配置服务器上的 `FOUNDRY_RESEARCH_DATABASE_URL` 时才启用。连接串保存在受保护的 `research.env`，不放入前端或 Actions Variables。当前仍等待 Paper Atlas 全量 PostgreSQL 迁移完成及真实数据验收，尚未启用线上论文接口。
+
 ## Actions 部署配置
 
 在仓库的 Actions Secrets 配置 `GOOGLE_CLIENT_ID` 和 `GOOGLE_CLIENT_SECRET`，在 Actions Variables 配置 `FOUNDRY_PROJECTS`。变量值为 JSON 对象，例如 `{"prompt":"http://localhost:17321/auth/callback"}`。部署读取服务器上预先配置的 `/opt/endoleap-infra/foundry/database.env`，OAuth 值由 Actions 在启动容器时注入。
@@ -212,6 +216,7 @@ Collapsing/scaling presentation does not rewrite child content. Projection is
 
 | Route | Required input / behavior |
 | --- | --- |
+| POST B/connections | `{source,target,edge,expected_versions}` + Idempotency-Key; atomically creates/reuses boundary port chains and stores one edge in the lowest common ancestor |
 | POST B/subgraphs | Parent If-Match; `{frame:{id,title,position,width,height,labelMode}}`; returns 201, child_board, parent/child snapshots and receipt |
 | POST B/subgraphs/{frame}/members | `{source_board_id,target_board_id,node_ids,expected_versions}`; parent/child or siblings coordinated by their common parent |
 | POST B/subgraphs/{frame}/dissolve | `{expected_versions}`; preserves content in the parent and archives the emptied child |
@@ -405,6 +410,109 @@ summary are retained with mode 0600 under
 on the ECS host, in addition to the database receipt. The frontend V2 integration
 remains a separate task; use a V2-aware client for this migrated test canvas.
 
+#### Creating new cross-boundary connections
+
+Ordinary Canvas PATCH preserves existing frame ports and cannot add or retarget
+ports. Migration/movement-generated ports are reusable, but creating a new
+external connection to an unexposed child node requires this dedicated command:
+
+`POST B/connections` with user session authorization,
+`Accept: application/vnd.foundry.canvas+json;version=2`, JSON content type and a
+UUID `Idempotency-Key`. Availability is advertised as
+`capabilities.subgraphs.connections: true`; clients must check the deployed
+capability before calling it.
+
+`B` identifies the **lowest common ancestor board** of the two actual endpoints.
+The coordinator may itself be a nested board. Both endpoints must be in this
+same project/workspace and inside its active subgraph tree. A direct endpoint on
+B is valid, as are siblings and descendants up to the existing depth limit.
+When both endpoints are in the same board, use that board as B.
+
+```json
+{
+  "source": {"board_id":"<child-board-uuid>","node_id":"child-text","handle":"right"},
+  "target": {"board_id":"<parent-board-uuid>","node_id":"parent-image","handle":"left"},
+  "edge": {
+    "id":"new-edge",
+    "data": {
+      "appearance":{"color":"#AA968B","width":2},
+      "animation":{"mode":"executing","effect":"flow","direction":"forward","durationMs":1600},
+      "execution":{"enabled":true}
+    }
+  },
+  "expected_versions": {"<parent-board-uuid>":210,"<child-board-uuid>":1}
+}
+```
+
+Use canonical board UUIDs for endpoints and version keys. Eligible real node
+kinds are text/script/plain-text/image/plain-image; handles must be one of
+`top/right/bottom/left`. Do not submit projected node IDs, frame IDs as generated
+content, or `port:*` handles. The server exposes a real endpoint through each
+containing frame, creating or reusing a port with the matching
+`direction/nodeId/handle` tuple. Existing matching ports retain their IDs,
+including migration-created or custom stable IDs. A source chain uses output
+ports, and a target chain uses input ports; these are distinct. No unbound port
+or writable arbitrary child reference is accepted.
+
+The expected_versions map must contain **exactly every board on both paths**,
+including the endpoint boards and B. For a grandchild-to-parent connection it
+contains parent, intermediate child and grandchild; for siblings it contains
+both children and their parent. No separate If-Match header is required for this
+multi-board operation. Missing participants return 428 `version_required`;
+when a nonempty map has missing/extra participant keys, the response includes
+`details.required_board_ids`. Any stale version returns 412 `version_conflict`.
+All participants require current editor/admin access. Calling a higher ancestor
+instead of the lowest common ancestor returns 409 `connection_owner_mismatch`.
+An unrelated/invalid ownership path returns 409 `ownership_conflict`; resources
+outside the caller's scope retain the normal 403/404 behavior.
+
+Only `edge.id` and optional typed V2 `edge.data` are accepted. Its appearance,
+animation, execution flag and bezier-chain path follow the normal V2 validation.
+All knots are already in B's local coordinates; port creation does not move node
+geometry or re-layout the edge. An existing edge ID in B returns 409
+`edge_id_conflict`. Graph, port and snapshot capacity failures return 413 without
+saving any ports, edges, versions or receipt.
+
+Success is 201 with the coordinator's ETag and:
+
+```json
+{
+  "transaction_id":"<uuid>",
+  "parent":{"board_id":"<coordinator>","version":211,"document":{}},
+  "snapshots":[
+    {"board_id":"<coordinator>","version":211,"document":{}},
+    {"board_id":"<child-board-uuid>","version":2,"document":{}}
+  ],
+  "connection":{
+    "board_id":"<coordinator>",
+    "edge":{"id":"new-edge","source":"<frame-id>","sourceHandle":"port:<stable-id>","target":"parent-image","targetHandle":"left","data":{}},
+    "source":{"board_id":"<child-board-uuid>","node_id":"child-text","handle":"right"},
+    "target":{"board_id":"<parent-board-uuid>","node_id":"parent-image","handle":"left"}
+  },
+  "created_ports":[{"board_id":"<coordinator>","frame_id":"<frame-id>","port_id":"<stable-id>"}],
+  "undo_receipt":{"transaction_id":"<uuid>","valid_for_seconds":604800,"expires_at":"<RFC3339 timestamp>"}
+}
+```
+
+The JSON above abbreviates document values and shows a one-level example; nested
+connections include every intermediate board in snapshots. The response edge is
+the stored owner-board edge, while source/target preserve the logical endpoint
+identities. created_ports contains only newly allocated ports; reused ports are
+available in the snapshots. Each participating board advances exactly one
+version (including endpoint boards whose node content remains unchanged),
+consistent with other cross-board commands. Flush pending saves before creating
+the connection, then absorb **all** returned versions/snapshots before resuming
+autosave. Retrying an identical body with the same key returns the original 201
+response and does not advance versions again. Changing the body, including its
+version map, with that key returns 409 `idempotency_conflict`.
+
+Undo uses the existing `POST B/transactions/{transaction_id}/undo` endpoint with
+a new key and all current participant versions. It restores the previous edge
+and port arrays atomically; subsequent incompatible edits cause 412. Existing
+ports reused by the command remain after undo. Redo submits the command with
+fresh versions and a new key. Ordinary PATCH can continue to edit/remove edges
+using already exposed ports, but cannot retarget or directly delete the ports.
+
 #### Reproducible session API examples
 
 These examples require Bash, curl and jq, an editor/admin login session, and an
@@ -500,3 +608,31 @@ api "$B/ai/runs/$(jq -r '.run_id' <<<"$propagated_run")?after_seq=0&limit=100"
 The live catalog, rather than the isolated verification IDs above, determines
 whether each request can run. A 202 response acknowledges a queued job; use its
 terminal status and authenticated canvas/asset rereads as completion evidence.
+
+
+### Managed image previews
+
+Original asset bytes remain immutable and are used by the ordinary asset URL,
+downloads and AI inputs. Authenticated `GET .../assets/{id}?variant=preview`
+returns one display rendition bounded to 1024 pixels on its longest edge. Opaque previews use JPEG quality 82, transparency stays PNG, GIF preserves
+the original animation, and JPEG EXIF orientation is applied. Small originals
+may be returned unchanged when this is more efficient. No preview URL belongs
+in node `data.src`: persist the original managed URL and optional original
+`assetId`, then select the display variant only in the renderer.
+
+Permissions are checked on every read, including cache hits. Originals are not
+replaced. Generation is serialized and the immutable encoded preview cache is
+bounded to 32 MiB / 128 entries. Current capabilities advertise `assets.previews`
+and `assets.preview_max_edge`. No database migration is needed. Asset reads and
+Canvas GET release their permission transaction before encoding/writing the
+response. Uploads release the initial workspace lock before receiving a large
+multipart body and recheck current write permission before committing.
+
+Legacy inline images must be migrated through an explicit editor action: upload
+unchanged bytes with idempotency keys, then save the same node IDs/geometry with
+managed asset references and the current canvas version. Opening a canvas does
+not migrate it or cause an implicit PATCH.
+
+#### Atomic member drop positions
+
+`POST B/subgraphs/{frame}/members` additionally accepts optional `positions`, an object with exactly one `{x,y}` entry for each `node_ids` item, expressed in the **source board's local coordinates**. Both coordinates are required, finite and within ±1,000,000. Capabilities advertise `subgraphs.member_positions: true`. Omission preserves legacy behavior. Positions, ownership, edge/port conversion, versions and undo receipt commit in one transaction; no preceding coordinate PATCH is needed. Validation failures or stale versions perform zero writes. The positions participate in the idempotency payload digest; retries must keep the entire body and key unchanged. Undo restores the pre-drop snapshot, including original positions.

cmd/foundry/main.go

@@ -13,6 +13,7 @@ import (
 	"gitea.entroia.com/Zayden-Jung/foundry-core/internal/auth"
 	"gitea.entroia.com/Zayden-Jung/foundry-core/internal/codex"
 	"gitea.entroia.com/Zayden-Jung/foundry-core/internal/platform"
+	"gitea.entroia.com/Zayden-Jung/foundry-core/internal/research"
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/stdlib"
 )
@@ -61,6 +62,33 @@ func main() {
 		log.Fatal(err)
 	}
 	mux := http.NewServeMux()
+	var papers research.Repository
+	if researchDSN := os.Getenv("FOUNDRY_RESEARCH_DATABASE_URL"); researchDSN != "" {
+		researchConfig, err := pgx.ParseConfig(researchDSN)
+		if err != nil {
+			log.Fatal("Invalid research database configuration")
+		}
+		researchConfig.RuntimeParams["default_transaction_read_only"] = "on"
+		researchConfig.RuntimeParams["application_name"] = "foundry-research"
+		researchDB := stdlib.OpenDB(*researchConfig)
+		defer researchDB.Close()
+		researchDB.SetMaxOpenConns(2)
+		researchDB.SetMaxIdleConns(2)
+		researchDB.SetConnMaxIdleTime(5 * time.Minute)
+		researchDB.SetConnMaxLifetime(30 * time.Minute)
+		repository, err := research.NewPostgresRepository(researchDB, env("FOUNDRY_RESEARCH_SCHEMA", "research"))
+		if err != nil {
+			log.Fatal("Invalid research database configuration")
+		}
+		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		err = repository.Check(ctx)
+		cancel()
+		if err != nil {
+			log.Fatal("Research database is unavailable or not published")
+		}
+		papers = repository
+	}
+	research.Register(mux, papers, callbacks)
 	if err := codex.Register(mux, callbacks, os.Getenv("FOUNDRY_CODEX_SOCKET"), os.Getenv("FOUNDRY_CODEX_GRANTS")); err != nil {
 		log.Fatal(err)
 	}

deploy/check-schema.sh

@@ -67,12 +67,14 @@ END $$;
 ROLLBACK;
 SQL
 printf 'Migration preserved existing logins and enforced project isolation.\n'
-for package in platform auth; do
+psql < internal/research/testdata/corpus.sql
+for package in platform auth research; do
   CGO_ENABLED=0 go test -mod=readonly -c -o "bin/$package-tests" "./internal/$package"
   docker run --rm --network "container:$container" --user 1000:1000 \
     --read-only --cap-drop ALL --security-opt no-new-privileges \
     --tmpfs /tmp:rw,nosuid,nodev,noexec,size=16m,mode=1777 \
     --mount "type=bind,source=$(pwd)/bin/$package-tests,target=/opt/integration-tests,readonly" \
     -e PLATFORM_TEST_DATABASE_URL=postgres://foundry_fixture@127.0.0.1:5432/postgres?sslmode=disable \
+    -e RESEARCH_TEST_DATABASE_URL=postgres://foundry_research_fixture@127.0.0.1:5432/postgres?sslmode=disable \
     --entrypoint /opt/integration-tests postgres:17-alpine -test.v
 done

deploy/migrate.py

@@ -2,12 +2,13 @@
 """Apply checked migrations atomically from the trusted deployment runner."""
 import argparse
 import hashlib
+import os
 from pathlib import Path
 import re
 import subprocess
 
 parser = argparse.ArgumentParser()
-parser.add_argument("--container", default="endoleap-dev-postgres-1")
+parser.add_argument("--container", default=os.environ.get("FOUNDRY_POSTGRES_CONTAINER", "foundry-postgres"))
 parser.add_argument("--database", default="foundry_auth")
 parser.add_argument("--role", default="foundry")
 args = parser.parse_args()

deploy/release.sh

@@ -53,6 +53,10 @@ mail_env=()
 if [[ -s "$foundry/mail.env" ]]; then
   mail_env=(--env-file "$foundry/mail.env")
 fi
+research_env=()
+if [[ -s "$foundry/research.env" ]]; then
+  research_env=(--env-file "$foundry/research.env")
+fi
 slack_env=()
 if [[ -n "${SLACK_CLIENT_ID:-}" || -n "${SLACK_CLIENT_SECRET:-}" ]]; then
   for name in SLACK_CLIENT_ID SLACK_CLIENT_SECRET; do
@@ -118,7 +122,7 @@ if [[ -d "$codex_dir" ]]; then
   )
 fi
 
-docker run -d --name "$candidate" "${codex_mounts[@]}" "${mail_env[@]}" "${slack_env[@]}" --network "$network" \
+docker run -d --name "$candidate" "${codex_mounts[@]}" "${mail_env[@]}" "${slack_env[@]}" "${research_env[@]}" --network "$network" \
   --env-file "$foundry/database.env" -e HOST -e PORT -e PUBLIC_ORIGIN \
   -e GOOGLE_CLIENT_ID -e GOOGLE_CLIENT_SECRET -e FOUNDRY_PROJECTS -e FOUNDRY_ONBOARDING \
   --user 1000:1000 --cap-drop ALL --security-opt no-new-privileges \

deploy/research/query-indexes.sql

@@ -0,0 +1,14 @@
+-- Manual operator step against the published Paper Atlas schema, outside a
+-- transaction. Core's SELECT-only runtime role must never execute this file.
+-- These indexes support the actual author search and exact category filter;
+-- the migration already owns the title/abstract English GIN index.
+SET lock_timeout = '2s';
+SET statement_timeout = '20min';
+SET maintenance_work_mem = '32MB';
+SET max_parallel_maintenance_workers = 0;
+CREATE INDEX CONCURRENTLY IF NOT EXISTS paper_tags_tag_paper
+ ON research.paper_tags(tag_id,paper_id);
+CREATE INDEX CONCURRENTLY IF NOT EXISTS papers_authors_search_gin
+ ON research.papers USING gin(to_tsvector('simple'::regconfig,authors_json));
+-- After an interrupted concurrent build, inspect pg_index.indisvalid before
+-- reuse. IF NOT EXISTS alone does not repair an invalid index.

docs/research-api.md

@@ -0,0 +1,145 @@
+# Shared research API
+
+This work belongs to Thesisly 研迹 and uses Foundry Core rather than a separate
+application backend. The corpus is public; account sessions and future private
+collections remain project scoped.
+
+## Current implementation boundary
+
+`internal/research` implements the HTTP contract and a PostgreSQL repository
+based on the fixed Paper Atlas `a3b677b` schema (TEXT/BIGINT columns, native
+English title/abstract search index). `cmd/foundry` enables it only when
+`FOUNDRY_RESEARCH_DATABASE_URL` is configured. The Paper Atlas migration is now published: 21 tables passed its independent
+count/content-hash acceptance. A dedicated `foundry_research_reader` role has
+SELECT on the ten tables consumed here, schema USAGE, and read-only defaults.
+Its connection is stored only in the protected server file below.
+The existing SQLite implementation was withdrawn. Real-corpus integration has
+been checked through a loopback Core and the local frontend. Formal production
+release still requires the user to merge and choose to deploy; this branch is
+not deployed by these checks. Very broad queries remain subject to the budget
+and performance limitation documented below.
+Do not add `FOUNDRY_RESEARCH_DATABASE`.
+
+The server-side deployment may load a protected
+`/opt/endoleap-infra/foundry/research.env` containing
+`FOUNDRY_RESEARCH_DATABASE_URL` and `FOUNDRY_RESEARCH_SCHEMA` (default `research`).
+Keep the connection string out of Git, Actions variables, browser configuration
+and command arguments. Use a dedicated role with schema USAGE and table SELECT
+only. Core uses a separate two-connection pool, read-only connection defaults and
+read-only transactions. Each research transaction disables JIT locally: the
+measured count plan spent about 2.85 seconds compiling JIT functions, and its
+end-to-end execution exceeded eight seconds. This does not alter shared
+PostgreSQL settings or replace exact counts with estimates. Startup checks the published schema/SELECT permissions;
+an explicitly enabled but invalid connection fails candidate readiness.
+
+Deployment migration defaults now follow the shared `foundry-postgres`
+container name. `FOUNDRY_POSTGRES_CONTAINER` or the existing `--container` option
+can select a different container without changing database/table names.
+
+## Routes
+
+- `GET /projects/{project}/research/papers`
+- `GET /projects/{project}/research/papers/{id}`
+
+Only configured Foundry projects are accepted. Reading the public corpus does
+not require a session; these routes do not expose private data or accept writes.
+
+List parameters are `q` (200 Unicode characters, 20 words), `searchIn`
+(`text` or `author`), `topic` (an exact arXiv category), `sort` (`recent` or
+`title`), `page` (1–1000), and `pageSize` (1–50, default 8). Unknown, malformed
+or repeated parameters return 400. `recent` means descending stable paper ID,
+not publication date. Search terms must become bound query parameters, never
+raw SQL or user-supplied full-text syntax.
+
+Title/abstract search uses the migration's English vector with
+`plainto_tsquery`; author search uses simple-language terms over the stored
+author JSON. Both treat input as terms, never full-text operators. Initial full-corpus measurements showed eight-second timeouts for text,
+author and topic counts; the unfiltered count took 7.25 seconds. Execution plans
+confirmed sequential scans for authors and category joins.
+`deploy/research/query-indexes.sql` supplies operator-run concurrent indexes for
+those two missing access paths. It is separate from Core startup and auth
+migrations; do not grant DDL to the runtime reader. Both indexes completed on the confirmed ECS (category 35.4 seconds, author
+436.47 seconds); measured end-to-end checks are documented below.
+
+List responses contain `items`, `total`, `topics`, `page`, and `pageSize`.
+Each paper preserves original title/abstract, authors, source dates, topics and
+source links. Publication decisions need source evidence; absent decisions stay
+unknown. No invented repositories, metrics or generated research conclusions.
+
+The repository counts and reads each list in one read-only snapshot.
+Paging first materializes matching IDs, then joins only that page to the wide
+paper rows. Filtered searches first materialize matching IDs without an
+ORDER BY/LIMIT, then sort/page those IDs, preventing sparse terms from choosing
+a backward heap scan with repeated text-vector computation. Activity sampling showed that selecting full rows before LIMIT
+caused sparse category queries to read large portions of the paper heap. The
+HTTP layer permits two in-flight queries with an eight-second context budget.
+Busy requests return 429 with `Retry-After: 1`; deadlines return 504; storage
+errors return a generic 503 without database details. Missing papers return 404.
+
+## Validation
+
+`go test -race -count=1 ./internal/research` covers routing, query limits,
+Unicode, anonymous reads, rejected mutations, project lookup, errors, saturation
+and timeout slot release. Run compilation/tests only on the user-confirmed ECS
+runner. These transport tests do not prove a database connection, full-corpus
+query performance or production availability.
+
+The PostgreSQL repository batches source/tag/identifier/submission queries in
+the same read-only repeatable-read transaction. Author tuples are displayed as
+given name, family name and suffix; an optional fourth affiliation field
+is excluded from the display name (observed in real arXiv records); null abstracts/dates remain null. Publication
+status comes from the latest observed decision, using decision date and ID as
+tie-breakers, and stays unknown when no decision exists. Unit checks cover
+author shapes, bound user values and schema qualification. The real Core detail
+and homepage HTTP routes have returned 200.
+The frontend has also rendered the actual 3,526,999-paper corpus through a
+loopback SSH tunnel: eight rows, matching detail title/source, desktop and
+mobile layout, no page exceptions or horizontal overflow in that check.
+Author search returned 146 records in 0.336 seconds. After paging/author fixes, category `cs.AR` returned 9,457 records in 4.906
+seconds, `CartoonGAN` returned three in 0.332 seconds, and the multi-term search
+`attention is all you need` returned 8,653 in 6.325 seconds. These were sequential
+manual HTTP checks, not a load test. Very broad text searches can still exceed
+the eight-second budget; clients retain search controls so users can narrow
+the query or retry. This known limit is not an estimated count or sample
+fallback. This is not a
+production deployment or an authentication acceptance result.
+
+The existing `deploy/check-schema.sh` also loads the synthetic fixture in
+`internal/research/testdata/corpus.sql` inside its disposable PostgreSQL
+container (`--network none`, no host port). The research test process receives a
+separate SELECT-only fixture role via `RESEARCH_TEST_DATABASE_URL`. These
+integration checks execute real SQL for filtering, stable paging, nullable
+content, source/tag/identifier joins, latest decision tie-breakers, permission
+denial and SQL-backed HTTP routes. They never connect to the shared migration
+instance and do not establish full-corpus performance or live-data correctness.
+
+## Text storage compatibility (repair 4e62710)
+
+The Paper Atlas repair uses `nul-soh-v1`: NUL is stored as SOH/SOH, and a
+literal SOH as SOH/STX, preserving binary text ordering. Core reads the schema
+comment `paper-atlas:text-codec=nul-soh-v1` through `pg_catalog` at startup,
+restores SQL text before JSON parsing/API responses, and encodes bound search
+terms. An absent/unrelated schema comment keeps legacy text unchanged; an
+unsupported marked codec or malformed encoded text fails explicitly.
+
+Schema comments are public metadata, so consumers need no privileges on the
+private migration controller. The research role still needs only schema USAGE
+and table SELECT. The codec adaptation is on the review branch, pushed with [skip ci] under the
+user's no-CI instruction; the published schema marker has been verified with
+the runtime reader.
+All research tests passed manually against disposable PostgreSQL on the
+confirmed ECS, including the encoded SQL fixture for schema detection, raw TEXT restoration,
+JSON author/metadata decoding, search and the HTTP round trip. Passing isolated
+fixtures does not establish full-corpus API performance.
+
+
+## Identity handoff
+
+The existing production Core received a configuration-only reload using its
+unchanged release binary: the original projects were retained and thesisly was
+registered. Both Google and Slack identity records were observed in this
+project. The user independently confirmed each provider's real authorization,
+reload/new-tab persistence and logout behavior. Browser control was unavailable,
+so these browser results are attributed to user acceptance rather than an
+automated real-account test. This does not deploy the research work branch;
+mail delivery was not exercised.

internal/platform/asset_preview.go

@@ -0,0 +1,242 @@
+package platform
+
+import (
+	"bytes"
+	"container/list"
+	"context"
+	"crypto/sha256"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"image"
+	"image/color"
+	"image/jpeg"
+	"image/png"
+	"sync"
+
+	"golang.org/x/image/draw"
+)
+
+type assetPreviewResult struct {
+	data []byte
+	mime string
+}
+type assetPreviewEntry struct {
+	key   string
+	value assetPreviewResult
+}
+
+var previewMemory = struct {
+	sync.Mutex
+	entries map[string]*list.Element
+	order   *list.List
+	bytes   int
+}{entries: make(map[string]*list.Element), order: list.New()}
+var previewSlots = make(chan struct{}, 1)
+
+func previewWidth(variant, width string) (int, error) {
+	if variant == "" && width == "" {
+		return 0, nil
+	}
+	if variant == "preview" && (width == "" || width == "1024") {
+		return 1024, nil
+	}
+	return 0, errors.New("invalid_asset_variant")
+}
+
+// Cache only immutable bytes, after the caller checks current asset permission.
+// Bound both decoded concurrency and retained encoded data.
+func cachedAssetPreview(ctx context.Context, encoded []byte, mime string, width int) (assetPreviewResult, error) {
+	key := fmt.Sprintf("%x:%d", sha256.Sum256(encoded), width)
+	previewMemory.Lock()
+	if entry := previewMemory.entries[key]; entry != nil {
+		previewMemory.order.MoveToFront(entry)
+		result := entry.Value.(assetPreviewEntry).value
+		previewMemory.Unlock()
+		return result, nil
+	}
+	previewMemory.Unlock()
+	select {
+	case previewSlots <- struct{}{}:
+	case <-ctx.Done():
+		return assetPreviewResult{}, ctx.Err()
+	}
+	defer func() { <-previewSlots }()
+	// Another request may have finished this same preview while we waited.
+	previewMemory.Lock()
+	if entry := previewMemory.entries[key]; entry != nil {
+		result := entry.Value.(assetPreviewEntry).value
+		previewMemory.order.MoveToFront(entry)
+		previewMemory.Unlock()
+		return result, nil
+	}
+	previewMemory.Unlock()
+	result, err := makeAssetPreview(encoded, mime, width)
+	if err != nil {
+		return result, err
+	}
+	if len(result.data) > 8<<20 {
+		return result, nil
+	}
+	previewMemory.Lock()
+	defer previewMemory.Unlock()
+	for previewMemory.bytes+len(result.data) > 32<<20 || previewMemory.order.Len() >= 128 {
+		tail := previewMemory.order.Back()
+		if tail == nil {
+			break
+		}
+		old := tail.Value.(assetPreviewEntry)
+		previewMemory.bytes -= len(old.value.data)
+		delete(previewMemory.entries, old.key)
+		previewMemory.order.Remove(tail)
+	}
+	previewMemory.entries[key] = previewMemory.order.PushFront(assetPreviewEntry{key, result})
+	previewMemory.bytes += len(result.data)
+	return result, nil
+}
+
+func makeAssetPreview(encoded []byte, mime string, width int) (assetPreviewResult, error) {
+	original := assetPreviewResult{encoded, mime}
+	// Keep animation intact. Original bytes remain the canonical AI/download input.
+	if mime == "image/gif" {
+		return original, nil
+	}
+	config, _, err := image.DecodeConfig(bytes.NewReader(encoded))
+	if err != nil {
+		return original, err
+	}
+	if config.Width <= 0 || config.Height <= 0 || int64(config.Width)*int64(config.Height) > assetPixelLimit {
+		return original, errors.New("invalid_image_dimensions")
+	}
+	if config.Width <= width && config.Height <= width && len(encoded) <= 64<<10 {
+		return original, nil
+	}
+	source, _, err := image.Decode(bytes.NewReader(encoded))
+	if err != nil {
+		return original, err
+	}
+	orientation := jpegOrientation(encoded)
+	if orientation != 1 {
+		source = orientedAsset{source, orientation}
+	}
+	bounds := source.Bounds()
+	w, h := bounds.Dx(), bounds.Dy()
+	if w > width || h > width {
+		if w >= h {
+			h = max(1, h*width/w)
+			w = width
+		} else {
+			w = max(1, w*width/h)
+			h = width
+		}
+	}
+	resized := image.NewNRGBA(image.Rect(0, 0, w, h))
+	draw.CatmullRom.Scale(resized, resized.Bounds(), source, source.Bounds(), draw.Src, nil)
+	var output bytes.Buffer
+	outputMime := "image/png"
+	if resized.Opaque() {
+		err = jpeg.Encode(&output, resized, &jpeg.Options{Quality: 82})
+		outputMime = "image/jpeg"
+	} else {
+		err = png.Encode(&output, resized)
+	}
+	if err != nil {
+		return original, err
+	}
+	if output.Len() >= len(encoded) && orientation == 1 {
+		return original, nil
+	}
+	return assetPreviewResult{output.Bytes(), outputMime}, nil
+}
+
+// JPEG EXIF orientation is applied to previews instead of silently rotating a
+// phone photograph after its EXIF metadata is removed by the image encoder.
+func jpegOrientation(data []byte) int {
+	if len(data) < 4 || data[0] != 0xff || data[1] != 0xd8 {
+		return 1
+	}
+	for at := 2; at+4 <= len(data); {
+		if data[at] != 0xff {
+			return 1
+		}
+		marker := data[at+1]
+		if marker == 0xda || marker == 0xd9 {
+			return 1
+		}
+		length := int(binary.BigEndian.Uint16(data[at+2 : at+4]))
+		if length < 2 || at+2+length > len(data) {
+			return 1
+		}
+		part := data[at+4 : at+2+length]
+		if marker == 0xe1 && len(part) >= 14 && string(part[:6]) == "Exif\x00\x00" {
+			tiff := part[6:]
+			var order binary.ByteOrder
+			switch string(tiff[:2]) {
+			case "II":
+				order = binary.LittleEndian
+			case "MM":
+				order = binary.BigEndian
+			default:
+				return 1
+			}
+			if order.Uint16(tiff[2:4]) != 42 {
+				return 1
+			}
+			offset := int(order.Uint32(tiff[4:8]))
+			if offset < 8 || offset+2 > len(tiff) {
+				return 1
+			}
+			count := int(order.Uint16(tiff[offset : offset+2]))
+			for i := 0; i < count; i++ {
+				entry := offset + 2 + i*12
+				if entry+12 > len(tiff) {
+					return 1
+				}
+				if order.Uint16(tiff[entry:entry+2]) == 0x112 && order.Uint16(tiff[entry+2:entry+4]) == 3 && order.Uint32(tiff[entry+4:entry+8]) == 1 {
+					value := int(order.Uint16(tiff[entry+8 : entry+10]))
+					if value >= 1 && value <= 8 {
+						return value
+					}
+					return 1
+				}
+			}
+		}
+		at += 2 + length
+	}
+	return 1
+}
+
+type orientedAsset struct {
+	image.Image
+	orientation int
+}
+
+func (im orientedAsset) Bounds() image.Rectangle {
+	b := im.Image.Bounds()
+	if im.orientation >= 5 {
+		return image.Rect(0, 0, b.Dy(), b.Dx())
+	}
+	return image.Rect(0, 0, b.Dx(), b.Dy())
+}
+func (im orientedAsset) ColorModel() color.Model { return im.Image.ColorModel() }
+func (im orientedAsset) At(x, y int) color.Color {
+	b := im.Image.Bounds()
+	w, h := b.Dx(), b.Dy()
+	switch im.orientation {
+	case 2:
+		x = w - 1 - x
+	case 3:
+		x, y = w-1-x, h-1-y
+	case 4:
+		y = h - 1 - y
+	case 5:
+		x, y = y, x
+	case 6:
+		x, y = y, h-1-x
+	case 7:
+		x, y = w-1-y, h-1-x
+	case 8:
+		x, y = w-1-y, x
+	}
+	return im.Image.At(b.Min.X+x, b.Min.Y+y)
+}

internal/platform/asset_preview_test.go

@@ -0,0 +1,137 @@
+package platform
+
+import (
+	"bytes"
+	"context"
+	"encoding/binary"
+	"image"
+	"image/color"
+	"image/jpeg"
+	"image/png"
+	"testing"
+)
+
+func TestAssetPreviewDownscalesWithoutChangingOriginal(t *testing.T) {
+	source := image.NewNRGBA(image.Rect(0, 0, 1600, 900))
+	for y := 0; y < 900; y++ {
+		for x := 0; x < 1600; x++ {
+			source.SetNRGBA(x, y, color.NRGBA{uint8(x), uint8(y), uint8(x * y), 255})
+		}
+	}
+	var encoded bytes.Buffer
+	if err := png.Encode(&encoded, source); err != nil {
+		t.Fatal(err)
+	}
+	original := append([]byte(nil), encoded.Bytes()...)
+	preview, err := makeAssetPreview(encoded.Bytes(), "image/png", 512)
+	if err != nil {
+		t.Fatal(err)
+	}
+	config, format, err := image.DecodeConfig(bytes.NewReader(preview.data))
+	if err != nil || config.Width != 512 || config.Height != 288 || format != "jpeg" {
+		t.Fatalf("bad preview %s %dx%d %v", format, config.Width, config.Height, err)
+	}
+	if len(preview.data) >= len(original) || !bytes.Equal(original, encoded.Bytes()) {
+		t.Fatal("preview must shrink without changing original")
+	}
+	cached, err := cachedAssetPreview(context.Background(), original, "image/png", 512)
+	if err != nil || !bytes.Equal(cached.data, preview.data) {
+		t.Fatal("cache result mismatch")
+	}
+}
+func TestAssetPreviewPreservesTransparencyAndAnimation(t *testing.T) {
+	source := image.NewNRGBA(image.Rect(0, 0, 800, 400))
+	source.SetNRGBA(1, 1, color.NRGBA{255, 0, 0, 255})
+	var encoded bytes.Buffer
+	_ = png.Encode(&encoded, source)
+	preview, err := makeAssetPreview(encoded.Bytes(), "image/png", 256)
+	if err != nil {
+		t.Fatal(err)
+	}
+	decoded, format, err := image.Decode(bytes.NewReader(preview.data))
+	if err != nil || format != "png" {
+		t.Fatal("transparent image must stay PNG")
+	}
+	_, _, _, alpha := decoded.At(200, 100).RGBA()
+	if alpha != 0 {
+		t.Fatal("transparent background was lost")
+	}
+	original := []byte("GIF original animation bytes")
+	gif, err := makeAssetPreview(original, "image/gif", 512)
+	if err != nil || !bytes.Equal(gif.data, original) {
+		t.Fatal("GIF animation changed")
+	}
+}
+func TestAssetPreviewRejectsUnboundedWidths(t *testing.T) {
+	for _, value := range []string{"0", "1", "256", "512", "2048", "999999", "-1", "abc"} {
+		if _, err := previewWidth("preview", value); err == nil {
+			t.Errorf("accepted width %s", value)
+		}
+	}
+	for _, value := range []string{"", "1024"} {
+		if _, err := previewWidth("preview", value); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if n, err := previewWidth("", ""); err != nil || n != 0 {
+		t.Fatal("original request changed")
+	}
+	if _, err := previewWidth("other", ""); err == nil {
+		t.Fatal("unknown variant accepted")
+	}
+	if _, err := makeAssetPreview([]byte("invalid"), "image/png", 512); err == nil {
+		t.Fatal("malformed image accepted")
+	}
+}
+func TestAssetPreviewAppliesJPEGOrientation(t *testing.T) {
+	source := image.NewNRGBA(image.Rect(0, 0, 800, 400))
+	for y := 0; y < 400; y++ {
+		for x := 0; x < 800; x++ {
+			source.SetNRGBA(x, y, color.NRGBA{255, 0, 0, 255})
+		}
+	}
+	var original bytes.Buffer
+	_ = jpeg.Encode(&original, source, nil)
+	tiff := make([]byte, 26)
+	copy(tiff, "II")
+	binary.LittleEndian.PutUint16(tiff[2:], 42)
+	binary.LittleEndian.PutUint32(tiff[4:], 8)
+	binary.LittleEndian.PutUint16(tiff[8:], 1)
+	binary.LittleEndian.PutUint16(tiff[10:], 0x112)
+	binary.LittleEndian.PutUint16(tiff[12:], 3)
+	binary.LittleEndian.PutUint32(tiff[14:], 1)
+	binary.LittleEndian.PutUint16(tiff[18:], 6)
+	exif := append([]byte("Exif\x00\x00"), tiff...)
+	segment := append([]byte{0xff, 0xe1, 0, byte(len(exif) + 2)}, exif...)
+	encoded := append(append(append([]byte{}, original.Bytes()[:2]...), segment...), original.Bytes()[2:]...)
+	if jpegOrientation(encoded) != 6 {
+		t.Fatal("EXIF orientation not decoded")
+	}
+	preview, err := makeAssetPreview(encoded, "image/jpeg", 256)
+	if err != nil {
+		t.Fatal(err)
+	}
+	config, _, err := image.DecodeConfig(bytes.NewReader(preview.data))
+	if err != nil || config.Width != 128 || config.Height != 256 {
+		t.Fatalf("rotated dimensions %dx%d %v", config.Width, config.Height, err)
+	}
+}
+func TestAllAssetOrientationMappings(t *testing.T) {
+	source := image.NewNRGBA(image.Rect(0, 0, 3, 2))
+	source.SetNRGBA(0, 0, color.NRGBA{255, 0, 0, 255})
+	for orientation := 1; orientation <= 8; orientation++ {
+		oriented := orientedAsset{source, orientation}
+		red := 0
+		for y := 0; y < oriented.Bounds().Dy(); y++ {
+			for x := 0; x < oriented.Bounds().Dx(); x++ {
+				r, _, _, _ := oriented.At(x, y).RGBA()
+				if r > 0 {
+					red++
+				}
+			}
+		}
+		if red != 1 {
+			t.Errorf("orientation %d loses/repeats pixels", orientation)
+		}
+	}
+}

internal/platform/assets.go

@@ -79,6 +79,10 @@ func (s *service) uploadAsset(w http.ResponseWriter, r *http.Request, project, a
 		reject(w, 403, "forbidden")
 		return nil
 	}
+	// Do not hold a workspace lock during a slow multipart upload. Recheck below.
+	if err = tx.Rollback(); err != nil {
+		return err
+	}
 	r.Body = http.MaxBytesReader(w, r.Body, assetLimit+(64<<10))
 	reader, err := r.MultipartReader()
 	if err != nil {
@@ -123,6 +127,19 @@ func (s *service) uploadAsset(w http.ResponseWriter, r *http.Request, project, a
 		reject(w, 400, "invalid_asset")
 		return nil
 	}
+	tx, err = s.db.BeginTx(r.Context(), nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	_, role, err = lockWorkspace(r.Context(), tx, project, workspace, account)
+	if err != nil {
+		return err
+	}
+	if role == "viewer" {
+		reject(w, 403, "forbidden")
+		return nil
+	}
 	digest := sha256.Sum256(encoded)
 	hash := hex.EncodeToString(digest[:])
 	info := assetInfo{MIME: mime, Bytes: len(encoded), Width: width, Height: height}
@@ -204,6 +221,11 @@ func (s *service) getAsset(w http.ResponseWriter, r *http.Request, project, acco
 		reject(w, 405, "method_not_allowed")
 		return nil
 	}
+	width, variantErr := previewWidth(r.URL.Query().Get("variant"), r.URL.Query().Get("width"))
+	if variantErr != nil {
+		reject(w, 400, "invalid_asset_variant")
+		return nil
+	}
 	id := r.PathValue("asset")
 	if !uuidSlugPattern.MatchString(id) {
 		reject(w, 404, "not_found")
@@ -235,6 +257,21 @@ func (s *service) getAsset(w http.ResponseWriter, r *http.Request, project, acco
 	if err != nil {
 		return err
 	}
+	// Permission and immutable bytes are now read consistently. Release locks
+	// before resizing or waiting for a slow public client to consume the body.
+	if err = tx.Rollback(); err != nil {
+		return err
+	}
+	variant := "original"
+	if width > 0 {
+		preview, e := cachedAssetPreview(r.Context(), encoded, mime, width)
+		if e != nil {
+			return e
+		}
+		encoded, mime = preview.data, preview.mime
+		variant = "preview"
+	}
+	w.Header().Set("X-Asset-Variant", variant)
 	w.Header().Set("Content-Type", mime)
 	w.Header().Set("Content-Length", strconv.Itoa(len(encoded)))
 	w.Header().Set("X-Content-Type-Options", "nosniff")

internal/platform/canvas.go

@@ -345,6 +345,9 @@ func (s *service) canvas(w http.ResponseWriter, r *http.Request, project, accoun
 		return nil
 	}
 	if r.Method == http.MethodGet {
+		if err = tx.Rollback(); err != nil {
+			return err
+		}
 		w.Header().Set("X-Canvas-Schema-Version", strconv.Itoa(canvasSchema(document)))
 		setVersion(w, version)
 		reply(w, 200, response{"version": version, "document": document})

internal/platform/canvas_capabilities.go

@@ -24,6 +24,9 @@ func (s *service) canvasCapabilities(w http.ResponseWriter, r *http.Request, pro
 		reject(w, 400, "invalid_material")
 		return nil
 	}
+	if err = tx.Rollback(); err != nil {
+		return err
+	}
 	ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
 	defer cancel()
 	models, modelErr := s.canvasAI.Models(ctx)
@@ -44,8 +47,9 @@ func (s *service) canvasCapabilities(w http.ResponseWriter, r *http.Request, pro
 		"contract_version":          "canvas-v2-2026-11-06",
 		"supported_schema_versions": []int{1, 2},
 		"edge_features":             response{"appearance": true, "markers": true, "bezier_chain": true, "animation_configuration": true, "execution_events": true},
-		"subgraphs":                 response{"create": true, "migrate": true, "undo": true, "members": true, "dissolve": true},
+		"subgraphs":                 response{"create": true, "migrate": true, "undo": true, "members": true, "dissolve": true, "connections": true, "member_positions": true},
 		"ai":                        ai,
+		"assets":                    response{"previews": true, "preview_max_edge": 1024, "originals_preserved": true},
 		"limits":                    response{"max_depth": 8, "max_nodes": 500, "max_edges": 2000, "max_frame_ports": 4000, "max_curve_knots": 64, "max_snapshot_bytes": canvasLimit, "max_active_tasks": 128, "max_run_steps": 64, "max_execution_depth": 32, "max_input_images": 4, "max_input_image_bytes": 20 << 20, "max_coordinate": 1000000},
 		"retention":                 response{"idempotency_min_seconds": 86400, "undo_receipt_seconds": 604800},
 	})

internal/platform/canvas_connections.go

@@ -0,0 +1,351 @@
+package platform
+
+import (
+	"crypto/sha256"
+	"database/sql"
+	"encoding/hex"
+	"encoding/json"
+	"net/http"
+	"sort"
+	"time"
+)
+
+// Endpoints name real nodes, never transient projected copies or client-created ports.
+type canvasConnectionEndpoint struct {
+	Board  string `json:"board_id"`
+	Node   string `json:"node_id"`
+	Handle string `json:"handle"`
+}
+type canvasConnectionInput struct {
+	Source canvasConnectionEndpoint `json:"source"`
+	Target canvasConnectionEndpoint `json:"target"`
+	Edge   struct {
+		ID   string         `json:"id"`
+		Data map[string]any `json:"data,omitempty"`
+	} `json:"edge"`
+	Expected map[string]int64 `json:"expected_versions"`
+}
+
+// Paths run from a leaf board upwards to (but excluding) the coordinator.
+func connectionPath(r *http.Request, tx *sql.Tx, project, workspace, coordinator, board string) ([]subgraphRelation, error) {
+	path := []subgraphRelation{}
+	seen := map[string]bool{}
+	for board != coordinator {
+		if seen[board] || len(path) >= 8 {
+			return nil, errGraphOwnership
+		}
+		seen[board] = true
+		relation, err := readSubgraphRelation(r, tx, project, workspace, board)
+		if err == sql.ErrNoRows {
+			return nil, errGraphOwnership
+		}
+		if err != nil {
+			return nil, err
+		}
+		if relation.Archived {
+			return nil, errGraphOwnership
+		}
+		path = append(path, relation)
+		board = relation.Parent
+	}
+	return path, nil
+}
+
+func connectionNode(doc map[string]any, id string) map[string]any {
+	for _, value := range doc["nodes"].([]any) {
+		node := value.(map[string]any)
+		if node["id"] == id {
+			return node
+		}
+	}
+	return nil
+}
+
+// Reuse ports by logical endpoint, including legacy ports with non-derived IDs.
+// A derived ID collision must never retarget an existing port.
+func connectionPort(data map[string]any, node, handle, direction string) (string, bool, error) {
+	ports, _ := data["ports"].([]any)
+	sum := sha256.Sum256([]byte(direction + "/" + node + "/" + handle))
+	id := "p-" + hex.EncodeToString(sum[:16])
+	collision := false
+	for _, value := range ports {
+		port := value.(map[string]any)
+		if port["direction"] == direction && port["nodeId"] == node && port["handle"] == handle {
+			return port["id"].(string), false, nil
+		}
+		collision = collision || port["id"] == id
+	}
+	if collision {
+		return "", false, errGraphOwnership
+	}
+	if len(ports) >= 4000 {
+		return "", false, errGraphCapacity
+	}
+	data["ports"] = append(ports, map[string]any{"id": id, "direction": direction, "nodeId": node, "handle": handle})
+	return id, true, nil
+}
+
+func exposeConnectionEndpoint(endpoint canvasConnectionEndpoint, direction string, path []subgraphRelation, documents map[string]map[string]any) (canvasConnectionEndpoint, []any, error) {
+	node := connectionNode(documents[endpoint.Board], endpoint.Node)
+	if node == nil {
+		return endpoint, nil, errGraphOwnership
+	}
+	data := node["data"].(map[string]any)
+	if !canvasChoice(data["kind"], "text", "script", "plain-text", "image", "plain-image") {
+		return endpoint, nil, errGraphOwnership
+	}
+	created := []any{}
+	for _, relation := range path {
+		frame := connectionNode(documents[relation.Parent], relation.Frame)
+		if frame == nil {
+			return endpoint, nil, errGraphOwnership
+		}
+		data := frame["data"].(map[string]any)
+		sub, _ := data["subgraph"].(map[string]any)
+		if relation.Child != endpoint.Board || data["kind"] != "frame" || sub["boardId"] != relation.Child {
+			return endpoint, nil, errGraphOwnership
+		}
+		id, added, err := connectionPort(data, endpoint.Node, endpoint.Handle, direction)
+		if err != nil {
+			return endpoint, nil, err
+		}
+		if added {
+			created = append(created, response{"board_id": relation.Parent, "frame_id": relation.Frame, "port_id": id})
+		}
+		endpoint = canvasConnectionEndpoint{Board: relation.Parent, Node: relation.Frame, Handle: "port:" + id}
+	}
+	return endpoint, created, nil
+}
+
+func (s *service) createCanvasConnection(w http.ResponseWriter, r *http.Request, project, account string) error {
+	if r.Method != http.MethodPost {
+		reject(w, 405, "method_not_allowed")
+		return nil
+	}
+	if !acceptsCanvasV2(r) {
+		reject(w, 409, "canvas_schema_upgrade_required")
+		return nil
+	}
+	key := canvasIdempotencyKey(r)
+	if !uuidSlugPattern.MatchString(key) {
+		reject(w, 400, "invalid_idempotency_key")
+		return nil
+	}
+	var input canvasConnectionInput
+	if !decode(w, r, &input) {
+		return nil
+	}
+	for _, endpoint := range []canvasConnectionEndpoint{input.Source, input.Target} {
+		if !uuidSlugPattern.MatchString(endpoint.Board) || !canvasID.MatchString(endpoint.Node) || !canvasChoice(endpoint.Handle, "top", "right", "bottom", "left") {
+			reject(w, 400, "invalid_connection_request")
+			return nil
+		}
+	}
+	if !canvasID.MatchString(input.Edge.ID) {
+		reject(w, 400, "invalid_edge")
+		return nil
+	}
+	limits, _ := json.Marshal(response{"edges": []any{input.Edge}})
+	if canvasExceededGraphLimits(limits) {
+		reject(w, 413, "graph_limit_exceeded")
+		return nil
+	}
+	if input.Edge.Data != nil && !validCanvasEdgeData(input.Edge.Data) {
+		reject(w, 400, "invalid_edge")
+		return nil
+	}
+	if len(input.Expected) == 0 {
+		reject(w, 428, "version_required")
+		return nil
+	}
+	workspace, coordinator := r.PathValue("workspace"), r.PathValue("material")
+	tx, err := s.db.BeginTx(r.Context(), nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	kind, role, err := lockMaterial(r.Context(), tx, project, workspace, coordinator, account)
+	if err != nil {
+		return err
+	}
+	if kind != "board" {
+		reject(w, 400, "invalid_material")
+		return nil
+	}
+	if role == "viewer" {
+		reject(w, 403, "forbidden")
+		return nil
+	}
+	action, hash := "create_connection", payloadDigest(input)
+	done, err := transactionRetry(w, r, tx, project, account, coordinator, action, key, hash)
+	if err != nil || done {
+		return err
+	}
+	participants := map[string]bool{coordinator: true}
+	paths := [][]subgraphRelation{}
+	for _, endpoint := range []canvasConnectionEndpoint{input.Source, input.Target} {
+		kind, role, err := lockMaterial(r.Context(), tx, project, workspace, endpoint.Board, account)
+		if err != nil {
+			return err
+		}
+		if role == "viewer" {
+			reject(w, 403, "forbidden")
+			return nil
+		}
+		if kind != "board" {
+			reject(w, 400, "invalid_material")
+			return nil
+		}
+		path, err := connectionPath(r, tx, project, workspace, coordinator, endpoint.Board)
+		if err == errGraphOwnership {
+			reject(w, 409, "ownership_conflict")
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		paths = append(paths, path)
+		participants[endpoint.Board] = true
+		for _, relation := range path {
+			participants[relation.Parent] = true
+		}
+	}
+	// Require the lowest common ancestor, so the logical edge has one owner.
+	left := map[string]bool{}
+	for _, relation := range paths[0] {
+		left[relation.Child] = true
+	}
+	for _, relation := range paths[1] {
+		if left[relation.Child] {
+			reject(w, 409, "connection_owner_mismatch")
+			return nil
+		}
+	}
+	ids := []string{}
+	for id := range participants {
+		ids = append(ids, id)
+	}
+	sort.Strings(ids)
+	before := []canvasSnapshot{}
+	snapshots := map[string]canvasSnapshot{}
+	documents := map[string]map[string]any{}
+	for _, id := range ids {
+		_, role, err := lockMaterial(r.Context(), tx, project, workspace, id, account)
+		if err != nil {
+			return err
+		}
+		if role == "viewer" {
+			reject(w, 403, "forbidden")
+			return nil
+		}
+		snapshot, err := loadCanvas(r, tx, project, workspace, id)
+		if err != nil {
+			return err
+		}
+		snapshots[id] = snapshot
+		before = append(before, snapshot)
+		var doc map[string]any
+		if err = json.Unmarshal(snapshot.Document, &doc); err != nil {
+			return err
+		}
+		documents[id] = doc
+	}
+	if len(input.Expected) != len(ids) {
+		reply(w, 428, response{"error": "version_required", "details": response{"required_board_ids": ids}})
+		return nil
+	}
+	for _, id := range ids {
+		expected, exists := input.Expected[id]
+		if !exists {
+			reply(w, 428, response{"error": "version_required", "details": response{"required_board_ids": ids}})
+			return nil
+		}
+		if expected != snapshots[id].Version {
+			reject(w, 412, "version_conflict")
+			return nil
+		}
+	}
+	for _, value := range documents[coordinator]["edges"].([]any) {
+		if value.(map[string]any)["id"] == input.Edge.ID {
+			reject(w, 409, "edge_id_conflict")
+			return nil
+		}
+	}
+	mapped := []canvasConnectionEndpoint{}
+	created := []any{}
+	for i, endpoint := range []canvasConnectionEndpoint{input.Source, input.Target} {
+		direction := "output"
+		if i == 1 {
+			direction = "input"
+		}
+		exposed, ports, err := exposeConnectionEndpoint(endpoint, direction, paths[i], documents)
+		if err == errGraphOwnership {
+			reject(w, 409, "ownership_conflict")
+			return nil
+		}
+		if err == errGraphCapacity {
+			reject(w, 413, "graph_limit_exceeded")
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		mapped = append(mapped, exposed)
+		created = append(created, ports...)
+	}
+	edge := map[string]any{"id": input.Edge.ID, "source": mapped[0].Node, "sourceHandle": mapped[0].Handle, "target": mapped[1].Node, "targetHandle": mapped[1].Handle}
+	if input.Edge.Data != nil {
+		edge["data"] = input.Edge.Data
+	}
+	documents[coordinator]["edges"] = append(documents[coordinator]["edges"].([]any), edge)
+	updated := []canvasSnapshot{}
+	versions := map[string]int64{}
+	for _, id := range ids {
+		documents[id]["schema_version"] = 2
+		raw, _ := json.Marshal(documents[id])
+		if len(raw) > canvasLimit {
+			reject(w, 413, "canvas_too_large")
+			return nil
+		}
+		if canvasExceededGraphLimits(raw) {
+			reject(w, 413, "graph_limit_exceeded")
+			return nil
+		}
+		normalized, ok := normalizeCanvas(raw)
+		if !ok {
+			reject(w, 400, "invalid_canvas")
+			return nil
+		}
+		snapshot := snapshots[id]
+		snapshot.Document = normalized
+		snapshot.Version++
+		updated = append(updated, snapshot)
+		snapshots[id] = snapshot
+		versions[id] = snapshot.Version
+	}
+	// Nothing is persisted until both complete port chains and all snapshots validate.
+	for _, snapshot := range updated {
+		if err = s.saveCanvas(r, tx, project, workspace, snapshot); err != nil {
+			return err
+		}
+	}
+	var id string
+	var expiry time.Time
+	if err = tx.QueryRowContext(r.Context(), `SELECT gen_random_uuid()::text,now()+interval '7 days'`).Scan(&id, &expiry); err != nil {
+		return err
+	}
+	result := response{"transaction_id": id, "parent": snapshots[coordinator], "snapshots": updated, "connection": response{"board_id": coordinator, "edge": edge, "source": input.Source, "target": input.Target}, "created_ports": created, "undo_receipt": response{"transaction_id": id, "valid_for_seconds": 604800, "expires_at": expiry}}
+	encoded, _ := json.Marshal(result)
+	prior, _ := json.Marshal(before)
+	after, _ := json.Marshal(versions)
+	_, err = tx.ExecContext(r.Context(), `INSERT INTO canvas_transactions(project,workspace_id,board_id,id,account_id,action,idempotency_key,payload_hash,response_status,response_body,before_state,after_versions) VALUES($1,$2,$3,$4,$5,$6,$7,$8,201,$9,$10,$11)`, project, workspace, coordinator, id, account, action, key, hash, encoded, prior, after)
+	if err != nil {
+		return err
+	}
+	if err = tx.Commit(); err != nil {
+		return err
+	}
+	setVersion(w, versions[coordinator])
+	reply(w, 201, result)
+	return nil
+}

internal/platform/canvas_connections_integration_test.go

@@ -0,0 +1,309 @@
+package platform
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"crypto/sha256"
+	"database/sql"
+	"encoding/base64"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"strconv"
+	"testing"
+)
+
+func TestCanvasConnectionTransactions(t *testing.T) {
+	dsn := os.Getenv("PLATFORM_TEST_DATABASE_URL")
+	if dsn == "" {
+		t.Skip("requires an isolated integration database")
+	}
+	db, err := sql.Open("pgx", dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	ctx := context.Background()
+	var random [12]byte
+	_, _ = rand.Read(random[:])
+	project := "connections-" + hex.EncodeToString(random[:])
+	defer func() {
+		for _, table := range []string{"canvas_transactions", "canvas_subgraphs", "workspaces", "project_accounts"} {
+			_, _ = db.ExecContext(ctx, "DELETE FROM "+table+" WHERE project=$1", project)
+		}
+	}()
+	tokens := []string{}
+	accounts := []string{}
+	for range 3 {
+		var account string
+		if err = db.QueryRowContext(ctx, `INSERT INTO project_accounts(project,id) VALUES($1,gen_random_uuid()) RETURNING id::text`, project).Scan(&account); err != nil {
+			t.Fatal(err)
+		}
+		var value [32]byte
+		_, _ = rand.Read(value[:])
+		token := base64.RawURLEncoding.EncodeToString(value[:])
+		hash := sha256.Sum256([]byte(token))
+		if _, err = db.ExecContext(ctx, `INSERT INTO project_sessions(token_hash,project,account_id,expires_at) VALUES($1,$2,$3,now()+interval '1 hour')`, base64.RawURLEncoding.EncodeToString(hash[:]), project, account); err != nil {
+			t.Fatal(err)
+		}
+		tokens = append(tokens, token)
+		accounts = append(accounts, account)
+	}
+	var workspace, root string
+	if err = db.QueryRowContext(ctx, `INSERT INTO workspaces(project,name,owner_id) VALUES($1,'Connections',$2) RETURNING id::text`, project, accounts[0]).Scan(&workspace); err != nil {
+		t.Fatal(err)
+	}
+	for i, role := range []string{"admin", "viewer"} {
+		if _, err = db.ExecContext(ctx, `INSERT INTO workspace_members VALUES($1,$2,$3,$4,now())`, project, workspace, accounts[i], role); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err = db.QueryRowContext(ctx, `INSERT INTO materials(project,workspace_id,kind,title,created_by,canvas_document) VALUES($1,$2,'board','Root',$3,$4) RETURNING id::text`, project, workspace, accounts[0], json.RawMessage(canvasFixture)).Scan(&root); err != nil {
+		t.Fatal(err)
+	}
+	mux := http.NewServeMux()
+	Register(mux, db, map[string]string{project: "https://app.example.com/auth/callback"}, "https://api.example.com", nil)
+	base := "/projects/" + project + "/workspaces/" + workspace + "/materials/"
+	path := func(board string) string { return base + board + "/canvas" }
+	do := func(method, url, token string, body any, version int64, key string) *httptest.ResponseRecorder {
+		raw, _ := json.Marshal(body)
+		r := httptest.NewRequest(method, url, bytes.NewReader(raw))
+		r.Header.Set("Authorization", "Bearer "+token)
+		r.Header.Set("Accept", "application/vnd.foundry.canvas+json;version=2")
+		if version > 0 {
+			r.Header.Set("If-Match", strconv.Quote(strconv.FormatInt(version, 10)))
+		}
+		if key != "" {
+			r.Header.Set("Idempotency-Key", key)
+		}
+		w := httptest.NewRecorder()
+		mux.ServeHTTP(w, r)
+		return w
+	}
+	call := func(method, url, token string, body any, status int, version int64, key string) response {
+		t.Helper()
+		w := do(method, url, token, body, version, key)
+		if w.Code != status {
+			t.Fatalf("%s %s got %d %s want %d", method, url, w.Code, w.Body.String(), status)
+		}
+		out := response{}
+		if err := json.Unmarshal(w.Body.Bytes(), &out); err != nil {
+			t.Fatal(err)
+		}
+		return out
+	}
+	key := func(n int) string { return fmt.Sprintf("a0000000-0000-4000-8000-%012d", n) }
+	versions := func(ids ...string) map[string]int64 {
+		t.Helper()
+		out := map[string]int64{}
+		for _, id := range ids {
+			out[id] = int64(call("GET", path(id), tokens[0], nil, 200, 0, "")["version"].(float64))
+		}
+		return out
+	}
+	create := func(board, frame string, n int) string {
+		t.Helper()
+		v := versions(board)[board]
+		result := call("POST", path(board)+"/subgraphs", tokens[0], response{"frame": response{"id": frame, "title": frame, "position": response{"x": 10, "y": 20}, "width": 600, "height": 400, "labelMode": "fixed"}}, 201, v, key(n))
+		return result["child_board"].(map[string]any)["id"].(string)
+	}
+	seed := func(board string, nodes ...string) {
+		t.Helper()
+		values := []any{}
+		for _, id := range nodes {
+			values = append(values, response{"id": id, "type": "material", "position": response{"x": 10, "y": 20}, "data": response{"kind": "text", "title": id, "text": "preserved"}})
+		}
+		call("PATCH", path(board), tokens[0], response{"document": response{"schema_version": 2, "nodes": values, "edges": []any{}, "viewport": response{"x": 0, "y": 0, "zoom": 1}}}, 200, 1, "")
+	}
+	left := create(root, "left", 1)
+	right := create(root, "right", 2)
+	seed(left, "l1", "l2")
+	seed(right, "r1")
+	leaf := create(left, "nested", 3)
+	seed(leaf, "deep")
+	command := func(source, node, target, targetNode, id string, ids ...string) response {
+		return response{"source": response{"board_id": source, "node_id": node, "handle": "right"}, "target": response{"board_id": target, "node_id": targetNode, "handle": "left"}, "edge": response{"id": id, "data": response{"appearance": response{"color": "#123456", "width": 2}, "path": response{"type": "bezier-chain", "knots": []any{response{"id": "k", "position": response{"x": 200, "y": 300}, "in": response{"x": -20, "y": 0}, "out": response{"x": 20, "y": 0}, "mode": "symmetric"}}}}}, "expected_versions": versions(ids...)}
+	}
+	firstBody := command(leaf, "deep", root, "a", "external", root, left, leaf)
+	beforeLeaf := call("GET", path(leaf), tokens[0], nil, 200, 0, "")["document"]
+	first := call("POST", path(root)+"/connections", tokens[0], firstBody, 201, 0, key(4))
+	if len(first["created_ports"].([]any)) != 2 {
+		t.Fatal("nested connection missing ports")
+	}
+	retry := call("POST", path(root)+"/connections", tokens[0], firstBody, 201, 0, key(4))
+	if retry["transaction_id"] != first["transaction_id"] {
+		t.Fatal("retry duplicated operation")
+	}
+	changedBody := command(leaf, "deep", root, "a", "different", root, left, leaf)
+	call("POST", path(root)+"/connections", tokens[0], changedBody, 409, 0, key(4))
+	afterLeaf := call("GET", path(leaf), tokens[0], nil, 200, 0, "")["document"]
+	if payloadDigest(beforeLeaf) != payloadDigest(afterLeaf) {
+		t.Fatal("connection changed leaf content")
+	}
+	parent := first["parent"].(map[string]any)
+	edge := first["connection"].(map[string]any)["edge"].(map[string]any)
+	if edge["source"] != "left" || edge["target"] != "a" || edge["data"].(map[string]any)["appearance"].(map[string]any)["color"] != "#123456" {
+		t.Fatal("connection edge lost mapping/style")
+	}
+	if len(parent["document"].(map[string]any)["edges"].([]any)) != 2 {
+		t.Fatal("existing edge lost")
+	}
+	undo := call("POST", path(root)+"/transactions/"+first["transaction_id"].(string)+"/undo", tokens[0], response{"expected_versions": versions(root, left, leaf)}, 200, 0, key(5))
+	if len(undo["snapshots"].([]any)) != 3 {
+		t.Fatal("undo did not restore complete chain")
+	}
+	for board, frame := range map[string]string{root: "left", left: "nested"} {
+		document := call("GET", path(board), tokens[0], nil, 200, 0, "")["document"]
+		raw, _ := json.Marshal(document)
+		if len(canvasNodeData(raw)[frame]["ports"].([]any)) != 0 {
+			t.Fatal("undo left ports behind")
+		}
+	}
+	redoBody := command(leaf, "deep", root, "a", "external", root, left, leaf)
+	redone := call("POST", path(root)+"/connections", tokens[0], redoBody, 201, 0, key(6))
+	if payloadDigest(redone["connection"]) != payloadDigest(first["connection"]) {
+		t.Fatal("redo changed stable ports")
+	}
+	reused := call("POST", path(root)+"/connections", tokens[0], command(leaf, "deep", root, "b", "second", root, left, leaf), 201, 0, key(7))
+	if len(reused["created_ports"].([]any)) != 0 {
+		t.Fatal("same source endpoint duplicated ports")
+	}
+	sibling := call("POST", path(root)+"/connections", tokens[0], command(left, "l1", right, "r1", "sibling", root, left, right), 201, 0, key(8))
+	if len(sibling["created_ports"].([]any)) != 2 {
+		t.Fatal("sibling connection missing boundary ports")
+	}
+	reverse := call("POST", path(root)+"/connections", tokens[0], command(root, "b", leaf, "deep", "reverse", root, left, leaf), 201, 0, key(9))
+	if len(reverse["created_ports"].([]any)) != 2 {
+		t.Fatal("input chain reused output ports")
+	}
+	// Missing/stale participant versions, bad endpoints and permission failures are atomic.
+	stable := versions(root, left, leaf)
+	missing := command(leaf, "deep", root, "a", "missing", root, left)
+	required := call("POST", path(root)+"/connections", tokens[0], missing, 428, 0, key(10))
+	if len(required["details"].(map[string]any)["required_board_ids"].([]any)) != 3 {
+		t.Fatal("version response omitted participants")
+	}
+	stale := command(leaf, "deep", root, "a", "stale", root, left, leaf)
+	stale["expected_versions"].(map[string]int64)[leaf]--
+	call("POST", path(root)+"/connections", tokens[0], stale, 412, 0, key(11))
+	call("POST", path(root)+"/connections", tokens[0], command(leaf, "absent", root, "a", "bad-node", root, left, leaf), 409, 0, key(12))
+	call("POST", path(root)+"/connections", tokens[0], command(root, "icon", leaf, "deep", "icon-edge", root, left, leaf), 409, 0, key(13))
+	call("POST", path(root)+"/connections", tokens[0], command(leaf, "deep", left, "l1", "wrong-owner", root, left, leaf), 409, 0, key(14))
+	call("POST", path(root)+"/connections", tokens[0], command(leaf, "deep", root, "a", "external", root, left, leaf), 409, 0, key(15))
+	call("POST", path(root)+"/connections", tokens[1], command(leaf, "deep", root, "a", "viewer", root, left, leaf), 403, 0, key(16))
+	call("POST", path(root)+"/connections", tokens[2], command(leaf, "deep", root, "a", "outsider", root, left, leaf), 404, 0, key(17))
+	invalid := command(leaf, "deep", root, "a", "handle", root, left, leaf)
+	invalid["source"].(response)["handle"] = "port:fake"
+	call("POST", path(root)+"/connections", tokens[0], invalid, 400, 0, key(18))
+	if payloadDigest(stable) != payloadDigest(versions(root, left, leaf)) {
+		t.Fatal("rejected transaction partially changed versions")
+	}
+	// The actual nested mapping is resolvable by the same reader used by AI propagation.
+	graph := map[string]canvasSnapshot{}
+	for _, board := range []string{root, left, leaf} {
+		read := call("GET", path(board), tokens[0], nil, 200, 0, "")
+		raw, _ := json.Marshal(read["document"])
+		graph[board] = canvasSnapshot{BoardID: board, Version: int64(read["version"].(float64)), Document: raw}
+	}
+	reader := readerFromSnapshots(graph, map[string]int{})
+	reader.snapshots = map[string]canvasSnapshot{}
+	reader.documents = map[string]map[string]any{}
+	reader.parents = map[string]subgraphRelation{}
+	reader.parentKnown = map[string]bool{}
+	resolved, err := reader.resolve(root, edge["source"].(string), edge["sourceHandle"].(string), "output", 0)
+	if err != nil || resolved != canvasNodeKey(leaf, "deep") {
+		t.Fatal("AI reader cannot resolve newly created port chain")
+	}
+	// Ordinary saves must still reject port retargeting.
+	readParent := call("GET", path(root), tokens[0], nil, 200, 0, "")
+	parentDoc := readParent["document"].(map[string]any)
+	for _, value := range parentDoc["nodes"].([]any) {
+		node := value.(map[string]any)
+		if node["id"] == "left" {
+			node["data"].(map[string]any)["ports"].([]any)[0].(map[string]any)["nodeId"] = "l2"
+		}
+	}
+	call("PATCH", path(root), tokens[0], response{"document": parentDoc}, 409, int64(readParent["version"].(float64)), "")
+	var otherWorkspace, otherBoard, unrelatedBoard string
+	if err = db.QueryRowContext(ctx, `INSERT INTO workspaces(project,name,owner_id) VALUES($1,'Other',$2) RETURNING id::text`, project, accounts[0]).Scan(&otherWorkspace); err != nil {
+		t.Fatal(err)
+	}
+	if _, err = db.ExecContext(ctx, `INSERT INTO workspace_members VALUES($1,$2,$3,'admin',now())`, project, otherWorkspace, accounts[0]); err != nil {
+		t.Fatal(err)
+	}
+	for _, target := range []struct {
+		workspace string
+		id        *string
+	}{{otherWorkspace, &otherBoard}, {workspace, &unrelatedBoard}} {
+		if err = db.QueryRowContext(ctx, `INSERT INTO materials(project,workspace_id,kind,title,created_by,canvas_document) VALUES($1,$2,'board','Other',$3,$4) RETURNING id::text`, project, target.workspace, accounts[0], json.RawMessage(canvasFixture)).Scan(target.id); err != nil {
+			t.Fatal(err)
+		}
+	}
+	call("POST", path(root)+"/connections", tokens[0], command(left, "l1", otherBoard, "a", "cross-workspace", root, left), 404, 0, key(21))
+	call("POST", path(root)+"/connections", tokens[0], command(left, "l1", unrelatedBoard, "a", "unrelated", root, left), 409, 0, key(22))
+	legacyRequest := httptest.NewRequest("POST", path(root)+"/connections", bytes.NewReader([]byte("{}")))
+	legacyRequest.Header.Set("Authorization", "Bearer "+tokens[0])
+	legacyResponse := httptest.NewRecorder()
+	mux.ServeHTTP(legacyResponse, legacyRequest)
+	if legacyResponse.Code != 409 {
+		t.Fatal("connection mutation accepted a legacy client")
+	}
+	caps := call("GET", path(root)+"/capabilities", tokens[0], nil, 200, 0, "")
+	if caps["subgraphs"].(map[string]any)["connections"] != true {
+		t.Fatal("capabilities omitted connection transaction")
+	}
+	// Four simultaneous retries create exactly one edge/receipt and advance versions once.
+	concurrent := command(left, "l2", root, "a", "concurrent", root, left)
+	prior := versions(root, left)
+	start := make(chan struct{})
+	results := make(chan *httptest.ResponseRecorder, 4)
+	for range 4 {
+		go func() { <-start; results <- do("POST", path(root)+"/connections", tokens[0], concurrent, 0, key(19)) }()
+	}
+	close(start)
+	transaction := ""
+	for range 4 {
+		w := <-results
+		if w.Code != 201 {
+			t.Fatalf("concurrent request: %d %s", w.Code, w.Body.String())
+		}
+		var result response
+		_ = json.Unmarshal(w.Body.Bytes(), &result)
+		id := result["transaction_id"].(string)
+		if transaction != "" && transaction != id {
+			t.Fatal("concurrent retry duplicated receipt")
+		}
+		transaction = id
+	}
+	for id, v := range versions(root, left) {
+		if v != prior[id]+1 {
+			t.Fatal("concurrent retry advanced version twice")
+		}
+	}
+	// A full coordinator rejects the entire operation, including new leaf ports.
+	snapshot := call("GET", path(root), tokens[0], nil, 200, 0, "")
+	doc := snapshot["document"].(map[string]any)
+	edges := []any{}
+	for i := 0; i < 2000; i++ {
+		edges = append(edges, response{"id": fmt.Sprint("full-", i), "source": "a", "target": "b", "sourceHandle": "right", "targetHandle": "left"})
+	}
+	doc["edges"] = edges
+	call("PATCH", path(root), tokens[0], response{"document": doc}, 200, int64(snapshot["version"].(float64)), "")
+	capacity := command(left, "l2", root, "a", "capacity", root, left)
+	capacity["source"].(response)["handle"] = "top"
+	beforeFailure := versions(root, left)
+	beforeLeft := call("GET", path(left), tokens[0], nil, 200, 0, "")["document"]
+	call("POST", path(root)+"/connections", tokens[0], capacity, 413, 0, key(20))
+	afterLeft := call("GET", path(left), tokens[0], nil, 200, 0, "")["document"]
+	if payloadDigest(beforeLeft) != payloadDigest(afterLeft) || payloadDigest(beforeFailure) != payloadDigest(versions(root, left)) {
+		t.Fatal("capacity failure left ports or versions behind")
+	}
+	var receipts int
+	if err = db.QueryRowContext(ctx, `SELECT count(*) FROM canvas_transactions WHERE project=$1 AND action='create_connection' AND idempotency_key=$2`, project, key(20)).Scan(&receipts); err != nil || receipts != 0 {
+		t.Fatal("rejected connection left a receipt")
+	}
+}

internal/platform/canvas_connections_test.go

@@ -0,0 +1,65 @@
+package platform
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+)
+
+func TestConnectionPortReuseCollisionAndCapacity(t *testing.T) {
+	data := map[string]any{"ports": []any{map[string]any{"id": "custom-stable", "nodeId": "n", "handle": "left", "direction": "input"}}}
+	id, created, err := connectionPort(data, "n", "left", "input")
+	if err != nil || created || id != "custom-stable" {
+		t.Fatal("existing logical port was not reused")
+	}
+	id, created, err = connectionPort(data, "n", "left", "output")
+	if err != nil || !created || id == "custom-stable" {
+		t.Fatal("input and output ports were conflated")
+	}
+	data["ports"].([]any)[1].(map[string]any)["nodeId"] = "other"
+	if _, _, err = connectionPort(data, "n", "left", "output"); err != errGraphOwnership {
+		t.Fatal("derived ID collision retargeted a port")
+	}
+	ports := []any{}
+	for i := 0; i < 4000; i++ {
+		ports = append(ports, map[string]any{"id": fmt.Sprint(i), "nodeId": "other", "handle": "left", "direction": "input"})
+	}
+	data["ports"] = ports
+	if _, _, err = connectionPort(data, "n", "right", "output"); err != errGraphCapacity || len(data["ports"].([]any)) != 4000 {
+		t.Fatal("port capacity partially mutated frame")
+	}
+}
+
+func TestConnectionExposureBuildsNestedDirectionChain(t *testing.T) {
+	frame := func(child string) map[string]any {
+		return map[string]any{"id": "frame", "data": map[string]any{"kind": "frame", "subgraph": map[string]any{"boardId": child}, "ports": []any{}}}
+	}
+	docs := map[string]map[string]any{
+		"root":   {"nodes": []any{frame("middle")}},
+		"middle": {"nodes": []any{frame("leaf")}},
+		"leaf":   {"nodes": []any{map[string]any{"id": "n", "data": map[string]any{"kind": "plain-image", "src": "https://example.com/image.png"}}}},
+	}
+	before, _ := json.Marshal(docs["leaf"])
+	path := []subgraphRelation{{Parent: "middle", Frame: "frame", Child: "leaf"}, {Parent: "root", Frame: "frame", Child: "middle"}}
+	endpoint := canvasConnectionEndpoint{Board: "leaf", Node: "n", Handle: "bottom"}
+	exposed, created, err := exposeConnectionEndpoint(endpoint, "output", path, docs)
+	if err != nil || exposed.Board != "root" || exposed.Node != "frame" || len(created) != 2 {
+		t.Fatalf("incomplete chain: %+v %v", exposed, err)
+	}
+	inner := connectionNode(docs["middle"], "frame")["data"].(map[string]any)["ports"].([]any)[0].(map[string]any)
+	outer := connectionNode(docs["root"], "frame")["data"].(map[string]any)["ports"].([]any)[0].(map[string]any)
+	if inner["nodeId"] != "n" || inner["handle"] != "bottom" || outer["nodeId"] != "frame" || outer["handle"] != "port:"+inner["id"].(string) {
+		t.Fatal("outer port skipped a child boundary")
+	}
+	again, created, err := exposeConnectionEndpoint(endpoint, "output", path, docs)
+	if err != nil || again != exposed || len(created) != 0 {
+		t.Fatal("repeated exposure duplicated ports")
+	}
+	if _, created, err = exposeConnectionEndpoint(endpoint, "input", path, docs); err != nil || len(created) != 2 {
+		t.Fatal("reverse direction did not create independent input chain")
+	}
+	after, _ := json.Marshal(docs["leaf"])
+	if string(before) != string(after) {
+		t.Fatal("exposure modified the actual child node")
+	}
+}

internal/platform/canvas_undo.go

@@ -71,7 +71,7 @@ func (s *service) undoCanvasTransaction(w http.ResponseWriter, r *http.Request,
 		reject(w, 409, "transaction_already_undone")
 		return nil
 	}
-	if originalAction != "create_subgraph" && originalAction != "migrate_subgraphs" && !strings.HasPrefix(originalAction, "dissolve_subgraph/") && !strings.HasPrefix(originalAction, "move_members/") {
+	if originalAction != "create_subgraph" && originalAction != "migrate_subgraphs" && originalAction != "create_connection" && !strings.HasPrefix(originalAction, "dissolve_subgraph/") && !strings.HasPrefix(originalAction, "move_members/") {
 		reject(w, 409, "undo_not_supported")
 		return nil
 	}

internal/platform/http.go

@@ -147,6 +147,7 @@ func Register(mux *http.ServeMux, db *sql.DB, callbacks map[string]string, origi
 	mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}/canvas/ai/runs/{run}/cancel", s.protected(callbacks, s.cancelCanvasAIRun))
 	mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}/canvas/ai/runs/{run}/apply", s.protected(callbacks, s.applyCanvasAIOutput))
 	mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}/canvas/capabilities", s.protected(callbacks, s.canvasCapabilities))
+	mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}/canvas/connections", s.protected(callbacks, s.createCanvasConnection))
 	mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}/canvas/subgraphs", s.protected(callbacks, s.createSubgraph))
 	mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}/canvas/subgraphs/migrate", s.protected(callbacks, s.migrateSubgraphs))
 	mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}/canvas/subgraphs/{frame}/dissolve", s.protected(callbacks, s.dissolveSubgraph))

internal/platform/integration_test.go

@@ -12,6 +12,7 @@ import (
 	"gitea.entroia.com/Zayden-Jung/foundry-core/internal/auth"
 	"image"
 	"image/png"
+	"io"
 	"mime/multipart"
 	"net/http"
 	"net/http/httptest"
@@ -19,7 +20,9 @@ import (
 	"regexp"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
+	"time"
 
 	_ "github.com/jackc/pgx/v5/stdlib"
 )
@@ -420,7 +423,8 @@ func TestOnboardingAndSharing(t *testing.T) {
 	}
 	parentID := migrationBoard["id"].(string)
 	memberPath := migrationPath + "/subgraphs/f/members"
-	moveCommand := response{"source_board_id": membersSource, "target_board_id": membersTarget, "node_ids": []string{"a"}, "expected_versions": map[string]int64{parentID: 5, membersSource: 1, membersTarget: 1}}
+	moveCommand := response{"source_board_id": membersSource, "target_board_id": membersTarget, "node_ids": []string{"a"}, "positions": map[string]memberDropPosition{"a": {X: 40, Y: 50}}, "expected_versions": map[string]int64{parentID: 5, membersSource: 1, membersTarget: 1}}
+	call("POST", memberPath, tokens[0], response{"source_board_id": membersSource, "target_board_id": membersTarget, "node_ids": []string{"a"}, "positions": response{"wrong": response{"x": 1, "y": 2}}, "expected_versions": map[string]int64{parentID: 5, membersSource: 1, membersTarget: 1}}, 400, "", accept, "eeeeeeee-1111-4111-8111-111111111111")
 	movedMembers := call("POST", memberPath, tokens[0], moveCommand, 200, "", accept, "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb")
 	memberRetry := call("POST", memberPath, tokens[0], moveCommand, 200, "", accept, "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb")
 	if movedMembers["transaction_id"] != memberRetry["transaction_id"] {
@@ -432,6 +436,10 @@ func TestOnboardingAndSharing(t *testing.T) {
 	targetRead := call("GET", targetCanvasPath, tokens[0], nil, 200, "", accept)
 	sourceJSON, _ := json.Marshal(sourceRead["document"])
 	targetJSON, _ := json.Marshal(targetRead["document"])
+	dropPosition := canvasNodesByID(targetJSON)["a"]["position"].(map[string]any)
+	if dropPosition["x"] != float64(140) || dropPosition["y"] != float64(250) {
+		t.Fatalf("atomic drop geometry %v", dropPosition)
+	}
 	if canvasNodeData(sourceJSON)["a"] != nil || canvasNodeData(targetJSON)["a"] == nil {
 		t.Fatal("member move duplicated or lost ownership")
 	}
@@ -530,6 +538,96 @@ func TestOnboardingAndSharing(t *testing.T) {
 			t.Fatal("asset bytes changed")
 		}
 	}
+	// Warm the immutable preview cache, then verify it never bypasses permission.
+	for _, sample := range []struct {
+		token  string
+		status int
+	}{{tokens[0], 200}, {tokens[1], 404}, {tokens[2], 404}} {
+		request := httptest.NewRequest("GET", workspacePath+"/assets/"+uploaded["id"].(string)+"?variant=preview", nil)
+		request.Header.Set("Authorization", "Bearer "+sample.token)
+		result := httptest.NewRecorder()
+		mux.ServeHTTP(result, request)
+		if result.Code != sample.status {
+			t.Fatalf("preview permission got %d want %d", result.Code, sample.status)
+		}
+		if result.Code == 200 && (!bytes.Equal(result.Body.Bytes(), assetPNG.Bytes()) || result.Header().Get("Cache-Control") != "private, no-store") {
+			t.Fatal("preview changed small original or privacy headers")
+		}
+	}
+	// Neither a snapshot nor an asset response may keep the workspace locked
+	// while a slow downstream consumer is accepting its body.
+	for _, path := range []string{canvasPath, workspacePath + "/assets/" + uploaded["id"].(string)} {
+		var probeErr error
+		writer := &readLockProbe{ResponseRecorder: httptest.NewRecorder(), probe: func() {
+			lockTx, e := db.BeginTx(ctx, nil)
+			if e != nil {
+				probeErr = e
+				return
+			}
+			defer lockTx.Rollback()
+			_, probeErr = lockTx.ExecContext(ctx, "SELECT id FROM workspaces WHERE project=$1 AND id=$2 FOR UPDATE NOWAIT", project, workspace)
+		}}
+		request := httptest.NewRequest("GET", path, nil)
+		request.Header.Set("Authorization", "Bearer "+tokens[0])
+		request.Header.Set("Accept", accept)
+		mux.ServeHTTP(writer, request)
+		if writer.Code != 200 || probeErr != nil {
+			t.Fatalf("read held workspace lock during response: status=%d error=%v", writer.Code, probeErr)
+		}
+	}
+	// A slow multipart body must not hold the workspace lock either.
+	func() {
+		reader, pipe := io.Pipe()
+		defer reader.Close()
+		defer pipe.Close()
+		form := multipart.NewWriter(pipe)
+		headers := make(chan struct{})
+		release := make(chan struct{})
+		done := make(chan struct{})
+		var once sync.Once
+		unblock := func() { once.Do(func() { close(release) }) }
+		defer unblock()
+		request := httptest.NewRequest("POST", workspacePath+"/assets", reader)
+		request.Header.Set("Authorization", "Bearer "+tokens[0])
+		request.Header.Set("Content-Type", form.FormDataContentType())
+		request.Header.Set("Idempotency-Key", "99999999-9999-4999-8999-999999999998")
+		result := httptest.NewRecorder()
+		go func() { mux.ServeHTTP(result, request); close(done) }()
+		go func() {
+			part, e := form.CreateFormFile("image", "slow.png")
+			close(headers)
+			if e != nil {
+				return
+			}
+			<-release
+			_, _ = part.Write(assetPNG.Bytes())
+			_ = form.Close()
+			_ = pipe.Close()
+		}()
+		select {
+		case <-headers:
+		case <-time.After(5 * time.Second):
+			t.Fatal("upload did not consume headers")
+		}
+		lockTx, e := db.BeginTx(ctx, nil)
+		if e != nil {
+			t.Fatal(e)
+		}
+		_, e = lockTx.ExecContext(ctx, "SELECT id FROM workspaces WHERE project=$1 AND id=$2 FOR UPDATE NOWAIT", project, workspace)
+		_ = lockTx.Rollback()
+		unblock()
+		if e != nil {
+			t.Fatalf("upload held workspace lock while receiving body: %v", e)
+		}
+		select {
+		case <-done:
+		case <-time.After(5 * time.Second):
+			t.Fatal("upload did not finish")
+		}
+		if result.Code != 201 {
+			t.Fatalf("slow upload failed: %d", result.Code)
+		}
+	}()
 	download(tokens[0], 200)
 	download(tokens[1], 404)
 	download(tokens[2], 404)
@@ -668,3 +766,19 @@ func TestOnboardingAndSharing(t *testing.T) {
 		t.Fatalf("click accounting incorrect %v", err)
 	}
 }
+
+// A real SQL NOWAIT probe at the response-write boundary catches locks that
+// would otherwise remain invisible in a buffered httptest recorder.
+type readLockProbe struct {
+	*httptest.ResponseRecorder
+	probe func()
+}
+
+func (w *readLockProbe) Write(body []byte) (int, error) {
+	if w.probe != nil {
+		probe := w.probe
+		w.probe = nil
+		probe()
+	}
+	return w.ResponseRecorder.Write(body)
+}

internal/platform/member_drop.go

@@ -0,0 +1,73 @@
+package platform
+
+import (
+	"encoding/json"
+	"fmt"
+	"math"
+)
+
+// Positions are expressed in the source board's local coordinates. The existing
+// ownership transaction converts them into destination coordinates exactly once.
+type memberDropPosition struct {
+	X float64 `json:"x"`
+	Y float64 `json:"y"`
+}
+
+func validMemberDropPositions(ids []string, positions map[string]memberDropPosition) bool {
+	if positions == nil {
+		return true
+	} // Backward-compatible old clients.
+	if len(positions) != len(ids) {
+		return false
+	}
+	seen := map[string]bool{}
+	for _, id := range ids {
+		p, ok := positions[id]
+		if !ok || seen[id] || math.IsNaN(p.X) || math.IsNaN(p.Y) || math.IsInf(p.X, 0) || math.IsInf(p.Y, 0) || math.Abs(p.X) > 1e6 || math.Abs(p.Y) > 1e6 {
+			return false
+		}
+		seen[id] = true
+	}
+	return true
+}
+
+func positionMemberDrop(raw json.RawMessage, positions map[string]memberDropPosition) (json.RawMessage, error) {
+	var doc map[string]any
+	if err := json.Unmarshal(raw, &doc); err != nil {
+		return nil, err
+	}
+	found := 0
+	nodes, ok := doc["nodes"].([]any)
+	if !ok {
+		return nil, errGraphOwnership
+	}
+	for _, v := range nodes {
+		n, ok := v.(map[string]any)
+		if !ok {
+			return nil, errGraphOwnership
+		}
+		id, _ := n["id"].(string)
+		if p, ok := positions[id]; ok {
+			n["position"] = map[string]any{"x": p.X, "y": p.Y}
+			found++
+		}
+	}
+	if found != len(positions) {
+		return nil, errGraphOwnership
+	}
+	return json.Marshal(doc)
+}
+
+func (p *memberDropPosition) UnmarshalJSON(raw []byte) error {
+	var fields map[string]json.RawMessage
+	if err := json.Unmarshal(raw, &fields); err != nil {
+		return err
+	}
+	if len(fields) != 2 || fields["x"] == nil || fields["y"] == nil || string(fields["x"]) == "null" || string(fields["y"]) == "null" {
+		return fmt.Errorf("invalid member position")
+	}
+	if err := json.Unmarshal(fields["x"], &p.X); err != nil {
+		return err
+	}
+	return json.Unmarshal(fields["y"], &p.Y)
+}

internal/platform/member_drop_test.go

@@ -0,0 +1,48 @@
+package platform
+
+import (
+	"encoding/json"
+	"math"
+	"testing"
+)
+
+func TestMemberDropPositions(t *testing.T) {
+	raw := memberFixture(t)
+	source := "11111111-1111-4111-8111-111111111111"
+	target := "22222222-2222-4222-8222-222222222222"
+	original := string(raw[source])
+	positions := map[string]memberDropPosition{"a": {X: 40, Y: 50}}
+	positioned, err := positionMemberDrop(raw[source], positions)
+	if err != nil {
+		t.Fatal(err)
+	}
+	raw[source] = positioned
+	moved, err := moveMembersDocuments("parent", source, target, []string{"a"}, raw)
+	if err != nil {
+		t.Fatal(err)
+	}
+	pos := canvasNodesByID(moved[target])["a"]["position"].(map[string]any)
+	if pos["x"] != float64(140) || pos["y"] != float64(250) {
+		t.Fatalf("final drop position %v", pos)
+	}
+	if original == string(positioned) {
+		t.Fatal("position not applied")
+	}
+	if _, err := positionMemberDrop(positioned, map[string]memberDropPosition{"missing": {}}); err == nil {
+		t.Fatal("missing member accepted")
+	}
+	if !validMemberDropPositions([]string{"a"}, nil) || !validMemberDropPositions([]string{"a"}, positions) {
+		t.Fatal("valid positions rejected")
+	}
+	for _, bad := range []map[string]memberDropPosition{{}, {"other": {}}, {"a": {X: 1e6 + 1}}, {"a": {Y: math.NaN()}}, {"a": {X: math.Inf(1)}}} {
+		if validMemberDropPositions([]string{"a"}, bad) {
+			t.Fatalf("bad positions accepted %v", bad)
+		}
+	}
+	for _, bad := range []string{`{"x":1}`, `{"x":null,"y":1}`, `{"x":1,"y":2,"z":3}`} {
+		var p memberDropPosition
+		if json.Unmarshal([]byte(bad), &p) == nil {
+			t.Fatal("malformed position accepted")
+		}
+	}
+}

internal/platform/subgraph_members_http.go

@@ -22,10 +22,11 @@ func (s *service) moveSubgraphMembers(w http.ResponseWriter, r *http.Request, pr
 		return nil
 	}
 	var input struct {
-		Source   string           `json:"source_board_id"`
-		Target   string           `json:"target_board_id"`
-		Nodes    []string         `json:"node_ids"`
-		Expected map[string]int64 `json:"expected_versions"`
+		Source    string                        `json:"source_board_id"`
+		Target    string                        `json:"target_board_id"`
+		Nodes     []string                      `json:"node_ids"`
+		Expected  map[string]int64              `json:"expected_versions"`
+		Positions map[string]memberDropPosition `json:"positions,omitempty"`
 	}
 	if !decode(w, r, &input) {
 		return nil
@@ -40,6 +41,10 @@ func (s *service) moveSubgraphMembers(w http.ResponseWriter, r *http.Request, pr
 			return nil
 		}
 	}
+	if !validMemberDropPositions(input.Nodes, input.Positions) {
+		reject(w, 400, "invalid_subgraph_request")
+		return nil
+	}
 	if len(input.Expected) == 0 {
 		reject(w, 428, "version_required")
 		return nil
@@ -194,6 +199,14 @@ func (s *service) moveSubgraphMembers(w http.ResponseWriter, r *http.Request, pr
 		before = append(before, snapshot)
 	}
 	raw := map[string]json.RawMessage{parent: snapshots[parent].Document, input.Source: snapshots[input.Source].Document, input.Target: snapshots[input.Target].Document}
+	if input.Positions != nil {
+		positioned, err := positionMemberDrop(raw[input.Source], input.Positions)
+		if err != nil {
+			reject(w, 409, "ownership_conflict")
+			return nil
+		}
+		raw[input.Source] = positioned
+	}
 	moved, err := moveMembersDocuments(parent, input.Source, input.Target, input.Nodes, raw)
 	if err != nil {
 		if err == errGraphOwnership {

internal/research/http.go

@@ -0,0 +1,210 @@
+// Package research serves the shared, public paper corpus. Account data and
+// private collections do not belong to these anonymous read-only endpoints.
+package research
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"net/url"
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+	"unicode/utf8"
+)
+
+type Source struct {
+	Label      string `json:"label"`
+	URL        string `json:"url"`
+	ExternalID string `json:"externalId"`
+}
+
+type Publication struct {
+	Venue       string `json:"venue"`
+	Year        int    `json:"year"`
+	Track       string `json:"track"`
+	Status      string `json:"status"`
+	EvidenceURL string `json:"evidenceUrl,omitempty"`
+}
+
+type Paper struct {
+	ID           string        `json:"id"`
+	Title        string        `json:"title"`
+	Abstract     *string       `json:"abstract"`
+	Authors      []string      `json:"authors"`
+	PublishedAt  *string       `json:"publishedAt"`
+	Topics       []string      `json:"topics"`
+	Sources      []Source      `json:"sources"`
+	Publications []Publication `json:"publications,omitempty"`
+	DOI          *string       `json:"doi,omitempty"`
+	Journal      *string       `json:"journal,omitempty"`
+	Comments     *string       `json:"comments,omitempty"`
+}
+
+type Filters struct {
+	Query    string
+	SearchIn string
+	Topic    string
+	Sort     string
+	Page     int
+	PageSize int
+}
+
+type Page struct {
+	Items    []Paper  `json:"items"`
+	Total    int64    `json:"total"`
+	Topics   []string `json:"topics"`
+	Page     int      `json:"page"`
+	PageSize int      `json:"pageSize"`
+}
+
+// Repository implementations must use a read-only transaction for each List,
+// so its count, topics and page describe the same database snapshot.
+type Repository interface {
+	List(context.Context, Filters) (Page, error)
+	Get(context.Context, int64) (Paper, error)
+}
+
+type service struct {
+	repository Repository
+	projects   map[string]string
+	slots      chan struct{}
+	timeout    time.Duration
+}
+
+// Register adds only GET endpoints; no credentials or database errors are sent
+// to anonymous callers. A nil repository leaves the existing server unchanged.
+func Register(mux *http.ServeMux, repository Repository, projects map[string]string) {
+	if repository == nil {
+		return
+	}
+	s := &service{repository: repository, projects: projects, slots: make(chan struct{}, 2), timeout: 8 * time.Second}
+	mux.HandleFunc("GET /projects/{project}/research/papers", s.serve)
+	mux.HandleFunc("GET /projects/{project}/research/papers/{id}", s.serve)
+}
+
+func reply(w http.ResponseWriter, status int, value any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.Header().Set("Cache-Control", "no-store")
+	w.Header().Set("X-Content-Type-Options", "nosniff")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(value)
+}
+
+func reject(w http.ResponseWriter, status int, code string) {
+	reply(w, status, map[string]string{"error": code})
+}
+
+var topicPattern = regexp.MustCompile(`^[A-Za-z][A-Za-z0-9.-]{0,79}$`)
+var positiveInteger = regexp.MustCompile(`^[1-9][0-9]*$`)
+
+func parseFilters(raw string) (Filters, error) {
+	f := Filters{SearchIn: "text", Sort: "recent", Page: 1, PageSize: 8}
+	values, err := url.ParseQuery(raw)
+	if err != nil {
+		return f, err
+	}
+	invalid := errors.New("invalid query")
+	for key, value := range values {
+		if len(value) != 1 {
+			return f, invalid
+		}
+		switch key {
+		case "q":
+			if !utf8.ValidString(value[0]) || utf8.RuneCountInString(value[0]) > 200 || len(strings.Fields(value[0])) > 20 || strings.ContainsRune(value[0], 0) {
+				return f, invalid
+			}
+			f.Query = strings.TrimSpace(value[0])
+		case "searchIn":
+			if value[0] != "text" && value[0] != "author" {
+				return f, invalid
+			}
+			f.SearchIn = value[0]
+		case "topic":
+			if value[0] != "" && !topicPattern.MatchString(value[0]) {
+				return f, invalid
+			}
+			f.Topic = value[0]
+		case "sort":
+			if value[0] != "recent" && value[0] != "title" {
+				return f, invalid
+			}
+			f.Sort = value[0]
+		case "page", "pageSize":
+			if !positiveInteger.MatchString(value[0]) {
+				return f, invalid
+			}
+			n, err := strconv.Atoi(value[0])
+			limit := 1000
+			if key == "pageSize" {
+				limit = 50
+			}
+			if err != nil || n > limit {
+				return f, invalid
+			}
+			if key == "page" {
+				f.Page = n
+			} else {
+				f.PageSize = n
+			}
+		default:
+			return f, invalid
+		}
+	}
+	return f, nil
+}
+
+func (s *service) serve(w http.ResponseWriter, r *http.Request) {
+	if _, exists := s.projects[r.PathValue("project")]; !exists {
+		reject(w, 404, "not_found")
+		return
+	}
+	var filters Filters
+	var id int64
+	var err error
+	if value := r.PathValue("id"); value != "" {
+		if !positiveInteger.MatchString(value) || r.URL.RawQuery != "" {
+			reject(w, 400, "invalid_request")
+			return
+		}
+		id, err = strconv.ParseInt(value, 10, 64)
+	} else {
+		filters, err = parseFilters(r.URL.RawQuery)
+	}
+	if err != nil {
+		reject(w, 400, "invalid_request")
+		return
+	}
+	select {
+	case s.slots <- struct{}{}:
+		defer func() { <-s.slots }()
+	default:
+		w.Header().Set("Retry-After", "1")
+		reject(w, 429, "research_busy")
+		return
+	}
+	ctx, cancel := context.WithTimeout(r.Context(), s.timeout)
+	defer cancel()
+	var result any
+	if id != 0 {
+		result, err = s.repository.Get(ctx, id)
+	} else {
+		result, err = s.repository.List(ctx, filters)
+	}
+	if err == nil {
+		reply(w, 200, result)
+		return
+	}
+	var state interface{ SQLState() string }
+	switch {
+	case errors.Is(err, sql.ErrNoRows):
+		reject(w, 404, "not_found")
+	case errors.Is(ctx.Err(), context.DeadlineExceeded), errors.Is(err, context.DeadlineExceeded), errors.As(err, &state) && state.SQLState() == "57014":
+		reject(w, 504, "research_timeout")
+	default:
+		reject(w, 503, "research_unavailable")
+	}
+}

internal/research/http_test.go

@@ -0,0 +1,145 @@
+package research
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+	"time"
+)
+
+type fixtureRepository struct {
+	filters Filters
+	id      int64
+	err     error
+	wait    bool
+	calls   int
+}
+
+func (f *fixtureRepository) List(ctx context.Context, filters Filters) (Page, error) {
+	f.calls++
+	f.filters = filters
+	if f.wait {
+		<-ctx.Done()
+		return Page{}, ctx.Err()
+	}
+	return Page{Items: []Paper{}, Topics: []string{}, Page: filters.Page, PageSize: filters.PageSize}, f.err
+}
+
+func (f *fixtureRepository) Get(ctx context.Context, id int64) (Paper, error) {
+	f.calls++
+	f.id = id
+	return Paper{ID: "42", Title: "Original title", Authors: []string{}, Topics: []string{}, Sources: []Source{}}, f.err
+}
+
+func request(handler http.Handler, method, path string) *httptest.ResponseRecorder {
+	w := httptest.NewRecorder()
+	handler.ServeHTTP(w, httptest.NewRequest(method, path, nil))
+	return w
+}
+
+func TestAnonymousReadRoutes(t *testing.T) {
+	f := &fixtureRepository{}
+	mux := http.NewServeMux()
+	Register(mux, f, map[string]string{"thesisly": "http://localhost:17322/auth/callback"})
+	w := request(mux, "GET", "/projects/thesisly/research/papers?q=graph&searchIn=author&sort=title&topic=cs.LG&page=2&pageSize=10")
+	if w.Code != 200 || f.filters != (Filters{Query: "graph", SearchIn: "author", Topic: "cs.LG", Sort: "title", Page: 2, PageSize: 10}) {
+		t.Fatalf("list: %d %+v", w.Code, f.filters)
+	}
+	var page Page
+	if err := json.Unmarshal(w.Body.Bytes(), &page); err != nil || page.Items == nil || page.Topics == nil {
+		t.Fatalf("invalid page: %s", w.Body.String())
+	}
+	if w.Header().Get("Cache-Control") != "no-store" || w.Header().Get("Content-Type") != "application/json" {
+		t.Fatal("missing response headers")
+	}
+	w = request(mux, "GET", "/projects/thesisly/research/papers/42")
+	if w.Code != 200 || f.id != 42 || !strings.Contains(w.Body.String(), `"abstract":null`) {
+		t.Fatalf("detail: %d %s", w.Code, w.Body.String())
+	}
+	if got := request(mux, "POST", "/projects/thesisly/research/papers").Code; got != 405 {
+		t.Fatalf("mutation status %d", got)
+	}
+	if got := request(mux, "GET", "/projects/other/research/papers").Code; got != 404 {
+		t.Fatalf("unknown project status %d", got)
+	}
+}
+
+func TestInvalidQueriesNeverReachRepository(t *testing.T) {
+	queries := []string{"unknown=1", "q=x&q=y", "q=%ff", "q=%00", "q=%zz", "page=0", "page=1001", "page=-1", "page=1.5", "page=01", "pageSize=51", "pageSize=0", "pageSize=9999999999999999999999", "searchIn=sql", "sort=published", "topic=arxiv:cs.LG", "q=" + url.QueryEscape(strings.Repeat("字", 201)), "q=" + url.QueryEscape(strings.Repeat("word ", 21))}
+	for _, query := range queries {
+		t.Run(query, func(t *testing.T) {
+			f := &fixtureRepository{}
+			mux := http.NewServeMux()
+			Register(mux, f, map[string]string{"thesisly": "unused"})
+			w := request(mux, "GET", "/projects/thesisly/research/papers?"+query)
+			if w.Code != 400 || f.calls != 0 {
+				t.Fatalf("status %d, calls %d", w.Code, f.calls)
+			}
+		})
+	}
+	for _, path := range []string{"0", "-1", "1x", "9223372036854775808", "1?extra=1"} {
+		mux := http.NewServeMux()
+		f := &fixtureRepository{}
+		Register(mux, f, map[string]string{"thesisly": "unused"})
+		if w := request(mux, "GET", "/projects/thesisly/research/papers/"+path); w.Code != 400 || f.calls != 0 {
+			t.Fatalf("invalid id %q: %d", path, w.Code)
+		}
+	}
+}
+
+func TestQueryDefaultsAndUnicode(t *testing.T) {
+	f, err := parseFilters("q=" + url.QueryEscape(" 论文 Graph "))
+	if err != nil || f.Query != "论文 Graph" || f.Page != 1 || f.PageSize != 8 || f.SearchIn != "text" || f.Sort != "recent" {
+		t.Fatalf("%+v %v", f, err)
+	}
+}
+
+type databaseFailure string
+
+func (failure databaseFailure) Error() string    { return "private database error" }
+func (failure databaseFailure) SQLState() string { return string(failure) }
+
+func TestStorageErrorsAreRedacted(t *testing.T) {
+	for _, test := range []struct {
+		err    error
+		status int
+		code   string
+	}{
+		{sql.ErrNoRows, 404, "not_found"},
+		{context.DeadlineExceeded, 504, "research_timeout"},
+		{errors.Join(databaseFailure("57014"), errors.New("private statement")), 504, "research_timeout"},
+		{databaseFailure("42501"), 503, "research_unavailable"},
+		{errors.New("postgres://private-user:private-password@db/corpus"), 503, "research_unavailable"},
+	} {
+		mux := http.NewServeMux()
+		Register(mux, &fixtureRepository{err: test.err}, map[string]string{"thesisly": "unused"})
+		w := request(mux, "GET", "/projects/thesisly/research/papers/42")
+		if w.Code != test.status || !strings.Contains(w.Body.String(), test.code) || strings.Contains(w.Body.String(), "private") {
+			t.Fatalf("%d %s", w.Code, w.Body.String())
+		}
+	}
+}
+
+func TestConcurrencyAndTimeoutReleaseSlots(t *testing.T) {
+	s := &service{repository: &fixtureRepository{wait: true}, projects: map[string]string{"thesisly": "unused"}, slots: make(chan struct{}, 2), timeout: time.Millisecond}
+	mux := http.NewServeMux()
+	mux.HandleFunc("GET /projects/{project}/research/papers", s.serve)
+	s.slots <- struct{}{}
+	s.slots <- struct{}{}
+	w := request(mux, "GET", "/projects/thesisly/research/papers")
+	if w.Code != 429 || w.Header().Get("Retry-After") != "1" {
+		t.Fatalf("busy %d", w.Code)
+	}
+	<-s.slots
+	<-s.slots
+	w = request(mux, "GET", "/projects/thesisly/research/papers")
+	if w.Code != 504 || len(s.slots) != 0 {
+		t.Fatalf("timeout %d, slots %d", w.Code, len(s.slots))
+	}
+}

internal/research/postgres.go

@@ -0,0 +1,389 @@
+package research
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"regexp"
+	"strconv"
+	"strings"
+)
+
+// The fixed Paper Atlas migration a3b677b preserves business columns as TEXT
+// and BIGINT and creates the English title/abstract GIN expression below.
+// No schema creation, indexing or other writes belong to this repository.
+type PostgresRepository struct {
+	db      *sql.DB
+	prefix  string
+	schema  string
+	encoded bool
+}
+
+var schemaName = regexp.MustCompile(`^[a-z][a-z0-9_]{0,35}$`)
+
+func NewPostgresRepository(db *sql.DB, schema string) (*PostgresRepository, error) {
+	if db == nil || !schemaName.MatchString(schema) {
+		return nil, errors.New("invalid research database configuration")
+	}
+	return &PostgresRepository{db: db, prefix: `"` + schema + `".`, schema: schema}, nil
+}
+
+func (r *PostgresRepository) query(value string) string {
+	return strings.ReplaceAll(value, "{schema}", r.prefix)
+}
+
+func (r *PostgresRepository) begin(ctx context.Context) (*sql.Tx, error) {
+	tx, err := r.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelRepeatableRead, ReadOnly: true})
+	if err != nil {
+		return nil, err
+	}
+	_, err = tx.ExecContext(ctx, `SELECT set_config('statement_timeout','8000',true),
+		set_config('lock_timeout','1000',true), set_config('work_mem','2MB',true),
+		set_config('max_parallel_workers_per_gather','0',true), set_config('jit','off',true)`)
+	if err != nil {
+		_ = tx.Rollback()
+		return nil, err
+	}
+	return tx, nil
+}
+
+// Check resolves required columns and SELECT permissions without scanning the
+// corpus. The migration publishes the complete schema atomically.
+func (r *PostgresRepository) Check(ctx context.Context) error {
+	tx, err := r.begin(ctx)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	var annotation sql.NullString
+	err = tx.QueryRowContext(ctx, `SELECT pg_catalog.obj_description(oid,'pg_namespace')
+		FROM pg_catalog.pg_namespace WHERE nspname=$1`, r.schema).Scan(&annotation)
+	if err != nil {
+		return err
+	}
+	encoded, err := encodedSchema(annotation)
+	if err != nil {
+		return err
+	}
+	rows, err := tx.QueryContext(ctx, r.query(`SELECT p.id,p.title,p.abstract,p.authors_json,p.first_published,p.title_key,
+		r.external_id,r.url,r.metadata_json,so.name,i.scheme,i.value,t.namespace,t.value,
+		s.id,e.year,e.track,v.acronym,d.status,d.evidence_url,d.observed_at,d.decision_date
+		FROM {schema}papers p LEFT JOIN {schema}source_records r ON r.paper_id=p.id
+		LEFT JOIN {schema}sources so ON so.id=r.source_id LEFT JOIN {schema}identifiers i ON i.paper_id=p.id
+		LEFT JOIN {schema}paper_tags pt ON pt.paper_id=p.id LEFT JOIN {schema}tags t ON t.id=pt.tag_id
+		LEFT JOIN {schema}submissions s ON s.paper_id=p.id LEFT JOIN {schema}editions e ON e.id=s.edition_id
+		LEFT JOIN {schema}venues v ON v.id=e.venue_id LEFT JOIN {schema}decisions d ON d.submission_id=s.id WHERE false`))
+	if err != nil {
+		return err
+	}
+	if err = rows.Close(); err != nil {
+		return err
+	}
+	if err = tx.Commit(); err != nil {
+		return err
+	}
+	r.encoded = encoded
+	return nil
+}
+
+func whereClause(filters Filters) (string, []any) {
+	clauses, values := []string{}, []any{}
+	if filters.Query != "" {
+		values = append(values, filters.Query)
+		if filters.SearchIn == "author" {
+			// JSON stores author strings or [family,given,suffix] tuples. Plain
+			// full-text terms search names regardless of tuple display order.
+			clauses = append(clauses, `to_tsvector('simple'::regconfig,p.authors_json) @@ plainto_tsquery('simple'::regconfig,$1)`)
+		} else {
+			clauses = append(clauses, `to_tsvector('english'::regconfig,coalesce(p.title,'') || ' ' || coalesce(p.abstract,'')) @@ plainto_tsquery('english'::regconfig,$1)`)
+		}
+	}
+	if filters.Topic != "" {
+		values = append(values, filters.Topic)
+		clauses = append(clauses, fmt.Sprintf(`EXISTS(SELECT 1 FROM {schema}paper_tags pt
+			JOIN {schema}tags t ON t.id=pt.tag_id
+			WHERE pt.paper_id=p.id AND t.namespace='arxiv:category' AND t.value=$%d)`, len(values)))
+	}
+	if len(clauses) == 0 {
+		return "", values
+	}
+	return " WHERE " + strings.Join(clauses, " AND "), values
+}
+
+type scanner interface{ Scan(...any) error }
+
+func (r *PostgresRepository) scanPaper(row scanner) (Paper, error) {
+	p := Paper{Authors: []string{}, Topics: []string{}, Sources: []Source{}}
+	var id int64
+	var abstract, published sql.NullString
+	var authors string
+	if err := row.Scan(&id, r.text(&p.Title), r.text(&abstract), r.text(&authors), r.text(&published)); err != nil {
+		return p, err
+	}
+	p.ID = strconv.FormatInt(id, 10)
+	if abstract.Valid {
+		p.Abstract = &abstract.String
+	}
+	if published.Valid {
+		p.PublishedAt = &published.String
+	}
+	var err error
+	p.Authors, err = decodeAuthors(authors)
+	return p, err
+}
+
+func decodeAuthors(raw string) ([]string, error) {
+	var entries []json.RawMessage
+	if err := json.Unmarshal([]byte(raw), &entries); err != nil {
+		return nil, errors.New("invalid stored authors")
+	}
+	names := []string{}
+	for _, entry := range entries {
+		var name string
+		if json.Unmarshal(entry, &name) == nil {
+			if name = strings.TrimSpace(name); name != "" {
+				names = append(names, name)
+			}
+			continue
+		}
+		var tuple []string
+		if json.Unmarshal(entry, &tuple) != nil || len(tuple) == 0 || len(tuple) > 4 {
+			return nil, errors.New("unsupported stored author")
+		}
+		// arXiv may append an affiliation after family, given and suffix.
+		// It is source metadata, not part of the display name.
+		if len(tuple) == 4 {
+			tuple = tuple[:3]
+		}
+		if len(tuple) >= 2 {
+			tuple[0], tuple[1] = tuple[1], tuple[0]
+		}
+		parts := []string{}
+		for _, part := range tuple {
+			if part = strings.TrimSpace(part); part != "" {
+				parts = append(parts, part)
+			}
+		}
+		if name = strings.Join(parts, " "); name != "" {
+			names = append(names, name)
+		}
+	}
+	return names, nil
+}
+
+func (r *PostgresRepository) List(ctx context.Context, filters Filters) (Page, error) {
+	result := Page{Items: []Paper{}, Topics: []string{}, Page: filters.Page, PageSize: filters.PageSize}
+	tx, err := r.begin(ctx)
+	if err != nil {
+		return result, err
+	}
+	defer tx.Rollback()
+	if r.encoded {
+		filters.Query = encodeText(filters.Query)
+		filters.Topic = encodeText(filters.Topic)
+	}
+	where, args := whereClause(filters)
+	if err = tx.QueryRowContext(ctx, r.query(`SELECT count(*) FROM {schema}papers p`+where), args...).Scan(&result.Total); err != nil {
+		return result, err
+	}
+	order := "p.id DESC"
+	if filters.Sort == "title" {
+		order = "p.title_key,p.id"
+	}
+	args = append(args, filters.PageSize, (filters.Page-1)*filters.PageSize)
+	// Select IDs before loading potentially TOASTed titles/abstracts. With
+	// sparse categories, selecting full rows here can make the planner walk
+	// hundreds of thousands of wide heap rows before finding eight matches.
+	candidateQuery := `SELECT p.id FROM {schema}papers p`
+	candidateOrder := order
+	prefix := ""
+	if where != "" {
+		columns := "p.id"
+		if filters.Sort == "title" {
+			columns += ",p.title_key"
+		}
+		// Apply search before LIMIT/ORDER BY so a sparse term cannot turn
+		// into a backward heap scan that recomputes vectors for millions.
+		prefix = fmt.Sprintf(`matches AS MATERIALIZED (SELECT %s FROM {schema}papers p%s), `, columns, where)
+		candidateQuery = "SELECT id FROM matches"
+		candidateOrder = strings.ReplaceAll(order, "p.", "")
+	}
+	query := fmt.Sprintf(`WITH %spage_ids AS MATERIALIZED (
+		%s ORDER BY %s LIMIT $%d OFFSET $%d
+		) SELECT p.id,p.title,p.abstract,p.authors_json,p.first_published
+		FROM page_ids JOIN {schema}papers p ON p.id=page_ids.id ORDER BY %s`, prefix, candidateQuery, candidateOrder, len(args)-1, len(args), order)
+	rows, err := tx.QueryContext(ctx, r.query(query), args...)
+	if err != nil {
+		return result, err
+	}
+	for rows.Next() {
+		p, err := r.scanPaper(rows)
+		if err != nil {
+			_ = rows.Close()
+			return result, err
+		}
+		result.Items = append(result.Items, p)
+	}
+	err = rows.Err()
+	_ = rows.Close()
+	if err != nil {
+		return result, err
+	}
+	if err = r.enrich(ctx, tx, result.Items); err != nil {
+		return result, err
+	}
+	rows, err = tx.QueryContext(ctx, r.query(`SELECT value FROM {schema}tags WHERE namespace='arxiv:category' ORDER BY value`))
+	if err != nil {
+		return result, err
+	}
+	for rows.Next() {
+		var topic string
+		if err = rows.Scan(r.text(&topic)); err != nil {
+			_ = rows.Close()
+			return result, err
+		}
+		result.Topics = append(result.Topics, topic)
+	}
+	err = rows.Err()
+	_ = rows.Close()
+	if err != nil {
+		return result, err
+	}
+	return result, tx.Commit()
+}
+
+func (r *PostgresRepository) Get(ctx context.Context, id int64) (Paper, error) {
+	tx, err := r.begin(ctx)
+	if err != nil {
+		return Paper{}, err
+	}
+	defer tx.Rollback()
+	p, err := r.scanPaper(tx.QueryRowContext(ctx, r.query(`SELECT id,title,abstract,authors_json,first_published FROM {schema}papers WHERE id=$1`), id))
+	if err != nil {
+		return p, err
+	}
+	papers := []Paper{p}
+	if err = r.enrich(ctx, tx, papers); err != nil {
+		return p, err
+	}
+	return papers[0], tx.Commit()
+}
+
+// Batch metadata queries avoid per-paper round trips and share the count/page
+// snapshot. User values are bound; only the validated schema is interpolated.
+func (r *PostgresRepository) enrich(ctx context.Context, tx *sql.Tx, papers []Paper) error {
+	if len(papers) == 0 {
+		return nil
+	}
+	ids := make([]int64, len(papers))
+	byID := map[int64]*Paper{}
+	for i := range papers {
+		id, err := strconv.ParseInt(papers[i].ID, 10, 64)
+		if err != nil {
+			return err
+		}
+		ids[i] = id
+		byID[id] = &papers[i]
+	}
+	rows, err := tx.QueryContext(ctx, r.query(`SELECT DISTINCT pt.paper_id,t.value
+		FROM {schema}paper_tags pt JOIN {schema}tags t ON t.id=pt.tag_id
+		WHERE pt.paper_id=ANY($1) AND t.namespace='arxiv:category' ORDER BY pt.paper_id,t.value`), ids)
+	if err != nil {
+		return err
+	}
+	for rows.Next() {
+		var id int64
+		var topic string
+		if err = rows.Scan(&id, r.text(&topic)); err != nil {
+			_ = rows.Close()
+			return err
+		}
+		byID[id].Topics = append(byID[id].Topics, topic)
+	}
+	err = rows.Err()
+	_ = rows.Close()
+	if err != nil {
+		return err
+	}
+	rows, err = tx.QueryContext(ctx, r.query(`SELECT r.paper_id,s.name,r.external_id,r.url,r.metadata_json
+		FROM {schema}source_records r JOIN {schema}sources s ON s.id=r.source_id
+		WHERE r.paper_id=ANY($1) ORDER BY r.paper_id,r.id`), ids)
+	if err != nil {
+		return err
+	}
+	for rows.Next() {
+		var id int64
+		var source Source
+		var metadata string
+		if err = rows.Scan(&id, r.text(&source.Label), r.text(&source.ExternalID), r.text(&source.URL), r.text(&metadata)); err != nil {
+			_ = rows.Close()
+			return err
+		}
+		p := byID[id]
+		switch source.Label {
+		case "arxiv":
+			source.Label = "arXiv"
+		case "openreview":
+			source.Label = "OpenReview"
+		case "pmlr":
+			source.Label = "PMLR"
+		}
+		p.Sources = append(p.Sources, source)
+		var fields map[string]json.RawMessage
+		if json.Unmarshal([]byte(metadata), &fields) == nil {
+			for key, target := range map[string]**string{"doi": &p.DOI, "journal-ref": &p.Journal, "comments": &p.Comments} {
+				var value string
+				if *target == nil && json.Unmarshal(fields[key], &value) == nil && strings.TrimSpace(value) != "" {
+					*target = &value
+				}
+			}
+		}
+	}
+	err = rows.Err()
+	_ = rows.Close()
+	if err != nil {
+		return err
+	}
+	rows, err = tx.QueryContext(ctx, r.query(`SELECT paper_id,value FROM {schema}identifiers WHERE paper_id=ANY($1) AND scheme='doi' ORDER BY paper_id,value`), ids)
+	if err != nil {
+		return err
+	}
+	for rows.Next() {
+		var id int64
+		var value string
+		if err = rows.Scan(&id, r.text(&value)); err != nil {
+			_ = rows.Close()
+			return err
+		}
+		if byID[id].DOI == nil && strings.TrimSpace(value) != "" {
+			byID[id].DOI = &value
+		}
+	}
+	err = rows.Err()
+	_ = rows.Close()
+	if err != nil {
+		return err
+	}
+	rows, err = tx.QueryContext(ctx, r.query(`SELECT s.paper_id,v.acronym,e.year,e.track,coalesce(d.status,'unknown'),coalesce(d.evidence_url,'')
+		FROM {schema}submissions s JOIN {schema}editions e ON e.id=s.edition_id JOIN {schema}venues v ON v.id=e.venue_id
+		LEFT JOIN LATERAL (SELECT status,evidence_url FROM {schema}decisions WHERE submission_id=s.id
+			ORDER BY observed_at DESC,coalesce(decision_date,observed_at) DESC,id DESC LIMIT 1) d ON true
+		WHERE s.paper_id=ANY($1) ORDER BY s.paper_id,e.year DESC,v.acronym,e.track,s.id`), ids)
+	if err != nil {
+		return err
+	}
+	for rows.Next() {
+		var id int64
+		var publication Publication
+		if err = rows.Scan(&id, r.text(&publication.Venue), &publication.Year, r.text(&publication.Track), r.text(&publication.Status), r.text(&publication.EvidenceURL)); err != nil {
+			_ = rows.Close()
+			return err
+		}
+		byID[id].Publications = append(byID[id].Publications, publication)
+	}
+	err = rows.Err()
+	_ = rows.Close()
+	return err
+}

internal/research/postgres_integration_test.go

@@ -0,0 +1,186 @@
+package research
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"os"
+	"reflect"
+	"testing"
+	"time"
+
+	_ "github.com/jackc/pgx/v5/stdlib"
+)
+
+func integrationRepository(t *testing.T) (*PostgresRepository, context.Context) {
+	t.Helper()
+	dsn := os.Getenv("RESEARCH_TEST_DATABASE_URL")
+	if dsn == "" {
+		t.Skip("requires the disposable research fixture database")
+	}
+	db, err := sql.Open("pgx", dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	db.SetMaxOpenConns(2)
+	repository, err := NewPostgresRepository(db, "research_fixture")
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	t.Cleanup(cancel)
+	return repository, ctx
+}
+
+func paperIDs(papers []Paper) []string {
+	ids := []string{}
+	for _, paper := range papers {
+		ids = append(ids, paper.ID)
+	}
+	return ids
+}
+
+func TestPostgresPageSearchAndFilters(t *testing.T) {
+	r, ctx := integrationRepository(t)
+	if err := r.Check(ctx); err != nil {
+		t.Fatal(err)
+	}
+	for _, test := range []struct {
+		filters Filters
+		total   int64
+		ids     []string
+	}{
+		{Filters{Page: 1, PageSize: 2}, 5, []string{"5", "4"}},
+		{Filters{Page: 2, PageSize: 2}, 5, []string{"3", "2"}},
+		{Filters{Page: 3, PageSize: 2}, 5, []string{"1"}},
+		{Filters{Page: 1, PageSize: 2, Sort: "title"}, 5, []string{"2", "5"}},
+		{Filters{Page: 1, PageSize: 8, Query: "graph"}, 2, []string{"3", "1"}},
+		{Filters{Page: 1, PageSize: 8, Query: "graph OR zebra"}, 0, []string{}},
+		{Filters{Page: 1, PageSize: 8, Query: "Ada Smith", SearchIn: "author"}, 1, []string{"1"}},
+		{Filters{Page: 1, PageSize: 8, Topic: "cs.LG"}, 1, []string{"1"}},
+		{Filters{Page: 1, PageSize: 8, Topic: "cs.DC", Query: "graph"}, 1, []string{"3"}},
+		{Filters{Page: 1, PageSize: 8, Query: `x'); DROP TABLE papers;--`}, 0, []string{}},
+	} {
+		page, err := r.List(ctx, test.filters)
+		if err != nil {
+			t.Fatalf("%+v: %v", test.filters, err)
+		}
+		if page.Total != test.total || !reflect.DeepEqual(paperIDs(page.Items), test.ids) {
+			t.Fatalf("%+v: total=%d ids=%v", test.filters, page.Total, paperIDs(page.Items))
+		}
+		if !reflect.DeepEqual(page.Topics, []string{"cs.AI", "cs.DC", "cs.LG"}) {
+			t.Fatalf("mixed tag namespaces: %v", page.Topics)
+		}
+		if page.Page != test.filters.Page || page.PageSize != test.filters.PageSize {
+			t.Fatal("lost paging metadata")
+		}
+	}
+}
+
+func TestPostgresMetadataPreservesEvidence(t *testing.T) {
+	r, ctx := integrationRepository(t)
+	p, err := r.Get(ctx, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if p.Title != "Graph & learning <original>" || p.Abstract == nil || *p.Abstract != "Neural graph models preserve original abstracts." || !reflect.DeepEqual(p.Authors, []string{"Ada Smith", "The Research Group"}) {
+		t.Fatalf("original content changed: %+v", p)
+	}
+	if !reflect.DeepEqual(p.Topics, []string{"cs.AI", "cs.LG"}) || len(p.Sources) != 1 || p.Sources[0].ExternalID != "fixture-1" || p.Sources[0].Label != "arXiv" {
+		t.Fatalf("incorrect source/tag joins: %+v", p)
+	}
+	if p.DOI == nil || *p.DOI != "10.fixture/canonical" || p.Journal == nil || *p.Journal != "Journal Fixture" || p.Comments == nil || *p.Comments != "Original comment" {
+		t.Fatalf("metadata missing: %+v", p)
+	}
+	if len(p.Publications) != 1 || p.Publications[0].Status != "withdrawn" || p.Publications[0].EvidenceURL != "https://example.test/decision/3" {
+		t.Fatalf("latest decision ordering changed: %+v", p.Publications)
+	}
+	p, err = r.Get(ctx, 3)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !reflect.DeepEqual(p.Authors, []string{"Carla Network"}) || len(p.Sources) != 2 || p.DOI == nil || *p.DOI != "10.fixture/other" || len(p.Publications) != 1 || p.Publications[0].Status != "unknown" || p.Publications[0].EvidenceURL != "" {
+		t.Fatalf("unknown decision/source mapping: %+v", p)
+	}
+	p, err = r.Get(ctx, 2)
+	if err != nil || p.Abstract != nil || p.PublishedAt != nil || p.Sources == nil || p.Topics == nil {
+		t.Fatalf("nullable data changed: %+v %v", p, err)
+	}
+	if _, err = r.Get(ctx, 99999); !errors.Is(err, sql.ErrNoRows) {
+		t.Fatalf("missing paper: %v", err)
+	}
+}
+
+func TestPostgresCorpusIsReadOnlyAndHTTPWorks(t *testing.T) {
+	r, ctx := integrationRepository(t)
+	tx, err := r.begin(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var readOnly, isolation, jit string
+	if err = tx.QueryRowContext(ctx, "SHOW transaction_read_only").Scan(&readOnly); err != nil {
+		t.Fatal(err)
+	}
+	if err = tx.QueryRowContext(ctx, "SHOW transaction_isolation").Scan(&isolation); err != nil {
+		t.Fatal(err)
+	}
+	if err = tx.QueryRowContext(ctx, "SHOW jit").Scan(&jit); err != nil {
+		t.Fatal(err)
+	}
+	_ = tx.Rollback()
+	if readOnly != "on" || isolation != "repeatable read" || jit != "off" {
+		t.Fatalf("unexpected read transaction settings: %s %s jit=%s", readOnly, isolation, jit)
+	}
+	_, err = r.db.ExecContext(ctx, `UPDATE research_fixture.papers SET title='unexpected write' WHERE id=1`)
+	var state interface{ SQLState() string }
+	if !errors.As(err, &state) || state.SQLState() != "42501" {
+		t.Fatalf("expected database permission denial, got %v", err)
+	}
+	mux := http.NewServeMux()
+	Register(mux, r, map[string]string{"thesisly": "http://localhost:17322/auth/callback"})
+	w := request(mux, "GET", "/projects/thesisly/research/papers?q=graph")
+	var page Page
+	if w.Code != 200 || json.Unmarshal(w.Body.Bytes(), &page) != nil || page.Total != 2 {
+		t.Fatalf("SQL-backed HTTP list: %d %s", w.Code, w.Body.String())
+	}
+	w = request(mux, "GET", "/projects/thesisly/research/papers/1")
+	var paper Paper
+	if w.Code != 200 || json.Unmarshal(w.Body.Bytes(), &paper) != nil || paper.ID != "1" {
+		t.Fatalf("SQL-backed HTTP detail: %d %s", w.Code, w.Body.String())
+	}
+	if w = request(mux, "GET", "/projects/thesisly/research/papers/99999"); w.Code != 404 {
+		t.Fatalf("missing HTTP status %d", w.Code)
+	}
+}
+
+func TestPostgresPublishedTextCodec(t *testing.T) {
+	legacy, ctx := integrationRepository(t)
+	r, err := NewPostgresRepository(legacy.db, "research_encoded_fixture")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = r.Check(ctx); err != nil {
+		t.Fatal(err)
+	}
+	p, err := r.Get(ctx, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if p.Title != "Graph\x00original\x01" || p.Abstract == nil || *p.Abstract != "Text\x00tail" || !reflect.DeepEqual(p.Authors, []string{"Ada\x00Smith", "Group\x01Name"}) || p.Comments == nil || *p.Comments != "Original\x00comment" {
+		t.Fatalf("published text was not restored: %+v", p)
+	}
+	mux := http.NewServeMux()
+	Register(mux, r, map[string]string{"thesisly": "http://localhost:17322/auth/callback"})
+	w := request(mux, "GET", "/projects/thesisly/research/papers/1")
+	var decoded Paper
+	if w.Code != 200 || json.Unmarshal(w.Body.Bytes(), &decoded) != nil || decoded.Title != p.Title || decoded.Abstract == nil || *decoded.Abstract != *p.Abstract {
+		t.Fatalf("HTTP lost original control characters: %d %s", w.Code, w.Body.String())
+	}
+	page, err := r.List(ctx, Filters{Page: 1, PageSize: 8, Query: "graph"})
+	if err != nil || page.Total != 2 || len(page.Items) != 2 || page.Items[1].Title != p.Title {
+		t.Fatalf("encoded corpus search: %+v %v", page, err)
+	}
+}

internal/research/postgres_test.go

@@ -0,0 +1,78 @@
+package research
+
+import (
+	"database/sql"
+	"encoding/json"
+	"reflect"
+	"strings"
+	"testing"
+)
+
+func TestStoredAuthorShapes(t *testing.T) {
+	for _, test := range []struct {
+		raw  string
+		want []string
+	}{
+		{`[]`, []string{}},
+		{`[["Buhrman","Harry","","CWI"],["Family","Given","Jr.","University"]]`, []string{"Harry Buhrman", "Given Family Jr."}},
+		{`["The BABAR Collaboration","B. Aubert"]`, []string{"The BABAR Collaboration", "B. Aubert"}},
+		{`[["Balázs","C.",""],["Berger","E. L.",""],["Smith","Ada","Jr."]]`, []string{"C. Balázs", "E. L. Berger", "Ada Smith Jr."}},
+		{`[" 张三 ",["Only family"],["","Only given",""],""]`, []string{"张三", "Only family", "Only given"}},
+	} {
+		got, err := decodeAuthors(test.raw)
+		if err != nil || !reflect.DeepEqual(got, test.want) {
+			t.Fatalf("%s: %#v, %v", test.raw, got, err)
+		}
+		encoded, _ := json.Marshal(got)
+		if string(encoded) == "null" {
+			t.Fatal("missing authors must remain an array")
+		}
+	}
+	for _, raw := range []string{`broken`, `{"unexpected":"object"}`, `[42]`, `[["a","b","c","d","e"]]`} {
+		if _, err := decodeAuthors(raw); err == nil {
+			t.Fatalf("accepted unsupported authors %s", raw)
+		}
+	}
+}
+
+func TestQueryValuesRemainBound(t *testing.T) {
+	query := `graph'); DROP TABLE papers; -- OR author:*`
+	where, args := whereClause(Filters{Query: query, Topic: "cs.LG", SearchIn: "text"})
+	if strings.Contains(where, query) || !reflect.DeepEqual(args, []any{query, "cs.LG"}) {
+		t.Fatalf("unbound user input: %s %#v", where, args)
+	}
+	if !strings.Contains(where, "plainto_tsquery('english'::regconfig,$1)") || !strings.Contains(where, "t.value=$2") {
+		t.Fatalf("unexpected search contract: %s", where)
+	}
+	where, args = whereClause(Filters{Query: "Ada Smith", SearchIn: "author"})
+	if !strings.Contains(where, "p.authors_json") || strings.Contains(where, "p.abstract") || !reflect.DeepEqual(args, []any{"Ada Smith"}) {
+		t.Fatalf("author search includes unrelated text: %s", where)
+	}
+	where, args = whereClause(Filters{Topic: "cs.AI"})
+	if !strings.Contains(where, "t.value=$1") || !reflect.DeepEqual(args, []any{"cs.AI"}) {
+		t.Fatalf("topic-only binding: %s %#v", where, args)
+	}
+	where, args = whereClause(Filters{})
+	if where != "" || len(args) != 0 {
+		t.Fatal("empty filters changed the corpus")
+	}
+}
+
+func TestSchemaCannotEscapeQualification(t *testing.T) {
+	for _, name := range []string{"", `research";DROP SCHEMA public;--`, "a.b", "Public", strings.Repeat("a", 37)} {
+		if _, err := NewPostgresRepository(&sql.DB{}, name); err == nil {
+			t.Fatalf("accepted schema %q", name)
+		}
+	}
+	if _, err := NewPostgresRepository(nil, "research"); err == nil {
+		t.Fatal("accepted nil database")
+	}
+	r, err := NewPostgresRepository(&sql.DB{}, "research_v2")
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := r.query(`SELECT p.id FROM {schema}papers p JOIN {schema}source_records s ON s.paper_id=p.id`)
+	if got != `SELECT p.id FROM "research_v2".papers p JOIN "research_v2".source_records s ON s.paper_id=p.id` {
+		t.Fatalf("unqualified corpus tables: %s", got)
+	}
+}

internal/research/testdata/corpus.sql

@@ -0,0 +1,81 @@
+-- Synthetic fixture for the disposable, network-isolated CI database only.
+-- Required columns/types follow Paper Atlas's fixed a3b677b migration.
+CREATE SCHEMA research_fixture;
+CREATE TABLE research_fixture.papers (
+ id bigint PRIMARY KEY, title text NOT NULL, abstract text, title_key text COLLATE "C" NOT NULL,
+ authors_json text NOT NULL, first_published text
+);
+CREATE INDEX papers_title_key ON research_fixture.papers(title_key);
+CREATE INDEX papers_search_gin ON research_fixture.papers USING gin
+ (to_tsvector('english'::regconfig,coalesce(title,'') || ' ' || coalesce(abstract,'')));
+CREATE TABLE research_fixture.sources (id bigint PRIMARY KEY,name text NOT NULL);
+CREATE TABLE research_fixture.source_records (
+ id bigint PRIMARY KEY,source_id bigint REFERENCES research_fixture.sources(id),
+ paper_id bigint REFERENCES research_fixture.papers(id),external_id text,url text,metadata_json text
+);
+CREATE TABLE research_fixture.identifiers (scheme text,value text,paper_id bigint REFERENCES research_fixture.papers(id),PRIMARY KEY(scheme,value));
+CREATE TABLE research_fixture.tags (id bigint PRIMARY KEY,namespace text,value text,UNIQUE(namespace,value));
+CREATE TABLE research_fixture.paper_tags (
+ paper_id bigint REFERENCES research_fixture.papers(id),tag_id bigint REFERENCES research_fixture.tags(id),
+ evidence_key text,PRIMARY KEY(paper_id,tag_id,evidence_key)
+);
+CREATE TABLE research_fixture.venues (id bigint PRIMARY KEY,acronym text);
+CREATE TABLE research_fixture.editions (id bigint PRIMARY KEY,venue_id bigint REFERENCES research_fixture.venues(id),year bigint,track text);
+CREATE TABLE research_fixture.submissions (
+ id bigint PRIMARY KEY,paper_id bigint REFERENCES research_fixture.papers(id),edition_id bigint REFERENCES research_fixture.editions(id)
+);
+CREATE TABLE research_fixture.decisions (
+ id bigint PRIMARY KEY,submission_id bigint REFERENCES research_fixture.submissions(id),
+ status text,evidence_url text,observed_at text,decision_date text
+);
+CREATE INDEX decisions_submission_order ON research_fixture.decisions(submission_id,observed_at DESC,id DESC);
+
+INSERT INTO research_fixture.papers VALUES
+ (1,'Graph & learning <original>','Neural graph models preserve original abstracts.','graph learning','[["Smith","Ada",""],"The Research Group"]','2024-01-02'),
+ (2,'Alpha topology',NULL,'alpha topology','["Bruno Park"]',NULL),
+ (3,'Graph systems','Graph systems and data.','graph systems','[["Network","Carla","","Fixture University"]]','2023-01-01'),
+ (4,'Zebra methods','Evidence and uncertainty.','zebra methods','[]','2022-01-01'),
+ (5,'Alpha Topology','Another independent record.','alpha topology','["Dana Other"]','2025-01-01');
+INSERT INTO research_fixture.sources VALUES(10,'arxiv'),(20,'pmlr');
+INSERT INTO research_fixture.source_records VALUES
+ (101,10,1,'fixture-1','https://example.test/papers/1','{"journal-ref":"Journal Fixture","comments":"Original comment"}'),
+ (202,20,3,'fixture-3','https://example.test/papers/3','{"doi":"10.fixture/other"}'),
+ (203,10,3,'fixture-3-arxiv','https://example.test/papers/3/arxiv','{}');
+INSERT INTO research_fixture.identifiers VALUES('doi','10.fixture/canonical',1);
+INSERT INTO research_fixture.tags VALUES(1,'arxiv:category','cs.LG'),(2,'arxiv:category','cs.AI'),(3,'arxiv:category','cs.DC'),(4,'different:namespace','cs.LG');
+INSERT INTO research_fixture.paper_tags VALUES(1,1,'first'),(1,1,'second'),(1,2,'first'),(3,3,'first'),(2,4,'first');
+INSERT INTO research_fixture.venues VALUES(1,'ICLR'),(2,'ICML');
+INSERT INTO research_fixture.editions VALUES(7,1,2024,'main'),(8,2,2023,'main');
+INSERT INTO research_fixture.submissions VALUES(11,1,7),(12,3,8);
+INSERT INTO research_fixture.decisions VALUES
+ (1,11,'accepted','https://example.test/decision/1','2024-02-01','2024-01-30'),
+ (2,11,'rejected','https://example.test/decision/2','2024-02-01','2024-01-31'),
+ (3,11,'withdrawn','https://example.test/decision/3','2024-02-01','2024-01-31'),
+ (4,11,'accepted','https://example.test/decision/4','2024-01-01','2099-01-01');
+
+CREATE ROLE foundry_research_fixture LOGIN;
+REVOKE ALL ON SCHEMA research_fixture FROM PUBLIC;
+GRANT USAGE ON SCHEMA research_fixture TO foundry_research_fixture;
+GRANT SELECT ON ALL TABLES IN SCHEMA research_fixture TO foundry_research_fixture;
+
+-- Separate encoded schema mirrors the published Paper Atlas nul-soh-v1 format.
+CREATE SCHEMA research_encoded_fixture;
+COMMENT ON SCHEMA research_encoded_fixture IS 'paper-atlas:text-codec=nul-soh-v1';
+DO $$
+DECLARE item text;
+BEGIN
+ FOREACH item IN ARRAY ARRAY['papers','sources','source_records','identifiers','tags','paper_tags','venues','editions','submissions','decisions'] LOOP
+  EXECUTE format('CREATE TABLE research_encoded_fixture.%I (LIKE research_fixture.%I INCLUDING ALL)',item,item);
+  EXECUTE format('INSERT INTO research_encoded_fixture.%I SELECT * FROM research_fixture.%I',item,item);
+ END LOOP;
+END $$;
+UPDATE research_encoded_fixture.papers
+SET title='Graph'||chr(1)||chr(1)||'original'||chr(1)||chr(2),
+    abstract='Text'||chr(1)||chr(1)||'tail',
+    authors_json='["Ada\u0000Smith","Group\u0001Name"]'
+WHERE id=1;
+UPDATE research_encoded_fixture.source_records
+SET metadata_json='{"comments":"Original\u0000comment"}' WHERE id=101;
+REVOKE ALL ON SCHEMA research_encoded_fixture FROM PUBLIC;
+GRANT USAGE ON SCHEMA research_encoded_fixture TO foundry_research_fixture;
+GRANT SELECT ON ALL TABLES IN SCHEMA research_encoded_fixture TO foundry_research_fixture;

internal/research/text.go

@@ -0,0 +1,88 @@
+package research
+
+import (
+	"database/sql"
+	"errors"
+	"strings"
+)
+
+// Matches Paper Atlas repair 4e62710's nul-soh-v1 protocol: NUL -> SOH/SOH and literal
+// SOH -> SOH/STX. All other bytes retain their original binary ordering.
+const textCodec = "nul-soh-v1"
+
+func encodedSchema(annotation sql.NullString) (bool, error) {
+	const prefix = "paper-atlas:text-codec="
+	if !annotation.Valid || !strings.HasPrefix(annotation.String, prefix) {
+		return false, nil
+	}
+	if strings.TrimPrefix(annotation.String, prefix) != textCodec {
+		return false, errors.New("unsupported research text codec")
+	}
+	return true, nil
+}
+
+func encodeText(value string) string {
+	return strings.ReplaceAll(strings.ReplaceAll(value, "\x01", "\x01\x02"), "\x00", "\x01\x01")
+}
+
+func decodeText(value string) (string, error) {
+	if !strings.ContainsRune(value, '\x01') {
+		return value, nil
+	}
+	var result strings.Builder
+	for {
+		position := strings.IndexByte(value, 1)
+		if position < 0 {
+			result.WriteString(value)
+			return result.String(), nil
+		}
+		result.WriteString(value[:position])
+		if position+1 >= len(value) {
+			return "", errors.New("invalid research text escape")
+		}
+		switch value[position+1] {
+		case 1:
+			result.WriteByte(0)
+		case 2:
+			result.WriteByte(1)
+		default:
+			return "", errors.New("invalid research text escape")
+		}
+		value = value[position+2:]
+	}
+}
+
+type textColumn struct {
+	encoded bool
+	target  any
+}
+
+func (column textColumn) Scan(source any) error {
+	var value sql.NullString
+	if err := value.Scan(source); err != nil {
+		return err
+	}
+	if value.Valid && column.encoded {
+		var err error
+		value.String, err = decodeText(value.String)
+		if err != nil {
+			return err
+		}
+	}
+	switch target := column.target.(type) {
+	case *string:
+		if !value.Valid {
+			return errors.New("unexpected null research text")
+		}
+		*target = value.String
+	case *sql.NullString:
+		*target = value
+	default:
+		return errors.New("unsupported research text destination")
+	}
+	return nil
+}
+
+func (r *PostgresRepository) text(target any) sql.Scanner {
+	return textColumn{encoded: r.encoded, target: target}
+}

internal/research/text_test.go

@@ -0,0 +1,69 @@
+package research
+
+import (
+	"database/sql"
+	"strings"
+	"testing"
+)
+
+func TestCorpusTextCodecRoundTripAndOrder(t *testing.T) {
+	values := []string{"", "\x00", "\x00\x01", "\x01", "\x01\x02", "\x02", "A", "A\x00", "A\x01", "Ada 张三 Ω", "Z\x00\x01\x02尾"}
+	for _, value := range values {
+		encoded := encodeText(value)
+		decoded, err := decodeText(encoded)
+		if err != nil || decoded != value || strings.ContainsRune(encoded, 0) {
+			t.Fatalf("round trip failed for %q", value)
+		}
+		for _, other := range values {
+			if strings.Compare(value, other) != strings.Compare(encoded, encodeText(other)) {
+				t.Fatalf("binary ordering changed for %q %q", value, other)
+			}
+		}
+	}
+	for _, invalid := range []string{"\x01", "prefix\x01x", "\x01\x00"} {
+		if _, err := decodeText(invalid); err == nil {
+			t.Fatalf("accepted malformed escape %q", invalid)
+		}
+	}
+}
+
+func TestCorpusTextColumnsRespectCodecAndNulls(t *testing.T) {
+	var required string
+	if err := (textColumn{target: &required}).Scan("\x01\x01"); err != nil || required != "\x01\x01" {
+		t.Fatal("legacy text was decoded")
+	}
+	if err := (textColumn{encoded: true, target: &required}).Scan([]byte("before\x01\x01after\x01\x02")); err != nil || required != "before\x00after\x01" {
+		t.Fatalf("encoded text did not decode: %q %v", required, err)
+	}
+	var nullable sql.NullString
+	if err := (textColumn{encoded: true, target: &nullable}).Scan(nil); err != nil || nullable.Valid {
+		t.Fatal("null changed into empty text")
+	}
+	if err := (textColumn{encoded: true, target: &nullable}).Scan("\x01\x01"); err != nil || !nullable.Valid || nullable.String != "\x00" {
+		t.Fatal("nullable text was not restored")
+	}
+	if err := (textColumn{encoded: true, target: &required}).Scan(nil); err == nil {
+		t.Fatal("accepted null for required text")
+	}
+	if err := (textColumn{encoded: true, target: &required}).Scan("\x01x"); err == nil {
+		t.Fatal("accepted corrupt encoded text")
+	}
+}
+
+func TestCorpusSchemaCodecMarker(t *testing.T) {
+	for _, annotation := range []sql.NullString{{}, {String: "unrelated schema description", Valid: true}} {
+		encoded, err := encodedSchema(annotation)
+		if encoded || err != nil {
+			t.Fatal("legacy schema changed encoding")
+		}
+	}
+	encoded, err := encodedSchema(sql.NullString{String: "paper-atlas:text-codec=nul-soh-v1", Valid: true})
+	if !encoded || err != nil {
+		t.Fatal("published codec was not recognized")
+	}
+	for _, name := range []string{"unknown", "", "legacy"} {
+		if _, err := encodedSchema(sql.NullString{String: "paper-atlas:text-codec=" + name, Valid: true}); err == nil {
+			t.Fatal("unknown codec was accepted")
+		}
+	}
+}