tests/contract/enrichment.ts
Rama main · réplica de lectura del repositorio original.
import assert from "node:assert/strict";
import frozenData from "../fixtures/enrichment-multilingual.json";
import { createApp } from "../../src/core/app";
import type { Db } from "../../src/core/db";
import { newId, sha256Hex } from "../../src/core/ids";
import type { VectorCapability } from "../../src/core/vectors";
import {
derivationClaimKey,
DERIVATION_SCHEMA,
drainEnrichment,
ENRICHMENT_LEASE_MS,
ENRICHMENT_MAX_LINK_JOBS_PER_EXPORT_OWNER,
REFLECTION_SCHEMA,
scheduleDerivationBackfill,
scheduleReflections,
} from "../../src/core/enrichment";
import {
ackEnrichmentJobs,
listEnrichmentJobs,
purgeEnrichmentJobs,
retryEnrichmentJobs,
} from "../../src/core/jobs";
import { MIGRATIONS } from "../../src/core/migrations";
import {
ExtractionProviderError,
type ExtractionCapability,
type ExtractionRequest,
} from "../../src/core/extraction";
import {
DEFAULT_SCOPES,
clientVia,
fakeVectors,
provisionWith,
} from "./harness";
interface FrozenCase {
id: string;
language: string;
content: string;
statement: string;
query: string;
}
const frozen = frozenData as {
version: number;
derivation: FrozenCase[];
reflection: { add_statement: string; link_relation: string };
};
function assertStrictObjects(value: unknown): void {
if (!value || typeof value !== "object") return;
const record = value as Record<string, unknown>;
if (record.type === "object") {
assert.equal(record.additionalProperties, false);
const properties = Object.keys((record.properties ?? {}) as Record<string, unknown>).sort();
assert.deepEqual([...(record.required as string[] ?? [])].sort(), properties);
}
for (const nested of Object.values(record)) assertStrictObjects(nested);
}
function observationId(request: ExtractionRequest): string {
return (request.input as any).observation.observation_id as string;
}
function observationContent(request: ExtractionRequest): string {
return (request.input as any).observation.content as string;
}
function requiredId(ids: ReadonlyMap<string, string>, id: string): string {
const mapped = ids.get(id);
assert.ok(mapped, "exported fixture IDs must have an import mapping");
return mapped;
}
function proposalFor(request: ExtractionRequest): unknown {
if (request.lane === "reflection") {
const premises = (request.input as any).premises as Array<{
claim_id: string;
statement: string;
status: string;
}>;
if (premises.some((premise) => premise.statement.includes("DISPUTED_REFLECTION_ADD"))) {
const active = premises.find((premise) => premise.status === "active");
const partial = premises.some((premise) => premise.statement.includes("PARTIAL"));
return {
action: "add",
claims: [{
kind: "semantic_fact",
statement: "A model must not launder an unresolved dispute.",
premise_ids: partial
? [active?.claim_id ?? premises[0]!.claim_id]
: premises.map((premise) => premise.claim_id),
valid_from: null,
valid_to: null,
}],
links: null,
};
}
const incomplete = premises.find((premise) =>
/(?:PARTIAL|REORDERED|DUPLICATE|FOREIGN)_REFLECTION_ADD/.test(premise.statement));
if (incomplete) {
const ids = premises.map((premise) => premise.claim_id);
const mode = incomplete.statement.match(
/(PARTIAL|REORDERED|DUPLICATE|FOREIGN)_REFLECTION_ADD/,
)![1];
const premiseIds = mode === "PARTIAL"
? ids.slice(0, -1)
: mode === "REORDERED"
? [...ids].reverse()
: mode === "DUPLICATE"
? [ids[0]!, ids[0]!]
: [ids[0]!, "claim_foreign"];
return {
action: "add",
claims: [{
kind: "semantic_fact",
statement: "Incomplete reflection provenance must not commit.",
premise_ids: premiseIds,
valid_from: null,
valid_to: null,
}],
links: null,
};
}
if (premises.some((premise) => premise.statement.includes("REFLECTION_STATUS_DRIFT")))
return {
action: "add",
claims: [{
kind: "semantic_fact",
statement: "A drifting premise must not commit.",
premise_ids: premises.map((premise) => premise.claim_id),
valid_from: null,
valid_to: null,
}],
links: null,
};
if (premises.some((premise) => premise.statement.includes("PORTABILITY_CYCLE_ADD")))
return {
action: "add",
claims: [{
kind: "semantic_fact",
statement: "Portable reflection cycle result.",
premise_ids: premises.map((premise) => premise.claim_id),
valid_from: null,
valid_to: null,
}],
links: null,
};
if (premises.some((premise) => premise.statement.includes("REFLECT_ADD")))
return {
action: "add",
claims: [{
kind: "relationship",
statement: frozen.reflection.add_statement,
premise_ids: premises.map((premise) => premise.claim_id),
valid_from: null,
valid_to: null,
}],
links: null,
};
if (premises.length > 1)
return {
action: "link",
claims: null,
links: [{
source_claim_id: premises[0]!.claim_id,
target_claim_id: premises[1]!.claim_id,
relation: frozen.reflection.link_relation,
}],
};
return { action: "abstain", claims: null, links: null };
}
const content = observationContent(request);
const id = observationId(request);
if (content.includes("MISSING_FIELD")) return { action: "abstain" };
if (content.includes("ABSTAIN")) return { action: "abstain", claims: null };
if (content.includes("AUTHORITY_INJECTION"))
return {
action: "add",
claims: [{
kind: "semantic_fact",
statement: "Unsafe authority output.",
evidence_ids: [id],
valid_from: null,
valid_to: null,
trust: "verified",
}],
};
if (content.includes("FOREIGN_ID"))
return {
action: "add",
claims: [{
kind: "semantic_fact",
statement: "Unsafe foreign source output.",
evidence_ids: ["obs_foreign"],
valid_from: null,
valid_to: null,
}],
};
if (content.includes("UNSUPPORTED_FUTURE"))
return {
action: "add",
claims: [{
kind: "semantic_fact",
statement: "An unsupported future boundary must not commit.",
evidence_ids: [id],
valid_from: "2099-01-01T00:00:00.000Z",
valid_to: null,
}],
};
if (content.includes("EXPLICIT_FUTURE"))
return {
action: "add",
claims: [{
kind: "episodic_event",
statement: "The cited maintenance window is explicitly scheduled.",
evidence_ids: [id],
valid_from: "2035-01-01T00:00:00.000Z",
valid_to: "2035-01-02T00:00:00.000Z",
}],
};
if (content.includes("INVALID_TIME"))
return {
action: "add",
claims: [{
kind: "semantic_fact",
statement: "Unsafe temporal output.",
evidence_ids: [id],
valid_from: "2026-08-02T00:00:00.000Z",
valid_to: "2026-08-01T00:00:00.000Z",
}],
};
if (content.includes("MALFORMED_TIME"))
return {
action: "add",
claims: [{
kind: "semantic_fact",
statement: "A date without a time is not RFC3339 date-time.",
evidence_ids: [id],
valid_from: "2026-08-01",
valid_to: null,
}],
};
if (content.includes("MALFORMED_CITED_TIME"))
return {
action: "add",
claims: [{
kind: "episodic_event",
statement: "An invalid calendar token cannot authorize a normalized date.",
evidence_ids: [id],
valid_from: "2026-03-02T00:00:00.000Z",
valid_to: null,
}],
};
if (content.includes("OVERBOUND_OUTPUT"))
return {
action: "add",
claims: [1, 2].map(() => ({
kind: "semantic_fact",
statement: "One proposal must not duplicate a semantic addition.",
evidence_ids: [id],
valid_from: null,
valid_to: null,
})),
};
if (content.includes("SEMANTIC_DUPLICATE"))
return {
action: "add",
claims: [{
kind: "semantic_fact",
statement: "The deployment window is Friday at 09:00 UTC.",
evidence_ids: [id],
valid_from: null,
valid_to: null,
}],
};
const fixture = frozen.derivation.find((entry) => entry.content === content);
return {
action: "add",
claims: [{
kind: "semantic_fact",
statement: fixture?.statement ?? content,
evidence_ids: [id],
valid_from: null,
valid_to: null,
}],
};
}
/** Same P0 enrichment replay against either canonical SQL adapter. */
export async function assertEnrichmentContract(db: Db, runtime: string): Promise<void> {
assert.equal(frozen.version, 1);
assertStrictObjects(DERIVATION_SCHEMA);
assertStrictObjects(REFLECTION_SCHEMA);
let current = new Date("2026-07-31T08:00:00.000Z");
let providerFailure: "typed" | "raw" | null = null;
let hold: Promise<void> | undefined;
let releaseHold: (() => void) | undefined;
let held: (() => void) | undefined;
let holdNext = false;
let calls = 0;
let concurrentDuplicateCalls = 0;
const retryRuns = new Map<string, number>();
const modelRevision = "a".repeat(64);
const providerIdentity = "https://models.example.test/v1";
const capability: ExtractionCapability = {
modelId: "frozen-sol",
modelFingerprint: modelRevision,
providerIdentity,
async generate(request) {
calls += 1;
if (
request.lane === "derivation"
&& observationContent(request) === "Concurrent duplicate drains must produce one semantic result."
) concurrentDuplicateCalls += 1;
const pendingHold = holdNext ? hold : undefined;
if (holdNext) {
holdNext = false;
held?.();
}
if (pendingHold) await pendingHold;
if (pendingHold && request.lane === "derivation"
&& observationContent(request).includes("STALE_INVALID_ONCE"))
return { action: "abstain" };
if (providerFailure === "typed")
throw new ExtractionProviderError("provider_unavailable", true);
if (providerFailure === "raw")
throw new Error("provider secret body must never persist");
if (request.lane === "derivation" && /RETRY_(?:VALID|REJECT|LATE)_OUTPUT/u.test(observationContent(request))) {
// Model output is not byte-stable: each run of one job returns a different proposal.
const content = observationContent(request);
const run = (retryRuns.get(content) ?? 0) + 1;
retryRuns.set(content, run);
const valid = (content.includes("RETRY_VALID_OUTPUT") && run > 1)
|| (content.includes("RETRY_LATE_OUTPUT") && run > 2);
return {
action: "add",
claims: [{
kind: "semantic_fact",
statement: `Retry run ${run} output.`,
evidence_ids: [valid ? observationId(request) : "obs_foreign"],
valid_from: null,
valid_to: null,
}],
};
}
if (request.lane === "derivation" && observationContent(request).includes("CYCLIC_OUTPUT")) {
const cyclic: Record<string, unknown> = {};
cyclic.self = cyclic;
return cyclic;
}
if (request.lane === "derivation" && observationContent(request).includes("OVERSIZED_OUTPUT"))
return { action: "abstain", claims: null, padding: "x".repeat(130 * 1024) };
if (request.lane === "derivation" && observationContent(request).includes("HASH_MISMATCH"))
return {
action: "add",
claims: [{
kind: "semantic_fact",
statement: "This getter-visible ADD must not escape its hashed JSON.",
evidence_ids: [observationId(request)],
valid_from: null,
valid_to: null,
}],
toJSON() { return { action: "abstain", claims: null }; },
};
return proposalFor(request);
},
};
const expectedModelFingerprint = await sha256Hex(JSON.stringify({
provider: providerIdentity,
model: capability.modelId,
revision: modelRevision,
}));
const createClient = async (
extraction: ExtractionCapability | null = capability,
orgId?: string,
vectors?: VectorCapability,
) => {
const principal = await provisionWith(db, {
orgId,
scopes: [
...DEFAULT_SCOPES,
"observations:purge",
"enrichment:write",
"export:read",
"export:all",
"import:write",
"workspaces:write",
"memberships:write",
],
});
await db.batch([{
sql: `INSERT INTO memberships
(id, org_id, workspace_id, principal_id, principal_kind, role, created_at)
VALUES (?, ?, NULL, ?, 'agent', 'admin', ?)`,
params: [newId("mbr"), principal.orgId, principal.principalId, current.toISOString()],
}]);
const app = createApp({
db,
runtime,
revision: "enrichment-contract",
extraction: extraction ?? undefined,
vectors,
semanticPrepared: Boolean(vectors),
now: () => current,
backgroundRepair: { configured: true, staleAfterMs: 60_000 },
migrationsReady: true,
secretStorageReady: true,
});
return {
...principal,
vectors,
client: clientVia(app, "http://enrichment.test"),
};
};
const primary = await createClient();
const observe = async (
target: Awaited<ReturnType<typeof createClient>>,
subjectId: string,
content: string,
scope?: { workspaceId: string; visibility: "private" | "team" },
occurredAt?: string,
) => {
const response = await target.client.call("POST", "/v1/observations", {
key: target.key,
body: {
subject_id: subjectId,
kind: "user_statement",
content,
visibility: scope?.visibility ?? "private",
...(scope ? { workspace_id: scope.workspaceId } : {}),
...(occurredAt ? { occurred_at: occurredAt } : {}),
source: { type: "contract_fixture", ref: "contract://enrichment" },
},
});
assert.equal(response.status, 201);
return response.body.data.observation_id as string;
};
const drainHttp = async (
target: Awaited<ReturnType<typeof createClient>>,
limit = 10,
) => ({
status: 200,
body: {
data: await drainEnrichment({
db,
capability,
limit,
now: () => current,
orgId: target.orgId,
vectors: target.vectors,
}),
},
});
// A first `unsafe_output` returns the job to pending; the second run decides.
const drainRejected = async (
target: Awaited<ReturnType<typeof createClient>>,
limit = 10,
) => {
await drainHttp(target, limit);
current = new Date(current.getTime() + 10_000);
return drainHttp(target, limit);
};
const portableRemapper = (
observations: any[],
claims: any[],
suffix: string,
) => {
const ids = [
...observations.map((row) => row.id),
...claims.flatMap((row) => [
row.id,
...row.enrichments.flatMap((bundle: any) => [
bundle.job.id,
...bundle.links.map((link: any) => link.id),
]),
]),
].sort((left, right) => right.length - left.length);
const idMap = new Map(ids.map((id) => [id, `${id}_${suffix}`]));
return {
idMap,
async remap(row: any) {
let serialized = JSON.stringify(row);
for (const [source, destination] of idMap)
serialized = serialized.replaceAll(source, destination);
const mapped = JSON.parse(serialized);
for (const bundle of mapped.enrichments ?? [])
bundle.job.input_hash = await sha256Hex(JSON.stringify({
lane: bundle.job.lane,
ids: bundle.job.input_ids,
}));
return mapped;
},
};
};
const invalidPrincipal = await provisionWith(db, {
scopes: [...DEFAULT_SCOPES, "enrichment:write"],
});
const invalidCapability = {
...capability,
modelFingerprint: "invalid",
} as ExtractionCapability;
const invalidClient = clientVia(createApp({
db,
runtime,
revision: "invalid-enrichment-contract",
extraction: invalidCapability,
now: () => current,
migrationsReady: true,
secretStorageReady: true,
}), "http://invalid-enrichment.test");
const invalidReady = await invalidClient.call("GET", "/readyz");
assert.equal(invalidReady.status, 503);
assert.equal(invalidReady.body.meta.capabilities.extraction, "configured_error");
const acceptedWithoutJob = await invalidClient.call("POST", "/v1/observations", {
key: invalidPrincipal.key,
body: {
subject_id: "subject_capability_recovery",
kind: "user_statement",
content: "Corrected extraction startup must recover this canonical evidence.",
source: { type: "contract_fixture", ref: "contract://enrichment" },
},
});
assert.equal(acceptedWithoutJob.status, 201);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM enrichment_jobs WHERE org_id = ?`,
[invalidPrincipal.orgId],
))[0]!.count), 0);
const correctedClient = clientVia(createApp({
db,
runtime,
revision: "corrected-enrichment-contract",
extraction: capability,
now: () => current,
migrationsReady: true,
secretStorageReady: true,
}), "http://corrected-enrichment.test");
const capabilityRecovery = await correctedClient.call("POST", "/v1/enrichment/drain?limit=1", {
key: invalidPrincipal.key,
});
assert.equal(capabilityRecovery.status, 200);
assert.equal(capabilityRecovery.body.data.scheduled, 1);
assert.equal(capabilityRecovery.body.data.completed, 1);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM enrichment_jobs
WHERE org_id = ? AND lane = 'derivation'`,
[invalidPrincipal.orgId],
))[0]!.count), 1);
await correctedClient.call("POST", "/v1/enrichment/drain?limit=1", {
key: invalidPrincipal.key,
});
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM enrichment_jobs
WHERE org_id = ? AND lane = 'derivation'`,
[invalidPrincipal.orgId],
))[0]!.count), 1);
const importedPrincipal = await provisionWith(db, {
scopes: [...DEFAULT_SCOPES, "enrichment:write", "import:write"],
});
const importedClient = clientVia(createApp({
db,
runtime,
extraction: capability,
now: () => current,
migrationsReady: true,
secretStorageReady: true,
}), "http://imported-enrichment-policy.test");
const importedObservationId = `obs_imported_backfill_${runtime.replace(/[^a-z0-9]/giu, "_")}`;
const importedContent = "A logical restore is provenance, not a new model instruction.";
const importedSourceActor = `source_actor_${runtime.replace(/[^a-z0-9]/giu, "_")}`;
const importedObservation = {
type: "observation",
id: importedObservationId,
subject_id: "subject_imported_backfill",
project_id: null,
workspace_id: null,
agent_id: null,
run_id: null,
actor_id: importedSourceActor,
kind: "user_statement",
content: importedContent,
content_hash: await sha256Hex(importedContent),
source_type: "logical_restore",
source_ref: null,
trust: "unverified",
visibility: "private",
occurred_at: null,
ingested_at: current.toISOString(),
};
const importedBody = [
{
type: "titen.export.header",
format_version: 3,
record_type: "observations",
org_id: "source_import_org",
},
{
type: "titen.import.actor_map",
source_org_id: "source_import_org",
source_actor_id: importedSourceActor,
destination_actor_id: importedPrincipal.principalId,
},
importedObservation,
].map((value) => JSON.stringify(value)).join("\n");
assert.equal((await importedClient.callRaw("POST", "/v1/import", {
key: importedPrincipal.key,
body: `${importedBody}\n`,
})).status, 200);
const importedDrain = await importedClient.call("POST", "/v1/enrichment/drain?limit=1", {
key: importedPrincipal.key,
});
assert.equal(importedDrain.status, 200);
assert.equal(importedDrain.body.data.scheduled, 0,
"logical imports must not become implicit model inputs");
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM enrichment_jobs
WHERE org_id = ? AND input_ids LIKE ?`,
[importedPrincipal.orgId, `%${importedObservationId}%`],
))[0]!.count), 0);
// AC-ENR-003, AC-ENR-004: backfill recovers each unscheduled observation once.
const backfillPrincipal = await provisionWith(db, {
scopes: [...DEFAULT_SCOPES, "enrichment:write"],
});
const backfillClient = clientVia(createApp({
db,
runtime,
now: () => current,
migrationsReady: true,
secretStorageReady: true,
}), "http://derivation-backfill.test");
const backfillObservation = await backfillClient.call("POST", "/v1/observations", {
key: backfillPrincipal.key,
body: {
subject_id: "subject_backfill_once",
kind: "user_statement",
content: "Accepted while extraction was unavailable.",
source: { type: "contract_fixture", ref: "contract://enrichment" },
},
});
assert.equal(backfillObservation.status, 201);
const backfillJobs = async () => db.all<{ source_observation_id: string | null }>(
`SELECT source_observation_id FROM enrichment_jobs WHERE org_id = ?`,
[backfillPrincipal.orgId],
);
assert.deepEqual(await backfillJobs(), []);
const runBackfill = () => scheduleDerivationBackfill({
db,
capability,
now: current,
limit: 10,
orgId: backfillPrincipal.orgId,
});
assert.equal(await runBackfill(), 1);
assert.deepEqual(await backfillJobs(), [
{ source_observation_id: backfillObservation.body.data.observation_id },
]);
assert.equal(await runBackfill(), 0, "a second backfill must schedule nothing");
// Simulate a store that existed before the column: clear it, then run the
// migration's own backfill statement.
await db.batch([{
sql: `UPDATE enrichment_jobs SET source_observation_id = NULL WHERE org_id = ?`,
params: [backfillPrincipal.orgId],
}]);
await db.batch(MIGRATIONS.find((migration) => migration.version === 28)!.statements
.filter((sql) => /^UPDATE\s+enrichment_jobs/i.test(sql))
.map((sql) => ({ sql })));
assert.deepEqual(await backfillJobs(), [
{ source_observation_id: backfillObservation.body.data.observation_id },
]);
assert.equal(await runBackfill(), 0, "pre-existing jobs must not be duplicated");
assert.equal((await backfillJobs()).length, 1);
const mutablePrincipal = await provisionWith(db, {
scopes: [...DEFAULT_SCOPES, "enrichment:write"],
});
let generatedAs = "";
const mutableCapability: ExtractionCapability = {
modelId: "frozen-at-startup",
modelFingerprint: "9".repeat(64),
async generate(request) {
generatedAs = this.modelId;
return proposalFor(request);
},
};
const mutableClient = clientVia(createApp({
db,
runtime,
extraction: mutableCapability,
now: () => current,
migrationsReady: true,
secretStorageReady: true,
}), "http://mutable-enrichment.test");
mutableCapability.modelId = "mutated-after-startup";
mutableCapability.modelFingerprint = "invalid";
const mutableObservation = await mutableClient.call("POST", "/v1/observations", {
key: mutablePrincipal.key,
body: {
subject_id: "subject_capability_snapshot",
kind: "user_statement",
content: "Capability provenance is frozen before traffic.",
source: { type: "contract_fixture", ref: "contract://enrichment" },
},
});
assert.equal(mutableObservation.status, 201);
assert.equal((await mutableClient.call("POST", "/v1/enrichment/drain?limit=1", {
key: mutablePrincipal.key,
})).status, 200);
assert.equal(generatedAs, "frozen-at-startup");
assert.deepEqual(await db.all<{ model_id: string }>(
`SELECT model_id FROM enrichment_jobs WHERE org_id = ?`,
[mutablePrincipal.orgId],
), [{ model_id: "frozen-at-startup" }]);
calls = 0;
for (const fixture of frozen.derivation)
await observe(primary, `subject_${fixture.id}`, fixture.content);
assert.equal(calls, 0, "canonical observation writes must not call the model");
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM enrichment_jobs
WHERE org_id = ? AND lane = 'derivation' AND state = 'pending'`,
[primary.orgId],
))[0]!.count), frozen.derivation.length);
const multilingual = await drainHttp(primary);
assert.equal(multilingual.status, 200);
assert.equal(multilingual.body.data.added, frozen.derivation.length);
for (const fixture of frozen.derivation) {
const context = await primary.client.call("POST", "/v1/context/compile", {
key: primary.key,
body: {
subject_id: `subject_${fixture.id}`,
task: fixture.query,
max_tokens: 800,
},
});
assert.equal(context.status, 200);
const item = context.body.data.items.find((entry: any) => entry.claim === fixture.statement);
assert.ok(item, `${fixture.language} derived claim must be recallable`);
const evidence = await primary.client.call(
"GET",
`/v1/claims/${item.claim_id}/evidence`,
{ key: primary.key },
);
assert.equal(evidence.status, 200);
assert.equal(evidence.body.data.evidence.supporting.length, 1);
assert.equal(evidence.body.data.claim.enrichment.lane, "derivation");
assert.equal(evidence.body.data.claim.enrichment.model_fingerprint, expectedModelFingerprint);
}
const persisted = await db.all<Record<string, unknown>>(
`SELECT * FROM enrichment_jobs
WHERE org_id = ? ORDER BY created_at, id`,
[primary.orgId],
);
assert.equal(persisted.length, frozen.derivation.length);
for (const row of persisted) {
const serialized = JSON.stringify(row);
assert.ok(!frozen.derivation.some((fixture) => serialized.includes(fixture.content)));
assert.doesNotMatch(serialized, /provider secret body/u);
assert.doesNotMatch(serialized, /models\.example\.test/u);
assert.equal(row.model_fingerprint, expectedModelFingerprint);
assert.match(String(row.output_hash), /^[a-f0-9]{64}$/u);
}
current = new Date(current.getTime() + 1_000);
const abstainObservation = await observe(primary, "subject_abstain", "ABSTAIN no durable memory");
const abstain = await drainHttp(primary);
assert.equal(abstain.status, 200);
const abstainJob = await db.all<{ state: string; result_ids: string; result_kind: string }>(
`SELECT j.state, j.result_ids, c.result_kind
FROM enrichment_jobs j JOIN enrichment_commits c ON c.job_id = j.id
WHERE j.org_id = ? AND j.input_ids LIKE ?`,
[primary.orgId, `%${abstainObservation}%`],
);
assert.deepEqual(abstainJob, [{ state: "done", result_ids: "[]", result_kind: "abstain" }]);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claims WHERE org_id = ? AND subject_id = 'subject_abstain'`,
[primary.orgId],
))[0]!.count), 0);
const duplicate = await createClient();
const callsBeforeDuplicate = calls;
await observe(duplicate, "subject_exact_duplicate", "Exact duplicate durable preference.");
await observe(duplicate, "subject_exact_duplicate", "Exact duplicate durable preference.");
await drainHttp(duplicate);
assert.equal(calls - callsBeforeDuplicate, 1, "exact duplicates must reuse without a model call");
const duplicateClaims = await db.all<{ id: string; version: number }>(
`SELECT id, version FROM claims WHERE org_id = ? AND subject_id = 'subject_exact_duplicate'`,
[duplicate.orgId],
);
assert.equal(duplicateClaims.length, 1);
assert.equal(duplicateClaims[0]!.version, 2);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claim_sources WHERE claim_id = ?`,
[duplicateClaims[0]!.id],
))[0]!.count), 2);
const duplicateJobs = await db.all<{ result_kind: string; output_hash: string | null }>(
`SELECT c.result_kind, j.output_hash
FROM enrichment_jobs j JOIN enrichment_commits c ON c.job_id = j.id
WHERE j.org_id = ? ORDER BY c.result_kind`,
[duplicate.orgId],
);
assert.deepEqual(duplicateJobs.map(({ result_kind }) => result_kind), ["add", "reuse"]);
assert.equal(duplicateJobs.filter(({ output_hash }) => output_hash === null).length, 1);
assert.match(duplicateJobs.find(({ output_hash }) => output_hash)?.output_hash ?? "", /^[a-f0-9]{64}$/u);
const semanticDuplicate = await createClient();
await observe(
semanticDuplicate,
"subject_semantic_duplicate",
"SEMANTIC_DUPLICATE first independently worded source.",
);
await observe(
semanticDuplicate,
"subject_semantic_duplicate",
"SEMANTIC_DUPLICATE second independently worded source.",
);
let semanticStarted!: () => void;
const semanticModelStarted = new Promise<void>((resolve) => { semanticStarted = resolve; });
hold = new Promise<void>((resolve) => { releaseHold = resolve; });
held = semanticStarted;
holdNext = true;
const semanticFirst = drainEnrichment({
db,
capability,
limit: 1,
now: () => current,
orgId: semanticDuplicate.orgId,
});
await semanticModelStarted;
const semanticSecond = await drainEnrichment({
db,
capability,
limit: 1,
now: () => current,
orgId: semanticDuplicate.orgId,
});
releaseHold!();
const semanticFirstResult = await semanticFirst;
hold = undefined;
held = undefined;
releaseHold = undefined;
assert.equal(semanticFirstResult.completed + semanticSecond.completed, 2);
const semanticClaims = await db.all<{ id: string; version: number }>(
`SELECT id, version FROM claims
WHERE org_id = ? AND subject_id = 'subject_semantic_duplicate'`,
[semanticDuplicate.orgId],
);
assert.equal(semanticClaims.length, 1);
assert.equal(semanticClaims[0]!.version, 2);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claim_sources WHERE claim_id = ?`,
[semanticClaims[0]!.id],
))[0]!.count), 2);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM observations WHERE org_id = ?`,
[semanticDuplicate.orgId],
))[0]!.count), 2, "semantic convergence must never delete source evidence");
assert.deepEqual((await db.all<{ result_kind: string; output_hash: string | null }>(
`SELECT ec.result_kind, job.output_hash FROM enrichment_commits ec
JOIN enrichment_jobs job ON job.id = ec.job_id
WHERE job.org_id = ? ORDER BY ec.result_kind`,
[semanticDuplicate.orgId],
)).map((row) => [row.result_kind, Boolean(row.output_hash)]), [
["add", true],
["reuse", true],
]);
const sameOrgPrincipal = await provisionWith(db, {
orgId: semanticDuplicate.orgId,
scopes: [...DEFAULT_SCOPES, "enrichment:write"],
});
const sameOrgActor = {
...sameOrgPrincipal,
vectors: undefined,
client: clientVia(createApp({
db,
runtime,
extraction: capability,
now: () => current,
migrationsReady: true,
secretStorageReady: true,
}), "http://same-org-private-enrichment.test"),
};
await observe(
sameOrgActor,
"subject_semantic_duplicate",
"SEMANTIC_DUPLICATE third source owned by another private principal.",
);
await drainHttp(sameOrgActor);
assert.deepEqual((await db.all<{ actor_id: string; source_count: number }>(
`SELECT c.actor_id,
(SELECT COUNT(*) FROM claim_sources source WHERE source.claim_id = c.id) AS source_count
FROM claims c
WHERE c.org_id = ? AND c.subject_id = 'subject_semantic_duplicate'
ORDER BY source_count`,
[semanticDuplicate.orgId],
)).map((row) => [row.actor_id, Number(row.source_count)]), [
[sameOrgActor.principalId, 1],
[semanticDuplicate.principalId, 2],
]);
const saturatedSemantic = await createClient();
for (let index = 0; index < 22; index += 1)
await observe(
saturatedSemantic,
"subject_semantic_saturation",
`SEMANTIC_DUPLICATE bounded source generation ${index}.`,
);
const saturatedDrain = await drainEnrichment({
db,
capability,
limit: 50,
now: () => current,
orgId: saturatedSemantic.orgId,
});
assert.equal(saturatedDrain.completed, 22);
const saturatedClaims = await db.all<{
id: string;
enrichment_key: string;
enrichment_key_archived: number;
source_count: number;
version: number;
}>(
`SELECT c.id, c.enrichment_key, c.enrichment_key_archived, c.version,
(SELECT COUNT(*) FROM claim_sources source
WHERE source.claim_id = c.id) AS source_count
FROM claims c
WHERE c.org_id = ? AND c.subject_id = 'subject_semantic_saturation'
ORDER BY source_count DESC`,
[saturatedSemantic.orgId],
);
assert.deepEqual(saturatedClaims.map((row) => [
Number(row.source_count),
Number(row.version),
Number(row.enrichment_key_archived),
]), [[20, 20, 1], [2, 2, 0]],
"post-cap observations must fill one new generation instead of creating one-source claims");
assert.equal(new Set(saturatedClaims.map(({ enrichment_key }) => enrichment_key)).size, 2);
assert.ok(saturatedClaims.every(({ enrichment_key }) => /^[a-f0-9]{64}$/u.test(enrichment_key)));
await assert.rejects(() => db.batch([{
sql: `UPDATE claims SET enrichment_key = ? WHERE id = ? AND org_id = ?`,
params: ["e".repeat(64), saturatedClaims[1]!.id, saturatedSemantic.orgId],
}]), /ENRICHMENT_KEY_IMMUTABLE/u,
"a not-yet-full generated claim cannot rotate its semantic key");
await assert.rejects(() => db.batch([{
sql: `UPDATE claims SET enrichment_key = ?, version = version + 1
WHERE id = ? AND org_id = ?`,
params: ["d".repeat(64), saturatedClaims[0]!.id, saturatedSemantic.orgId],
}]), /ENRICHMENT_KEY_IMMUTABLE/u,
"an archived full generation cannot rotate its key twice");
const directKeyGuardId = `claim_direct_key_guard_${runtime.replace(/[^a-z0-9]/giu, "_")}`;
await db.batch([{
sql: `INSERT INTO claims
(id, org_id, subject_id, project_id, workspace_id, observer_id,
actor_id, kind, statement, confidence, trust, visibility, status,
version, valid_from, valid_to, created_at, enrichment_job_id,
enrichment_key)
VALUES (?, ?, 'subject_direct_key_guard', NULL, NULL, NULL, ?,
'semantic_fact', 'Direct claims never own semantic enrichment keys.',
0.5, 'unverified', 'private', 'active', 1, ?, NULL, ?, NULL, NULL)`,
params: [
directKeyGuardId,
saturatedSemantic.orgId,
saturatedSemantic.principalId,
current.toISOString(),
current.toISOString(),
],
}]);
await assert.rejects(() => db.batch([{
sql: `UPDATE claims SET enrichment_key = ? WHERE id = ? AND org_id = ?`,
params: ["e".repeat(64), directKeyGuardId, saturatedSemantic.orgId],
}]), /ENRICHMENT_KEY_IMMUTABLE/u,
"a direct claim cannot acquire a generated semantic key");
const exportedObservations = await duplicate.client.call("GET", "/v1/export?type=observations", {
key: duplicate.key,
});
const exportedClaims = await duplicate.client.call("GET", "/v1/export?type=claims", {
key: duplicate.key,
});
assert.equal(exportedObservations.status, 200);
assert.equal(exportedClaims.status, 200);
const observationRows = String(exportedObservations.body).trim().split("\n").map((value) => JSON.parse(value));
const claimRows = String(exportedClaims.body).trim().split("\n").map((value) => JSON.parse(value));
assert.equal(claimRows[0]!.format_version, 4);
assert.equal(claimRows.length, 2);
assert.match(claimRows[1]!.enrichment_job_id, /^enr_/u);
assert.deepEqual(
claimRows[1]!.enrichments.map((bundle: any) => bundle.commit.result_kind).sort(),
["add", "reuse"],
);
assert.ok(claimRows[1]!.enrichments.every((bundle: any) => Array.isArray(bundle.job.input_ids)));
const portableAddJob = claimRows[1]!.enrichments.find(
(bundle: any) => bundle.commit.result_kind === "add",
).job;
await db.batch([{
sql: `UPDATE enrichment_jobs SET result_ids = ? WHERE id = ? AND org_id = ?`,
params: [JSON.stringify(["claim_missing_primary"]), portableAddJob.id, duplicate.orgId],
}]);
const corruptedPrimaryExport = await duplicate.client.call("GET", "/v1/export?type=claims", {
key: duplicate.key,
});
assert.equal(corruptedPrimaryExport.status, 400,
"export must reject a job that no longer owns its primary result");
await db.batch([{
sql: `UPDATE enrichment_jobs SET result_ids = ? WHERE id = ? AND org_id = ?`,
params: [JSON.stringify([claimRows[1]!.id]), portableAddJob.id, duplicate.orgId],
}]);
const foreignEvidenceId = (await db.all<{ id: string }>(
`SELECT id FROM observations WHERE org_id = ? ORDER BY id LIMIT 1`,
[primary.orgId],
))[0]!.id;
await db.batch([{
sql: `INSERT INTO claim_sources
(claim_id, observation_id, relation, created_at)
VALUES (?, ?, 'contradicts', ?)`,
params: [claimRows[1]!.id, foreignEvidenceId, current.toISOString()],
}]);
for (const scope of ["", "&all=true"]) {
const corruptEvidenceExport = await duplicate.client.call(
"GET",
`/v1/export?type=claims${scope}`,
{ key: duplicate.key },
);
assert.equal(corruptEvidenceExport.status, 400,
"claim export must reject an incomplete authorized evidence graph");
}
await db.batch([{
sql: `DELETE FROM claim_sources
WHERE claim_id = ? AND observation_id = ? AND relation = 'contradicts'`,
params: [claimRows[1]!.id, foreignEvidenceId],
}]);
const exactReplay = await duplicate.client.callRaw("POST", "/v1/import", {
key: duplicate.key,
body: `${String(exportedObservations.body)}${String(exportedClaims.body)}`,
});
assert.equal(exactReplay.status, 200);
assert.equal(exactReplay.body.data.inserted.enrichment, 0);
const portableIds = [
...observationRows.slice(1).map((row) => row.id),
claimRows[1]!.id,
...claimRows[1]!.enrichments.map((bundle: any) => bundle.job.id),
];
const portableIdMap = new Map(portableIds.map((id) => [id, `${id}_restored`]));
const remap = (row: any) => {
let serialized = JSON.stringify(row);
for (const [source, destination] of portableIdMap)
serialized = serialized.replaceAll(source, destination);
return JSON.parse(serialized);
};
const restoredObservations = [observationRows[0], ...observationRows.slice(1).map(remap)];
const restoredClaim = remap(claimRows[1]);
for (const bundle of restoredClaim.enrichments)
bundle.job.input_hash = await sha256Hex(JSON.stringify({
lane: bundle.job.lane,
ids: bundle.job.input_ids,
}));
const restoredClaims = [claimRows[0], restoredClaim];
const restoreTarget = await createClient();
const actorMap = {
type: "titen.import.actor_map",
source_org_id: duplicate.orgId,
source_actor_id: duplicate.principalId,
destination_actor_id: restoreTarget.principalId,
};
const restoredBody = `${restoredObservations.map((value) => JSON.stringify(value)).join("\n")}\n${JSON.stringify(actorMap)}\n${restoredClaims.map((value) => JSON.stringify(value)).join("\n")}\n`;
const missingPrimary = structuredClone(restoredClaim);
missingPrimary.enrichments = [];
const rejectedMissing = await restoreTarget.client.callRaw("POST", "/v1/import", {
key: restoreTarget.key,
body: `${restoredObservations.map((value) => JSON.stringify(value)).join("\n")}\n${JSON.stringify(actorMap)}\n${JSON.stringify(claimRows[0])}\n${JSON.stringify(missingPrimary)}\n`,
});
assert.equal(rejectedMissing.status, 400);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM observations WHERE id LIKE '%_restored'`,
))[0]!.count), 0, "an incomplete provenance bundle must write nothing");
const unrelatedEvidence = structuredClone(restoredObservations[1]);
unrelatedEvidence.id = `${unrelatedEvidence.id}_uncited`;
unrelatedEvidence.content = "Same-domain evidence that no enrichment job cited.";
unrelatedEvidence.content_hash = await sha256Hex(unrelatedEvidence.content);
const wrongSourceClaim = structuredClone(restoredClaim);
wrongSourceClaim.sources[0].observation_id = unrelatedEvidence.id;
const rejectedWrongSource = await restoreTarget.client.callRaw("POST", "/v1/import", {
key: restoreTarget.key,
body: `${[...restoredObservations, unrelatedEvidence].map((value) => JSON.stringify(value)).join("\n")}\n${
JSON.stringify(actorMap)
}\n${JSON.stringify(claimRows[0])}\n${JSON.stringify(wrongSourceClaim)}\n`,
});
assert.equal(rejectedWrongSource.status, 400,
"same-domain evidence cannot replace the exact cited source graph");
const unsupportedFutureClaim = structuredClone(restoredClaim);
unsupportedFutureClaim.valid_from = "2099-01-01T00:00:00.000Z";
unsupportedFutureClaim.valid_to = null;
unsupportedFutureClaim.enrichment_key = await derivationClaimKey({
subjectId: unsupportedFutureClaim.subject_id,
projectId: unsupportedFutureClaim.project_id,
workspaceId: unsupportedFutureClaim.workspace_id,
visibility: unsupportedFutureClaim.visibility,
kind: unsupportedFutureClaim.kind,
statement: unsupportedFutureClaim.statement,
validFrom: unsupportedFutureClaim.valid_from,
validTo: unsupportedFutureClaim.valid_to,
});
const rejectedUnsupportedFuture = await restoreTarget.client.callRaw("POST", "/v1/import", {
key: restoreTarget.key,
body: `${restoredObservations.map((value) => JSON.stringify(value)).join("\n")}\n${
JSON.stringify(actorMap)
}\n${JSON.stringify(claimRows[0])}\n${JSON.stringify(unsupportedFutureClaim)}\n`,
});
assert.equal(rejectedUnsupportedFuture.status, 400,
"a recomputed semantic key cannot bypass cited temporal bounds");
for (const invalidArchiveFlag of [null, "0"]) {
const invalidArchiveClaim = structuredClone(restoredClaim);
invalidArchiveClaim.enrichment_key_archived = invalidArchiveFlag;
const rejectedArchiveFlag = await restoreTarget.client.callRaw("POST", "/v1/import", {
key: restoreTarget.key,
body: `${restoredObservations.map((value) => JSON.stringify(value)).join("\n")}\n${
JSON.stringify(actorMap)
}\n${JSON.stringify(claimRows[0])}\n${JSON.stringify(invalidArchiveClaim)}\n`,
});
assert.equal(rejectedArchiveFlag.status, 400,
"v3 archive state must be the numeric 0/1 wire value");
}
const supersededWithoutReplacement = structuredClone(restoredClaim);
supersededWithoutReplacement.status = "superseded";
supersededWithoutReplacement.version += 1;
supersededWithoutReplacement.superseded_by = null;
const rejectedMissingReplacement = await restoreTarget.client.callRaw("POST", "/v1/import", {
key: restoreTarget.key,
body: `${restoredObservations.map((value) => JSON.stringify(value)).join("\n")}\n${
JSON.stringify(actorMap)
}\n${JSON.stringify(claimRows[0])}\n${JSON.stringify(supersededWithoutReplacement)}\n`,
});
assert.equal(rejectedMissingReplacement.status, 400,
"superseded status and replacement must be present together");
const restored = await restoreTarget.client.callRaw("POST", "/v1/import", {
key: restoreTarget.key,
body: restoredBody,
});
assert.equal(restored.status, 200, JSON.stringify(restored.body));
assert.equal(restored.body.data.inserted.observation, 2);
assert.equal(restored.body.data.inserted.claim, 1);
assert.equal(restored.body.data.inserted.enrichment, 2);
assert.equal(restored.body.data.inserted.enrichment_commit, 2);
const restoredReplay = await restoreTarget.client.callRaw("POST", "/v1/import", {
key: restoreTarget.key,
body: restoredBody,
});
assert.equal(restoredReplay.status, 200);
assert.equal(restoredReplay.body.data.inserted.enrichment, 0);
const tampered = structuredClone(restoredClaim);
tampered.enrichments.find((bundle: any) => bundle.commit.result_kind === "add").job.output_hash = "f".repeat(64);
const rejectedTamper = await restoreTarget.client.callRaw("POST", "/v1/import", {
key: restoreTarget.key,
body: `${JSON.stringify(claimRows[0])}\n${JSON.stringify(actorMap)}\n${JSON.stringify(tampered)}\n`,
});
assert.equal(rejectedTamper.status, 409);
const privateReplacementImporter = await createClient(capability, restoreTarget.orgId);
await observe(
privateReplacementImporter,
"subject_private_replacement_probe",
"A private replacement reference must remain fail-closed.",
);
await drainHttp(privateReplacementImporter);
const privateObservationExport = await privateReplacementImporter.client.call(
"GET",
"/v1/export?type=observations",
{ key: privateReplacementImporter.key },
);
const privateClaimExport = await privateReplacementImporter.client.call(
"GET",
"/v1/export?type=claims",
{ key: privateReplacementImporter.key },
);
assert.equal(privateObservationExport.status, 200);
assert.equal(privateClaimExport.status, 200);
const privateObservationRows = String(privateObservationExport.body)
.trim().split("\n").map((value) => JSON.parse(value));
const privateClaimRows = String(privateClaimExport.body)
.trim().split("\n").map((value) => JSON.parse(value));
const privateRemapper = portableRemapper(
privateObservationRows.slice(1),
privateClaimRows.slice(1),
"private_reference_probe",
);
const mappedPrivateObservations = await Promise.all(
privateObservationRows.slice(1).map(privateRemapper.remap),
);
const mappedPrivateClaim = await privateRemapper.remap(privateClaimRows[1]);
mappedPrivateClaim.status = "superseded";
mappedPrivateClaim.version += 1;
mappedPrivateClaim.superseded_by = portableIdMap.get(claimRows[1]!.id);
const hiddenReplacementRejected = await privateReplacementImporter.client.callRaw(
"POST",
"/v1/import",
{
key: privateReplacementImporter.key,
body: `${JSON.stringify(privateObservationRows[0])}\n${
mappedPrivateObservations.map((value) => JSON.stringify(value)).join("\n")
}\n${JSON.stringify(privateClaimRows[0])}\n${JSON.stringify(mappedPrivateClaim)}\n`,
},
);
assert.equal(hiddenReplacementRejected.status, 422,
"another principal's private replacement must be indistinguishable from an absent claim");
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM observations WHERE id = ?`,
[mappedPrivateObservations[0]!.id],
))[0]!.count), 0, "hidden replacement rejection must remain atomic");
const episodic = await createClient();
const callsBeforeEpisodic = calls;
await observe(
episodic,
"subject_episodic",
"The same words can describe separate temporal evidence.",
undefined,
"2026-07-30T08:00:00.000Z",
);
await observe(
episodic,
"subject_episodic",
"The same words can describe separate temporal evidence.",
undefined,
"2026-07-31T08:00:00.000Z",
);
await drainHttp(episodic);
assert.equal(calls - callsBeforeEpisodic, 2);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claims WHERE org_id = ? AND subject_id = 'subject_episodic'`,
[episodic.orgId],
))[0]!.count), 2);
const implicitTemporal = await createClient();
await observe(
implicitTemporal,
"subject_implicit_temporal",
"Identical evidence ingested at different times stays temporally distinct.",
);
await drainHttp(implicitTemporal);
current = new Date(current.getTime() + 86_400_000);
await observe(
implicitTemporal,
"subject_implicit_temporal",
"Identical evidence ingested at different times stays temporally distinct.",
);
await drainHttp(implicitTemporal);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claims
WHERE org_id = ? AND subject_id = 'subject_implicit_temporal'`,
[implicitTemporal.orgId],
))[0]!.count), 2);
current = new Date(current.getTime() + 1_000);
await observe(primary, "subject_reflect_add", "REFLECT_ADD rollback is required.");
await observe(primary, "subject_reflect_add", "REFLECT_ADD approval is required.");
await drainHttp(primary);
current = new Date(current.getTime() + 1_000);
await drainHttp(primary);
const reflected = await db.all<{
id: string;
enrichment_job_id: string;
}>(
`SELECT id, enrichment_job_id FROM claims
WHERE org_id = ? AND subject_id = 'subject_reflect_add'
AND statement = ?`,
[primary.orgId, frozen.reflection.add_statement],
);
assert.equal(reflected.length, 1, "one canonical premise set must produce one reflection claim");
const derivedLinks = await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claim_links
WHERE org_id = ? AND source_claim_id = ? AND relation = 'derived_from'`,
[primary.orgId, reflected[0]!.id],
);
assert.equal(Number(derivedLinks[0]!.count), 2);
const rejectIncompleteReflection = async (
mode: "PARTIAL" | "REORDERED" | "DUPLICATE" | "FOREIGN",
) => {
const target = await createClient();
const marker = `${mode}_REFLECTION_ADD`;
await observe(target, `subject_${marker}`, `${marker} first active premise.`);
await observe(target, `subject_${marker}`, `${marker} second active premise.`);
await drainHttp(target, 2);
current = new Date(current.getTime() + 1_000);
const rejected = await drainRejected(target, 1);
assert.equal(rejected.body.data.failed, 1);
assert.equal(rejected.body.data.added, 0);
assert.deepEqual(await db.all<{ state: string; error_class: string }>(
`SELECT state, error_class FROM enrichment_jobs
WHERE org_id = ? AND lane = 'reflection'`,
[target.orgId],
), [{ state: "failed", error_class: "unsafe_output" }]);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claims
WHERE org_id = ? AND statement = 'Incomplete reflection provenance must not commit.'`,
[target.orgId],
))[0]!.count), 0);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claim_links
WHERE org_id = ? AND relation = 'derived_from'`,
[target.orgId],
))[0]!.count), 0);
};
for (const mode of ["PARTIAL", "REORDERED", "DUPLICATE", "FOREIGN"] as const)
await rejectIncompleteReflection(mode);
const rejectDisputedReflection = async (marker: string) => {
const target = await createClient();
const supporting = await observe(target, `subject_${marker}`, `${marker} supporting evidence.`);
const contradicting = await observe(target, `subject_${marker}`, `${marker} contradicting evidence.`);
await drainHttp(target, 2);
const consolidated = await target.client.call("POST", "/v1/consolidations", {
key: target.key,
body: {
subject_id: `subject_${marker}`,
claims: [{
kind: "semantic_fact",
statement: `${marker} unresolved canonical dispute.`,
sources: [
{ observation_id: supporting, relation: "supports" },
{ observation_id: contradicting, relation: "contradicts" },
],
}],
},
});
assert.equal(consolidated.status, 201);
assert.equal(consolidated.body.data.claims[0].status, "disputed");
current = new Date(current.getTime() + 1_000);
const rejected = await drainRejected(target, 1);
assert.equal(rejected.body.data.failed, 1);
assert.equal(rejected.body.data.added, 0);
assert.deepEqual(await db.all<{ state: string; error_class: string }>(
`SELECT state, error_class FROM enrichment_jobs
WHERE org_id = ? AND lane = 'reflection'`,
[target.orgId],
), [{ state: "failed", error_class: "unsafe_output" }]);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claims
WHERE org_id = ? AND statement = 'A model must not launder an unresolved dispute.'`,
[target.orgId],
))[0]!.count), 0);
assert.deepEqual(await db.all<{ relation: string }>(
`SELECT relation FROM claim_sources WHERE claim_id = ? ORDER BY relation`,
[consolidated.body.data.claims[0].claim_id],
), [{ relation: "contradicts" }, { relation: "supports" }]);
};
await rejectDisputedReflection("DISPUTED_REFLECTION_ADD_CITED");
await rejectDisputedReflection("DISPUTED_REFLECTION_ADD_PARTIAL");
const statusDrift = await createClient();
await observe(
statusDrift,
"subject_reflection_status_drift",
"REFLECTION_STATUS_DRIFT starts active.",
);
await drainHttp(statusDrift, 1);
current = new Date(current.getTime() + 1_000);
let driftStarted!: () => void;
const driftModelStarted = new Promise<void>((resolve) => { driftStarted = resolve; });
hold = new Promise<void>((resolve) => { releaseHold = resolve; });
held = driftStarted;
holdNext = true;
const driftingDrain = drainEnrichment({
db,
capability,
limit: 1,
now: () => current,
orgId: statusDrift.orgId,
});
await driftModelStarted;
const driftingPremise = (await db.all<{ id: string; version: number }>(
`SELECT id, version FROM claims
WHERE org_id = ? AND subject_id = 'subject_reflection_status_drift'
AND status = 'active'`,
[statusDrift.orgId],
))[0]!;
await db.batch([{
sql: `UPDATE claims SET status = 'disputed', version = version + 1
WHERE id = ? AND org_id = ? AND version = ?`,
params: [driftingPremise.id, statusDrift.orgId, driftingPremise.version],
}]);
releaseHold!();
const drifted = await driftingDrain;
hold = undefined;
held = undefined;
releaseHold = undefined;
assert.equal(drifted.failed, 1);
assert.equal(drifted.added, 0);
assert.deepEqual(await db.all<{ state: string; error_class: string }>(
`SELECT state, error_class FROM enrichment_jobs
WHERE org_id = ? AND lane = 'reflection'`,
[statusDrift.orgId],
), [{ state: "failed", error_class: "source_changed" }]);
current = new Date(current.getTime() + 1_000);
await observe(primary, "subject_reflect_link", "Link candidate one.");
await observe(primary, "subject_reflect_link", "Link candidate two.");
await drainHttp(primary);
current = new Date(current.getTime() + 1_000);
await drainHttp(primary);
const candidateLinks = await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claim_links
WHERE org_id = ? AND relation = 'duplicate_candidate'`,
[primary.orgId],
);
assert.equal(Number(candidateLinks[0]!.count), 1);
const reflectedObservationExport = await primary.client.call("GET", "/v1/export?type=observations", {
key: primary.key,
});
const reflectedClaimExport = await primary.client.call("GET", "/v1/export?type=claims", {
key: primary.key,
});
assert.equal(reflectedObservationExport.status, 200);
assert.equal(reflectedClaimExport.status, 200);
const reflectedClaimLines = String(reflectedClaimExport.body).trim().split("\n").map((value) => JSON.parse(value));
const reflectedClaimHeader = reflectedClaimLines[0]!;
const reflectedClaimRows = reflectedClaimLines.slice(1);
const portableReflection = reflectedClaimRows.find((row) => row.id === reflected[0]!.id);
assert.ok(portableReflection);
const reflectionAddBundle = portableReflection.enrichments.find(
(bundle: any) => bundle.job.id === reflected[0]!.enrichment_job_id,
);
assert.equal(reflectionAddBundle.commit.result_kind, "add");
assert.equal(reflectionAddBundle.links.length, 2);
assert.ok(reflectionAddBundle.links.every((link: any) => link.relation === "derived_from"));
const duplicateReflectionInputs = structuredClone(portableReflection);
const duplicateReflectionBundle = duplicateReflectionInputs.enrichments.find(
(bundle: any) => bundle.job.id === reflected[0]!.enrichment_job_id,
);
duplicateReflectionBundle.job.input_ids[1] = structuredClone(
duplicateReflectionBundle.job.input_ids[0],
);
duplicateReflectionBundle.job.input_hash = await sha256Hex(JSON.stringify({
lane: duplicateReflectionBundle.job.lane,
ids: duplicateReflectionBundle.job.input_ids,
}));
const duplicateReflectionRejected = await primary.client.callRaw("POST", "/v1/import", {
key: primary.key,
body: `${JSON.stringify(reflectedClaimHeader)}\n${JSON.stringify(duplicateReflectionInputs)}\n`,
});
assert.equal(duplicateReflectionRejected.status, 400,
"reflection provenance cannot repeat a premise id");
const nestedReflection = structuredClone(portableReflection);
const nestedBundle = nestedReflection.enrichments.find(
(bundle: any) => bundle.job.id === reflected[0]!.enrichment_job_id,
);
nestedBundle.job.input_ids = [{ id: nestedReflection.id, version: nestedReflection.version }];
nestedBundle.job.input_hash = await sha256Hex(JSON.stringify({
lane: nestedBundle.job.lane,
ids: nestedBundle.job.input_ids,
}));
const nestedReflectionRejected = await primary.client.callRaw("POST", "/v1/import", {
key: primary.key,
body: `${JSON.stringify(reflectedClaimHeader)}\n${JSON.stringify(nestedReflection)}\n`,
});
assert.equal(nestedReflectionRejected.status, 400,
"a reflection-generated claim cannot become another reflection premise");
const linkBundles = reflectedClaimRows.flatMap((row) => row.enrichments)
.filter((bundle: any) => bundle.commit.result_kind === "link");
assert.equal(linkBundles.length, 1, "a link-only job must have one deterministic owner claim");
assert.deepEqual(
linkBundles[0]!.job.result_ids,
linkBundles[0]!.links.map((link: any) => link.id),
);
await db.batch([{
sql: `UPDATE enrichment_jobs SET result_ids = ? WHERE id = ? AND org_id = ?`,
params: [JSON.stringify(["link_missing_result"]), linkBundles[0]!.job.id, primary.orgId],
}]);
const corruptedLinkExport = await primary.client.call("GET", "/v1/export?type=claims", {
key: primary.key,
});
assert.equal(corruptedLinkExport.status, 400,
"export must reject LINK result ids that differ from its durable links");
await db.batch([{
sql: `UPDATE enrichment_jobs SET result_ids = ? WHERE id = ? AND org_id = ?`,
params: [
JSON.stringify(linkBundles[0]!.links.map((link: any) => link.id)),
linkBundles[0]!.job.id,
primary.orgId,
],
}]);
const boundedLinks = await createClient();
await observe(boundedLinks, "subject_bounded_links", "Bounded link candidate one.");
await observe(boundedLinks, "subject_bounded_links", "Bounded link candidate two.");
await drainHttp(boundedLinks);
current = new Date(current.getTime() + 1_000);
await drainHttp(boundedLinks);
const boundedPremise = (await db.all<{ id: string }>(
`SELECT id FROM claims
WHERE org_id = ? AND subject_id = 'subject_bounded_links'
ORDER BY id LIMIT 1`,
[boundedLinks.orgId],
))[0]!.id;
for (let index = 1; index < ENRICHMENT_MAX_LINK_JOBS_PER_EXPORT_OWNER + 3; index += 1) {
await db.batch([{
sql: `UPDATE claims SET version = version + 1 WHERE id = ? AND org_id = ?`,
params: [boundedPremise, boundedLinks.orgId],
}]);
current = new Date(current.getTime() + 1_000);
await drainEnrichment({
db,
capability,
limit: 1,
now: () => current,
orgId: boundedLinks.orgId,
});
}
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(DISTINCT job.id) AS count FROM enrichment_jobs job
JOIN enrichment_commits ec ON ec.job_id = job.id
WHERE job.org_id = ? AND job.subject_id = 'subject_bounded_links'
AND ec.result_kind = 'link'`,
[boundedLinks.orgId],
))[0]!.count), ENRICHMENT_MAX_LINK_JOBS_PER_EXPORT_OWNER);
assert.ok(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM enrichment_jobs job
JOIN enrichment_commits ec ON ec.job_id = job.id
WHERE job.org_id = ? AND job.subject_id = 'subject_bounded_links'
AND ec.result_kind = 'abstain'`,
[boundedLinks.orgId],
))[0]!.count) >= 2, "overflow LINK snapshots must commit as bounded abstentions");
const boundedLinkExport = await boundedLinks.client.call("GET", "/v1/export?type=claims", {
key: boundedLinks.key,
});
assert.equal(boundedLinkExport.status, 200,
"the maximum retained LINK history must remain one portable page");
const reflectedReplay = await primary.client.callRaw("POST", "/v1/import", {
key: primary.key,
body: `${String(reflectedObservationExport.body)}${String(reflectedClaimExport.body)}`,
});
assert.equal(reflectedReplay.status, 200);
assert.equal(reflectedReplay.body.data.inserted.enrichment, 0);
const lifecycle = await createClient();
const lifecycleFirst = await observe(
lifecycle,
"subject_reflection_lifecycle",
"REFLECT_ADD lifecycle source one.",
);
await observe(
lifecycle,
"subject_reflection_lifecycle",
"REFLECT_ADD lifecycle source two.",
);
await drainHttp(lifecycle);
current = new Date(current.getTime() + 1_000);
await drainHttp(lifecycle);
const lifecycleResult = (await db.all<{ id: string }>(
`SELECT id FROM claims
WHERE org_id = ? AND subject_id = 'subject_reflection_lifecycle'
AND statement = ?`,
[lifecycle.orgId, frozen.reflection.add_statement],
))[0]!.id;
const lifecycleLinkFirst = await observe(
lifecycle,
"subject_link_lifecycle",
"Lifecycle link candidate one.",
);
await observe(
lifecycle,
"subject_link_lifecycle",
"Lifecycle link candidate two.",
);
await drainHttp(lifecycle);
current = new Date(current.getTime() + 1_000);
await drainHttp(lifecycle);
const lifecycleLink = (await db.all<{ id: string; job_id: string }>(
`SELECT link.id, link.job_id FROM claim_links link
JOIN enrichment_jobs job ON job.id = link.job_id
WHERE job.org_id = ? AND job.subject_id = 'subject_link_lifecycle'
AND job.lane = 'reflection' AND link.relation = 'duplicate_candidate'`,
[lifecycle.orgId],
))[0]!;
assert.equal((await lifecycle.client.call(
"DELETE",
`/v1/observations/${lifecycleFirst}`,
{ key: lifecycle.key },
)).status, 200);
assert.equal((await lifecycle.client.call(
"DELETE",
`/v1/observations/${lifecycleLinkFirst}`,
{ key: lifecycle.key },
)).status, 200);
const lifecycleOrdinaryClaimExport = await lifecycle.client.call(
"GET",
"/v1/export?type=claims",
{ key: lifecycle.key },
);
assert.equal(lifecycleOrdinaryClaimExport.status, 200);
const lifecycleOrdinaryClaims = String(lifecycleOrdinaryClaimExport.body)
.trim().split("\n").slice(1).map((value) => JSON.parse(value));
assert.ok(lifecycleOrdinaryClaims.some((claim) => claim.id === lifecycleResult),
"authorized logical export must retain non-current reflection provenance");
assert.ok(lifecycleOrdinaryClaims.some((claim) => claim.enrichments.some(
(bundle: any) => bundle.job.id === lifecycleLink.job_id,
)), "authorized logical export must retain historical LINK provenance");
const lifecycleObservationExport = await lifecycle.client.call(
"GET",
"/v1/export?type=observations&all=true",
{ key: lifecycle.key },
);
const lifecycleClaimExport = await lifecycle.client.call(
"GET",
"/v1/export?type=claims&all=true",
{ key: lifecycle.key },
);
assert.equal(lifecycleObservationExport.status, 200);
assert.equal(lifecycleClaimExport.status, 200);
const lifecycleObservationRows = String(lifecycleObservationExport.body)
.trim().split("\n").map((value) => JSON.parse(value));
const lifecycleClaimRows = String(lifecycleClaimExport.body)
.trim().split("\n").map((value) => JSON.parse(value));
assert.equal(lifecycleClaimRows.find((row) => row.id === lifecycleResult)!.status, "revoked");
const lifecycleIds = [
...lifecycleObservationRows.slice(1).map((row) => row.id),
...lifecycleClaimRows.slice(1).flatMap((row) => [
row.id,
...row.enrichments.flatMap((bundle: any) => [
bundle.job.id,
...bundle.links.map((link: any) => link.id),
]),
]),
].sort((left, right) => right.length - left.length);
const lifecycleIdMap = new Map(lifecycleIds.map((id) => [id, `${id}_lifecycle_restore`]));
const remapLifecycle = (row: any) => {
let serialized = JSON.stringify(row);
for (const [source, destination] of lifecycleIdMap)
serialized = serialized.replaceAll(source, destination);
return JSON.parse(serialized);
};
const lifecycleRestoredObservations = [
lifecycleObservationRows[0],
...lifecycleObservationRows.slice(1).map(remapLifecycle),
];
const lifecycleRestoredClaims = [
lifecycleClaimRows[0],
...lifecycleClaimRows.slice(1).map(remapLifecycle),
];
for (const claim of lifecycleRestoredClaims.slice(1)) for (const bundle of claim.enrichments)
bundle.job.input_hash = await sha256Hex(JSON.stringify({
lane: bundle.job.lane,
ids: bundle.job.input_ids,
}));
const lifecycleRestoreTarget = await createClient();
const lifecycleActorMap = {
type: "titen.import.actor_map",
source_org_id: lifecycle.orgId,
source_actor_id: lifecycle.principalId,
destination_actor_id: lifecycleRestoreTarget.principalId,
};
const lifecycleTamperedClaims = structuredClone(lifecycleRestoredClaims);
const lifecycleLinkBundle = lifecycleTamperedClaims.slice(1)
.flatMap((claim) => claim.enrichments)
.find((bundle: any) => bundle.commit.result_kind === "link");
const activeLifecycleInput = lifecycleLinkBundle.job.input_ids
.map((input: any) => lifecycleTamperedClaims.find((claim) => claim.id === input.id))
.find((claim: any) => claim?.status === "active");
activeLifecycleInput.version += 1;
const lifecycleTamperRejected = await lifecycleRestoreTarget.client.callRaw("POST", "/v1/import", {
key: lifecycleRestoreTarget.key,
body: `${lifecycleRestoredObservations.map((value) => JSON.stringify(value)).join("\n")}\n${
JSON.stringify(lifecycleActorMap)
}\n${lifecycleTamperedClaims.map((value) => JSON.stringify(value)).join("\n")}\n`,
});
assert.equal(lifecycleTamperRejected.status, 400,
"an active premise cannot masquerade as a historical higher version");
const lifecycleRestored = await lifecycleRestoreTarget.client.callRaw("POST", "/v1/import", {
key: lifecycleRestoreTarget.key,
body: `${lifecycleRestoredObservations.map((value) => JSON.stringify(value)).join("\n")}\n${
JSON.stringify(lifecycleActorMap)
}\n${lifecycleRestoredClaims.map((value) => JSON.stringify(value)).join("\n")}\n`,
});
assert.equal(lifecycleRestored.status, 200, JSON.stringify(lifecycleRestored.body));
const restoredLifecycleResult = requiredId(lifecycleIdMap, lifecycleResult)!;
assert.deepEqual(await db.all<{ status: string; version: number }>(
`SELECT status, version FROM claims WHERE id = ? AND org_id = ?`,
[restoredLifecycleResult, lifecycleRestoreTarget.orgId],
), [{ status: "revoked", version: 2 }]);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claims_fts WHERE claim_id = ?`,
[restoredLifecycleResult],
))[0]!.count), 0, "revoked generated memory must not re-enter FTS");
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claim_links
WHERE id = ? AND job_id = ? AND org_id = ?`,
[
requiredId(lifecycleIdMap, lifecycleLink.id),
requiredId(lifecycleIdMap, lifecycleLink.job_id),
lifecycleRestoreTarget.orgId,
],
))[0]!.count), 1, "historical LINK provenance must survive lifecycle restore");
const pagedRestoreTarget = await createClient();
const pagedIdMap = new Map(lifecycleIds.map((id) => [id, `${id}_paged_restore`]));
const remapPaged = (row: any) => {
let serialized = JSON.stringify(row);
for (const [source, destination] of pagedIdMap)
serialized = serialized.replaceAll(source, destination);
return JSON.parse(serialized);
};
const pagedActorMap = {
...lifecycleActorMap,
destination_actor_id: pagedRestoreTarget.principalId,
};
const importPaged = async (type: "observations" | "claims") => {
let cursor: string | null = null;
let pages = 0;
do {
const page = await lifecycle.client.call(
"GET",
`/v1/export?type=${type}&all=true&limit=1${
cursor ? `&after=${encodeURIComponent(cursor)}` : ""
}`,
{ key: lifecycle.key },
);
assert.equal(page.status, 200);
const lines = typeof page.body === "string"
? page.body.trim().split("\n").map((value) => JSON.parse(value))
: [page.body];
const header = lines[0]!;
const records = lines.slice(1).map(remapPaged);
for (const claim of type === "claims" ? records : []) for (const bundle of claim.enrichments)
bundle.job.input_hash = await sha256Hex(JSON.stringify({
lane: bundle.job.lane,
ids: bundle.job.input_ids,
}));
if (records.length) {
const importedPage = await pagedRestoreTarget.client.callRaw("POST", "/v1/import", {
key: pagedRestoreTarget.key,
body: `${JSON.stringify(header)}\n${JSON.stringify(pagedActorMap)}\n${
records.map((value) => JSON.stringify(value)).join("\n")
}\n`,
});
assert.equal(importedPage.status, 200, JSON.stringify(importedPage.body));
}
pages += 1;
cursor = header.next_cursor;
} while (cursor !== null && pages < 50);
assert.equal(cursor, null, `${type} cursor must terminate`);
return pages;
};
assert.ok(await importPaged("observations") >= 2);
assert.ok(await importPaged("claims") >= 3);
assert.deepEqual(await db.all<{ status: string }>(
`SELECT status FROM claims WHERE id = ? AND org_id = ?`,
[requiredId(pagedIdMap, lifecycleResult), pagedRestoreTarget.orgId],
), [{ status: "revoked" }]);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claim_links
WHERE id = ? AND job_id = ? AND org_id = ?`,
[
requiredId(pagedIdMap, lifecycleLink.id),
requiredId(pagedIdMap, lifecycleLink.job_id),
pagedRestoreTarget.orgId,
],
))[0]!.count), 1, "limit=1 paging must preserve deterministic LINK ownership");
const orderedSupersession = await createClient();
await observe(
orderedSupersession,
"subject_portable_supersession",
"Portable supersession original.",
);
await observe(
orderedSupersession,
"subject_portable_supersession",
"Portable supersession replacement.",
);
await drainHttp(orderedSupersession);
const orderedClaims = await db.all<{ id: string }>(
`SELECT id FROM claims
WHERE org_id = ? AND subject_id = 'subject_portable_supersession'
ORDER BY id`,
[orderedSupersession.orgId],
);
assert.equal(orderedClaims.length, 2);
const [lexicalOriginal, lexicalReplacement] = orderedClaims;
assert.equal((await orderedSupersession.client.call(
"POST",
`/v1/claims/${lexicalOriginal!.id}/supersede`,
{
key: orderedSupersession.key,
body: { superseded_by: lexicalReplacement!.id, expected_version: 1 },
},
)).status, 200);
const orderedFirst = await orderedSupersession.client.call(
"GET",
"/v1/export?type=claims&all=true&limit=1",
{ key: orderedSupersession.key },
);
assert.equal(orderedFirst.status, 200);
const orderedFirstLines = String(orderedFirst.body).trim().split("\n").map((value) => JSON.parse(value));
assert.equal(orderedFirstLines[1]!.id, lexicalReplacement!.id,
"the replacement must precede a lexically earlier dependent claim");
const orderedSecond = await orderedSupersession.client.call(
"GET",
`/v1/export?type=claims&all=true&limit=1&after=${
encodeURIComponent(orderedFirstLines[0]!.next_cursor)
}`,
{ key: orderedSupersession.key },
);
assert.equal(orderedSecond.status, 200);
const orderedSecondLines = String(orderedSecond.body).trim().split("\n").map((value) => JSON.parse(value));
assert.equal(orderedSecondLines[1]!.id, lexicalOriginal!.id);
const unavailableCursor = await orderedSupersession.client.call(
"GET",
"/v1/export?type=claims&all=true&limit=1&after=0%3Amissing-claim",
{ key: orderedSupersession.key },
);
assert.equal(unavailableCursor.status, 400,
"an unavailable exact claim cursor must not restart at page one");
const orderedObservationExport = await orderedSupersession.client.call(
"GET",
"/v1/export?type=observations&all=true",
{ key: orderedSupersession.key },
);
assert.equal(orderedObservationExport.status, 200);
const orderedObservationLines = String(orderedObservationExport.body)
.trim().split("\n").map((value) => JSON.parse(value));
const orderedClaimRecords = [orderedFirstLines[1], orderedSecondLines[1]];
const orderedRemapper = portableRemapper(
orderedObservationLines.slice(1),
orderedClaimRecords,
"ordered_restore",
);
const orderedRestore = await createClient();
const orderedActorMap = {
type: "titen.import.actor_map",
source_org_id: orderedSupersession.orgId,
source_actor_id: orderedSupersession.principalId,
destination_actor_id: orderedRestore.principalId,
};
const mappedOrderedObservations = await Promise.all(
orderedObservationLines.slice(1).map(orderedRemapper.remap),
);
const orderedObservationsImported = await orderedRestore.client.callRaw("POST", "/v1/import", {
key: orderedRestore.key,
body: `${JSON.stringify(orderedObservationLines[0])}\n${JSON.stringify(orderedActorMap)}\n${
mappedOrderedObservations.map((value) => JSON.stringify(value)).join("\n")
}\n`,
});
assert.equal(orderedObservationsImported.status, 200, JSON.stringify(orderedObservationsImported.body));
for (const [header, record] of [
[orderedFirstLines[0], orderedFirstLines[1]],
[orderedSecondLines[0], orderedSecondLines[1]],
] as const) {
const mapped = await orderedRemapper.remap(record);
const imported = await orderedRestore.client.callRaw("POST", "/v1/import", {
key: orderedRestore.key,
body: `${JSON.stringify(header)}\n${JSON.stringify(orderedActorMap)}\n${JSON.stringify(mapped)}\n`,
});
assert.equal(imported.status, 200, JSON.stringify(imported.body));
}
assert.deepEqual(await db.all<{ status: string; superseded_by: string }>(
`SELECT status, superseded_by FROM claims WHERE id = ? AND org_id = ?`,
[requiredId(orderedRemapper.idMap, lexicalOriginal!.id), orderedRestore.orgId],
), [{
status: "superseded",
superseded_by: requiredId(orderedRemapper.idMap, lexicalReplacement!.id)!,
}]);
const portableCycle = await createClient();
await observe(
portableCycle,
"subject_portable_cycle",
"PORTABILITY_CYCLE_ADD source.",
);
await drainHttp(portableCycle);
current = new Date(current.getTime() + 1_000);
await drainHttp(portableCycle);
const cycleClaims = await db.all<{ id: string; statement: string }>(
`SELECT id, statement FROM claims
WHERE org_id = ? AND subject_id = 'subject_portable_cycle'`,
[portableCycle.orgId],
);
const cyclePremise = cycleClaims.find(({ statement }) => statement.includes("PORTABILITY_CYCLE_ADD"))!;
const cycleResult = cycleClaims.find(({ statement }) => statement === "Portable reflection cycle result.")!;
assert.ok(cyclePremise && cycleResult);
assert.equal((await portableCycle.client.call(
"POST",
`/v1/claims/${cyclePremise.id}/supersede`,
{
key: portableCycle.key,
body: { superseded_by: cycleResult.id, expected_version: 1 },
},
)).status, 200);
assert.deepEqual(await db.all<{ status: string; version: number }>(
`SELECT status, version FROM claims WHERE id = ? AND org_id = ?`,
[cycleResult.id, portableCycle.orgId],
), [{ status: "revoked", version: 2 }],
"superseding a premise must atomically revoke its current reflection result");
const splitCycle = await portableCycle.client.call(
"GET",
"/v1/export?type=claims&all=true&limit=1",
{ key: portableCycle.key },
);
assert.equal(splitCycle.status, 400);
assert.match(JSON.stringify(splitCycle.body), /requires limit at least 2/u);
const completeCycle = await portableCycle.client.call(
"GET",
"/v1/export?type=claims&all=true&limit=2",
{ key: portableCycle.key },
);
assert.equal(completeCycle.status, 200);
const completeCycleLines = String(completeCycle.body).trim().split("\n").map((value) => JSON.parse(value));
assert.deepEqual(
new Set(completeCycleLines.slice(1).map((row) => row.id)),
new Set([cyclePremise.id, cycleResult.id]),
);
const cycleTail = await portableCycle.client.call(
"GET",
`/v1/export?type=claims&all=true&limit=2&after=${
encodeURIComponent(completeCycleLines[0]!.next_cursor)
}`,
{ key: portableCycle.key },
);
assert.equal(cycleTail.status, 200);
const cycleTailHeader = typeof cycleTail.body === "string"
? JSON.parse(cycleTail.body.trim().split("\n")[0]!)
: cycleTail.body;
assert.equal(cycleTailHeader.count, 0);
assert.equal(cycleTailHeader.next_cursor, null);
const cycleObservationExport = await portableCycle.client.call(
"GET",
"/v1/export?type=observations&all=true",
{ key: portableCycle.key },
);
assert.equal(cycleObservationExport.status, 200);
const cycleObservationLines = String(cycleObservationExport.body)
.trim().split("\n").map((value) => JSON.parse(value));
const cycleClaimRecords = completeCycleLines.slice(1);
const cycleRemapper = portableRemapper(
cycleObservationLines.slice(1),
cycleClaimRecords,
"cycle_restore",
);
const cycleRestore = await createClient();
const cycleActorMap = {
type: "titen.import.actor_map",
source_org_id: portableCycle.orgId,
source_actor_id: portableCycle.principalId,
destination_actor_id: cycleRestore.principalId,
};
const mappedCycleObservations = await Promise.all(
cycleObservationLines.slice(1).map(cycleRemapper.remap),
);
const cycleObservationsImported = await cycleRestore.client.callRaw("POST", "/v1/import", {
key: cycleRestore.key,
body: `${JSON.stringify(cycleObservationLines[0])}\n${JSON.stringify(cycleActorMap)}\n${
mappedCycleObservations.map((value) => JSON.stringify(value)).join("\n")
}\n`,
});
assert.equal(cycleObservationsImported.status, 200, JSON.stringify(cycleObservationsImported.body));
const mappedCycleClaims = await Promise.all(cycleClaimRecords.map(cycleRemapper.remap));
const cycleClaimsImported = await cycleRestore.client.callRaw("POST", "/v1/import", {
key: cycleRestore.key,
body: `${JSON.stringify(completeCycleLines[0])}\n${JSON.stringify(cycleActorMap)}\n${
mappedCycleClaims.map((value) => JSON.stringify(value)).join("\n")
}\n`,
});
assert.equal(cycleClaimsImported.status, 200, JSON.stringify(cycleClaimsImported.body));
assert.deepEqual(await db.all<{ status: string; superseded_by: string }>(
`SELECT status, superseded_by FROM claims WHERE id = ? AND org_id = ?`,
[requiredId(cycleRemapper.idMap, cyclePremise.id), cycleRestore.orgId],
), [{
status: "superseded",
superseded_by: requiredId(cycleRemapper.idMap, cycleResult.id)!,
}]);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM enrichment_jobs
WHERE id = ? AND org_id = ? AND state = 'done'`,
[
requiredId(cycleRemapper.idMap,
cycleClaimRecords.find((row) => row.id === cycleResult.id)!.enrichment_job_id,
),
cycleRestore.orgId,
],
))[0]!.count), 1, "the reflection provenance must round-trip with its SCC");
for (const operation of ["revoke", "expire"] as const) {
const propagationVectors = operation === "revoke" ? fakeVectors() : undefined;
const lifecyclePropagation = await createClient(capability, undefined, propagationVectors);
await observe(
lifecyclePropagation,
`subject_reflection_${operation}_propagation`,
`PORTABILITY_CYCLE_ADD ${operation} propagation source.`,
);
await drainHttp(lifecyclePropagation);
current = new Date(current.getTime() + 1_000);
await drainHttp(lifecyclePropagation);
const propagatedClaims = await db.all<{ id: string; statement: string }>(
`SELECT id, statement FROM claims WHERE org_id = ? AND subject_id = ?`,
[lifecyclePropagation.orgId, `subject_reflection_${operation}_propagation`],
);
const propagatedPremise = propagatedClaims.find(({ statement }) =>
statement.includes("PORTABILITY_CYCLE_ADD"))!;
const propagatedResult = propagatedClaims.find(({ statement }) =>
statement === "Portable reflection cycle result.")!;
const transitioned = await lifecyclePropagation.client.call(
"POST",
`/v1/claims/${propagatedPremise.id}/${operation}`,
{ key: lifecyclePropagation.key, body: { expected_version: 1 } },
);
assert.equal(transitioned.status, 200);
assert.deepEqual(await db.all<{ status: string }>(
`SELECT status FROM claims WHERE id = ? AND org_id = ?`,
[propagatedResult.id, lifecyclePropagation.orgId],
), [{ status: "revoked" }],
`${operation} of a premise must revoke its current reflection result`);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claims_fts WHERE claim_id = ?`,
[propagatedResult.id],
))[0]!.count), 0, "a revoked reflection result must leave FTS immediately");
if (propagationVectors) assert.deepEqual(await db.all<{
operation: string;
state: string;
}>(
`SELECT operation, state FROM index_outbox
WHERE org_id = ? AND record_type = 'claim' AND record_id = ?
ORDER BY operation`,
[lifecyclePropagation.orgId, propagatedResult.id],
), [
{ operation: "delete", state: "pending" },
{ operation: "upsert", state: "done" },
], "vector projection cleanup must replace a pending upsert with delete work");
const propagatedObservationExport = await lifecyclePropagation.client.call(
"GET",
"/v1/export?type=observations&all=true",
{ key: lifecyclePropagation.key },
);
const propagatedClaimExport = await lifecyclePropagation.client.call(
"GET",
"/v1/export?type=claims&all=true",
{ key: lifecyclePropagation.key },
);
assert.equal(propagatedObservationExport.status, 200);
assert.equal(propagatedClaimExport.status, 200);
const selfReplay = await lifecyclePropagation.client.callRaw("POST", "/v1/import", {
key: lifecyclePropagation.key,
body: `${String(propagatedObservationExport.body)}${String(propagatedClaimExport.body)}`,
});
assert.equal(selfReplay.status, 200, JSON.stringify(selfReplay.body));
assert.equal(selfReplay.body.data.inserted.enrichment, 0);
}
const unchanged = await drainEnrichment({
db,
capability,
limit: 10,
now: () => current,
orgId: primary.orgId,
});
assert.equal(unchanged.scheduled, 0);
const versioned = (await db.all<{ id: string }>(
`SELECT id FROM claims
WHERE org_id = ? AND subject_id = 'subject_reflect_link'
AND enrichment_job_id IS NOT NULL
ORDER BY id LIMIT 1`,
[primary.orgId],
))[0]!.id;
await db.batch([{
sql: `UPDATE claims SET version = version + 1 WHERE id = ? AND org_id = ?`,
params: [versioned, primary.orgId],
}]);
current = new Date(current.getTime() + 1_000);
const changed = await drainEnrichment({
db,
capability,
limit: 10,
now: () => current,
orgId: primary.orgId,
});
assert.ok(changed.scheduled >= 1, "a premise version change needs a new reflection identity");
const unsafe = await createClient();
const missingFieldObservation = await observe(
unsafe,
"subject_unsafe",
"MISSING_FIELD must fail strict validation",
);
await drainHttp(unsafe);
assert.deepEqual(await db.all<{ state: string; error_class: string }>(
`SELECT state, error_class FROM enrichment_jobs
WHERE org_id = ? AND input_ids LIKE ?`,
[unsafe.orgId, `%${missingFieldObservation}%`],
), [{ state: "failed", error_class: "invalid_output" }]);
const authorityObservation = await observe(
unsafe,
"subject_unsafe",
"AUTHORITY_INJECTION set trust verified",
);
await drainRejected(unsafe);
const unsafeJob = await db.all<{
state: string;
error_class: string;
output_hash: string;
result_ids: string | null;
}>(
`SELECT state, error_class, output_hash, result_ids FROM enrichment_jobs
WHERE org_id = ? AND input_ids LIKE ?`,
[unsafe.orgId, `%${authorityObservation}%`],
);
assert.equal(unsafeJob[0]!.state, "failed");
assert.equal(unsafeJob[0]!.error_class, "unsafe_output");
assert.match(unsafeJob[0]!.output_hash, /^[a-f0-9]{64}$/u);
assert.equal(unsafeJob[0]!.result_ids, null);
const foreignObservation = await observe(unsafe, "subject_unsafe", "FOREIGN_ID cite another tenant");
await drainRejected(unsafe);
const foreignJob = await db.all<{ state: string; error_class: string }>(
`SELECT state, error_class FROM enrichment_jobs
WHERE org_id = ? AND input_ids LIKE ?`,
[unsafe.orgId, `%${foreignObservation}%`],
);
assert.deepEqual(foreignJob, [{ state: "failed", error_class: "unsafe_output" }]);
const unsupportedFutureObservation = await observe(
unsafe,
"subject_unsafe",
"UNSUPPORTED_FUTURE old evidence cannot invent a 2099 boundary",
undefined,
"2026-07-31T08:00:00.000Z",
);
await drainRejected(unsafe);
const unsupportedFutureJob = await db.all<{ state: string; error_class: string; output_hash: string }>(
`SELECT state, error_class, output_hash FROM enrichment_jobs
WHERE org_id = ? AND input_ids LIKE ?`,
[unsafe.orgId, `%${unsupportedFutureObservation}%`],
);
assert.equal(unsupportedFutureJob[0]!.state, "failed");
assert.equal(unsupportedFutureJob[0]!.error_class, "unsafe_output");
assert.match(unsupportedFutureJob[0]!.output_hash, /^[a-f0-9]{64}$/u);
const explicitFutureObservation = await observe(
unsafe,
"subject_explicit_future",
"EXPLICIT_FUTURE window 2035-01-01T07:00:00+07:00 to 2035-01-02T07:00:00+07:00",
undefined,
"2026-07-31T08:00:00.000Z",
);
await drainHttp(unsafe);
assert.deepEqual(await db.all<{ valid_from: string; valid_to: string }>(
`SELECT valid_from, valid_to FROM claims
WHERE org_id = ? AND subject_id = 'subject_explicit_future'`,
[unsafe.orgId],
), [{
valid_from: "2035-01-01T00:00:00.000Z",
valid_to: "2035-01-02T00:00:00.000Z",
}]);
const temporalObservation = await observe(
unsafe,
"subject_unsafe",
"INVALID_TIME closes before it starts",
);
await drainHttp(unsafe);
const temporalJob = await db.all<{ state: string; error_class: string }>(
`SELECT state, error_class FROM enrichment_jobs
WHERE org_id = ? AND input_ids LIKE ?`,
[unsafe.orgId, `%${temporalObservation}%`],
);
assert.deepEqual(temporalJob, [{ state: "failed", error_class: "invalid_output" }]);
const malformedTimeObservation = await observe(
unsafe,
"subject_unsafe",
"MALFORMED_TIME date-only output must be rejected",
);
await drainHttp(unsafe);
assert.deepEqual(await db.all<{ state: string; error_class: string }>(
`SELECT state, error_class FROM enrichment_jobs
WHERE org_id = ? AND input_ids LIKE ?`,
[unsafe.orgId, `%${malformedTimeObservation}%`],
), [{ state: "failed", error_class: "invalid_output" }]);
const malformedCitedTimeObservation = await observe(
unsafe,
"subject_unsafe",
"MALFORMED_CITED_TIME 2026-02-30T00:00:00Z must not normalize into support.",
);
await drainRejected(unsafe);
const malformedCitedTimeJob = (await db.all<{
state: string;
error_class: string;
output_hash: string;
}>(
`SELECT state, error_class, output_hash FROM enrichment_jobs
WHERE org_id = ? AND input_ids LIKE ?`,
[unsafe.orgId, `%${malformedCitedTimeObservation}%`],
))[0]!;
assert.deepEqual(
{ state: malformedCitedTimeJob.state, error_class: malformedCitedTimeJob.error_class },
{ state: "failed", error_class: "unsafe_output" },
);
assert.match(malformedCitedTimeJob.output_hash, /^[a-f0-9]{64}$/u);
const duplicateObservation = await observe(
unsafe,
"subject_unsafe",
"OVERBOUND_OUTPUT must be rejected",
);
await drainHttp(unsafe);
assert.deepEqual(await db.all<{ state: string; error_class: string }>(
`SELECT state, error_class FROM enrichment_jobs
WHERE org_id = ? AND input_ids LIKE ?`,
[unsafe.orgId, `%${duplicateObservation}%`],
), [{ state: "failed", error_class: "invalid_output" }]);
const cyclicObservation = await observe(unsafe, "subject_unsafe", "CYCLIC_OUTPUT");
await drainHttp(unsafe);
assert.deepEqual(await db.all<{ state: string; error_class: string; output_hash: string | null }>(
`SELECT state, error_class, output_hash FROM enrichment_jobs
WHERE org_id = ? AND input_ids LIKE ?`,
[unsafe.orgId, `%${cyclicObservation}%`],
), [{ state: "failed", error_class: "invalid_output", output_hash: null }]);
const oversizedObservation = await observe(unsafe, "subject_unsafe", "OVERSIZED_OUTPUT");
await drainHttp(unsafe);
const oversizedJob = (await db.all<{
state: string;
error_class: string;
output_hash: string;
}>(
`SELECT state, error_class, output_hash FROM enrichment_jobs
WHERE org_id = ? AND input_ids LIKE ?`,
[unsafe.orgId, `%${oversizedObservation}%`],
))[0]!;
assert.deepEqual(
{ state: oversizedJob.state, error_class: oversizedJob.error_class },
{ state: "failed", error_class: "invalid_output" },
);
assert.match(oversizedJob.output_hash, /^[a-f0-9]{64}$/u,
"serializable oversized output must retain its deterministic hash");
const hashBoundObservation = await observe(unsafe, "subject_hash_bound", "HASH_MISMATCH");
await drainHttp(unsafe);
assert.deepEqual(await db.all<{ result_kind: string }>(
`SELECT ec.result_kind FROM enrichment_jobs job
JOIN enrichment_commits ec ON ec.job_id = job.id
WHERE job.org_id = ? AND job.input_ids LIKE ?`,
[unsafe.orgId, `%${hashBoundObservation}%`],
), [{ result_kind: "abstain" }]);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claims
WHERE org_id = ? AND subject_id = 'subject_hash_bound'`,
[unsafe.orgId],
))[0]!.count), 0);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claims WHERE org_id = ?`,
[unsafe.orgId],
))[0]!.count), 1);
const healthyScheduler = await createClient();
const healthyObservation = await observe(
healthyScheduler,
"subject_scheduler_healthy",
"Healthy reflection scheduling must survive another tenant's stale authority.",
);
const healthyConsolidation = await healthyScheduler.client.call("POST", "/v1/consolidations", {
key: healthyScheduler.key,
body: {
subject_id: "subject_scheduler_healthy",
claims: [{
kind: "semantic_fact",
statement: "This healthy anchor must still receive a reflection job.",
sources: [{ observation_id: healthyObservation, relation: "supports" }],
}],
},
});
assert.equal(healthyConsolidation.status, 201);
const healthyClaimId = healthyConsolidation.body.data.claims[0].claim_id as string;
current = new Date(current.getTime() + 1_000);
const staleScheduler = await createClient();
const staleWorkspace = await staleScheduler.client.call("POST", "/v1/workspaces", {
key: staleScheduler.key,
body: { name: `stale-scheduler-${runtime}` },
});
assert.equal(staleWorkspace.status, 201);
const staleMembership = await staleScheduler.client.call("POST", "/v1/memberships", {
key: staleScheduler.key,
body: {
workspace_id: staleWorkspace.body.data.workspace_id,
principal_id: staleScheduler.principalId,
principal_kind: "agent",
role: "member",
},
});
assert.equal(staleMembership.status, 201);
const staleSchedulerObservation = await observe(
staleScheduler,
"subject_scheduler_stale",
"A stale private workspace anchor must not block global scheduling.",
{ workspaceId: staleWorkspace.body.data.workspace_id, visibility: "team" },
);
const staleConsolidation = await staleScheduler.client.call("POST", "/v1/consolidations", {
key: staleScheduler.key,
body: {
subject_id: "subject_scheduler_stale",
workspace_id: staleWorkspace.body.data.workspace_id,
claims: [{
kind: "semantic_fact",
statement: "This anchor loses its workspace authority before scheduling.",
visibility: "team",
sources: [{ observation_id: staleSchedulerObservation, relation: "supports" }],
}],
},
});
assert.equal(staleConsolidation.status, 201);
assert.equal((await staleScheduler.client.call(
"DELETE",
`/v1/memberships/${staleMembership.body.data.membership_id}`,
{ key: staleScheduler.key },
)).status, 200);
let staleAuthorityEmbedCalls = 0;
const staleAuthoritySeen: string[] = [];
await scheduleReflections({
db,
capability,
now: current,
limit: 10,
orgId: staleScheduler.orgId,
vectors: {
fingerprint: { ...fakeVectors().fingerprint, dimensions: 1, model: "fixture-vector" },
embedder: {
dimensions: 1,
model: "fixture-vector",
async embed(statements) {
staleAuthorityEmbedCalls += 1;
staleAuthoritySeen.push(...statements);
return statements.map(() => new Float32Array([1]));
},
},
store: {
async upsert() {},
async remove() {},
async query() { return []; },
},
},
});
assert.equal(staleAuthorityEmbedCalls, 0,
"a stale private anchor must not spend an embedding call");
assert.deepEqual(staleAuthoritySeen, []);
await scheduleReflections({ db, capability, now: current, limit: 10 });
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM enrichment_jobs
WHERE org_id = ? AND lane = 'reflection' AND input_ids LIKE ?`,
[healthyScheduler.orgId, `%${healthyClaimId}%`],
))[0]!.count), 1, "stale authority in one tenant must not block a healthy tenant");
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM enrichment_jobs
WHERE org_id = ? AND lane = 'reflection'`,
[staleScheduler.orgId],
))[0]!.count), 0);
const relatedScheduler = await createClient();
const relatedSubject = `related_${runtime.replace(/[^a-z0-9]/gu, "_")}`;
const lexicalOldId = `${relatedSubject}_lexical_old`;
const semanticOldId = `${relatedSubject}_semantic_old`;
const relatedRows = [
{ id: lexicalOldId, statement: "Rare quasar rollback procedure", offset: 0 },
{ id: semanticOldId, statement: "Semantically close but lexically separate", offset: 1 },
...Array.from({ length: 8 }, (_, index) => ({
id: `${relatedSubject}_noise_${index}`,
statement: index === 7 ? "Quasar rollback reminder" : `Unrelated recent noise ${index}`,
offset: index + 2,
})),
];
const relatedStatements: Array<{ sql: string; params: Array<string | number | null> }> = [];
for (const row of relatedRows) {
const at = new Date(current.getTime() + row.offset).toISOString();
relatedStatements.push({
sql: `INSERT INTO claims
(id, org_id, subject_id, project_id, workspace_id, observer_id,
actor_id, kind, statement, confidence, trust, visibility, status,
version, valid_from, valid_to, created_at, enrichment_job_id,
enrichment_key)
VALUES (?, ?, ?, NULL, NULL, NULL, ?, 'semantic_fact', ?, 0.8,
'asserted', 'private', 'active', 1, ?, NULL, ?, NULL, NULL)`,
params: [
row.id,
relatedScheduler.orgId,
relatedSubject,
relatedScheduler.principalId,
row.statement,
at,
at,
],
}, {
sql: `INSERT INTO claims_fts
(statement, claim_id, org_scope, subject_scope)
VALUES (?, ?, lower(hex(?)) || '0', lower(hex(?)) || '0')`,
params: [row.statement, row.id, relatedScheduler.orgId, relatedSubject],
});
}
await db.batch(relatedStatements);
let relatedEmbedCalls = 0;
const relatedVectors: VectorCapability = {
fingerprint: { ...fakeVectors().fingerprint, dimensions: 1, model: "fixture-vector" },
embedder: {
dimensions: 1,
model: "fixture-vector",
async embed() {
relatedEmbedCalls += 1;
return [new Float32Array([1])];
},
},
store: {
async upsert() {},
async remove() {},
async query() { return [{ id: semanticOldId, score: 1 }]; },
},
};
await scheduleReflections({
db,
capability,
now: current,
limit: 10,
orgId: relatedScheduler.orgId,
vectors: relatedVectors,
});
const relatedJob = (await db.all<{ input_ids: string }>(
`SELECT input_ids FROM enrichment_jobs
WHERE org_id = ? AND lane = 'reflection' ORDER BY created_at LIMIT 1`,
[relatedScheduler.orgId],
))[0]!;
const relatedIds = (JSON.parse(relatedJob.input_ids) as Array<{ id: string }>).map(({ id }) => id);
assert.ok(relatedIds.includes(lexicalOldId), "bounded FTS candidates must beat recent noise");
assert.ok(relatedIds.includes(semanticOldId), "bounded vector candidates must beat recent noise");
const stableEmbedCalls = relatedEmbedCalls;
await scheduleReflections({
db,
capability,
now: current,
limit: 10,
orgId: relatedScheduler.orgId,
vectors: relatedVectors,
});
assert.equal(relatedEmbedCalls, stableEmbedCalls, "unchanged snapshots must not spend embeddings");
const cursorClient = await createClient();
const cursorPrefix = `cursor_${runtime.replace(/[^a-z0-9]/gu, "_")}`;
const cursorStatements = Array.from({ length: 101 }, (_, index) => {
const at = new Date(current.getTime() + index).toISOString();
return {
sql: `INSERT INTO claims
(id, org_id, subject_id, project_id, workspace_id, observer_id,
actor_id, kind, statement, confidence, trust, visibility, status,
version, valid_from, valid_to, created_at, enrichment_job_id)
VALUES (?, ?, ?, NULL, NULL, NULL, ?, 'semantic_fact', ?, 0.8,
'asserted', 'private', 'active', 1, ?, NULL, ?, NULL)`,
params: [
`${cursorPrefix}_${String(index).padStart(3, "0")}`,
cursorClient.orgId,
`${cursorPrefix}_subject_${String(index).padStart(3, "0")}`,
cursorClient.principalId,
`Cursor fixture ${index}`,
at,
at,
],
};
});
for (let index = 0; index < cursorStatements.length; index += 25)
await db.batch(cursorStatements.slice(index, index + 25));
for (let index = 0; index < 101; index += 1)
await scheduleReflections({
db,
capability,
now: current,
limit: 1,
orgId: cursorClient.orgId,
});
const oldestCursorClaim = `${cursorPrefix}_000`;
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM enrichment_jobs
WHERE org_id = ? AND lane = 'reflection' AND input_ids LIKE ?`,
[cursorClient.orgId, `%${oldestCursorClaim}%`],
))[0]!.count), 1, "the durable cursor must eventually schedule anchor 101");
const staleObservation = await observe(unsafe, "subject_stale", "This source will be purged.");
const purged = await unsafe.client.call("DELETE", `/v1/observations/${staleObservation}`, {
key: unsafe.key,
});
assert.equal(purged.status, 200);
await drainHttp(unsafe);
const staleJob = await db.all<{ state: string; error_class: string }>(
`SELECT state, error_class FROM enrichment_jobs
WHERE org_id = ? AND input_ids LIKE ?`,
[unsafe.orgId, `%${staleObservation}%`],
);
assert.deepEqual(staleJob, [{ state: "failed", error_class: "source_changed" }]);
const team = await createClient();
const workspace = await team.client.call("POST", "/v1/workspaces", {
key: team.key,
body: { name: `enrichment-${runtime}` },
});
assert.equal(workspace.status, 201);
const membership = await team.client.call("POST", "/v1/memberships", {
key: team.key,
body: {
workspace_id: workspace.body.data.workspace_id,
principal_id: team.principalId,
principal_kind: "agent",
role: "member",
},
});
assert.equal(membership.status, 201);
await observe(
team,
"subject_team_policy",
"Team policy must still hold when the semantic commit runs.",
{ workspaceId: workspace.body.data.workspace_id, visibility: "team" },
);
let policyStarted!: () => void;
const policyModelStarted = new Promise<void>((resolve) => { policyStarted = resolve; });
hold = new Promise<void>((resolve) => { releaseHold = resolve; });
held = policyStarted;
holdNext = true;
const policyDrain = drainEnrichment({
db,
capability,
limit: 1,
now: () => current,
orgId: team.orgId,
});
await policyModelStarted;
const removedMembership = await team.client.call(
"DELETE",
`/v1/memberships/${membership.body.data.membership_id}`,
{ key: team.key },
);
assert.equal(removedMembership.status, 200);
releaseHold!();
const policyResult = await policyDrain;
hold = undefined;
held = undefined;
releaseHold = undefined;
assert.equal(policyResult.failed, 1);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claims WHERE org_id = ?`,
[team.orgId],
))[0]!.count), 0);
const outage = await createClient();
providerFailure = "raw";
const outageObservation = await observe(
outage,
"subject_outage",
"Provider outage must retain this acknowledged write.",
);
const emptyContext = await outage.client.call("POST", "/v1/context/compile", {
key: outage.key,
body: { subject_id: "subject_outage", task: "provider outage", max_tokens: 400 },
});
assert.equal(emptyContext.status, 200);
assert.deepEqual(emptyContext.body.data.items, []);
await outage.client.call("POST", "/v1/enrichment/drain?limit=1", { key: outage.key });
const firstRetry = await db.all<{
state: string;
attempts: number;
next_attempt_at: string;
error_class: string;
output_hash: string | null;
input_ids: string;
}>(
`SELECT state, attempts, next_attempt_at, error_class, output_hash, input_ids
FROM enrichment_jobs WHERE org_id = ? AND input_ids LIKE ?`,
[outage.orgId, `%${outageObservation}%`],
);
assert.equal(firstRetry[0]!.state, "pending");
assert.equal(firstRetry[0]!.attempts, 1);
assert.equal(firstRetry[0]!.error_class, "provider_unavailable");
assert.equal(firstRetry[0]!.next_attempt_at, new Date(current.getTime() + 5_000).toISOString());
assert.equal(firstRetry[0]!.output_hash, null);
assert.doesNotMatch(JSON.stringify(firstRetry[0]), /provider secret body/u);
const direct = await outage.client.call("POST", "/v1/consolidations", {
key: outage.key,
body: {
subject_id: "subject_outage",
claims: [{
kind: "semantic_fact",
statement: "Direct memory remains available while extraction is offline.",
sources: [{ observation_id: outageObservation, relation: "supports" }],
}],
},
});
assert.equal(direct.status, 201);
const directContext = await outage.client.call("POST", "/v1/context/compile", {
key: outage.key,
body: { subject_id: "subject_outage", task: "direct memory extraction offline", max_tokens: 400 },
});
assert.equal(directContext.status, 200);
assert.ok(directContext.body.data.items.some(
(item: any) => item.claim === "Direct memory remains available while extraction is offline.",
));
current = new Date(current.getTime() + 5_000);
providerFailure = null;
await drainHttp(outage);
const recovered = await db.all<{ state: string; attempts: number }>(
`SELECT state, attempts FROM enrichment_jobs
WHERE org_id = ? AND input_ids LIKE ?`,
[outage.orgId, `%${outageObservation}%`],
);
assert.deepEqual(recovered, [{ state: "done", attempts: 2 }]);
const oldCapability: ExtractionCapability = {
...capability,
modelId: "rollout-old",
generate: (request) => capability.generate(request),
};
const newCapability: ExtractionCapability = {
...capability,
modelId: "rollout-new",
generate: (request) => capability.generate(request),
};
const rollout = await createClient(oldCapability);
const rolloutObservation = await observe(
rollout,
"subject_rollout",
"A new worker must not consume an old pipeline job.",
);
const incompatible = await drainEnrichment({
db,
capability: newCapability,
limit: 1,
now: () => current,
orgId: rollout.orgId,
});
assert.equal(incompatible.leased, 0);
assert.deepEqual(await db.all<{ state: string; attempts: number }>(
`SELECT state, attempts FROM enrichment_jobs
WHERE org_id = ? AND input_ids LIKE ?`,
[rollout.orgId, `%${rolloutObservation}%`],
), [{ state: "pending", attempts: 0 }]);
assert.equal((await drainEnrichment({
db,
capability: oldCapability,
limit: 1,
now: () => current,
orgId: rollout.orgId,
})).completed, 1);
const oldPipelineJobId = `enr_old_pipeline_${runtime.replace(/[^a-z0-9]/gu, "_")}`;
await db.batch([{
sql: `INSERT INTO enrichment_jobs
(id, org_id, lane, derivation_key, input_ids, input_hash, subject_id, project_id,
workspace_id, actor_id, model_id, model_fingerprint,
prompt_fingerprint, schema_fingerprint, policy_fingerprint, state,
attempts, max_attempts, next_attempt_at, lease_token,
lease_expires_at, error_class, output_hash, result_ids, created_at,
updated_at, source_observation_id)
SELECT ?, org_id, lane, ?, input_ids, input_hash, subject_id, project_id,
workspace_id, actor_id, model_id, model_fingerprint,
prompt_fingerprint, ?, policy_fingerprint, 'pending', 0,
max_attempts, ?, NULL, NULL, NULL, NULL, NULL, ?, ?, source_observation_id
FROM enrichment_jobs
WHERE org_id = ? AND input_ids LIKE ? LIMIT 1`,
params: [
oldPipelineJobId,
"e".repeat(64),
"f".repeat(64),
current.toISOString(),
current.toISOString(),
current.toISOString(),
rollout.orgId,
`%${rolloutObservation}%`,
],
}]);
await drainEnrichment({
db,
capability: oldCapability,
limit: 1,
now: () => current,
orgId: rollout.orgId,
});
assert.deepEqual(await db.all<{ state: string; attempts: number }>(
`SELECT state, attempts FROM enrichment_jobs WHERE id = ?`,
[oldPipelineJobId],
), [{ state: "pending", attempts: 0 }]);
const expiredRollout = await createClient(oldCapability);
const expiredRolloutObservation = await observe(
expiredRollout,
"subject_rollout_expired",
"An expired old lease must recover without new-pipeline consumption.",
);
await db.batch([{
sql: `UPDATE enrichment_jobs
SET state = 'leased', attempts = 1, lease_token = 'old-worker-token',
lease_expires_at = ?, updated_at = ?
WHERE org_id = ? AND input_ids LIKE ?`,
params: [
new Date(current.getTime() - 1).toISOString(),
current.toISOString(),
expiredRollout.orgId,
`%${expiredRolloutObservation}%`,
],
}]);
assert.equal((await drainEnrichment({
db,
capability: newCapability,
limit: 1,
now: () => current,
orgId: expiredRollout.orgId,
})).leased, 0);
assert.deepEqual(await db.all<{ state: string; attempts: number; lease_token: string | null }>(
`SELECT state, attempts, lease_token FROM enrichment_jobs
WHERE org_id = ? AND input_ids LIKE ?`,
[expiredRollout.orgId, `%${expiredRolloutObservation}%`],
), [{ state: "pending", attempts: 1, lease_token: null }]);
assert.equal((await drainEnrichment({
db,
capability: oldCapability,
limit: 1,
now: () => current,
orgId: expiredRollout.orgId,
})).completed, 1);
const advancingCapability: ExtractionCapability = {
...capability,
modelId: "sequential-fresh-lease",
async generate(request) {
current = new Date(current.getTime() + 25_000);
return proposalFor(request);
},
};
const sequential = await createClient(advancingCapability);
for (const index of [1, 2, 3])
await observe(
sequential,
`subject_sequential_${index}`,
`Sequential provider call ${index} needs a fresh lease.`,
);
const sequentialResult = await drainEnrichment({
db,
capability: advancingCapability,
limit: 3,
now: () => current,
orgId: sequential.orgId,
});
assert.equal(sequentialResult.completed, 3);
assert.equal(sequentialResult.leased, 3);
providerFailure = "typed";
const exhaustedClient = await createClient();
const exhaustedObservation = await observe(
exhaustedClient,
"subject_exhausted",
"A bounded provider failure must stop after four attempts.",
);
for (const advance of [0, 5_000, 10_000, 20_000]) {
current = new Date(current.getTime() + advance);
await exhaustedClient.client.call("POST", "/v1/enrichment/drain?limit=1", {
key: exhaustedClient.key,
});
}
const exhausted = await db.all<{ state: string; attempts: number; error_class: string }>(
`SELECT state, attempts, error_class FROM enrichment_jobs
WHERE org_id = ? AND input_ids LIKE ?`,
[exhaustedClient.orgId, `%${exhaustedObservation}%`],
);
assert.deepEqual(exhausted, [{
state: "failed",
attempts: 4,
error_class: "provider_unavailable",
}]);
providerFailure = null;
const concurrent = await createClient();
const concurrentObservation = await observe(
concurrent,
"subject_concurrent",
"Concurrent duplicate drains must produce one semantic result.",
);
const concurrentDuplicateObservation = await observe(
concurrent,
"subject_concurrent",
"Concurrent duplicate drains must produce one semantic result.",
);
let started!: () => void;
const modelStarted = new Promise<void>((resolve) => { started = resolve; });
hold = new Promise<void>((resolve) => { releaseHold = resolve; });
held = started;
holdNext = true;
const first = drainEnrichment({
db,
capability,
limit: 1,
now: () => current,
orgId: concurrent.orgId,
});
await modelStarted;
const second = await drainEnrichment({
db,
capability,
limit: 1,
now: () => current,
orgId: concurrent.orgId,
});
releaseHold!();
const firstResult = await first;
hold = undefined;
held = undefined;
releaseHold = undefined;
assert.equal(firstResult.leased, 1);
assert.equal(second.leased, 0, "a duplicate job must wait behind the leased canonical job");
assert.equal(concurrentDuplicateCalls, 1);
const reuseResult = await drainEnrichment({
db,
capability,
limit: 2,
now: () => current,
orgId: concurrent.orgId,
});
assert.ok(reuseResult.leased >= 1);
assert.ok(reuseResult.completed >= 1);
assert.equal(
concurrentDuplicateCalls,
1,
"the waiting duplicate must reuse without a second derivation model call",
);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claims
WHERE org_id = ? AND subject_id = 'subject_concurrent'`,
[concurrent.orgId],
))[0]!.count), 1);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM enrichment_commits c
JOIN enrichment_jobs j ON j.id = c.job_id
WHERE j.org_id = ? AND (j.input_ids LIKE ? OR j.input_ids LIKE ?)`,
[
concurrent.orgId,
`%${concurrentObservation}%`,
`%${concurrentDuplicateObservation}%`,
],
))[0]!.count), 2);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claim_sources s
JOIN claims c ON c.id = s.claim_id
WHERE c.org_id = ? AND c.subject_id = 'subject_concurrent'`,
[concurrent.orgId],
))[0]!.count), 2);
const expired = await createClient();
const expiredObservation = await observe(
expired,
"subject_expired_lease",
"STALE_INVALID_ONCE an expired semantic lease must fence stale failure output.",
);
let expiredStarted!: () => void;
const expiredModelStarted = new Promise<void>((resolve) => { expiredStarted = resolve; });
hold = new Promise<void>((resolve) => { releaseHold = resolve; });
held = expiredStarted;
holdNext = true;
const staleDrain = drainEnrichment({
db,
capability,
limit: 1,
now: () => current,
orgId: expired.orgId,
});
await expiredModelStarted;
current = new Date(current.getTime() + ENRICHMENT_LEASE_MS + 1);
const recoveredDrain = await drainEnrichment({
db,
capability,
limit: 1,
now: () => current,
orgId: expired.orgId,
});
releaseHold!();
const staleResult = await staleDrain;
hold = undefined;
held = undefined;
releaseHold = undefined;
assert.equal(recoveredDrain.completed, 1);
assert.deepEqual(staleResult.errors, ["enrichment:lease_lost"]);
assert.deepEqual(await db.all<{ state: string; attempts: number }>(
`SELECT state, attempts FROM enrichment_jobs
WHERE org_id = ? AND input_ids LIKE ?`,
[expired.orgId, `%${expiredObservation}%`],
), [{ state: "done", attempts: 2 }]);
assert.equal(Number((await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM claims
WHERE org_id = ? AND subject_id = 'subject_expired_lease'`,
[expired.orgId],
))[0]!.count), 1);
const readinessNoise = await createClient();
assert.notEqual(readinessNoise.orgId, primary.orgId);
await observe(
readinessNoise,
"subject_readiness_privacy",
"One pending job must not expose a deployment-wide count.",
);
const readinessWithOne = await primary.client.call("GET", "/readyz");
assert.equal(readinessWithOne.status, 200);
const publicEnrichmentState = readinessWithOne.body.data.checks.enrichment_jobs;
assert.deepEqual(Object.keys(publicEnrichmentState), ["state"]);
assert.equal(typeof publicEnrichmentState.state, "string");
await observe(
readinessNoise,
"subject_readiness_privacy",
"A second pending job must not change public queue cardinality metadata.",
);
const readinessWithTwo = await primary.client.call("GET", "/readyz");
assert.equal(readinessWithTwo.status, 200);
assert.deepEqual(readinessWithTwo.body.data.checks.enrichment_jobs, publicEnrichmentState);
const immutable = (await db.all<{ id: string }>(
`SELECT id FROM enrichment_jobs WHERE org_id = ? LIMIT 1`,
[primary.orgId],
))[0]!.id;
const foreignClaim = (await db.all<{ id: string }>(
`SELECT id FROM claims WHERE org_id = ? LIMIT 1`,
[concurrent.orgId],
))[0]!.id;
await assert.rejects(
() => db.batch([{
sql: `UPDATE claims SET enrichment_job_id = ? WHERE id = ?`,
params: [immutable, foreignClaim],
}]),
/ENRICHMENT_JOB_SCOPE/u,
);
await assert.rejects(
() => db.batch([{
sql: `UPDATE enrichment_jobs SET model_fingerprint = ? WHERE id = ?`,
params: ["b".repeat(64), immutable],
}]),
/ENRICHMENT_INPUT_IMMUTABLE/u,
);
await assert.rejects(
() => db.batch([{
sql: `UPDATE enrichment_jobs SET output_hash = ? WHERE id = ?`,
params: ["c".repeat(64), immutable],
}]),
/ENRICHMENT_OUTPUT_IMMUTABLE/u,
);
// A retried job records the output of its new run (AC-RTY-001 to AC-RTY-004).
const retryTarget = await createClient();
const jobFor = async (observation: string) => (await db.all<{
id: string;
state: string;
attempts: number;
error_class: string | null;
output_hash: string | null;
result_ids: string | null;
}>(
`SELECT id, state, attempts, error_class, output_hash, result_ids FROM enrichment_jobs
WHERE org_id = ? AND lane = 'derivation' AND input_ids LIKE ?`,
[retryTarget.orgId, `%${observation}%`],
))[0]!;
const retryJob = async (id: string) => retryEnrichmentJobs(
db,
"failed",
current.toISOString(),
retryTarget.orgId,
id,
);
// AC-RDY-003 and AC-RDY-004: a first rejected output returns the job to pending.
const validObservation = await observe(
retryTarget,
"subject_retry_valid",
"RETRY_VALID_OUTPUT the first run is unsafe.",
);
await drainHttp(retryTarget);
const secondRunPending = await jobFor(validObservation);
assert.equal(secondRunPending.state, "pending", "AC-RDY-003: the first rejection retries once");
assert.equal(secondRunPending.error_class, "unsafe_output");
assert.equal(secondRunPending.attempts, 1);
assert.equal(secondRunPending.output_hash, null, "AC-RDY-003: the pending job records no hash");
current = new Date(current.getTime() + 10_000);
const secondRunDrain = await drainHttp(retryTarget);
assert.deepEqual(secondRunDrain.body.data.errors, []);
const completedSecondRun = await jobFor(validObservation);
assert.equal(completedSecondRun.state, "done", "AC-RDY-004: the second run completes");
assert.equal(completedSecondRun.error_class, null);
assert.match(completedSecondRun.output_hash ?? "", /^[a-f0-9]{64}$/u, "AC-RDY-004: the hash is recorded");
assert.notEqual(completedSecondRun.result_ids, null);
// AC-RTY-001 and AC-RTY-002: an operator retry of a failed job clears its hash and the job completes.
const lateObservation = await observe(
retryTarget,
"subject_retry_late",
"RETRY_LATE_OUTPUT the first two runs are unsafe.",
);
await drainRejected(retryTarget);
const failedFirst = await jobFor(lateObservation);
assert.equal(failedFirst.state, "failed");
assert.equal(failedFirst.error_class, "unsafe_output");
assert.match(failedFirst.output_hash ?? "", /^[a-f0-9]{64}$/u);
const retryQueued = await retryJob(failedFirst.id);
assert.deepEqual(retryQueued.job_ids, [failedFirst.id]);
const requeued = await jobFor(lateObservation);
assert.equal(requeued.state, "pending");
assert.equal(requeued.attempts, 0);
assert.equal(requeued.output_hash, null, "AC-RTY-001: a retry clears the output hash");
current = new Date(current.getTime() + 1_000);
const retriedDrain = await drainHttp(retryTarget);
assert.deepEqual(retriedDrain.body.data.errors, []);
const completedRetry = await jobFor(lateObservation);
assert.equal(completedRetry.state, "done", "AC-RTY-002: the retried job completes");
assert.equal(completedRetry.error_class, null);
assert.match(completedRetry.output_hash ?? "", /^[a-f0-9]{64}$/u);
assert.notEqual(completedRetry.output_hash, failedFirst.output_hash);
assert.notEqual(completedRetry.result_ids, null);
// AC-RTY-003 and AC-RDY-005: a second rejected output fails the job and the pass continues.
const rejectObservation = await observe(
retryTarget,
"subject_retry_reject",
"RETRY_REJECT_OUTPUT every run is unsafe.",
);
await drainRejected(retryTarget);
const rejectedFirst = await jobFor(rejectObservation);
assert.equal(rejectedFirst.state, "failed", "AC-RDY-005: the second rejection is terminal");
assert.match(rejectedFirst.output_hash ?? "", /^[a-f0-9]{64}$/u, "AC-RDY-005: the second hash is recorded");
await retryJob(rejectedFirst.id);
current = new Date(current.getTime() + 1_000);
const followObservation = await observe(
retryTarget,
"subject_retry_follow",
"A job queued after the retried job still completes in the same pass.",
);
current = new Date(current.getTime() + 1_000);
const rejectedDrain = await drainHttp(retryTarget);
assert.deepEqual(rejectedDrain.body.data.errors, []);
assert.equal((await jobFor(followObservation)).state, "done");
assert.equal((await jobFor(rejectObservation)).state, "pending");
assert.equal((await jobFor(rejectObservation)).output_hash, null);
current = new Date(current.getTime() + 10_000);
const rejectedAgain = await drainHttp(retryTarget);
assert.deepEqual(rejectedAgain.body.data.errors, []);
const rejectedSecond = await jobFor(rejectObservation);
assert.equal(rejectedSecond.state, "failed");
assert.equal(rejectedSecond.error_class, "unsafe_output");
assert.match(rejectedSecond.output_hash ?? "", /^[a-f0-9]{64}$/u);
assert.notEqual(rejectedSecond.output_hash, rejectedFirst.output_hash);
// AC-RTY-004: only the retry reset may change a recorded output hash.
const setHash = (id: string, hash: string | null) => db.batch([{
sql: `UPDATE enrichment_jobs SET output_hash = ? WHERE id = ?`,
params: [hash, id],
}]);
await assert.rejects(() => setHash(completedRetry.id, "d".repeat(64)), /ENRICHMENT_OUTPUT_IMMUTABLE/u);
await assert.rejects(() => setHash(completedRetry.id, null), /ENRICHMENT_OUTPUT_IMMUTABLE/u);
await assert.rejects(() => setHash(rejectedSecond.id, "d".repeat(64)), /ENRICHMENT_OUTPUT_IMMUTABLE/u);
await assert.rejects(() => setHash(rejectedSecond.id, null), /ENRICHMENT_OUTPUT_IMMUTABLE/u);
await assert.rejects(
() => db.batch([{
sql: `UPDATE enrichment_jobs SET state = 'pending', output_hash = ? WHERE id = ?`,
params: ["d".repeat(64), rejectedSecond.id],
}]),
/ENRICHMENT_OUTPUT_IMMUTABLE/u,
);
await assert.rejects(
() => db.batch([{
sql: `UPDATE enrichment_jobs SET state = 'pending', attempts = 1, output_hash = NULL
WHERE id = ?`,
params: [rejectedSecond.id],
}]),
/ENRICHMENT_OUTPUT_IMMUTABLE/u,
);
const afterRejects = await jobFor(rejectObservation);
assert.equal(afterRejects.state, "failed");
assert.equal(afterRejects.output_hash, rejectedSecond.output_hash);
// AC-RDY-001, AC-RDY-002, AC-RDY-006: readiness marks only failures that an operator can fix.
const readiness = await createClient();
const readyState = async () => (await readiness.client.call("GET", "/readyz"))
.body.data.checks.enrichment_jobs.state as string;
const readyJob = async (observation: string) => (await db.all<{
id: string;
state: string;
error_class: string | null;
}>(
`SELECT id, state, error_class FROM enrichment_jobs
WHERE org_id = ? AND lane = 'derivation' AND input_ids LIKE ?`,
[readiness.orgId, `%${observation}%`],
))[0]!;
// Readiness covers every organization: acknowledge the failures that earlier cases left.
await db.batch([{
sql: `UPDATE enrichment_jobs SET acked_at = ? WHERE state = 'failed' AND acked_at IS NULL`,
params: [current.toISOString()],
}]);
assert.notEqual(await readyState(), "terminal_error");
const markFailed = (id: string, errorClass: string) => db.batch([{
sql: `UPDATE enrichment_jobs SET state = 'failed', error_class = ? WHERE id = ?`,
params: [errorClass, id],
}]);
const readyRejected = await observe(readiness, "subject_ready", "RETRY_REJECT_OUTPUT readiness content failure.");
await drainHttp(readiness);
assert.equal((await readyJob(readyRejected)).state, "pending");
assert.notEqual(await readyState(), "terminal_error");
current = new Date(current.getTime() + 10_000);
await drainHttp(readiness);
const readyUnsafe = await readyJob(readyRejected);
assert.deepEqual(
{ state: readyUnsafe.state, error_class: readyUnsafe.error_class },
{ state: "failed", error_class: "unsafe_output" },
);
assert.notEqual(await readyState(), "terminal_error", "AC-RDY-002: unsafe_output does not mark readiness");
const readyInvalid = await observe(readiness, "subject_ready", "MISSING_FIELD readiness content failure.");
await drainHttp(readiness);
const readyInvalidJob = await readyJob(readyInvalid);
assert.deepEqual(
{ state: readyInvalidJob.state, error_class: readyInvalidJob.error_class },
{ state: "failed", error_class: "invalid_output" },
);
assert.notEqual(await readyState(), "terminal_error", "AC-RDY-002: invalid_output does not mark readiness");
const readySource = await observe(readiness, "subject_ready", "A job whose source changed.");
const readySourceJob = await readyJob(readySource);
await markFailed(readySourceJob.id, "source_changed");
assert.notEqual(await readyState(), "terminal_error", "AC-RDY-002: source_changed does not mark readiness");
const readyProvider = await observe(readiness, "subject_ready", "A job whose provider failed.");
const readyProviderJob = await readyJob(readyProvider);
for (const errorClass of ["provider_unavailable", "provider_rejected", "provider_protocol"]) {
await markFailed(readyProviderJob.id, errorClass);
assert.equal(await readyState(), "terminal_error", `AC-RDY-001: ${errorClass} marks readiness`);
}
// AC-RDY-006: the job commands select a failed job of every class.
const listed = await listEnrichmentJobs(db, "failed", readiness.orgId);
assert.deepEqual(
listed.map((row) => row.error_class).sort(),
["invalid_output", "provider_protocol", "source_changed", "unsafe_output"],
);
const retried = await retryEnrichmentJobs(
db,
"failed",
current.toISOString(),
readiness.orgId,
readySourceJob.id,
);
assert.deepEqual(retried.job_ids, [readySourceJob.id]);
const acked = await ackEnrichmentJobs(db, "failed", current.toISOString(), readiness.orgId);
assert.equal(acked.matched, 3, "AC-RDY-006: ack selects every failed class");
assert.notEqual(await readyState(), "terminal_error", "an acknowledged provider failure does not mark readiness");
const purgedJobs = await purgeEnrichmentJobs(db, "failed", readiness.orgId);
assert.equal(purgedJobs.matched, 3, "AC-RDY-006: purge selects every failed class");
}