feat(registry): labeler discovery and assessment orchestration#1978
Conversation
Independent Jetstream discovery of release records: a per-labeler discovery DO maintains the subscription (cursor in ingest_state), and the queue consumer verifies each event's exact URI+CID against the publisher's signed record before creating a subject and issuing assessment-pending — a forged or unverifiable event dead-letters and produces no label. The assessment orchestrator drives pending through the stage pipeline to an atomic finalization that composes the current-pointer update with label issuance and negates only this run's own prior automated labels, never a manually-issued one. Production wiring stops at assessment-pending; the orchestrator and its stage adapters ship with deterministic stubs exercised only by tests, enforced by a static production-boundary test. Scheduled reconciliation flags stuck runs and unassessed subjects.
…nance Adversarial review. A thrown DID-document resolution is now a transient RecordVerificationError the consumer retries, rather than permanently dead-lettering a release on a directory blip at publish time; a document that resolves but lacks a PDS or key stays permanent. getNegatableAutomated Labels computes the active stream head across all issuers, so a val whose latest event is manual is never returned for automated negation. The production-boundary test also catches bare side-effect imports. The finalize() currency-check comment no longer claims a lock it doesn't hold and documents the in-batch guard / workflow lock required to wire the orchestrator live in W7/W8.
Re-check subject currency inside finalize() with no signing between the check and db.batch, shrinking the delete/cancel window from every label's signing round-trip to two adjacent D1 ops. Full closure still needs the workflow lock deferred to W7/W8; a mid-run delete injected via a stage now proves the re-check aborts before issuing any label.
|
Scope checkThis PR changes 3,692 lines across 23 files. Large PRs are harder to review and more likely to be closed without review. If this scope is intentional, no action needed. A maintainer will review it. If not, please consider splitting this into smaller PRs. See CONTRIBUTING.md for contribution guidelines. |
Deploying with
|
| Status | Name | Latest Commit | Updated (UTC) |
|---|---|---|---|
| ✅ Deployment successful! View logs |
emdash-playground | 70d5cc9 | Jul 12 2026, 02:41 PM |
Deploying with
|
| Status | Name | Latest Commit | Updated (UTC) |
|---|---|---|---|
| ✅ Deployment successful! View logs |
emdash-demo-do | 70d5cc9 | Jul 12 2026, 02:42 PM |
Deploying with
|
| Status | Name | Latest Commit | Updated (UTC) |
|---|---|---|---|
| ✅ Deployment successful! View logs |
emdash-demo-cache | 70d5cc9 | Jul 12 2026, 02:42 PM |
![emdashbot[bot]](https://avatars.githubusercontent.com/in/3255304?s=60&v=4){kind=small}
There was a problem hiding this comment.
Approach judgment
This PR is the right next slice for W6: it keeps the production path at assessment-pending, ships the orchestrator as test-only code behind a clear boundary test, and adds the independent Jetstream discovery consumer, reconciliation, and verification plumbing. The structure matches the aggregator precedent and the persistence model from #1976. Test coverage is extensive and the persistence-layer atomicity contract is well documented.
However, there are several real correctness and security gaps in the production-bound discovery path and the finalization/negation plumbing that should be fixed before this merges. The most important are:
- Negation candidates are not scoped to this labeler.
getNegatableAutomatedLabelsqueries across allsrcvalues, so a label stream from another issuer can incorrectly drive negations in this labeler's own stream, or suppress negation of this labeler's own active labels. - Mid-download network failures are treated as permanent. In
pds-verify.ts, a stream read error after headers are received is rethrown as a genericErrorand reaches the consumer as an unexpected failure, which dead-letters instead of retrying — contradicting the stated transient-policy. - Delete events are not verified. The discovery consumer trusts Jetstream delete events and immediately cancels non-terminal assessments. A forged or relay-compromised delete can suppress labels without the PDS verification that create/update events require.
- Only one blocking label is issued.
resolvePolicyOutcomeusesfindings.findfor critical blockers, so a run with multiple blocking findings only emits the first one; warnings correctly dedupe and emit all.
I also note the acknowledged deferrals from the PR description (finalization/delete race, create-after-delete tombstone resurrection) but do not re-file them.
| export async function getNegatableAutomatedLabels( | ||
| db: D1Database, | ||
| input: { uri: string; cid: string }, | ||
| ): Promise<NegatableAutomatedLabel[]> { | ||
| const rows = await db | ||
| .prepare( | ||
| // The inner MAX reflects the TRUE stream head across all issuers, so a | ||
| // val whose latest event was a manual action is never returned — only | ||
| // a val whose current active event is an automated non-negation is a | ||
| // candidate for automated negation (§10). | ||
| `SELECT l.val | ||
| FROM issued_labels l | ||
| JOIN issuance_actions a ON a.id = l.action_id | ||
| WHERE l.uri = ? AND l.cid = ? AND a.type = 'automated-assessment' AND l.neg = 0 | ||
| AND l.sequence = ( | ||
| SELECT MAX(l2.sequence) FROM issued_labels l2 | ||
| WHERE l2.uri = l.uri AND l2.cid = l.cid AND l2.val = l.val | ||
| )`, | ||
| ) |
There was a problem hiding this comment.
[needs fixing] getNegatableAutomatedLabels returns automated positive labels without filtering by src. In AT Protocol each labeler maintains its own stream of labels and can only negate labels it issued. Because the outer query is not scoped to this.config.labelerDid, another issuer's later automated positive can become the "active" candidate, causing this labeler to emit a negation for its own stream for a label it may not have issued, or worse, fail to negate its own active label when another issuer has a newer one. Both the outer query and the inner MAX(sequence) lookup should be scoped to the calling source, and the caller in assessment-orchestrator.ts should pass src: this.config.labelerDid.
| export async function getNegatableAutomatedLabels( | |
| db: D1Database, | |
| input: { uri: string; cid: string }, | |
| ): Promise<NegatableAutomatedLabel[]> { | |
| const rows = await db | |
| .prepare( | |
| // The inner MAX reflects the TRUE stream head across all issuers, so a | |
| // val whose latest event was a manual action is never returned — only | |
| // a val whose current active event is an automated non-negation is a | |
| // candidate for automated negation (§10). | |
| `SELECT l.val | |
| FROM issued_labels l | |
| JOIN issuance_actions a ON a.id = l.action_id | |
| WHERE l.uri = ? AND l.cid = ? AND a.type = 'automated-assessment' AND l.neg = 0 | |
| AND l.sequence = ( | |
| SELECT MAX(l2.sequence) FROM issued_labels l2 | |
| WHERE l2.uri = l.uri AND l2.cid = l.cid AND l2.val = l.val | |
| )`, | |
| ) | |
| export async function getNegatableAutomatedLabels( | |
| db: D1Database, | |
| input: { src: string; uri: string; cid: string }, | |
| ): Promise<NegatableAutomatedLabel[]> { | |
| const rows = await db | |
| .prepare( | |
| // The inner MAX reflects the stream head for this labeler's own | |
| // source, so a manually-negated or manually-issued val from this | |
| // source is never returned as a candidate. We only negate labels we | |
| // issued ourselves (spec §10). | |
| `SELECT l.val | |
| FROM issued_labels l | |
| JOIN issuance_actions a ON a.id = l.action_id | |
| WHERE l.src = ? AND l.uri = ? AND l.cid = ? AND a.type = 'automated-assessment' AND l.neg = 0 | |
| AND l.sequence = ( | |
| SELECT MAX(l2.sequence) FROM issued_labels l2 | |
| WHERE l2.src = l.src AND l2.uri = l.uri AND l2.cid = l.cid AND l2.val = l.val | |
| )`, | |
| ) | |
| .bind(input.src, input.uri, input.cid) | |
| .all<{ val: string }>(); | |
| return (rows.results ?? []).map((row) => ({ val: row.val })); | |
| } |
Call site in assessment-orchestrator.ts should also be updated to pass src: this.config.labelerDid.
| } catch (err) { | ||
| // Cancel the stream so the underlying socket isn't left dangling. | ||
| await reader.cancel().catch(() => { | ||
| /* swallow — we already have a primary error to surface */ | ||
| }); | ||
| throw err; |
There was a problem hiding this comment.
[needs fixing] The fetchCar catch block rethrows whatever reader.read() threw. If the PDS connection drops or the stream is aborted after the response headers were received, that error surfaces as a generic Error in discovery-consumer.ts, is caught by the unexpected-error branch, and is dead-lettered as UNEXPECTED_ERROR. The PR's stated policy is that network/timeout failures are transient and should retry. Wrap non-PdsVerificationError stream errors as PDS_NETWORK_ERROR.
| } catch (err) { | |
| // Cancel the stream so the underlying socket isn't left dangling. | |
| await reader.cancel().catch(() => { | |
| /* swallow — we already have a primary error to surface */ | |
| }); | |
| throw err; | |
| } catch (err) { | |
| // Cancel the stream so the underlying socket isn't left dangling. | |
| await reader.cancel().catch(() => { | |
| /* swallow — we already have a primary error to surface */ | |
| }); | |
| // A network drop or stream abort after headers is a transient PDS | |
| // failure, not a permanent verification failure. | |
| if (!(err instanceof PdsVerificationError)) { | |
| throw new PdsVerificationError( | |
| "PDS_NETWORK_ERROR", | |
| err instanceof Error ? err.message : String(err), | |
| undefined, | |
| err, | |
| ); | |
| } | |
| throw err; | |
| } finally { |
| const blocking = findings.find( | ||
| (f) => f.severity === "critical" && blockCategories.has(f.category), | ||
| ); | ||
| if (blocking) { | ||
| return { | ||
| toState: "blocked", | ||
| labels: [ | ||
| { val: blocking.category, findingCategory: blocking.category, severity: "critical" }, | ||
| ], |
There was a problem hiding this comment.
[needs fixing] resolvePolicyOutcome only emits the first critical blocking finding. A release could trigger several distinct automated-block findings (for example, both malware and supply-chain-compromise), in which case the final outcome should carry all applicable blocking labels, not just whichever appears first. The warning branch already collects all unique warning values; the blocking branch should do the same.
| const blocking = findings.find( | |
| (f) => f.severity === "critical" && blockCategories.has(f.category), | |
| ); | |
| if (blocking) { | |
| return { | |
| toState: "blocked", | |
| labels: [ | |
| { val: blocking.category, findingCategory: blocking.category, severity: "critical" }, | |
| ], | |
| const blocking = findings.filter( | |
| (f) => f.severity === "critical" && blockCategories.has(f.category), | |
| ); | |
| const blockingVals = [...new Set(blocking.map((f) => f.category))]; | |
| if (blockingVals.length > 0) { | |
| return { | |
| toState: "blocked", | |
| labels: blockingVals.map((val) => ({ | |
| val, | |
| findingCategory: val, | |
| severity: "critical", | |
| })), | |
| }; | |
| } |
| const now = deps.now ?? (() => new Date()); | ||
| const uri = jobUri(job); | ||
|
|
||
| if (job.operation === "delete") { |
There was a problem hiding this comment.
[needs fixing] The delete path in processDiscoveryMessage trusts the Jetstream event and immediately tombstones subjects and cancels non-terminal runs. Create/update events are verified against the publisher's PDS, but a forged or relay-compromised delete event can suppress assessments and labels without any cryptographic or PDS check. At minimum, the consumer should verify the delete (for example, by resolving the DID/PDS and confirming the record is absent or deleted, or by requiring a prior verified subject before cancelling runs) before mutating assessment state.
github-actions
Bot
added
review/awaiting-author
…all blocks Bot review of #1978. getNegatableAutomatedLabels is scoped to this labeler's own src (a labeler only negates labels it issued). A delete event now confirms the record is genuinely absent at the PDS before suppressing any work — a still-resolving record dead-letters as a forged/premature delete, matching the distrust the create path already applies. A PDS stream error after headers is re-wrapped as a transient PdsVerificationError so it retries instead of dead-lettering. And resolvePolicyOutcome emits every distinct critical blocking finding, not just the first.
github-actions
Bot
added
review/needs-rereview
…all blocks Bot review of #1978. getNegatableAutomatedLabels is scoped to this labeler's own src (a labeler only negates labels it issued). A delete event now confirms the record is genuinely absent at the PDS before suppressing any work — a still-resolving record dead-letters as a forged/premature delete, matching the distrust the create path already applies. A PDS stream error after headers is re-wrapped as a transient PdsVerificationError so it retries instead of dead-lettering. And resolvePolicyOutcome emits every distinct critical blocking finding, not just the first.
ascorbic
force-pushed
the
feat/labeler-discovery
branch
from
d281a04 to
ee843c2
Compare
|
All four addressed in ee843c2:
124 tests pass; typecheck and lint clean. |
There was a problem hiding this comment.
Re-review of the discovery/orchestration slice. The four prior blockers are all addressed:
- Negation candidates scoped to this labeler —
getNegatableAutomatedLabelsnow filters onl.src = ?and computes the stream head within that source. - Mid-download transient failures —
pds-verify.tsnow rewraps stream-read errors asPdsVerificationError("PDS_NETWORK_ERROR")so the consumer retries. - Delete-event verification —
discovery-consumer.tscallsconfirmRecordAbsentbefore acting on a delete and dead-letters forged/premature deletes whose record still resolves. - Multiple blocking labels —
resolvePolicyOutcomecollects all distinct critical block categories and emits one label per category.
The approach remains sound: production wiring still stops at assessment-pending, the orchestrator is kept behind the static production-boundary test, and the Jetstream/verification plumbing matches the aggregator precedent.
I found one new correctness issue: the delete-handling path in discovery-consumer.ts retries permanent verification failures (INVALID_URI, DID_RESOLUTION_FAILED, and any unexpected error) instead of dead-lettering them. This contradicts the stated policy that "a forged or unverifiable event is ALWAYS a dead letter, never a retry" and diverges from the create/update path, which maps permanent RecordVerificationError reasons straight to dead_letters and acks. It should mirror that classification logic. Because the queue has max_retries: 5, a malformed delete will eventually land in the DLQ, but it will burn retries and lose its specific reason code.
Test coverage is otherwise extensive and the atomicity contract is well documented.
| } catch (err) { | ||
| if (err instanceof PdsVerificationError && isTransient(err.reason, err.status)) { | ||
| controller.retry(); | ||
| return; | ||
| } | ||
| if (err instanceof RecordVerificationError && err.transient) { | ||
| controller.retry(); | ||
| return; | ||
| } | ||
| console.error("[labeler] discovery delete failed", { | ||
| did: job.did, | ||
| collection: job.collection, | ||
| rkey: job.rkey, | ||
| error: err instanceof Error ? err.message : String(err), | ||
| }); | ||
| controller.retry(); | ||
| } |
There was a problem hiding this comment.
[needs fixing] The delete path retries permanent verification failures, while the create/update path dead-letters them. confirmRecordAbsent can throw non-transient RecordVerificationErrors such as INVALID_URI or DID_RESOLUTION_FAILED; the catch block currently falls through to controller.retry() (line 233), which is the wrong signal. It should classify failures the same way the create branch does: transient errors retry, permanent RecordVerificationErrors write a dead_letters row and ack, and unexpected errors are also dead-lettered/acked rather than retried.
| } catch (err) { | |
| if (err instanceof PdsVerificationError && isTransient(err.reason, err.status)) { | |
| controller.retry(); | |
| return; | |
| } | |
| if (err instanceof RecordVerificationError && err.transient) { | |
| controller.retry(); | |
| return; | |
| } | |
| console.error("[labeler] discovery delete failed", { | |
| did: job.did, | |
| collection: job.collection, | |
| rkey: job.rkey, | |
| error: err instanceof Error ? err.message : String(err), | |
| }); | |
| controller.retry(); | |
| } | |
| } catch (err) { | |
| if (err instanceof PdsVerificationError && isTransient(err.reason, err.status)) { | |
| controller.retry(); | |
| return; | |
| } | |
| if (err instanceof RecordVerificationError && err.transient) { | |
| controller.retry(); | |
| return; | |
| } | |
| if (err instanceof RecordVerificationError) { | |
| await writeDeadLetter(deps.db, job, err.reason, err.message, now()); | |
| controller.ack(); | |
| return; | |
| } | |
| console.error("[labeler] discovery delete failed", { | |
| did: job.did, | |
| collection: job.collection, | |
| rkey: job.rkey, | |
| error: err instanceof Error ? (err.stack ?? err.message) : String(err), | |
| }); | |
| await writeDeadLetter( | |
| deps.db, | |
| job, | |
| "UNEXPECTED_ERROR", | |
| err instanceof Error ? err.message : String(err), | |
| now(), | |
| ); | |
| controller.ack(); | |
| } |
github-actions
Bot
added
review/awaiting-author
Bot review: the delete path retried permanent verification failures (INVALID_URI, DID_RESOLUTION_FAILED) instead of dead-lettering them, diverging from the create path. Both paths now route through one shared classifier — transient retries, permanent dead-letters and acks — so they can't drift.
github-actions
Bot
added
review/needs-rereview
|
Fixed in 5afe847. Extracted a single |
ascorbic
added
bot:review
…vation Bot review. issueAutomatedAssessmentLabel's signing-pause/stale/refresh conditions now throw a typed LabelIssuanceUnavailableError (identical messages) that the discovery consumer retries instead of dead-lettering — so a discovery or delete event arriving mid key-rotation doesn't permanently lose its label. Adds the two-consecutive-error-runs regression test for the assessment-error keep-set fix, and a paused- issuance retry test.
github-actions
Bot
added
review/needs-rereview
|
Both addressed in c68729c: Issuance-during-rotation classification (needs fixing) — the signing-pause/stale/refresh/state-changed conditions in Missing error-preservation test (needs fixing) — added a two-consecutive-error-runs test: the second error run for the same subject leaves 129 tests pass; typecheck and lint clean. |
ascorbic
added
bot:review
There was a problem hiding this comment.
The approach is sound: an independent Jetstream discovery cursor, PDS exact-record verification before trusting events, idempotent run creation, and a deliberately test-only orchestrator/finalization layer behind an explicit production-boundary test. It matches RFC #694 and the architecture conventions in AGENTS.md.
Both items from the prior review are addressed in this revision:
applyDiscoveryDeletenow negates theassessment-pendinglabel it issued, so a deleted release stops advertising an in-progress assessment.AssessmentOrchestrator.finalizenow keepsassessment-errorin thekeepset, so back-to-back error runs do not accidentally supersede their own active error label.
I found two remaining gaps:
-
Production-reachable correctness gap on delete retry.
applyDiscoveryDeletetombstones the subject and cancels every non-terminal run beforenegatePendingForDeletedRunssigns theassessment-pendingnegations. If the signer throws a transient error after the cancellations have committed, the message retries; on the retry the now-cancelled runs are invisible tolistNonTerminalAssessmentsForUri, so no pending negation is produced and the deleted release keeps a positiveassessment-pendinglabel. -
Test-only orchestrator stale path leaves pending labels active. When
AssessmentOrchestratorfinds the subject no longer current and finalizes asstale, it discards the entire signed batch (including its ownassessment-pendingnegation). The discovery consumer already issued a positiveassessment-pendinglabel, so a deleted/superseded subject would keep advertising an in-progress assessment. The existing tests hide this becausependingRunnever issues the discovery label.
Fix #1 and add a test that simulates a transient signer failure during the delete negation and asserts the retry still produces the negative label. For #2, add the pending negation before each running -> stale transition and update the orchestrator tests to seed a real discovery-issued pending label. Once those are addressed this should be good to merge.
| const cancelled = await applyDiscoveryDelete(deps.db, uri, now()); | ||
| await negatePendingForDeletedRuns(deps, cancelled, now()); |
There was a problem hiding this comment.
[needs fixing] applyDiscoveryDelete transitions all non-terminal runs to cancelled before negatePendingForDeletedRuns signs the matching assessment-pending negations. If the signer throws a transient error after the cancellations have committed, the queue message retries; on the second attempt the runs are now terminal, listNonTerminalAssessmentsForUri returns an empty array, and the positive assessment-pending label is never retracted. queryLabels then continues to advertise an in-progress assessment for a deleted release.
Negate the pending labels while the runs are still non-terminal, then tombstone/cancel. A regression test that fails the signer on the first delete attempt and asserts the retry still produces neg=1 is needed to close the loop.
| const cancelled = await applyDiscoveryDelete(deps.db, uri, now()); | |
| await negatePendingForDeletedRuns(deps, cancelled, now()); | |
| const nonTerminal = await listNonTerminalAssessmentsForUri(deps.db, uri); | |
| const hadPending = nonTerminal.filter( | |
| (run) => run.state === "pending" || run.state === "running", | |
| ); | |
| await negatePendingForDeletedRuns(deps, hadPending, now()); | |
| await deleteSubjectsByUri(deps.db, { uri, now }); | |
| for (const run of nonTerminal) { | |
| if ( | |
| run.state !== "observed" && | |
| run.state !== "verifying" && | |
| run.state !== "pending" && | |
| run.state !== "running" | |
| ) | |
| continue; | |
| try { | |
| await transitionAssessmentState(deps.db, { | |
| id: run.id, | |
| from: run.state, | |
| to: "cancelled", | |
| now, | |
| }); | |
| } catch (err) { | |
| if (!(err instanceof AssessmentTransitionConflictError)) throw err; | |
| } | |
| } | |
| controller.ack(); |
| if (!current) { | ||
| return transitionAssessmentState(this.db, { id: runId, from: "running", to: "stale", now }); |
There was a problem hiding this comment.
[needs fixing] When the run is no longer current, both runAssessment (line 198-199) and finalize (line 326-331) transition the run to stale and discard the signed label batch. The discovery consumer already issued a positive assessment-pending label for this run, so a deleted or superseded subject keeps advertising an in-progress assessment. The tests pass because pendingRun never seeds that label.
Import issueAutomatedAssessmentLabel from ./service.js and issue a negative assessment-pending label for the run before each running -> stale transition. Then update pendingRun in test/assessment-orchestrator.test.ts to issue the discovery assessment-pending label so the stale paths are actually exercised.
github-actions
Bot
added
review/awaiting-author
There was a problem hiding this comment.
The approach remains the right one: the labeler gets its own Jetstream cursor and PDS-verified discovery path, while the orchestrator/finalization machinery stays deliberately behind the production-boundary test. The PR builds on an approved Discussion (RFC #694) and fits the existing aggregator-style patterns.
Both findings from the prior review are addressed in this revision:
AssessmentOrchestrator.finalizenow addsassessment-errorto thekeepset whentoState === "error"(line 311 ofassessment-orchestrator.ts), and a regression test exercises a second error run for the same subject to ensure the priorassessment-erroris not self-negated.discovery-consumer.tsnow catchesLabelIssuanceUnavailableErrorand callscontroller.retry()instead of dead-lettering/acking, and the test suite adds a regression test for paused signing on the create path.
AGENTS.md conventions look good for this private Worker app: parameterized D1 statements, .js internal imports, no admin UI strings, and no published-package changeset. I also re-read the production-boundary test — it now catches bare side-effect imports as the author noted.
I have two minor, non-blocking suggestions: one test gap on the delete path and one edge-case note on finalization consistency under a mid-batch signing-state flip.
| .first<{ n: number }>(); | ||
| expect(dl?.n).toBe(0); | ||
| }); | ||
| }); |
There was a problem hiding this comment.
[suggestion] The create path gained a regression test for paused signing-state retry, but the delete path's negatePendingForDeletedRuns also calls issueAutomatedAssessmentLabel and can throw the same LabelIssuanceUnavailableError. Add a mirror test inside the "processDiscoveryMessage: delete" describe block that pauses signing_state, processes a confirmed delete, and asserts msg.retried === 1 with no dead_letters row.
| }); | ||
| } | ||
|
|
||
| await this.db.batch(statements); |
There was a problem hiding this comment.
[suggestion] buildFinalizationStatements commits the assessment state transition before the label-issuance statements. Each buildIssuanceStatements result guards the label inserts on signing state, but the state-update statement itself does not. If signing state flips after the last issue() await but before db.batch, the terminal transition can commit while some label inserts no-op; postCommit then throws and leaves a terminal run with missing labels. Because the orchestrator is still test-only this is narrow and deferred, but consider either guarding the state transition on signing state as well, or documenting this sub-window alongside the existing finalization/deletion-race deferral.
|
Both suggestions: Delete-path paused-issuance test — added in 3ceeeb9, mirroring the create-path one: a confirmed delete with signing paused retries (no ack, no dead-letter), since Mid-batch signing-state flip during finalization — this is the same test-only atomicity family as the finalization/delete-race already documented in 130 tests pass; typecheck and lint clean. |
ascorbic
added
bot:review
ascorbic
force-pushed
the
feat/labeler-discovery
branch
from
3ceeeb9 to
9550832
Compare
ascorbic
added
bot:review
github-actions
Bot
added
review/needs-rereview
There was a problem hiding this comment.
This is a re-review of the labeler discovery/orchestration PR. The approach still fits EmDash: the discovery consumer gets its own Jetstream cursor and PDS-verified subject creation, while the orchestrator/finalization machinery stays deliberately behind the production-boundary test. The prior review’s two blocking points are resolved here: assessment-error is now in the keep set during error finalization, and LabelIssuanceUnavailableError on the create path triggers a retry rather than a dead-letter.
I checked the new orchestrator, discovery consumer, record verification, Jetstream ingestor, reconciliation, tests, and wrangler/package wiring. Most of it is solid — parameterized D1 statements, no admin UI i18n or Tailwind concerns, no public-package changeset needed. I do see two correctness issues in AssessmentOrchestrator.finalize that are worth fixing before this code is relied on in production, plus a suggestion for the concurrent-delete race:
- Stale finalization leaves a dangling
assessment-pendinglabel. WhenisSubjectCurrent()returnsfalse, the method returns after only transitioning the run tostale; the pending-negation statement that was already built is never committed. A CID-superseded (but not deleted) subject will therefore keep advertising an in-progress assessment. This contradicts spec §9.9 point 9, which says the run must always negate its ownassessment-pending. - The finalization CAS is not validated after
db.batch. The transition out ofrunninguses aWHERE state = ?guard, but D1 batches do not abort on a zero-row update. If a concurrent delete/cancel moved the state, the remaining label INSERTs still commit and the function returns a run that never actually exitedrunning. - Unreified
AssessmentTransitionConflictErrorunder concurrent state changes.runAssessmentand the stale branch calltransitionAssessmentStatewithout catching the conflict exception. Once this is wired to production, a delete racing with the orchestrator can propagate an unhandled exception.
None of these touch the production path today (the production-boundary test still holds), but they are real data-shape/logic bugs in the shipped orchestrator code.
| // The same lock also closes a signing-state flip during the batch: the | ||
| // label inserts are guarded on active signing state and no-op if it | ||
| // flips, but the state CAS below is not, so a flip after this point | ||
| // could commit the terminal state without its labels. Deferred to the |
There was a problem hiding this comment.
[needs fixing] The !stillCurrent early return finalizes the run as stale but never commits the assessment-pending negation that was built at line 298. For a CID-superseded (but not deleted) subject, the old revision will now carry a dangling, never-negated assessment-pending label, which violates spec §9.9 point 9 ("always negate this run's own assessment-pending").
The stale branch should still clean up its own pending label. The minimal fix is to commit a reduced batch on the stale path that contains the state transition plus the pending negation, while omitting the outcome labels that the spec says a stale run must not issue.
| if (!stillCurrent) { | ||
| return transitionAssessmentState(this.db, { | ||
| id: assessment.id, | ||
| from: "running", |
There was a problem hiding this comment.
[needs fixing] After await this.db.batch(statements), the batch result array is not inspected. The CAS statement that exits running is guarded only by WHERE id = ? AND state = ?; if a concurrent delete or cancel has already changed the state, that UPDATE returns meta.changes === 0 but D1 continues executing the subsequent label INSERTs. The function can then return a still-non-terminal assessment while outcome labels have already been committed.
Capture the batch result and verify results[finalization.assessmentUpdateIndex].meta.changes === 1 before running postCommit. If the CAS lost, do not run post-commit hooks and surface the race rather than returning success.
| this.retryDelayMs = opts.retryDelayMs ?? 0; | ||
| } | ||
|
|
||
| async runAssessment(runId: string): Promise<Assessment> { |
There was a problem hiding this comment.
[suggestion] The pending → running transition and the stale-path running → stale transition both call transitionAssessmentState without catching AssessmentTransitionConflictError. Once the orchestrator is wired to production, a delete that races with these transitions will propagate an unhandled exception. Consider catching the conflict, re-reading the current assessment state, and returning the terminal run instead of throwing.
github-actions
Bot
added
review/awaiting-author
The three finalize() concurrency gaps (stale run leaving its pending label, unchecked finalization CAS, uncaught transition conflict) are all closed by the per-subject workflow lock the spec assigns to production wiring (§14.1, W7/W8). The production-boundary test proves nothing reaches this method in production until then; documenting them together so the lock work closes the whole family.
github-actions
Bot
added
review/needs-rereview
|
Confirmed real, and thank you for the thorough trace — all three are genuine, and they're one family: Decision (maintainer): defer all three to W7/W8, where the orchestrator is wired to production behind the per-subject workflow lock the spec assigns to finalization (§14.1). That lock serialises finalization per subject and closes the whole family — including the currency-window and signing-flip cases already noted. Fixing them piecemeal here (partial CAS checks, in-code negations) would be superseded by the lock and risks a false sense of closure. The production-boundary test proves nothing wires this method live until that work, so none is reachable today. Documented together in Everything production-reachable in this PR — discovery, PDS verification, delete verification, issuance retry-under-rotation, negation provenance — is fixed and tested. Merging on that basis. |
ascorbic
merged commit ee3ecef
into
feat/plugin-registry-labelling-service
1 participant
What does this PR do?
Adds independent Jetstream discovery and assessment orchestration to the labeler — plan workstream W6, on top of the assessment persistence layer (#1976). It wires the production path up to
assessment-pending; the full analysis pipeline ships with deterministic stub adapters, exercised only by tests, until W7/W8 provides real analyses.LabelerDiscoveryDOmaintains an independent Jetstream subscription to release records (cursor persisted iningest_state). The queue consumer verifies each event's exact URI+CID against the publisher's signed record (fetchAndVerifyExactRecord) before creating a subject and issuing a signedassessment-pendinglabel. A forged or unverifiable event dead-letters and produces nothing; a CID mismatch (the PDS serving a newer version than the event named) is a permanent dead-letter; a transient PDS or DID-directory failure retries. Redelivery converges on one run and one pending label via the deterministic runKey and idempotency key.AssessmentOrchestratordrivespending → running → stages → finalization. Finalization composes the assessment CAS + current-pointer update with all label statements (outcome labels,assessment-erroron transient exhaustion, negation of this run's ownassessment-pending, and negation of prior automated labels this outcome no longer supports) into one atomicdb.batch. It never negates a manually-issued label (getNegatableAutomatedLabelsreturns only automated-provenanced labels whose current stream head is automated).Adversarial review (opus) findings addressed: transient DID-resolution failures now retry instead of permanently dropping a release on a directory blip (#2);
getNegatableAutomatedLabelscomputes the active stream head across all issuers so a manually-negated val is never a candidate (#4); the production-boundary test also catches bare side-effect imports (#6); and the finalization currency re-check was moved to immediately before the commit, shrinking the delete/cancel window from "spans every label's signing round-trip" to two adjacent D1 ops, with a mid-run-delete test.Two deferrals, called out for the maintainer (both provably unreachable in production — the production-boundary test enforces that no production path constructs the orchestrator until W7/W8):
postCommitdiagnosis). The window is tightened and documented; the lock lands with the orchestrator's production wiring in W7/W8. The overstated "under the same lock" comment is corrected.If you'd rather either be fully closed before this merges, say so.
Targets
feat/plugin-registry-labelling-service. Related to #1909 and RFC #694.Type of change
Checklist
pnpm typecheckpassespnpm lintpassespnpm testpasses (or targeted tests for my change)pnpm formathas been runAdmin i18n and changeset are n/a —
apps/labeleris a private, unpublished Worker app.AI-generated code disclosure
Screenshots / test output
pnpm --filter @emdash-cms/labeler test: 121 pass (75 from the persistence layer + 46 new: discovery consumer, record verification, orchestrator with stub stages, reconciliation, and the static production-boundary test)pnpm lint:quick0 diagnosticsTry this PR
Open a fresh playground →
A full working EmDash site, deployed from this branch. Each visit gets its own session-scoped sandbox: no login needed and no shared state. Try the admin, edit content, hit the public site.
Tracks
feat/labeler-discovery. Updated automatically when the playground redeploys.