fix(labeler): keep assessment-pending live until the last concurrent run resolves (held review finding #13)

apps/labeler/src/assessment-orchestrator.ts

@@ -32,6 +32,7 @@ import type { ModerationPolicy } from "./policy.js";
 import {
 	buildIssuanceStatements,
 	markPublicationAccepted,
+	readIssuedLabelByActionKey,
 	type AutomatedIssuanceAction,
 	type AutomatedLabelProposal,
 	type IssuedLabel,
@@ -351,8 +352,40 @@ export class AssessmentOrchestrator {
 		if (toState === "error") {
 			await issue("assessment-error", false);
 		}
-		// Spec §9.9 point 9: always negate this run's own assessment-pending.
-		await issue("assessment-pending", true);
+		// Spec §9.9 point 9: negate this run's own assessment-pending — but only when
+		// this is the last run in flight for the subject. `requireNoOtherActiveRun`
+		// no-ops the negation while a sibling run for the same (uri, cid) is still
+		// non-terminal, so the positive assessment-pending stays the stream head and
+		// the release stays gated; the last run to finalize clears it.
+		const pendingNegationKey = automatedIdempotencyKey(
+			assessment.runKey,
+			"assessment-pending",
+			true,
+		);
+		const pendingNegation = await buildIssuanceStatements(
+			this.db,
+			this.config,
+			this.signer,
+			{
+				actor: this.config.labelerDid,
+				type: "automated-assessment",
+				assessmentId: assessment.id,
+				reason: `assessment ${toState}`,
+				idempotencyKey: pendingNegationKey,
+			},
+			{ uri: assessment.uri, cid: assessment.cid, val: "assessment-pending", neg: true },
+			now,
+			this.publisher !== undefined,
+			{
+				requireAssessmentState: toState,
+				requireNoOtherActiveRun: {
+					uri: assessment.uri,
+					cid: assessment.cid,
+					assessmentId: assessment.id,
+				},
+			},
+		);
+		statements.push(...pendingNegation.statements);
 
 		// Spec §9.9 point 8 / §10: negate prior active automated labels this
 		// outcome no longer supports.
@@ -428,6 +461,12 @@ export class AssessmentOrchestrator {
 		}
 		const issued: IssuedLabel[] = [];
 		for (const postCommit of postCommits) issued.push(await postCommit());
+		// The pending negation is suppressed (writes no row) while a sibling run is
+		// still in flight; broadcast it only when it committed. It is the only label
+		// here that can legitimately be absent after the CAS succeeded — the CAS shares
+		// its signing guard — so its absence needs no signing diagnosis.
+		const pendingNegated = await readIssuedLabelByActionKey(this.db, pendingNegationKey);
+		if (pendingNegated) issued.push(pendingNegated);
 		await this.publishLabels(issued);
 
 		const finalised = await getAssessment(this.db, assessment.id);

apps/labeler/src/assessment-store.ts

@@ -387,25 +387,14 @@ export async function listNonTerminalAssessmentsForUri(
 	return (rows.results ?? []).map(rowToAssessment);
 }
 
-/** States a run can be in and still carry a live, un-negated positive
- * `assessment-pending`: every non-terminal state, plus terminal `stale` (which,
- * unlike the other terminals, does not negate its own pending on transition).
- * `cancelled` is excluded — the delete negates before cancelling; the decision
- * outcomes negate their own pending at finalization. */
-const PENDING_BEARING_STATES: readonly AssessmentState[] = [
-	"observed",
-	"verifying",
-	"pending",
-	"running",
-	"stale",
-];
-
 /**
- * Runs for a URI that could still carry a live positive `assessment-pending`
- * label — the delete cleanup's scan set (spec §9.1). Widens
- * `listNonTerminalAssessmentsForUri` to include terminal `stale` runs: a run that
- * self-transitioned to `stale` on detecting a deleted/superseded subject keeps
- * its committed positive, so the delete must still reach it to negate.
+ * Runs for a URI the delete cleanup must reach (spec §9.1): every non-terminal
+ * run (to cancel it and negate any pending it issued) plus any run — of ANY state,
+ * terminal included — still holding a committed, un-negated positive
+ * `assessment-pending`. A run that self-stales on a deleted/superseded subject, or
+ * a decision run that suppressed its own pending-negation while a sibling was still
+ * in flight (`requireNoOtherActiveRun`), stays terminal yet keeps its positive, so
+ * its state alone does not identify it — the label stream does.
  */
 export async function listPendingBearingAssessmentsForUri(
 	db: D1Database,
@@ -416,10 +405,19 @@ export async function listPendingBearingAssessmentsForUri(
 			`SELECT id, run_key, uri, cid, artifact_id, artifact_checksum, state, trigger, trigger_id,
 			 policy_version, model_id, prompt_hash, public_summary, coverage_json,
 			 supersedes_assessment_id, started_at, completed_at, created_at
-			 FROM assessments
-			 WHERE uri = ? AND state IN (${PENDING_BEARING_STATES.map(() => "?").join(", ")})`,
+			 FROM assessments a
+			 WHERE a.uri = ?
+			   AND (
+				 a.state NOT IN (${Array.from(TERMINAL_STATES, () => "?").join(", ")})
+				 OR (
+				   EXISTS (SELECT 1 FROM issued_labels l JOIN issuance_actions act ON act.id = l.action_id
+				           WHERE act.assessment_id = a.id AND l.val = 'assessment-pending' AND l.neg = 0)
+				   AND NOT EXISTS (SELECT 1 FROM issued_labels l JOIN issuance_actions act ON act.id = l.action_id
+				           WHERE act.assessment_id = a.id AND l.val = 'assessment-pending' AND l.neg = 1)
+				 )
+			   )`,
 		)
-		.bind(uri, ...PENDING_BEARING_STATES)
+		.bind(uri, ...TERMINAL_STATES)
 		.all<AssessmentRow>();
 	return (rows.results ?? []).map(rowToAssessment);
 }

apps/labeler/src/discovery-consumer.ts

@@ -45,6 +45,7 @@ import {
 	automatedIdempotencyKey,
 	computeRunKey,
 	initialTriggerId,
+	TERMINAL_STATES,
 } from "./assessment-lifecycle.js";
 import {
 	buildAssessmentRunStatement,
@@ -596,22 +597,23 @@ async function transitionOrObserve(
 
 /**
  * Tombstones the subject (advancing its `delete_generation`) and retires every
- * run that could still carry a live positive `assessment-pending` — including
- * terminal `stale` runs, which self-transition on detecting a deleted subject and
- * do NOT negate their own pending. For each such run the negation is issued
- * BEFORE any cancellation, so a failed or paused negation (signing mid-rotation)
- * leaves a non-terminal run non-terminal and re-discoverable on redelivery;
- * cancelling first would drop it from the scan set, stranding the pending live
- * once the message acks.
+ * run that could still carry a live positive `assessment-pending` — including any
+ * terminal run still holding a committed, un-negated positive (a `stale` run that
+ * self-transitioned on a deleted/superseded subject, or a decision run that
+ * suppressed its own pending-negation while a sibling was in flight). For each such
+ * run the negation is issued BEFORE any cancellation, so a failed or paused
+ * negation (signing mid-rotation) leaves a non-terminal run non-terminal and
+ * re-discoverable on redelivery; cancelling first would drop it from the scan set,
+ * stranding the pending live once the message acks.
  *
  * The negation is keyed on the run having committed a live positive — NOT on its
  * lifecycle state — because an operator rerun issues its positive while the run is
- * still `observed`, and a stale run keeps its positive after going terminal.
- * Cancellation applies only to non-terminal runs (a stale run is already
- * terminal). The invariant: a run is cancelled only after any positive it
- * committed has been negated, and the message cannot ack while a negation is still
- * owed (a throw propagates to the delete handler's mutation-phase catch, which
- * always retries), so no active `assessment-pending` survives an acked delete.
+ * still `observed`, and a terminal run keeps its positive after finalizing.
+ * Cancellation applies only to non-terminal runs (a terminal run needs none). The
+ * invariant: a run is cancelled only after any positive it committed has been
+ * negated, and the message cannot ack while a negation is still owed (a throw
+ * propagates to the delete handler's mutation-phase catch, which always retries),
+ * so no active `assessment-pending` survives an acked delete.
  */
 async function applyDiscoveryDelete(
 	deps: DiscoveryConsumerDeps,
@@ -621,8 +623,8 @@ async function applyDiscoveryDelete(
 	await deleteSubjectsByUri(deps.db, { uri, now });
 	const runs = await listPendingBearingAssessmentsForUri(deps.db, uri);
 	for (const run of runs) {
-		// Negate (before any cancel) any run — non-terminal OR terminal `stale` —
-		// that committed a positive pending and has not already been negated.
+		// Negate (before any cancel) any run — non-terminal or terminal — that
+		// committed a positive pending and has not already been negated.
 		const positive = await readIssuedLabelByActionKey(
 			deps.db,
 			automatedIdempotencyKey(run.runKey, "assessment-pending", false),
@@ -634,7 +636,7 @@ async function applyDiscoveryDelete(
 			);
 			if (!negated) await negateRunPendingLabel(deps, run, now);
 		}
-		if (run.state === "stale") continue; // already terminal — negated, nothing to cancel
+		if (TERMINAL_STATES.has(run.state)) continue; // already terminal — negated, nothing to cancel
 		try {
 			await transitionAssessmentState(deps.db, {
 				id: run.id,

apps/labeler/src/service.ts

@@ -1,6 +1,6 @@
 import type { LabelSigner, SignedLabel } from "@emdash-cms/registry-moderation";
 
-import { ASSESSMENT_ID, type AssessmentState } from "./assessment-lifecycle.js";
+import { ASSESSMENT_ID, TERMINAL_STATES, type AssessmentState } from "./assessment-lifecycle.js";
 import { getNegatableAutomatedLabels } from "./assessment-store.js";
 import type { LabelerConfig } from "./config.js";
 import type { FindingSeverity } from "./evidence.js";
@@ -132,6 +132,18 @@ export interface BuildIssuanceOptions {
 	 * action (not the label) leaves no orphan label, same as the state guard.
 	 */
 	requireSubjectNotDeleted?: { uri: string; cid: string; generation: number };
+	/**
+	 * Gate the action insert (and thus its label) on NO OTHER assessment run for the
+	 * same subject `(uri, cid)` being non-terminal at commit time. Finalization pairs
+	 * this with `requireAssessmentState` on the `assessment-pending` negation so the
+	 * shared pending gate clears only when the LAST run resolves: while any sibling
+	 * run is still in flight, this run's negation no-ops and the positive
+	 * `assessment-pending` stays the stream head, holding the release gated.
+	 * `assessmentId` excludes this run itself from the sibling scan (by id), so
+	 * self-exclusion holds regardless of batch ordering; siblings are read as their
+	 * committed terminal state. Requires an automated-assessment action.
+	 */
+	requireNoOtherActiveRun?: { uri: string; cid: string; assessmentId: string };
 }
 
 /**
@@ -232,6 +244,15 @@ export async function buildIssuanceStatements(
 			? ""
 			: `\n\t\t\t\t AND EXISTS (SELECT 1 FROM subjects
 				 WHERE uri = ? AND cid = ? AND deleted_at IS NULL AND delete_generation = ?)`;
+	const requireNoOtherActiveRun = options.requireNoOtherActiveRun;
+	if (requireNoOtherActiveRun !== undefined && action.type !== "automated-assessment")
+		throw new TypeError("requireNoOtherActiveRun requires an automated-assessment action");
+	const activeRunGuardSql =
+		requireNoOtherActiveRun === undefined
+			? ""
+			: `\n\t\t\t\t AND NOT EXISTS (SELECT 1 FROM assessments
+				 WHERE uri = ? AND cid = ? AND id != ?
+				   AND state NOT IN (${Array.from(TERMINAL_STATES, () => "?").join(", ")}))`;
 	const actionBinds: unknown[] = [
 		action.actor,
 		action.type,
@@ -253,6 +274,13 @@ export async function buildIssuanceStatements(
 	if (requireState !== undefined) actionBinds.push(assessmentId, requireState);
 	if (requireSubject !== undefined)
 		actionBinds.push(requireSubject.uri, requireSubject.cid, requireSubject.generation);
+	if (requireNoOtherActiveRun !== undefined)
+		actionBinds.push(
+			requireNoOtherActiveRun.uri,
+			requireNoOtherActiveRun.cid,
+			requireNoOtherActiveRun.assessmentId,
+			...TERMINAL_STATES,
+		);
 
 	const statements: D1PreparedStatement[] = [
 		db
@@ -277,7 +305,7 @@ export async function buildIssuanceStatements(
 						  )
 						  AND l2.neg = 0 AND a2.type <> 'automated-assessment'
 					)
-				 )${stateGuardSql}${subjectGuardSql}
+				 )${stateGuardSql}${subjectGuardSql}${activeRunGuardSql}
 				 ON CONFLICT(idempotency_key) DO NOTHING`,
 			)
 			.bind(...actionBinds),

apps/labeler/test/assessment-orchestrator.test.ts

@@ -12,7 +12,12 @@ import {
 	type AcquisitionHolder,
 	type AcquisitionTarget,
 } from "../src/artifact-acquisition.js";
-import { computeRunKey, initialTriggerId, operatorTriggerId } from "../src/assessment-lifecycle.js";
+import {
+	automatedIdempotencyKey,
+	computeRunKey,
+	initialTriggerId,
+	operatorTriggerId,
+} from "../src/assessment-lifecycle.js";
 import {
 	AssessmentFinalizationConflictError,
 	AssessmentOrchestrator,
@@ -26,14 +31,20 @@ import {
 	createAssessmentRun,
 	createSubject,
 	deleteSubject,
+	getActiveLabelState,
 	getAssessment,
 	getCurrentAssessment,
 	transitionAssessmentState,
 } from "../src/assessment-store.js";
 import { FindingValidationError, HISTORY_FINDING_CATEGORIES } from "../src/findings.js";
 import { analyzeHistory } from "../src/history-context.js";
 import { MODERATION_POLICY } from "../src/policy.js";
-import { issueManualLabel, type IssuedLabel } from "../src/service.js";
+import {
+	issueAutomatedAssessmentLabel,
+	issueManualLabel,
+	readIssuedLabelByActionKey,
+	type IssuedLabel,
+} from "../src/service.js";
 import {
 	abortRoutineKeyRotation,
 	beginRoutineKeyRotation,
@@ -1388,6 +1399,84 @@ describe("AssessmentOrchestrator: delete racing finalization (Blocker 1)", () =>
 	});
 });
 
+/** Issues a run's initial positive `assessment-pending`, the way the discovery
+ * consumer does, so the subject carries a live pending gate before a run finalizes. */
+async function issuePendingPositive(run: {
+	id: string;
+	uri: string;
+	cid: string;
+	runKey: string;
+}): Promise<void> {
+	await issueAutomatedAssessmentLabel(
+		testEnv.DB,
+		config,
+		await signer(),
+		{
+			actor: LABELER_DID,
+			type: "automated-assessment",
+			assessmentId: run.id,
+			reason: "initial discovery",
+			idempotencyKey: automatedIdempotencyKey(run.runKey, "assessment-pending", false),
+		},
+		{ uri: run.uri, cid: run.cid, val: "assessment-pending" },
+	);
+}
+
+describe("AssessmentOrchestrator: concurrent runs share one pending gate", () => {
+	it("keeps assessment-pending live while a sibling run for the same subject is in flight, clearing it only when the last run finalizes", async () => {
+		const cidValue = await cid("shared-pending");
+		const runA = await pendingRun({
+			name: "shared-pending",
+			cidValue,
+			triggerId: initialTriggerId(cidValue),
+		});
+		const runB = await pendingRun({
+			name: "shared-pending",
+			cidValue,
+			triggerId: operatorTriggerId("op-shared-pending"),
+		});
+		await issuePendingPositive(runA);
+		await issuePendingPositive(runB);
+
+		const orchestrator = await buildOrchestrator();
+
+		// A finalizes while B is still `pending`. Its own pending-negation is
+		// suppressed, so the gate B also depends on stays live.
+		const a = await orchestrator.runAssessment(runA.id);
+		expect(a.state).toBe("passed");
+
+		const aNegation = await readIssuedLabelByActionKey(
+			testEnv.DB,
+			automatedIdempotencyKey(runA.runKey, "assessment-pending", true),
+		);
+		expect(aNegation).toBeNull();
+
+		const gateWhileBActive = await getActiveLabelState(testEnv.DB, {
+			src: LABELER_DID,
+			uri: runA.uri,
+			cid: cidValue,
+		});
+		expect(gateWhileBActive.get("assessment-pending")?.active).toBe(true);
+
+		// B is the last run in flight; finalizing it clears the shared gate.
+		const b = await orchestrator.runAssessment(runB.id);
+		expect(b.state).toBe("passed");
+
+		const bNegation = await readIssuedLabelByActionKey(
+			testEnv.DB,
+			automatedIdempotencyKey(runB.runKey, "assessment-pending", true),
+		);
+		expect(bNegation).not.toBeNull();
+
+		const gateAfterLast = await getActiveLabelState(testEnv.DB, {
+			src: LABELER_DID,
+			uri: runB.uri,
+			cid: cidValue,
+		});
+		expect(gateAfterLast.get("assessment-pending")?.active).toBe(false);
+	});
+});
+
 describe("AssessmentOrchestrator: automation pause between prep and commit", () => {
 	it("leaves the run running and issues no labels when automation pauses before the batch commits", async () => {
 		const run = await pendingRun({