diff --git a/apps/labeler/src/assessment-orchestrator.ts b/apps/labeler/src/assessment-orchestrator.ts index 1384622274..bb27f7e694 100644 --- a/apps/labeler/src/assessment-orchestrator.ts +++ b/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); diff --git a/apps/labeler/src/assessment-store.ts b/apps/labeler/src/assessment-store.ts index 6b28b705bb..da2d7e20a2 100644 --- a/apps/labeler/src/assessment-store.ts +++ b/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(); return (rows.results ?? []).map(rowToAssessment); } diff --git a/apps/labeler/src/discovery-consumer.ts b/apps/labeler/src/discovery-consumer.ts index f481e26663..e8e4f563b3 100644 --- a/apps/labeler/src/discovery-consumer.ts +++ b/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, diff --git a/apps/labeler/src/service.ts b/apps/labeler/src/service.ts index 7c3fb62dde..58c4e4c8e2 100644 --- a/apps/labeler/src/service.ts +++ b/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), diff --git a/apps/labeler/test/assessment-orchestrator.test.ts b/apps/labeler/test/assessment-orchestrator.test.ts index 8b8cdff7c5..31d7d1a51d 100644 --- a/apps/labeler/test/assessment-orchestrator.test.ts +++ b/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,6 +31,7 @@ import { createAssessmentRun, createSubject, deleteSubject, + getActiveLabelState, getAssessment, getCurrentAssessment, transitionAssessmentState, @@ -33,7 +39,12 @@ import { 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 { + 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({ diff --git a/apps/labeler/test/discovery-consumer.test.ts b/apps/labeler/test/discovery-consumer.test.ts index 0aacd97439..e0c28ca27d 100644 --- a/apps/labeler/test/discovery-consumer.test.ts +++ b/apps/labeler/test/discovery-consumer.test.ts @@ -900,6 +900,45 @@ describe("processDiscoveryMessage: delete", () => { expect((await latestPending())?.neg).toBe(1); }); + it("negates a live pending positive still held by a terminal decision run on delete", async () => { + const job = await jobFor({ rkey: rkey() }); + const deps = await buildDeps(); + await processDiscoveryMessage(job, new FakeMessage(), { ...deps, verify: verifiedFor(job) }); + const runKey = await runKeyFor(job); + const run = await getAssessmentByRunKey(testEnv.DB, runKey); + + // A finalization that suppressed its own pending-negation (a sibling run was + // still in flight) leaves a terminal `passed` run that still holds a live + // positive assessment-pending — the orchestrator produces exactly this state. + await testEnv.DB.prepare( + `UPDATE assessments SET state = 'passed', completed_at = ?, completed_at_epoch_ms = ? WHERE id = ?`, + ) + .bind(new Date().toISOString(), Date.now(), run!.id) + .run(); + + const latestPending = () => + testEnv.DB.prepare( + `SELECT neg FROM issued_labels WHERE uri = ? AND val = 'assessment-pending' + ORDER BY sequence DESC LIMIT 1`, + ) + .bind(uriFor(job)) + .first<{ neg: number }>(); + expect((await latestPending())?.neg).toBe(0); + + const deleteJob: DiscoveryJob = { ...job, operation: "delete", cid: "" }; + const msg = new FakeMessage(); + await processDiscoveryMessage(deleteJob, msg, { + ...deps, + confirmDeleted: () => Promise.resolve(true), + }); + + expect(msg.acked).toBe(1); + expect(msg.retried).toBe(0); + // The delete cleanup reaches the terminal run's un-negated positive and clears + // it, so no active assessment-pending survives the delete. + expect((await latestPending())?.neg).toBe(1); + }); + it("dead-letters a forged/premature delete whose record still resolves, suppressing nothing", async () => { const job = await jobFor({ rkey: rkey() }); const deps = await buildDeps();