feat(labeler): publisher-history context stage (W8.4 slice 1)

.opencode/plans/plugin-registry-labelling-service/implementation-plan.md

@@ -1045,7 +1045,7 @@ History may recommend operator review but cannot automatically produce package/p
 
 Dependencies: `W2.3`, registry publisher/profile data.
 
-Decisions (2026-07-14): history is not a new contract surface — `FindingSource` already includes `"history"` and the resolver already drops it (`policy-resolver.ts` excludes `source: "history"` before any category→label mapping; `isBlockingFinding` returns false for it), so "never auto-label" is structural, not a new guard. "Recommend review" is passive: a history finding surfaces through the existing assessment-detail projection an operator already opens; no `operational_events` push and no new "flag-for-review" operator action (D3). Two of the five inputs — recent handle/profile changes and verification state — live only in the aggregator's D1, which the labeler has no binding to reach (and `constants.ts` deliberately avoids ingesting profiles); they are deferred to a later slice gated on a ratified read-only labeler→aggregator read path (service-binding preferred over a second D1 binding), not built now (D1). Repeated-checksum detection is global (cross-publisher): a checksum reused under a different DID is the sharper signal, needing a new `idx_assessments_artifact_checksum` and a cross-DID lookup (D2). Prior releases, active manual labels (reuse `getActiveLabelState`, filter `!automated && active`), and repeated checksums/findings come from the labeler's own D1. The finding contract is amended (touches W8.1): `source: "history"` findings cite a dedicated history/review category set exempt from the automated-block ∪ warning constraint that `validateFinding` enforces, rather than borrowing a warn category (D4). Neutral display facts (prior-release count, verification badge) are assembled at console read-time in the existing subject-history endpoint, not persisted as a pipeline `ContextBundle` (D5). The stage is DB-bound (`analyzeHistory(db, assessment, opts) → NormalizedFinding[]`), wired into `OrchestratorStages`/`STAGE_ORDER`/`stubStages` and run last, wrapping D1 failures as `StageTransientError`. Sliced: (1) own-D1 history findings + the W8.1 amendment + the checksum index — unblocked; (2) console neutral-context surfacing — unblocked, optional; (3) the two aggregator-sourced inputs — blocked on D1's read path.
+Decisions (2026-07-14): history is not a new contract surface — `FindingSource` already includes `"history"` and the resolver already drops it (`policy-resolver.ts` excludes `source: "history"` before any category→label mapping; `isBlockingFinding` returns false for it), so "never auto-label" is structural, not a new guard. "Recommend review" is passive: a history finding surfaces through the existing assessment-detail projection an operator already opens; no `operational_events` push and no new "flag-for-review" operator action (D3). Two of the five inputs — recent handle/profile changes and verification state — live only in the aggregator's D1, which the labeler has no binding to reach (and `constants.ts` deliberately avoids ingesting profiles); they are deferred to a later slice gated on a ratified read-only labeler→aggregator read path (service-binding preferred over a second D1 binding), not built now (D1). Repeated-checksum detection is global (cross-publisher): a checksum reused under a different DID is the sharper signal, needing a new `idx_assessments_artifact_checksum` and a cross-DID lookup (D2). Prior releases, active manual labels (reuse `getActiveLabelState`, filter `!automated && active`), and repeated checksums/findings come from the labeler's own D1. The finding contract is amended (touches W8.1): `source: "history"` findings cite a dedicated history/review category set exempt from the automated-block ∪ warning constraint that `validateFinding` enforces, rather than borrowing a warn category (D4). Neutral display facts (prior-release count, verification badge) are assembled at console read-time in the existing subject-history endpoint, not persisted as a pipeline `ContextBundle` (D5). The stage is DB-bound (`analyzeHistory(db, assessment, opts) → NormalizedFinding[]`), wired into `OrchestratorStages`/`STAGE_ORDER`/`stubStages` and run last. Because history is operator-only context that never becomes a label, the stage is best-effort — on any error it logs and returns no findings rather than throwing, so it can never fail the run and discard the decision-relevant findings from the other stages (adversary-flagged 2026-07-14). History findings keep sensitive specifics (cross-publisher checksum reuse, active manual-label identities/existence including redactions) in `privateDetail` only, never in the public-facing `title`/`publicSummary`; slice (2)'s surfacing must additionally render history findings in the operator projection only and exclude them from any public serializer. Sliced: (1) own-D1 history findings + the W8.1 amendment + the checksum index — unblocked; (2) console neutral-context surfacing — unblocked, optional; (3) the two aggregator-sourced inputs — blocked on D1's read path.
 
 ### `W8.5` Implement versioned policy resolver
 

apps/labeler/migrations/0006_assessment_checksum_index.sql

@@ -0,0 +1,9 @@
+-- Indexes the artifact checksum so the publisher-history stage (plan W8.4) can
+-- run its global, cross-publisher checksum-repeat lookup — "this exact artifact
+-- was also submitted under another DID" — without scanning the assessments
+-- table. Partial (`WHERE artifact_checksum IS NOT NULL`) because pre-artifact
+-- lifecycle states leave the column null.
+
+CREATE INDEX idx_assessments_artifact_checksum
+	ON assessments(artifact_checksum)
+	WHERE artifact_checksum IS NOT NULL;

apps/labeler/src/assessment-orchestrator.ts

@@ -60,9 +60,23 @@ export interface OrchestratorStages {
 	dependency: StageAdapter;
 	codeAi: StageAdapter;
 	imageAi: StageAdapter;
+	/** Publisher-history context (plan W8.4). Runs last: its findings are
+	 * context the resolver drops, never labels, so nothing downstream depends on
+	 * them. Must be best-effort — a history stage that throws would fail the whole
+	 * run and discard every other stage's findings; `analyzeHistory` swallows its
+	 * own errors and returns `[]`. DB-bound — a real wiring passes a closure over
+	 * `analyzeHistory`. */
+	history: StageAdapter;
 }
 
-const STAGE_ORDER = ["acquire", "deterministic", "dependency", "codeAi", "imageAi"] as const;
+const STAGE_ORDER = [
+	"acquire",
+	"deterministic",
+	"dependency",
+	"codeAi",
+	"imageAi",
+	"history",
+] as const;
 
 export interface AssessmentOrchestratorOptions {
 	db: D1Database;
@@ -336,4 +350,5 @@ export const stubStages: OrchestratorStages = {
 	dependency: () => Promise.resolve([]),
 	codeAi: () => Promise.resolve([]),
 	imageAi: () => Promise.resolve([]),
+	history: () => Promise.resolve([]),
 };

apps/labeler/src/assessment-store.ts

@@ -1128,6 +1128,54 @@ export async function getAllLabelsForAssessment(
 	}));
 }
 
+/**
+ * Distinct release URIs previously observed under a publishing DID, newest
+ * observation first, excluding the URI under assessment — the "prior releases
+ * from the DID" input to the publisher-history stage (plan W8.4). Capped at
+ * `limit`; the caller reports the count as bounded rather than exact.
+ */
+export async function getPriorReleaseUrisForDid(
+	db: D1Database,
+	input: { did: string; excludeUri: string; limit: number },
+): Promise<string[]> {
+	const rows = await db
+		.prepare(
+			`SELECT uri, MAX(observed_at_epoch_ms) AS observed
+			 FROM subjects
+			 WHERE did = ? AND uri != ?
+			 GROUP BY uri
+			 ORDER BY observed DESC
+			 LIMIT ?`,
+		)
+		.bind(input.did, input.excludeUri, input.limit)
+		.all<{ uri: string; observed: number }>();
+	return (rows.results ?? []).map((row) => row.uri);
+}
+
+/**
+ * Distinct OTHER publishing DIDs that have submitted an assessment for the same
+ * exact artifact checksum (plan W8.4 D2) — the cross-publisher reuse signal.
+ * Global: it spans every publisher's assessments, not just the queried
+ * subject's. Capped at `limit`.
+ */
+export async function getPublishersSharingChecksum(
+	db: D1Database,
+	input: { checksum: string; excludeDid: string; limit: number },
+): Promise<string[]> {
+	const rows = await db
+		.prepare(
+			`SELECT DISTINCT s.did
+			 FROM assessments a
+			 JOIN subjects s ON s.uri = a.uri AND s.cid = a.cid
+			 WHERE a.artifact_checksum = ? AND s.did != ?
+			 ORDER BY s.did
+			 LIMIT ?`,
+		)
+		.bind(input.checksum, input.excludeDid, input.limit)
+		.all<{ did: string }>();
+	return (rows.results ?? []).map((row) => row.did);
+}
+
 function rowToAssessment(row: AssessmentRow): Assessment {
 	return {
 		id: row.id,

apps/labeler/src/findings.ts

@@ -35,11 +35,30 @@ export interface ModelFindingMetadata {
  * finding; model/image stages report the model and prompt version. */
 export type FindingSourceMetadata = ToolFindingMetadata | ModelFindingMetadata;
 
+/**
+ * The category vocabulary a `source: "history"` finding may cite (plan W8.4
+ * D4). History findings are context, never labels — the resolver drops them
+ * before any category→label mapping — so they cite this dedicated set rather
+ * than the policy's automated-block ∪ warning label values, and
+ * `validateFinding` holds them to it exclusively.
+ */
+export type HistoryFindingCategory =
+	| "publisher-history"
+	| "shared-artifact"
+	| "active-manual-label";
+
+export const HISTORY_FINDING_CATEGORIES: ReadonlySet<string> = new Set<HistoryFindingCategory>([
+	"publisher-history",
+	"shared-artifact",
+	"active-manual-label",
+]);
+
 export interface NormalizedFinding {
 	source: FindingSource;
-	/** A label value from the policy vocabulary — validated against
-	 * `allowedFindingCategories` (automated-block ∪ warning), never the
-	 * eligibility or manual-system label values. */
+	/** For non-history sources, a label value from the policy vocabulary —
+	 * validated against `allowedFindingCategories` (automated-block ∪ warning),
+	 * never the eligibility or manual-system label values. For `source:
+	 * "history"`, one of `HISTORY_FINDING_CATEGORIES` instead (plan W8.4 D4). */
 	category: string;
 	severity: FindingSeverity;
 	confidence?: number;
@@ -77,8 +96,14 @@ export function validateFinding(finding: unknown, opts: ValidateFindingOptions):
 	if (typeof source !== "string" || !isFindingSource(source))
 		throw new FindingValidationError(`finding.source is invalid: ${String(source)}`);
 
+	// History findings are exempt from the automated-block ∪ warning constraint
+	// (plan W8.4 D4): they cite the dedicated `HISTORY_FINDING_CATEGORIES` set,
+	// and only that set — a history finding may not borrow a block/warn label
+	// value, nor a non-history finding a history category.
+	const allowedCategories =
+		source === "history" ? HISTORY_FINDING_CATEGORIES : opts.allowedCategories;
 	const category = finding.category;
-	if (typeof category !== "string" || !opts.allowedCategories.has(category))
+	if (typeof category !== "string" || !allowedCategories.has(category))
 		throw new FindingValidationError(
 			`finding.category is not an allowed finding category: ${String(category)}`,
 		);

apps/labeler/src/history-context.ts

@@ -0,0 +1,168 @@
+/**
+ * Publisher-history context stage (plan W8.4, slice 1). Produces
+ * `source: "history"` normalized findings from the labeler's OWN D1 — prior
+ * releases from the publishing DID, the same artifact checksum submitted under
+ * other DIDs, and existing active manual labels on the subject.
+ *
+ * These findings are bounded, factual context an operator reads through the
+ * assessment projection. They are structurally never turned into labels: the
+ * resolver drops every `source: "history"` finding before any category→label
+ * mapping (`policy-resolver.ts`), and `validateFinding` holds them to the
+ * dedicated `HISTORY_FINDING_CATEGORIES` set (`findings.ts`), disjoint from the
+ * policy's label vocabulary.
+ *
+ * The two deferred inputs — recent handle/profile changes and verification
+ * state — live only in the aggregator's D1, which the labeler has no binding to
+ * reach; they are gated on a ratified read path (plan W8.4 D1) and not built
+ * here.
+ */
+
+import {
+	getActiveLabelState,
+	getCurrentSubjectByUri,
+	getPriorReleaseUrisForDid,
+	getPublishersSharingChecksum,
+	type Assessment,
+} from "./assessment-store.js";
+import type { NormalizedFinding } from "./findings.js";
+
+const HISTORY_TOOL = "publisher-history";
+const HISTORY_TOOL_VERSION = "1";
+
+const DEFAULT_PRIOR_RELEASE_LIMIT = 20;
+const DEFAULT_SHARED_PUBLISHER_LIMIT = 20;
+/** How many URIs/DIDs to name in a finding's private detail. */
+const SAMPLE_SIZE = 5;
+
+export interface HistoryContextOptions {
+	/** The labeler's own DID (`src`) — the stream whose active manual labels
+	 * count as context. */
+	src: string;
+	priorReleaseLimit?: number;
+	sharedPublisherLimit?: number;
+	now?: Date;
+}
+
+/**
+ * DB-bound stage adapter matching the orchestrator's stage contract: returns
+ * `NormalizedFinding[]`, one per input that has something to report. History is
+ * operator-only context that never becomes a label, so a failure to gather it
+ * must never fail the assessment run — that would discard the decision-relevant
+ * findings from every other stage. The stage is best-effort: on any error it
+ * logs and returns no findings rather than throwing, so it can never gate the
+ * run regardless of its position in `STAGE_ORDER`.
+ */
+export async function analyzeHistory(
+	db: D1Database,
+	assessment: Assessment,
+	opts: HistoryContextOptions,
+): Promise<NormalizedFinding[]> {
+	const priorReleaseLimit = opts.priorReleaseLimit ?? DEFAULT_PRIOR_RELEASE_LIMIT;
+	const sharedPublisherLimit = opts.sharedPublisherLimit ?? DEFAULT_SHARED_PUBLISHER_LIMIT;
+
+	try {
+		const findings: NormalizedFinding[] = [];
+		const subject = await getCurrentSubjectByUri(db, assessment.uri);
+
+		if (subject) {
+			const priorUris = await getPriorReleaseUrisForDid(db, {
+				did: subject.did,
+				excludeUri: assessment.uri,
+				limit: priorReleaseLimit,
+			});
+			if (priorUris.length > 0)
+				findings.push(priorReleasesFinding(subject.did, priorUris, priorReleaseLimit));
+
+			if (assessment.artifactChecksum) {
+				const otherDids = await getPublishersSharingChecksum(db, {
+					checksum: assessment.artifactChecksum,
+					excludeDid: subject.did,
+					limit: sharedPublisherLimit,
+				});
+				if (otherDids.length > 0)
+					findings.push(
+						sharedArtifactFinding(assessment.artifactChecksum, otherDids, sharedPublisherLimit),
+					);
+			}
+		}
+
+		const labelState = await getActiveLabelState(db, {
+			src: opts.src,
+			uri: assessment.uri,
+			cid: assessment.cid,
+			...(opts.now !== undefined ? { now: opts.now } : {}),
+		});
+		const activeManualLabels = [...labelState.values()]
+			.filter((winner) => !winner.automated && winner.active)
+			.map((winner) => winner.val);
+		if (activeManualLabels.length > 0) findings.push(activeManualLabelFinding(activeManualLabels));
+
+		return findings;
+	} catch (err) {
+		console.error(
+			`[history-context] lookup failed, skipping context for this run: ${err instanceof Error ? err.message : String(err)}`,
+		);
+		return [];
+	}
+}
+
+function toolMetadata() {
+	return { kind: "tool", tool: HISTORY_TOOL, version: HISTORY_TOOL_VERSION } as const;
+}
+
+function countText(count: number, capped: boolean): string {
+	return capped ? `at least ${count}` : `${count}`;
+}
+
+function pluralize(count: number, noun: string): string {
+	return count === 1 ? noun : `${noun}s`;
+}
+
+function priorReleasesFinding(did: string, uris: string[], limit: number): NormalizedFinding {
+	const capped = uris.length >= limit;
+	const count = countText(uris.length, capped);
+	return {
+		source: "history",
+		category: "publisher-history",
+		severity: "info",
+		title: `Publisher has ${count} prior ${pluralize(uris.length, "release")}`,
+		publicSummary: `The publishing account has ${count} other ${pluralize(uris.length, "release")} known to the labeler.`,
+		privateDetail: `Publisher ${did} has ${count} prior ${pluralize(uris.length, "release")}. Sample: ${uris.slice(0, SAMPLE_SIZE).join(", ")}.`,
+		evidenceRefs: [],
+		sourceMetadata: toolMetadata(),
+	};
+}
+
+function sharedArtifactFinding(checksum: string, dids: string[], limit: number): NormalizedFinding {
+	const capped = dids.length >= limit;
+	const count = countText(dids.length, capped);
+	// Cross-publisher artifact reuse is a correlation/deanonymization signal, so
+	// the title and publicSummary stay non-revealing — the specifics live only in
+	// privateDetail.
+	return {
+		source: "history",
+		category: "shared-artifact",
+		severity: "low",
+		title: "Artifact provenance context",
+		publicSummary: "Operator-only artifact-provenance context is recorded for this release.",
+		privateDetail: `Artifact checksum ${checksum} also submitted by ${count} other ${pluralize(dids.length, "publisher")}: ${dids.slice(0, SAMPLE_SIZE).join(", ")}.`,
+		evidenceRefs: [],
+		sourceMetadata: toolMetadata(),
+	};
+}
+
+function activeManualLabelFinding(vals: string[]): NormalizedFinding {
+	// The existence and identity of manual labels (including redactions) must not
+	// leak into a public projection, so the title and publicSummary reveal
+	// nothing — the label values live only in privateDetail.
+	return {
+		source: "history",
+		category: "active-manual-label",
+		severity: "info",
+		title: "Operator label context",
+		publicSummary: "Operator-only label context is recorded for this subject.",
+		privateDetail: `Active manual labels: ${vals.join(", ")}.`,
+		evidenceRefs: [],
+		sourceMetadata: toolMetadata(),
+	};
+}