// serveDeviceUds — a LOCAL hyper-svc control surface for a SwarmDevice over a UNIX SOCKET, so a
// CO-LOCATED client (favela-web on the same box) reaches the device instantly via local IPC instead of a
// DHT rendezvous + hole-punch. The swarm surface (serve-device.js) stays for the REMOTE keyless bridge.
//
// AUTH: a UDS connection is owner-gated by the filesystem (only a local process with access to the socket
// path can connect), so the local surface is TRUSTED WITHOUT the writer-allowlist — the same model pgh /
// pgv3 / atproto-bridge use for their local hyper-svc sockets. (The swarm surface keeps canWrite because
// remote peers are untrusted.) Same op set as serveDevice; a client calls `c.call('<op>', params)`.
import b4a from 'b4a'
import fs from 'bare-fs'
import { Service } from 'hyper-svc'
import { conflictDiag } from './conflict-watch.js'
export async function serveDeviceUds (device, { uds, blobs = null, onRelease = null, labels = null } = {}) {
if (!uds) throw new Error('serveDeviceUds: uds path required')
const svc = new Service({ name: 'hyper-proto-peer', uds })
// OTA — the app half of the push contract. The sidecar performs the FULL install (tree + dep closure +
// materialize) and then calls this; our only job is to exit cleanly so the supervisor respawns onto the
// new tree. Injected rather than exiting here: this module serves a device and must not own the
// process lifecycle.
//
// ⚠ NOT pgh-runtime's fetch path — it fetches root trees WITHOUT dep closures, which produces a broken
// install (the MODULE_NOT_FOUND respawn loop that took the membrane down on 2026-07-31).
if (onRelease) {
svc.handle('release-ready', async (ctx, params = {}) => {
const ver = (params && params.version) || '?'
console.log(JSON.stringify({ msg: 'ota: release-ready — sidecar installed; respawning', version: ver }))
// ⛔ A BARE catch {} here is what hid this twice. The handler logged "respawning", threw a
// TypeError (0.3.100-0.3.103: `this` undefined), and the silence read as success on every surface.
setImmediate(() => {
try { onRelease(ver) } catch (e) {
console.error(JSON.stringify({ msg: 'ota: respawn handler THREW — the daemon did NOT respawn', version: ver, error: (e && e.message) || String(e) }))
}
})
return { ok: true, willRespawn: true }
})
}
svc.handle('get-state', async () => device.state())
// conflict-diag: hypercore conflicts seen SINCE BOOT. A conflict freezes a core and stops it
// replicating while the served view stays correct, so a forked node looks healthy on every other
// surface — loser-club logged conflicts for weeks under a clean `diag`. Per-process because
// hypercore's own freeze is: `frozen` is never persisted, so a restart re-arms and re-detects it.
svc.handle('conflict-diag', async () => conflictDiag())
// members-diag: read-only membership/admission diagnostic (why a writer can't be admitted). reproject-
// members: force-re-project __members__ (writable creator only). Both non-secret, UDS is owner-gated.
svc.handle('members-diag', async () => device.membersDiag())
// pointer-diag: does this node's LOCAL byte set match the converged pointer set? Hashes each record
// file and compares it to the pointer's $oid. Read-only (no writes, no fetches), so it is safe on a
// live daemon. This is the check that was missing when two wj62 co-owners held identical converged
// pointers but different bytes for one record — invisible from every surface until two signed CARs
// were diffed by hand. UDS-gated like the rest; it reads record keys + hashes, no record content.
svc.handle('pointer-diag', async (ctx, op = {}) => device.pointerDiag(op || {}))
// migrate-inline-pointers: OPERATOR MIGRATION for a DID whose autobase holds full RECORDS where
// { $oid, size } pointers belong (written through the legacy substrate, now running the pointer
// design). Such a repo reads fine locally and is unfetchable by every peer, so the DID cannot be
// co-owned until this runs. Writable-only, idempotent, preserves published content (the pointer
// addresses the bytes that already resolve, asserted by hash). WRITES — hence operator-triggered
// over the owner-gated UDS rather than anything automatic.
svc.handle('migrate-inline-pointers', async () => device.migrateInlinePointers())
svc.handle('reproject-members', async () => device.reprojectMembers())
// get-repo-car: export this device's current signed repo as an atproto CAR (com.atproto.sync.getRepo
// shape). The input a PEER needs for fork recovery (fetch the authoritative node's CAR → reconcileFromCar
// into a fresh line). Public (UDS is filesystem-owner-gated). Bytes as hex, matching get-blob.
svc.handle('get-repo-car', async () => { const r = await device.exportCar(); return { car: b4a.toString(r.carBytes, 'hex'), commitCid: r.commitCid, dataCid: r.dataCid, rev: r.rev, recordCount: r.recordCount } })
svc.handle('apply-write', async (ctx, op) => device.applyWrite(op))
// label-intent / label-list — a LABELER DID's moderation surface (member labels, hyper-social #80).
// `labels` is the DID's LabelRoom. A projector (the gateway's roster → member badges) re-asserts its whole
// set on every change, so label-intent is IDEMPOTENT: an unchanged state writes nothing and keeps its
// cts (LabelRoom.ensure). The write lands in the StateBase, whose watch fires the tick that signs it —
// nothing to trigger here. Registered on every DID so a non-labeler answers "not a labeler", not
// "unknown method".
const needLabels = () => { if (!labels) throw new Error('this DID is not a labeler (set labeler: true in its config)'); return labels }
svc.handle('label-intent', async (ctx, { uri, val, neg = false } = {}) => {
const l = needLabels()
if (!uri || !val) throw new Error('label-intent: uri and val required')
return l.ensure(uri, val, { neg: !!neg })
})
// label-list — also the labeler's DIAGNOSTIC (2026-10-14: member labels never re-signed and nothing said why).
// `holds` is the exact gate the label tick asks (LabelRoom.holds → device.isSigner(), election WITH the network
// head — get-state's holdsSigner asks without it and can disagree). `labels` is the signed log itself, `ver`
// included, so "is it signed correctly" is a read, not an inference. Read-only.
svc.handle('label-list', async () => {
const l = needLabels()
const log = await l.signedLog()
let holds = null
try { holds = l.holds ? !!(await l.holds()) : null } catch { holds = null }
return {
intents: await l.base.intents(),
signed: log.length,
holds,
// Did the signed log reach the label core the membrane serves? (LabelRoom.lastProjection — null until the
// first projectTo since boot.)
projection: l.lastProjection === undefined ? null : l.lastProjection,
labels: log.map((x) => ({ uri: x.uri, val: x.val, cts: x.cts, neg: !!x.neg, ver: x.ver === undefined ? null : x.ver }))
}
})
// get-record: read ONE record back. get-state lists {collection, rkey} and no values, so a client
// could see THAT a record exists and never what it says — which makes "is what I would write
// already there?" unanswerable, and the only alternative is a local fingerprint that drifts from
// the repo. A projector reconciling desired-against-actual needs the actual.
//
// Read-only and UDS-gated like the diagnostics beside it. Returns null rather than throwing for an
// absent record: "not written yet" is an ordinary state for a caller deciding whether to write.
svc.handle('get-record', async (ctx, { collection, rkey } = {}) => {
if (!collection || !rkey) throw new Error('get-record: collection and rkey required')
try { return (await device.repo.getRecord(collection, rkey)) || null } catch { return null }
})
// reconcile-bytes: ATTEMPT to pull the authoritative bytes for every converged pointer this node
// cannot resolve. The daemon already runs this on a loop; exposing it gives an operator a way to
// demand one NOW and read the result, which is what makes "is this record recoverable?" answerable.
// `repair-pointers` depends on exactly that: a record is only unrecoverable once a real fetch has
// been attempted and failed, and tombstoning one that was merely un-fetched would destroy content a
// peer still holds. PULL-ONLY — never appends to the autobase, so it is safe on a non-writable
// replica and cannot alter the record set.
svc.handle('reconcile-bytes', async (ctx, op = {}) => device.reconcileBytes(op || {}))
// reindex-overlay: OPERATOR REPAIR for a node whose overlay index is missing or stale. Re-registers
// every record file AND every blob file so peers can fetch them by oid, then announces any local blob
// that never got a shared pointer. Byte resolution gates signing, so an impoverished index silently
// stops this device signing — this is the way to fix that without a restart.
//
// UDS-ONLY BY DESIGN: it re-hashes every file, so exposing it on the peer-facing swarm surface would
// hand any remote peer a free CPU-exhaustion trigger. The UDS is filesystem-owner-gated, which is the
// right authority for an operator action. Purely additive — re-registration re-asserts identical
// entries and the pointer backfill never clobbers one a co-owner already published.
svc.handle('reindex-overlay', async () => {
const records = await device.reindexOverlay()
let blobFiles = { skipped: 'no-blob-store' }
let pointers = { skipped: 'no-blob-store' }
if (blobs) {
if (typeof blobs.reindexOverlay === 'function') blobFiles = await blobs.reindexOverlay()
// Pointer backfill APPENDS to the autobase, so only an admitted writer may run it; peers converge
// the pointers by replication instead (same gate as migratePointers).
if (typeof blobs.backfillPointers === 'function') {
const writable = !!(device.sb && device.sb.base && device.sb.base.writable)
pointers = writable ? await blobs.backfillPointers() : { skipped: 'not-writable' }
}
}
return { records, blobs: blobFiles, pointers }
})
// get-service-auth — mint a short-lived ES256K service-auth JWT so the KEYLESS membrane can make a
// USER-SCOPED AppView read as this account (the bridge's proxyToAppView already calls a serviceAuth
// hook and attaches the Bearer; it simply had nothing to call). The primitive is
// SwarmDevice.getServiceAuth — deliberately general, since it is atproto's com.atproto.server
// .getServiceAuth. The AUTHORITY BOUNDARY is here, at the exposed capability.
//
// A minted token authenticates AS the DID, signed by the same #atproto key that signs every repo
// commit (verified: it matches the key published in both hosted DIDs' PLC documents). So this handler
// is a signing oracle and is constrained accordingly:
//
// - lxm is REQUIRED and ALLOWLISTED. The primitive treats lxm as optional, and a token with no lxm
// is valid for ANY method at that audience — a far broader grant than anything we need. One entry
// for now; widening it must be a visible edit to this list, not a caller's choice.
// - aud must look like a DID (the AppView's service DID, e.g. did:web:api.bsky.app).
// - TTL is capped short. A leaked token then expires on its own.
// - UDS ONLY. This is deliberately NOT added to lib/serve-device.js's swarm surface: that is
// reachable by any peer that knows the DID, and pointer-diag is already writer-gated there for a
// much weaker reason (CPU cost). A remotely-reachable signing oracle would be the single worst
// thing on that surface. The UDS is owner-gated by filesystem permissions.
const SERVICE_AUTH_LXM = ['app.bsky.notification.listNotifications']
const SERVICE_AUTH_MAX_TTL_SEC = 120
svc.handle('get-service-auth', async (ctx, op = {}) => {
const aud = op && op.aud
const lxm = op && op.lxm
if (typeof aud !== 'string' || !aud.startsWith('did:')) throw new Error('get-service-auth: aud must be a service DID')
if (typeof lxm !== 'string' || !lxm) throw new Error('get-service-auth: lxm required (a token without one is valid for ANY method)')
if (!SERVICE_AUTH_LXM.includes(lxm)) throw new Error(`get-service-auth: lxm "${lxm}" is not allowlisted (allowed: ${SERVICE_AUTH_LXM.join(', ')})`)
const now = Math.floor(Date.now() / 1000)
const ttl = Math.min(Number(op.ttlSec) > 0 ? Number(op.ttlSec) : SERVICE_AUTH_MAX_TTL_SEC, SERVICE_AUTH_MAX_TTL_SEC)
// Log the GRANT, never the token.
console.log(JSON.stringify({ msg: 'service-auth minted', did: device.did, aud, lxm, ttlSec: ttl }))
return device.getServiceAuth({ aud, lxm, exp: now + ttl })
})
svc.handle('delegate', async (ctx, op) => device.delegate(op || {}))
svc.handle('list-devices', async () => ({ devices: await device.listDevices() }))
svc.handle('revoke-device', async (ctx, op = {}) => {
if (!op.subKey) throw new Error('revoke-device: subKey required')
return op.restore ? device.restoreDevice(op.subKey) : device.revokeDevice(op.subKey, { reason: op.reason || null })
})
svc.handle('is-device-revoked', async (ctx, op = {}) => ({ revoked: await device.isDeviceRevoked(op.subKey) }))
if (blobs) {
// get-blob: local read counterpart to upload-blob — the co-located client (favela-web) renders images it
// uploaded. The UDS had upload-blob but no read path, so locally-stored blobs were unreadable. Public (a
// UDS connection is already filesystem-owner-gated). Returns { mimeType, size, bytes(hex) } or null.
svc.handle('list-blobs', async (ctx, { limit, cursor } = {}) => {
if (!blobs || typeof blobs.list !== 'function') return { cids: [], cursor: null, skipped: 'not-supported' }
return blobs.list({ limit, cursor })
})
svc.handle('get-blob', async (ctx, { cid } = {}) => {
if (!cid) throw new Error('get-blob: cid required')
const blob = await blobs.get(String(cid))
if (!blob) return null
return { cid: String(cid), mimeType: blob.mimeType, size: blob.size, bytes: b4a.toString(blob.bytes, 'hex') }
})
// ⛔ A PATH IS ACCEPTED AS WELL AS BYTES, and for a co-located client it is the better one. This
// is a unix socket: both ends are on one machine, and hex-encoding a picture to hand it to a
// process that will write it straight back to disk doubles it in transit for nothing. favela-peer
// settled the same question the same way ("PATHS cross this socket, never bytes"), and a client
// that has to hex-encode needs a buffer library it may not otherwise depend on — which is a
// MODULE_NOT_FOUND at boot waiting to happen, not a detail.
//
// Bytes stay supported: a REMOTE client has no shared filesystem and no path to offer.
svc.handle('upload-blob', async (ctx, { bytes, path, mimeType } = {}) => {
if (!bytes && !path) throw new Error('upload-blob: bytes (hex) or path required')
let buf
if (path) {
try { buf = fs.readFileSync(path) } catch { throw new Error('upload-blob: no file at ' + path) }
} else {
buf = b4a.from(bytes, 'hex')
}
if (!buf || !buf.length) throw new Error('upload-blob: that file is empty')
const { cid, size } = await blobs.put(buf, mimeType || 'application/octet-stream')
return { cid: cid.toString(), mimeType: mimeType || 'application/octet-stream', size }
})
}
await svc.listen()
return { uds, close: async () => { try { await svc.close() } catch {} } }
}