test(aggregator): prove historical policy input

.opencode/plans/delegated-release-service/evidence/w0.6-aggregator-history.md

@@ -0,0 +1,95 @@
+# W0.6 Aggregator Historical Input Evidence
+
+Outcome: **proved for normal-size commit frames**.
+
+## Source selection
+
+Selected source: the relay `com.atproto.sync.subscribeRepos` firehose, decoded by the installed `@atcute/firehose` client against `ComAtprotoSyncSubscribeRepos.mainSchema`.
+
+Jetstream alone is **not sufficient**. Its commit event preserves record JSON, record CID, repo revision, and `time_us`, but it does not provide the signed repo commit or CAR/MST blocks that bind that JSON to the publisher repository at that event. A later `com.atproto.sync.getRecord` call is also insufficient: after `strict -> relaxed -> strict`, it returns only the final strict record and proof. Jetstream JSON can be retained as a diagnostic comparison, not as authoritative historical state.
+
+The selected `subscribeRepos#commit` frame provides:
+
+- `seq`: relay-wide stream sequence.
+- `ops[]`: ordered repository operations, each with path and record CID.
+- `repo` and `rev`: publisher DID and repository revision.
+- `commit`: signed repository commit CID.
+- `blocks`: a CAR diff containing the commit, changed MST path, and changed record blocks.
+
+The prototype copies the CAR into every queued operation before acknowledging the source frame. Under the spike's single-relay, single-cursor assumption, `(seq, operationIndex)` is a deterministic total order of arrival from that relay. It is not repository commit-chain order. `operationIndex` distinguishes multiple operations in one commit frame.
+
+## Executable proof
+
+`apps/aggregator/src/history-spike/firehose-history.ts` is a standalone source/proof adapter. It is not wired into production ingestion.
+
+For each create/update operation it:
+
+1. Rejects `tooBig` frames because their event-specific block set is incomplete.
+2. Copies the frame CAR into the queued event together with DID, rev, commit CID, record CID, path, `seq`, and operation index.
+3. Requires the CAR's sole root to equal the frame commit CID.
+4. Recomputes every supplied block CID.
+5. Decodes the root as an atproto repo commit and binds its DID and rev to the frame.
+6. Uses `@atcute/repo` `verifyRecord` with the publisher's DID signing key to verify the commit signature and MST inclusion for the exact collection/rkey.
+7. Requires the verified record CID to equal the operation CID.
+
+The deterministic fixture creates real P-256 signed repository commits and minimal commit/MST/record CAR proofs using the installed `@atcute/crypto`, `@atcute/mst`, `@atcute/car`, and `@atcute/repo` implementations. It emits:
+
+1. profile strict at `(101, 0)`;
+2. profile relaxed at `(102, 0)`;
+3. release at `(103, 0)`;
+4. profile strict at `(104, 0)`.
+
+No event is processed until all four commits exist. Delivery is reversed and the relaxed event is redelivered. Verification, exact-redelivery deduplication, and sorting recover all four snapshots. The release resolves to the greatest matching profile ordering key below its own, `(102, 0)`, so its publication policy is relaxed. A separate assertion verifies that the current profile proof has the final strict CID and cannot recover the relaxed CID.
+
+The first and final strict records intentionally have the same content CID. They remain separate transitions because their commit CIDs, revisions, and ordering keys differ. Production deduplication therefore cannot use record CID alone.
+
+Negative cases reject a substituted rev, substituted operation CID, changed CAR bytes, a commit checked against an unrelated public key, and two byte-different events claiming the same `(seq, operationIndex)`. Exact redelivery uses a separately cloned, byte-equal CAR so deduplication exercises content comparison rather than object identity.
+
+Delete events are retained by extraction and then fail closed at verification. This spike proves create/update inclusion only; it does not construct or verify historical deletion/non-inclusion proof material. Delete transitions must not be silently skipped.
+
+## Limits and W10.1 handoff
+
+- Production must resolve the publisher's signing key at the event's DID-key epoch. Verifying old commits only against a later rotated DID document can reject valid history; blindly accepting an old key is not safe. W10.1 needs a verified DID-operation/key-history strategy.
+- Relay cursor retention is finite. Production must durably advance a cursor only after event jobs and proof bytes are accepted, monitor lag, and define backfill/recovery when the cursor is outside retention.
+- `tooBig` frames are intentionally rejected by this prototype. W10.1 must prove an event-specific sync/repo recovery path or fail closed; fetching the current record is not a substitute.
+- Historical deletes are outside this proof. W10.1 must select and prove deletion/non-inclusion material and its queue envelope before delete transitions can affect policy history.
+- `(seq, operationIndex)` is single-relay arrival order. The spike does not verify commit `prev` links or rev-chain contiguity. A profile update and release in the same atomic repository commit do not necessarily have meaningful temporal policy order merely because one op appears first. W10.1 or the RFC must define or forbid that case before policy-at-publication behavior ships.
+- The spike assumes one relay and one cursor epoch. If production admits multiple relays, relay replacement, or sequence epochs, source identity must participate in deduplication and ordering/collision handling. This spike does not choose that production identity or key.
+- Relay trust is limited to availability, completeness, and arrival ordering. Record authority comes from the signed commit and MST proof, not from relay or Jetstream JSON. Production still needs fork/rebase, commit-chain, and sequence-gap handling.
+- This spike does not choose a D1 history schema, cooldown duration, downgrade semantics, profile-policy shape, or production Worker wiring.
+
+## Command evidence
+
+Baseline:
+
+```text
+pnpm lint:json | jq '.diagnostics | length'
+0
+```
+
+Focused history test:
+
+```text
+pnpm --filter @emdash-cms/aggregator exec vitest run test/history-spike/firehose-history.test.ts
+1 file passed, 6 tests passed
+```
+
+Existing source-ingest tests:
+
+```text
+pnpm --filter @emdash-cms/aggregator exec vitest run test/jetstream-ingestor.test.ts test/jetstream-client.test.ts
+2 files passed, 17 tests passed
+```
+
+Completion checks:
+
+```text
+pnpm build
+passed (existing unresolved virtual-import and deliberate eval warnings)
+
+pnpm --filter @emdash-cms/aggregator typecheck
+passed
+
+pnpm lint
+0 warnings, 0 errors
+```

apps/aggregator/src/history-spike/firehose-history.ts

@@ -0,0 +1,220 @@
+import * as CAR from "@atcute/car";
+import * as CBOR from "@atcute/cbor";
+import * as CID from "@atcute/cid";
+import type { CidLink } from "@atcute/cid";
+import type { PublicKey } from "@atcute/crypto";
+import { type AtprotoDid, isDid } from "@atcute/lexicons/syntax";
+import { isCommit, verifyRecord } from "@atcute/repo";
+
+export const FIREHOSE_HISTORY_SOURCE = "com.atproto.sync.subscribeRepos" as const;
+
+export interface FirehoseRepoOp {
+	action: "create" | "update" | "delete";
+	cid: CidLink | null;
+	path: string;
+}
+
+/** The history-bearing subset of a decoded subscribeRepos commit frame. */
+export interface FirehoseCommitFrame {
+	repo: string;
+	seq: number;
+	rev: string;
+	commit: CidLink;
+	blocks: CBOR.Bytes;
+	ops: readonly FirehoseRepoOp[];
+	tooBig: boolean;
+}
+
+export interface HistoricalRecordEvent {
+	source: typeof FIREHOSE_HISTORY_SOURCE;
+	did: AtprotoDid;
+	sequence: number;
+	operationIndex: number;
+	orderingKey: readonly [sequence: number, operationIndex: number];
+	rev: string;
+	commitCid: string;
+	collection: string;
+	rkey: string;
+	operation: FirehoseRepoOp["action"];
+	recordCid: string | null;
+	/** Event-specific commit, MST proof, and record blocks copied before enqueue. */
+	carBytes: Uint8Array;
+}
+
+export interface VerifiedHistoricalRecordEvent extends HistoricalRecordEvent {
+	record: unknown;
+}
+
+/**
+ * Converts one relay commit frame into queue-safe, event-specific record jobs.
+ * A frame can contain multiple writes, so `operationIndex` completes `seq` into
+ * a total ordering and idempotency key within one relay cursor epoch.
+ */
+export function extractHistoricalRecordEvents(frame: FirehoseCommitFrame): HistoricalRecordEvent[] {
+	if (!Number.isSafeInteger(frame.seq) || frame.seq < 0) {
+		throw new Error("firehose sequence must be a non-negative safe integer");
+	}
+	if (!isAtprotoDid(frame.repo))
+		throw new Error(`unsupported atproto repository DID: ${frame.repo}`);
+	const did = frame.repo;
+	if (frame.tooBig) {
+		throw new Error("tooBig firehose commits do not contain complete event-specific blocks");
+	}
+
+	const carBytes = CBOR.fromBytes(frame.blocks);
+	return frame.ops.map((op, operationIndex) => {
+		const separator = op.path.indexOf("/");
+		if (separator <= 0 || separator === op.path.length - 1) {
+			throw new Error(`invalid repository operation path: ${op.path}`);
+		}
+		if (op.action !== "delete" && op.cid === null) {
+			throw new Error(`${op.action} operation is missing its record CID`);
+		}
+		if (op.action === "delete" && op.cid !== null) {
+			throw new Error("delete operation unexpectedly has a record CID");
+		}
+
+		return {
+			source: FIREHOSE_HISTORY_SOURCE,
+			did,
+			sequence: frame.seq,
+			operationIndex,
+			orderingKey: [frame.seq, operationIndex] as const,
+			rev: frame.rev,
+			commitCid: frame.commit.$link,
+			collection: op.path.slice(0, separator),
+			rkey: op.path.slice(separator + 1),
+			operation: op.action,
+			recordCid: op.cid?.$link ?? null,
+			// Every job owns its proof bytes; queue delay cannot turn this into a
+			// later PDS snapshot and callers cannot mutate sibling jobs in memory.
+			carBytes: carBytes.slice(),
+		};
+	});
+}
+
+export async function verifyHistoricalRecordEvent(
+	event: HistoricalRecordEvent,
+	publicKey: PublicKey,
+): Promise<VerifiedHistoricalRecordEvent> {
+	if (event.operation === "delete") {
+		throw new Error(
+			"historical delete/non-inclusion proof is not implemented by the W0.6 prototype",
+		);
+	}
+
+	const car = CAR.fromUint8Array(event.carBytes);
+	if (car.roots.length !== 1 || car.roots[0]?.$link !== event.commitCid) {
+		throw new Error("CAR root does not match the firehose commit CID");
+	}
+
+	let commitBytes: Uint8Array | undefined;
+	for (const block of car) {
+		const codec = block.cid.codec;
+		if (codec !== CID.CODEC_DCBOR && codec !== CID.CODEC_RAW) {
+			throw new Error(`unsupported CAR block codec: ${codec}`);
+		}
+		const actualCid = CID.toString(await CID.create(codec, Uint8Array.from(block.bytes)));
+		if (CID.toString(block.cid) !== actualCid) {
+			throw new Error("CAR block bytes do not match their CID");
+		}
+		if (actualCid === event.commitCid) commitBytes = block.bytes;
+	}
+	if (commitBytes === undefined) throw new Error("CAR does not contain its root commit block");
+
+	const commit = CBOR.decode(commitBytes);
+	if (!isCommit(commit)) throw new Error("CAR root is not an atproto repo commit");
+	if (commit.did !== event.did) throw new Error("commit DID does not match the firehose repo");
+	if (commit.rev !== event.rev) throw new Error("commit rev does not match the firehose rev");
+
+	const verified = await verifyRecord({
+		did: event.did,
+		collection: event.collection,
+		rkey: event.rkey,
+		publicKey,
+		carBytes: event.carBytes,
+	});
+	if (verified.cid !== event.recordCid) {
+		throw new Error("verified record CID does not match the firehose operation CID");
+	}
+
+	return { ...event, record: verified.record };
+}
+
+/** Verifies, collision-checks, deduplicates, and orders one relay's delayed jobs. */
+export async function recoverOrderedHistory(
+	events: readonly HistoricalRecordEvent[],
+	publicKey: PublicKey,
+): Promise<VerifiedHistoricalRecordEvent[]> {
+	const unique = new Map<string, HistoricalRecordEvent>();
+	for (const event of events) {
+		const key = `${event.sequence}:${event.operationIndex}`;
+		const previous = unique.get(key);
+		if (previous !== undefined) {
+			if (!sameEvent(previous, event)) {
+				throw new Error(`conflicting firehose redelivery at ${key}`);
+			}
+			continue;
+		}
+		unique.set(key, event);
+	}
+
+	const verified = await Promise.all(
+		Array.from(unique.values(), (event) => verifyHistoricalRecordEvent(event, publicKey)),
+	);
+	verified.sort(
+		(left, right) => left.sequence - right.sequence || left.operationIndex - right.operationIndex,
+	);
+	return verified;
+}
+
+/** Finds the profile state in force immediately before a release event. */
+export function precedingProfileEvent(
+	history: readonly VerifiedHistoricalRecordEvent[],
+	release: VerifiedHistoricalRecordEvent,
+	profileCollection: string,
+	profileRkey: string,
+): VerifiedHistoricalRecordEvent | undefined {
+	let preceding: VerifiedHistoricalRecordEvent | undefined;
+	for (const event of history) {
+		if (compareOrder(event, release) >= 0) break;
+		if (
+			event.did === release.did &&
+			event.collection === profileCollection &&
+			event.rkey === profileRkey
+		) {
+			preceding = event;
+		}
+	}
+	return preceding;
+}
+
+function compareOrder(left: HistoricalRecordEvent, right: HistoricalRecordEvent): number {
+	return left.sequence - right.sequence || left.operationIndex - right.operationIndex;
+}
+
+function sameEvent(left: HistoricalRecordEvent, right: HistoricalRecordEvent): boolean {
+	return (
+		left.source === right.source &&
+		left.did === right.did &&
+		left.rev === right.rev &&
+		left.commitCid === right.commitCid &&
+		left.collection === right.collection &&
+		left.rkey === right.rkey &&
+		left.operation === right.operation &&
+		left.recordCid === right.recordCid &&
+		equalBytes(left.carBytes, right.carBytes)
+	);
+}
+
+function equalBytes(left: Uint8Array, right: Uint8Array): boolean {
+	if (left.length !== right.length) return false;
+	for (let index = 0; index < left.length; index++) {
+		if (left[index] !== right[index]) return false;
+	}
+	return true;
+}
+
+function isAtprotoDid(value: string): value is AtprotoDid {
+	return isDid(value) && (value.startsWith("did:plc:") || value.startsWith("did:web:"));
+}