tests/integration/maintenance.test.ts
Rama main · réplica de lectura del repositorio original.
import { afterAll, beforeAll, test } from "bun:test";
import assert from "node:assert/strict";
import { mkdtempSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
import type { Stmt } from "../../src/core/db";
import { runMaintenance } from "../../src/core/maintenance";
import { migrate } from "../../src/core/migrations";
import { createSqliteDb, openDatabase } from "../../src/runtime/bun/sqlite";
import { serve } from "../../src/runtime/bun/server";
import { fakeVectors, provisionWith } from "../contract/harness";
/**
* Indexing has to happen on its own.
*
* Before the maintenance timer existed, a deployment with a vector capability
* searched an index nothing ever populated: every write queued an outbox row and
* no code consumed it, so semantic retrieval silently returned nothing forever
* unless an operator remembered to call the drain endpoint. This asserts the
* service closes that loop itself, and that a caller never has to know.
*/
const directory = mkdtempSync(join(tmpdir(), "titen-maint-"));
const vectors = fakeVectors();
let running: Awaited<ReturnType<typeof serve>>;
let key: string;
async function api(method: string, path: string, body?: unknown) {
const res = await fetch(`${running.url}${path}`, {
method,
headers: { "content-type": "application/json", authorization: `Bearer ${key}` },
body: body === undefined ? undefined : JSON.stringify(body),
});
const json = (await res.json()) as any;
if (!res.ok) throw new Error(`${method} ${path} -> ${res.status} ${JSON.stringify(json)}`);
return json.data;
}
const pendingCount = async (db: ReturnType<typeof createSqliteDb>) => {
const rows = await db.all<{ count: number }>(
`SELECT COUNT(*) AS count FROM index_outbox WHERE state = 'pending'`,
);
return Number(rows[0]!.count);
};
let db: ReturnType<typeof createSqliteDb>;
beforeAll(async () => {
const dbPath = join(directory, "titen.db");
// A short interval so the test observes real elapsed behavior, not a mock clock.
running = await serve({
dbPath,
port: 0,
hostname: "127.0.0.1",
quiet: true,
revision: "maint",
maintenanceIntervalMs: 250,
vectors,
});
db = createSqliteDb(openDatabase(dbPath));
const provisioned = await provisionWith(db, { scopes: ["*"] });
key = provisioned.key;
});
afterAll(async () => {
await running.stop();
rmSync(directory, { recursive: true, force: true });
});
test("a written claim is indexed without anyone calling the drain endpoint", async () => {
const observation = await api("POST", "/v1/observations", {
subject_id: "user_maint",
kind: "tool_result",
content: "Evidence whose claim must be indexed automatically.",
source: { type: "tool", ref: "maint#1" },
trust: "verified",
});
const consolidated = await api("POST", "/v1/consolidations", {
subject_id: "user_maint",
claims: [
{
kind: "procedural",
statement: "Automatic indexing must require no operator action.",
sources: [{ observation_id: observation.observation_id, relation: "supports" }],
},
],
});
const claimId = consolidated.claims[0].claim_id as string;
// The write queued outbox rows. Nothing below calls /v1/index/drain.
assert.ok((await pendingCount(db)) > 0, "the write must queue indexing work");
const deadline = Date.now() + 5_000;
while (Date.now() < deadline && (await pendingCount(db)) > 0)
await new Promise((resolve) => setTimeout(resolve, 100));
assert.equal(
await pendingCount(db),
0,
"the maintenance pass must drain the queue on its own",
);
// Drained is not the same as indexed: the claim must be searchable by vector.
const hits = await vectors.store.query(new Float32Array([1, 0, 0, 0]), { topK: 10 });
assert.ok(
hits.some((hit) => hit.id === claimId),
"the claim must be present in the vector index",
);
});
test("readiness reports the maintenance timer that is actually running", async () => {
const response = await fetch(`${running.url}/readyz`);
const body = (await response.json()) as any;
assert.equal(response.status, 200);
assert.equal(body.data.capabilities.background_repair, "enabled");
});
test("a claim that stopped being retrievable is retired, not embedded forever", async () => {
const observation = await api("POST", "/v1/observations", {
subject_id: "user_maint_retire",
kind: "tool_result",
content: "Evidence for a claim that gets revoked before indexing settles.",
source: { type: "tool", ref: "maint#2" },
trust: "verified",
});
const consolidated = await api("POST", "/v1/consolidations", {
subject_id: "user_maint_retire",
claims: [
{
kind: "procedural",
statement: "This claim will be revoked.",
sources: [{ observation_id: observation.observation_id, relation: "supports" }],
},
],
});
await api("POST", `/v1/claims/${consolidated.claims[0].claim_id}/revoke`, {
reason: "no longer applicable",
expected_version: 1,
});
const deadline = Date.now() + 5_000;
while (Date.now() < deadline && (await pendingCount(db)) > 0)
await new Promise((resolve) => setTimeout(resolve, 100));
assert.equal(await pendingCount(db), 0, "the queue must not stall on an ineligible claim");
});
test("background repair removes vectors queued by evidence purge", async () => {
const observation = await api("POST", "/v1/observations", {
subject_id: "user_maint_purge",
kind: "tool_result",
content: "Evidence whose indexed claim will be purged.",
source: { type: "tool", ref: "maint#purge" },
trust: "verified",
});
const consolidated = await api("POST", "/v1/consolidations", {
subject_id: "user_maint_purge",
claims: [{
kind: "procedural",
statement: "Background repair must remove this vector after purge.",
sources: [{ observation_id: observation.observation_id, relation: "supports" }],
}],
});
const claimId = consolidated.claims[0].claim_id as string;
let deadline = Date.now() + 5_000;
while (Date.now() < deadline && !vectors.metadataFor(claimId))
await new Promise((resolve) => setTimeout(resolve, 100));
assert.ok(vectors.metadataFor(claimId), "fixture must observe the indexed vector before purge");
await api("DELETE", `/v1/observations/${observation.observation_id}`);
deadline = Date.now() + 5_000;
while (Date.now() < deadline && vectors.metadataFor(claimId))
await new Promise((resolve) => setTimeout(resolve, 100));
assert.equal(vectors.metadataFor(claimId), undefined);
assert.equal(await pendingCount(db), 0);
});
test("the queue drains again after new work arrives", async () => {
const before = vectors.embedCalls();
const observation = await api("POST", "/v1/observations", {
subject_id: "user_maint_again",
kind: "tool_result",
content: "A second batch of evidence arriving later.",
source: { type: "tool", ref: "maint#3" },
trust: "verified",
});
await api("POST", "/v1/consolidations", {
subject_id: "user_maint_again",
claims: [
{
kind: "procedural",
statement: "Later work is picked up by a later pass.",
sources: [{ observation_id: observation.observation_id, relation: "supports" }],
},
],
});
const deadline = Date.now() + 5_000;
while (Date.now() < deadline && (await pendingCount(db)) > 0)
await new Promise((resolve) => setTimeout(resolve, 100));
assert.equal(await pendingCount(db), 0, "a later pass must drain later work");
assert.ok(
vectors.embedCalls() > before,
"the embedder must have been called again for the new batch",
);
});
test("maintenance deletes only one bounded page of expired execution state", async () => {
const handle = openDatabase(join(directory, "ephemeral.db"));
const isolated = createSqliteDb(handle);
await migrate(isolated);
const actor = await provisionWith(isolated, { scopes: ["*"] });
const now = new Date("2026-07-31T12:00:00.000Z");
const expired = [
"2026-07-30T00:00:00.000Z",
"2026-07-30T00:01:00.000Z",
"2026-07-30T00:02:00.000Z",
];
const future = "2026-08-01T00:00:00.000Z";
const statements: Stmt[] = [];
for (let index = 0; index < expired.length; index += 1) {
statements.push(
{
sql: `INSERT INTO idempotency_v3
(org_id, principal_id, key_id, request_identity, key_hash, request_hash, status, response, created_at, expires_at)
VALUES (?, ?, ?, ?, ?, ?, 201, '{}', ?, ?)`,
params: [
actor.orgId, actor.principalId, actor.keyId, `POST /expired/${index}`, `expired-key-${index}`,
`request-${index}`, expired[index]!, expired[index]!,
],
},
{
sql: `INSERT INTO checkpoints
(id, org_id, subject_id, agent_id, kind, state, state_hash, ttl_seconds,
expires_at, created_at, updated_at)
VALUES (?, ?, ?, ?, 'cursor', '{}', 'hash', 60, ?, ?, ?)`,
params: [
`ckpt_expired_${index}`, actor.orgId, `cleanup-subject-${index}`, actor.principalId,
expired[index]!, expired[index]!, expired[index]!,
],
},
);
}
statements.push(
{
sql: `INSERT INTO handoffs
(id, org_id, from_principal, to_principal, subject_id, checkpoint_id, status, created_at)
VALUES ('handoff_expired_checkpoint', ?, ?, 'recipient', 'cleanup-subject-0',
'ckpt_expired_0', 'accepted', ?)`,
params: [actor.orgId, actor.principalId, expired[0]!],
},
{
sql: `INSERT INTO idempotency_v3
(org_id, principal_id, key_id, request_identity, key_hash, request_hash, status, response, created_at, expires_at)
VALUES (?, ?, ?, 'POST /active', 'active-key', 'active-request', 201, '{}', ?, ?)`,
params: [actor.orgId, actor.principalId, actor.keyId, now.toISOString(), future],
},
{
sql: `INSERT INTO checkpoints
(id, org_id, subject_id, agent_id, kind, state, state_hash, ttl_seconds,
expires_at, created_at, updated_at)
VALUES ('ckpt_active', ?, 'cleanup-subject-active', ?, 'cursor', '{}', 'hash', 60, ?, ?, ?)`,
params: [actor.orgId, actor.principalId, future, now.toISOString(), now.toISOString()],
},
{
sql: `INSERT INTO leases
(id, org_id, resource_type, resource_id, holder_id, purpose, ttl_seconds,
expires_at, created_at, released_at)
VALUES ('lease_released_first', ?, 'task', 'released-first', ?, 'test', 60, ?, ?, ?)`,
params: [actor.orgId, actor.principalId, future, expired[0]!, expired[0]!],
},
{
sql: `INSERT INTO leases
(id, org_id, resource_type, resource_id, holder_id, purpose, ttl_seconds,
expires_at, created_at, released_at)
VALUES ('lease_expired', ?, 'task', 'expired', ?, 'test', 60, ?, ?, NULL)`,
params: [actor.orgId, actor.principalId, expired[1]!, expired[1]!],
},
{
sql: `INSERT INTO leases
(id, org_id, resource_type, resource_id, holder_id, purpose, ttl_seconds,
expires_at, created_at, released_at)
VALUES ('lease_released_later', ?, 'task', 'released-later', ?, 'test', 60, ?, ?, ?)`,
params: [actor.orgId, actor.principalId, future, expired[2]!, expired[2]!],
},
{
sql: `INSERT INTO leases
(id, org_id, resource_type, resource_id, holder_id, purpose, ttl_seconds,
expires_at, created_at, released_at)
VALUES ('lease_active', ?, 'task', 'active', ?, 'test', 60, ?, ?, NULL)`,
params: [actor.orgId, actor.principalId, future, now.toISOString()],
},
{
sql: `INSERT INTO events
(id, org_id, kind, actor_id, resource_type, resource_id, payload, created_at)
VALUES ('evt_cleanup_sentinel', ?, 'cleanup.sentinel', ?, 'test', 'sentinel', '{}', ?)`,
params: [actor.orgId, actor.principalId, expired[0]!],
},
{
sql: `INSERT INTO record_history
(id, org_id, record_type, record_id, version, change_kind, actor_id, snapshot_hash, changed_at)
VALUES ('hist_cleanup_sentinel', ?, 'test', 'sentinel', 1, 'append', ?, 'hash', ?)`,
params: [actor.orgId, actor.principalId, expired[0]!],
},
);
await isolated.batch(statements);
const result = await runMaintenance({
db: isolated,
now,
limit: 2,
deliverWebhooks: false,
});
assert.deepEqual(result.errors, []);
const counts = (await isolated.all<Record<string, number>>(
`SELECT
(SELECT COUNT(*) FROM idempotency_v3 WHERE expires_at <= ?) AS expired_idempotency,
(SELECT COUNT(*) FROM idempotency_v3 WHERE expires_at > ?) AS active_idempotency,
(SELECT COUNT(*) FROM checkpoints WHERE expires_at <= ?) AS expired_checkpoints,
(SELECT COUNT(*) FROM checkpoints WHERE expires_at > ?) AS active_checkpoints,
(SELECT COUNT(*) FROM leases WHERE released_at IS NOT NULL OR expires_at <= ?) AS inactive_leases,
(SELECT COUNT(*) FROM leases WHERE released_at IS NULL AND expires_at > ?) AS active_leases,
(SELECT COUNT(*) FROM events WHERE id = 'evt_cleanup_sentinel') AS events,
(SELECT COUNT(*) FROM record_history WHERE id = 'hist_cleanup_sentinel') AS history`,
[now.toISOString(), now.toISOString(), now.toISOString(), now.toISOString(), now.toISOString(), now.toISOString()],
))[0]!;
assert.deepEqual(
Object.fromEntries(Object.entries(counts).map(([key, value]) => [key, Number(value)])),
{
expired_idempotency: 1,
active_idempotency: 1,
expired_checkpoints: 1,
active_checkpoints: 1,
inactive_leases: 1,
active_leases: 1,
events: 1,
history: 1,
},
);
assert.deepEqual(await isolated.all(
`SELECT checkpoint_id FROM handoffs WHERE id = 'handoff_expired_checkpoint'`,
), [{ checkpoint_id: null }]);
handle.close();
});