@@ -302,7 +302,7 @@ export default function DocsPage() {
The certified organization labeler exposes a small REST API for the dashboard as well as the standard AT Protocol
- labeler XRPC endpoint. Labels are still published against the organization record URI.
+ labeler XRPC endpoint. Quality labels are published against actor DIDs.
{API_ENDPOINTS.map(({ method, path, description, curl }) => (
@@ -360,7 +360,7 @@ export default function DocsPage() {
—
- Each label is signed with ed25519 and includes: source DID, record URI, label value,
+ Each label is signed with ed25519 and includes: source DID, subject DID, label value,
timestamp, and a cryptographic signature.
@@ -368,15 +368,15 @@ export default function DocsPage() {
—
Apps can subscribe to the labeler to automatically receive organization quality signals for
- app.certified.actor.organization records and filter or sort them by tier. The related profile record is
- used for merged actor context, not for the label target.
+ actor DIDs and filter or sort them by tier. The related profile and organization records are
+ used for merged actor context, not as the label target.
—
- Only one quality label is active per record URI at a time. When a record is updated and
- re-scored, the previous label is negated before the new one is applied.
+ Only one quality label is active per actor DID at a time. When actor context changes and
+ is re-scored, the previous label is negated before the new one is applied.
diff --git a/src/labeler/server.ts b/src/labeler/server.ts
index e4ff986..01f2ae7 100644
--- a/src/labeler/server.ts
+++ b/src/labeler/server.ts
@@ -5,11 +5,16 @@ import logger from './logger'
export const labelerServer = new LabelerServer({ did: DID, signingKey: SIGNING_KEY, dbPath: LABELS_DB_PATH })
-export async function fetchCurrentLabels(uri: string): Promise> {
+/**
+ * Returns the active labels for one AT Protocol label subject.
+ * Subjects are normally actor DIDs; negated labels remove earlier active
+ * values from the returned set.
+ */
+export async function fetchCurrentLabels(subjectUri: string): Promise> {
await labelerServer.db.execute('SELECT 1') // ensure db is ready
const result = await labelerServer.db.execute({
sql: 'SELECT val, neg FROM labels WHERE uri = ? ORDER BY id ASC',
- args: [uri],
+ args: [subjectUri],
})
const active = new Set()
@@ -25,94 +30,79 @@ export async function fetchCurrentLabels(uri: string): Promise> {
return active
}
-export async function negateAllDIDLabels(): Promise {
- await labelerServer.db.execute('SELECT 1') // ensure db is ready
- const result = await labelerServer.db.execute({
- sql: 'SELECT DISTINCT uri FROM labels WHERE uri LIKE \'did:%\'',
- args: [],
- })
-
- let negatedCount = 0
- for (const row of result.rows) {
- const did = row['uri'] as string
- try {
- const activeLabels = await fetchCurrentLabels(did)
- const activeQualityLabels = [...activeLabels].filter(l => QUALITY_LABEL_IDENTIFIERS.includes(l))
- if (activeQualityLabels.length > 0) {
- logger.info({ did, negating: activeQualityLabels }, 'Negating stale DID-level quality labels')
- await labelerServer.createLabels({ uri: did }, { negate: activeQualityLabels })
- negatedCount++
- }
- } catch (err) {
- logger.error({ err, did }, 'Failed to negate DID-level labels, continuing')
- }
- }
- return negatedCount
-}
-
-export async function negateQualityLabels(recordUri: string): Promise {
- const existing = uriLocks.get(recordUri)
+/**
+ * Negates all active quality labels on one label subject.
+ * Use this when an organization actor is deleted and its DID should no longer
+ * carry an active quality tier.
+ */
+export async function negateQualityLabels(subjectUri: string): Promise {
+ const existing = uriLocks.get(subjectUri)
const operation = (existing ?? Promise.resolve()).then(async () => {
await labelerServer.db.execute('SELECT 1') // ensure db is ready
- const currentLabels = await fetchCurrentLabels(recordUri)
+ const currentLabels = await fetchCurrentLabels(subjectUri)
const activeQualityLabels = [...currentLabels].filter(l => QUALITY_LABEL_IDENTIFIERS.includes(l))
if (activeQualityLabels.length === 0) {
return 0
}
- logger.info({ uri: recordUri, negating: activeQualityLabels }, 'Negating quality labels for deleted record')
- await labelerServer.createLabels({ uri: recordUri }, { negate: activeQualityLabels })
+ logger.info({ uri: subjectUri, negating: activeQualityLabels }, 'Negating quality labels')
+ await labelerServer.createLabels({ uri: subjectUri }, { negate: activeQualityLabels })
return activeQualityLabels.length
}).finally(() => {
- if (uriLocks.get(recordUri) === operation) {
- uriLocks.delete(recordUri)
+ if (uriLocks.get(subjectUri) === operation) {
+ uriLocks.delete(subjectUri)
}
})
- uriLocks.set(recordUri, operation)
+ uriLocks.set(subjectUri, operation)
return operation as Promise
}
const uriLocks = new Map>()
-export async function applyQualityLabel(recordUri: string, labelIdentifier: string): Promise {
+/**
+ * Ensures one quality label is active on a label subject.
+ * Existing quality labels on the same DID are negated before the new value is
+ * written, so each actor has at most one active quality tier.
+ */
+export async function applyQualityLabel(subjectUri: string, labelIdentifier: string): Promise {
// Validate labelIdentifier before any DB operations
if (!QUALITY_LABEL_IDENTIFIERS.includes(labelIdentifier)) {
- logger.error({ uri: recordUri, label: labelIdentifier }, 'Rejected unknown label identifier')
+ logger.error({ uri: subjectUri, label: labelIdentifier }, 'Rejected unknown label identifier')
return
}
// Wait for any in-flight operation on this URI
- const existing = uriLocks.get(recordUri)
+ const existing = uriLocks.get(subjectUri)
const operation = (existing ?? Promise.resolve()).then(async () => {
- // 1. Fetch current labels for the record URI
- const currentLabels = await fetchCurrentLabels(recordUri)
+ // 1. Fetch current labels for the subject URI
+ const currentLabels = await fetchCurrentLabels(subjectUri)
// 2. Filter to only QUALITY_LABEL_IDENTIFIERS
const currentQualityLabels = [...currentLabels].filter(l => QUALITY_LABEL_IDENTIFIERS.includes(l))
- // 3. If record already has the same quality label → return (no change needed)
+ // 3. If subject already has the same quality label → return (no change needed)
if (currentQualityLabels.includes(labelIdentifier)) {
- logger.info({ uri: recordUri, label: labelIdentifier }, 'Record already has label, skipping')
+ logger.info({ uri: subjectUri, label: labelIdentifier }, 'Subject already has label, skipping')
return
}
- // 4. If record has a different quality label → negate it first
+ // 4. If subject has a different quality label → negate it first
if (currentQualityLabels.length > 0) {
- logger.info({ uri: recordUri, negating: currentQualityLabels }, 'Negating existing quality labels')
- await labelerServer.createLabels({ uri: recordUri }, { negate: currentQualityLabels })
+ logger.info({ uri: subjectUri, negating: currentQualityLabels }, 'Negating existing quality labels')
+ await labelerServer.createLabels({ uri: subjectUri }, { negate: currentQualityLabels })
}
// 5. Create the new label
- logger.info({ uri: recordUri, label: labelIdentifier }, 'Applying quality label')
- await labelerServer.createLabel({ uri: recordUri, val: labelIdentifier })
+ logger.info({ uri: subjectUri, label: labelIdentifier }, 'Applying quality label')
+ await labelerServer.createLabel({ uri: subjectUri, val: labelIdentifier })
}).finally(() => {
// Clean up lock if we're still the latest
- if (uriLocks.get(recordUri) === operation) {
- uriLocks.delete(recordUri)
+ if (uriLocks.get(subjectUri) === operation) {
+ uriLocks.delete(subjectUri)
}
})
- uriLocks.set(recordUri, operation)
+ uriLocks.set(subjectUri, operation)
return operation as Promise
}
diff --git a/src/labeler/start.ts b/src/labeler/start.ts
index 4c73c0d..f27e900 100644
--- a/src/labeler/start.ts
+++ b/src/labeler/start.ts
@@ -2,7 +2,7 @@ import 'dotenv/config'
import fs from 'node:fs'
import { HOST, LABELER_PORT, METRICS_PORT, TAP_URL, APP_DB_PATHS, TAP_ADMIN_PASSWORD } from '../lib/config'
import { getPendingActivities, deleteActivity } from '../lib/db'
-import { labelerServer, negateAllDIDLabels, applyQualityLabel } from './server'
+import { labelerServer, applyQualityLabel } from './server'
import {
startTapConsumer,
backfillHfClassification,
@@ -103,24 +103,15 @@ async function main() {
})
})
- // Wire HF reclassification to also update ATProto labels
- setReclassifyCallback(applyQualityLabel)
+ // Wire HF reclassification to update DID labels.
+ setReclassifyCallback(async (did, newTier) => {
+ await applyQualityLabel(did, newTier)
+ })
// 2. Start metrics server
startMetricsServer(METRICS_PORT)
logger.info({ port: METRICS_PORT }, 'Metrics server started')
- // 2b. Negate any stale DID-level labels from previous deployments
- // Fix 1: wrap in try/catch so a failure doesn't crash startup
- try {
- const negatedCount = await negateAllDIDLabels()
- if (negatedCount > 0) {
- logger.info({ count: negatedCount }, 'Negated stale DID-level labels')
- }
- } catch (err) {
- logger.error({ err }, 'Failed to negate stale DID-level labels — continuing startup')
- }
-
// Fix 7: clean up stale pending records BEFORE starting tap consumer
// so a backfill event can't re-score a record that we're about to delete
const pendingRecords = getPendingActivities()
@@ -149,7 +140,7 @@ async function main() {
logger.info('Tap consumer started — receiving backfill + live events')
backfillHfClassification()
- // One-time sync: fix any records where DB tier disagrees with ATProto label
+ // One-time sync: fix any actors where DB tier disagrees with ATProto label
// (caused by pre-fix HF reclassifications that only updated the DB)
syncLabelsWithDb().catch(err => {
logger.warn({ err }, 'Label sync failed — will retry on next restart')
diff --git a/src/labeler/tap-consumer.ts b/src/labeler/tap-consumer.ts
index 2c9b6be..4ac493b 100644
--- a/src/labeler/tap-consumer.ts
+++ b/src/labeler/tap-consumer.ts
@@ -16,6 +16,7 @@ import {
failRecomputeJob,
getRecomputeJobCounts,
hasPendingOrganizationDelete,
+ hasMatchingPendingOrganizationDelete,
recoverStaleRunningRecomputeJobs,
getProfileSnapshot,
getOrganizationSnapshot,
@@ -247,7 +248,7 @@ export async function recomputeLabeledOrganizationRow(did: string): Promise {
for (const activity of activities) {
try {
- const currentLabels = await fetchCurrentLabels(activity.uri)
- const currentQuality = [...currentLabels].filter(isRuntimeLabelTier)
if (!isRuntimeLabelTier(activity.tier)) {
continue
}
+ const currentLabels = await fetchCurrentLabels(activity.did)
+ const currentQuality = [...currentLabels].filter(isRuntimeLabelTier)
const dbTier = activity.tier as RuntimeLabelTier
- // If the ATProto label doesn't match the DB tier, update it
+ // If the ATProto DID label doesn't match the DB tier, update it
if (!currentQuality.includes(dbTier)) {
- logger.info({ uri: activity.uri, dbTier, atprotoLabels: currentQuality }, 'Syncing mismatched label')
- await applyQualityLabel(activity.uri, dbTier)
+ logger.info({ did: activity.did, dbTier, atprotoLabels: currentQuality }, 'Syncing mismatched DID label')
+ await applyQualityLabel(activity.did, dbTier)
synced++
}
+
} catch (err) {
- logger.warn({ err, uri: activity.uri }, 'Failed to sync label, continuing')
+ logger.warn({ err, did: activity.did, uri: activity.uri }, 'Failed to sync label, continuing')
}
}
if (synced > 0) {
- logger.info({ count: synced }, 'Synced mismatched ATProto labels with DB tiers')
+ logger.info({ count: synced }, 'Synced mismatched DID labels with DB tiers')
} else {
- logger.info('All ATProto labels match DB tiers')
+ logger.info('All DID labels match DB tiers')
}
}
diff --git a/src/lib/db.ts b/src/lib/db.ts
index f8bd339..ee086bf 100644
--- a/src/lib/db.ts
+++ b/src/lib/db.ts
@@ -589,6 +589,22 @@ export function hasPendingOrganizationDelete(did: string): boolean {
return Boolean(row)
}
+/**
+ * Checks whether a queued organization delete still targets the same record.
+ * Use this after async label cleanup to avoid applying actor-level cleanup for
+ * a delete that was superseded by a newer organization upsert.
+ */
+export function hasMatchingPendingOrganizationDelete(did: string, rkey: string, recordUri: string): boolean {
+ const db = getDb()
+ const row = db.prepare(`
+ SELECT 1
+ FROM pending_organization_deletes
+ WHERE did = @did AND rkey = @rkey AND record_uri = @recordUri
+ LIMIT 1
+ `).get({ did, rkey, recordUri })
+ return Boolean(row)
+}
+
export function deletePendingOrganizationDelete(did: string): void {
const db = getDb()
db.prepare('DELETE FROM pending_organization_deletes WHERE did = ?').run(did)
diff --git a/src/lib/hf-classifier.ts b/src/lib/hf-classifier.ts
index ee3c6aa..8660ba2 100644
--- a/src/lib/hf-classifier.ts
+++ b/src/lib/hf-classifier.ts
@@ -30,9 +30,13 @@ interface QueueItem {
const queue: QueueItem[] = []
let processing = false
-type ReclassifyCallback = (uri: string, newTier: string) => Promise
+type ReclassifyCallback = (did: string, newTier: string) => Promise
let _onReclassify: ReclassifyCallback | null = null
+/**
+ * Registers the labeler-side hook used when HF signals move an actor to a new
+ * quality tier after the main scoring pass has already stored an activity row.
+ */
export function setReclassifyCallback(cb: ReclassifyCallback): void {
_onReclassify = cb
}
@@ -105,7 +109,7 @@ function reclassifyWithHfSignal(did: string, rkey: string, classification: Conte
})
if (tier !== row.tier && _onReclassify) {
- void _onReclassify(row.uri, tier).catch(err => {
+ void _onReclassify(row.did, tier).catch(err => {
console.warn('[hf-classifier] label update failed:', err instanceof Error ? err.message : err)
})
}
diff --git a/src/lib/types.ts b/src/lib/types.ts
index fbad899..fc299ce 100644
--- a/src/lib/types.ts
+++ b/src/lib/types.ts
@@ -122,7 +122,7 @@ export interface ActivityLogEntry {
id?: number
did: string
rkey: string
- uri: string
+ uri: string // Organization record URI retained for dashboard links; labels target did.
displayName: string
score: number
tier: LabelTier
diff --git a/tests/did-label-subjects.test.ts b/tests/did-label-subjects.test.ts
new file mode 100644
index 0000000..083e954
--- /dev/null
+++ b/tests/did-label-subjects.test.ts
@@ -0,0 +1,106 @@
+import assert from 'node:assert/strict'
+import { after, beforeEach, test } from 'node:test'
+import { mkdtempSync, rmSync } from 'node:fs'
+import { join } from 'node:path'
+import { tmpdir } from 'node:os'
+import type { OrganizationSnapshot } from '../src/lib/types'
+
+const testDir = mkdtempSync(join(tmpdir(), 'orglabeler-did-label-subjects-'))
+process.env.ACTIVITY_DB_PATH = join(testDir, 'activity-log.db')
+process.env.LABELS_DB_PATH = join(testDir, 'labels.db')
+process.env.DID = 'did:example:labeler'
+process.env.SIGNING_KEY = '01'.repeat(32)
+process.env.TAP_URL = 'http://127.0.0.1:65535'
+process.env.HF_TOKEN = ''
+
+const {
+ closeDb,
+ getDb,
+ logActivity,
+ upsertOrganizationSnapshot,
+} = await import('../src/lib/db')
+const { fetchCurrentLabels, labelerServer } = await import('../src/labeler/server')
+const { recomputeLabeledOrganizationRow, syncLabelsWithDb } = await import('../src/labeler/tap-consumer')
+
+async function resetLabelDb(): Promise {
+ await (labelerServer as unknown as { dbInitLock?: Promise }).dbInitLock
+ await labelerServer.db.execute('DELETE FROM labels')
+}
+
+beforeEach(async () => {
+ getDb().exec(`
+ DELETE FROM activities;
+ DELETE FROM organization_snapshots;
+ DELETE FROM profile_snapshots;
+ DELETE FROM pending_organization_deletes;
+ DELETE FROM recompute_jobs;
+ DELETE FROM actor_pds_cache;
+ DELETE FROM url_checks;
+ `)
+ await resetLabelDb()
+})
+
+after(() => {
+ labelerServer.db.close()
+ closeDb()
+ rmSync(testDir, { recursive: true, force: true })
+})
+
+function organizationPayload(): OrganizationSnapshot['payload'] {
+ return {
+ $type: 'app.certified.actor.organization',
+ organizationType: ['ngo'],
+ createdAt: '2024-01-01T00:00:00.000Z',
+ }
+}
+
+async function storedLabelSubjects(): Promise {
+ const result = await labelerServer.db.execute('SELECT uri FROM labels ORDER BY id ASC')
+ return result.rows.map(row => String(row.uri))
+}
+
+test('recompute applies quality labels to the actor DID, not the organization record URI', async () => {
+ const did = 'did:example:org-recompute'
+ const rkey = 'self'
+ const recordUri = `at://${did}/app.certified.actor.organization/${rkey}`
+
+ upsertOrganizationSnapshot({
+ did,
+ rkey,
+ recordUri,
+ payload: organizationPayload(),
+ })
+
+ const outcome = await recomputeLabeledOrganizationRow(did)
+
+ assert.ok(outcome)
+ assert.equal(outcome.tier, 'standard')
+ assert.deepEqual([...await fetchCurrentLabels(did)], ['standard'])
+ assert.deepEqual([...await fetchCurrentLabels(recordUri)], [])
+ assert.deepEqual(await storedLabelSubjects(), [did])
+})
+
+test('DB label sync repairs missing actor DID labels without labeling record URIs', async () => {
+ const did = 'did:example:org-sync'
+ const rkey = 'self'
+ const recordUri = `at://${did}/app.certified.actor.organization/${rkey}`
+
+ logActivity({
+ did,
+ rkey,
+ uri: recordUri,
+ title: 'Sync Example Organization',
+ score: 80,
+ tier: 'high-quality',
+ breakdown: '{}',
+ testSignals: '[]',
+ validationNotes: [],
+ labeledAt: new Date().toISOString(),
+ })
+
+ await syncLabelsWithDb()
+
+ assert.deepEqual([...await fetchCurrentLabels(did)], ['high-quality'])
+ assert.deepEqual([...await fetchCurrentLabels(recordUri)], [])
+ assert.deepEqual(await storedLabelSubjects(), [did])
+})
diff --git a/tests/pending-organization-deletes.test.ts b/tests/pending-organization-deletes.test.ts
index a0fcfd5..6186e1d 100644
--- a/tests/pending-organization-deletes.test.ts
+++ b/tests/pending-organization-deletes.test.ts
@@ -15,6 +15,7 @@ const {
deletePendingOrganizationDelete,
getDb,
getOrganizationSnapshot,
+ hasMatchingPendingOrganizationDelete,
hasPendingOrganizationDelete,
upsertOrganizationSnapshot,
upsertPendingOrganizationDelete,
@@ -59,6 +60,18 @@ test('completing a pending delete removes only the matching local organization s
assert.equal(hasPendingOrganizationDelete(did), false)
})
+test('matching pending delete check requires the same record URI and rkey', () => {
+ const did = 'did:example:delete-match'
+ const rkey = 'self'
+ const recordUri = `at://${did}/app.certified.actor.organization/${rkey}`
+
+ upsertPendingOrganizationDelete(did, rkey, recordUri)
+
+ assert.equal(hasMatchingPendingOrganizationDelete(did, rkey, recordUri), true)
+ assert.equal(hasMatchingPendingOrganizationDelete(did, 'other', recordUri), false)
+ assert.equal(hasMatchingPendingOrganizationDelete(did, rkey, `${recordUri}-old`), false)
+})
+
test('stale pending delete cleanup does not remove a newer replacement snapshot', () => {
const did = 'did:example:delete-race'
const staleRkey = 'old'