baa-conductor


baa-conductor / packages / db / src
codex@macbookpro  ·  2026-03-31

index.test.js

  1import assert from "node:assert/strict";
  2import { readFileSync } from "node:fs";
  3import test from "node:test";
  4
  5import {
  6  BROWSER_LOGIN_STATE_STATUS_VALUES,
  7  D1ControlPlaneRepository,
  8  GLOBAL_LEASE_NAME,
  9  SqliteD1Database,
 10  buildControllerHeartbeatRecord,
 11  buildLeaderLeaseAcquireResult,
 12  buildLeaderLeaseRecord,
 13  getLeaderLeaseOperation,
 14  isLeaderLeaseExpired
 15} from "../dist/index.js";
 16
 17const CONTROL_PLANE_SCHEMA_SQL = readFileSync(new URL("../../../ops/sql/schema.sql", import.meta.url), "utf8");
 18
 19test("buildControllerHeartbeatRecord preserves heartbeat and startup timestamps", () => {
 20  const record = buildControllerHeartbeatRecord({
 21    controllerId: "mini-main",
 22    host: "mini",
 23    role: "primary",
 24    priority: 100,
 25    status: "alive",
 26    version: "0.1.0",
 27    heartbeatAt: 120,
 28    startedAt: 90,
 29    metadataJson: "{\"slot\":\"main\"}"
 30  });
 31
 32  assert.deepEqual(record, {
 33    controllerId: "mini-main",
 34    host: "mini",
 35    role: "primary",
 36    priority: 100,
 37    status: "alive",
 38    version: "0.1.0",
 39    lastHeartbeatAt: 120,
 40    lastStartedAt: 90,
 41    metadataJson: "{\"slot\":\"main\"}"
 42  });
 43});
 44
 45test("getLeaderLeaseOperation returns renew only for the active holder", () => {
 46  const currentLease = {
 47    leaseName: GLOBAL_LEASE_NAME,
 48    holderId: "mini-main",
 49    holderHost: "mini",
 50    term: 7,
 51    leaseExpiresAt: 150,
 52    renewedAt: 120,
 53    preferredHolderId: "mini-main",
 54    metadataJson: null
 55  };
 56
 57  assert.equal(getLeaderLeaseOperation(currentLease, "mini-main", 140), "renew");
 58  assert.equal(getLeaderLeaseOperation(currentLease, "mac-standby", 140), "acquire");
 59  assert.equal(getLeaderLeaseOperation(currentLease, "mini-main", 151), "acquire");
 60  assert.equal(isLeaderLeaseExpired(currentLease, 151), true);
 61});
 62
 63test("buildLeaderLeaseRecord increments term for takeover and marks standby responses", () => {
 64  const previousLease = {
 65    leaseName: GLOBAL_LEASE_NAME,
 66    holderId: "mini-main",
 67    holderHost: "mini",
 68    term: 3,
 69    leaseExpiresAt: 110,
 70    renewedAt: 80,
 71    preferredHolderId: "mini-main",
 72    metadataJson: "{\"role\":\"primary\"}"
 73  };
 74
 75  const desiredLease = buildLeaderLeaseRecord(previousLease, {
 76    controllerId: "mac-standby",
 77    host: "mac",
 78    ttlSec: 30,
 79    preferred: false,
 80    now: 111
 81  });
 82
 83  assert.deepEqual(desiredLease, {
 84    leaseName: GLOBAL_LEASE_NAME,
 85    holderId: "mac-standby",
 86    holderHost: "mac",
 87    term: 4,
 88    leaseExpiresAt: 141,
 89    renewedAt: 111,
 90    preferredHolderId: "mini-main",
 91    metadataJson: "{\"role\":\"primary\"}"
 92  });
 93
 94  const response = buildLeaderLeaseAcquireResult(previousLease, previousLease, {
 95    controllerId: "mac-standby",
 96    host: "mac",
 97    ttlSec: 30,
 98    now: 100
 99  });
100
101  assert.equal(response.isLeader, false);
102  assert.equal(response.operation, "acquire");
103  assert.equal(response.holderId, "mini-main");
104});
105
106test("control-plane automation system_state timestamps are initialized and backfilled in milliseconds", async () => {
107  const db = new SqliteD1Database(":memory:", {
108    schemaSql: CONTROL_PLANE_SCHEMA_SQL
109  });
110  const repository = new D1ControlPlaneRepository(db);
111
112  try {
113    const initialState = await repository.getAutomationState();
114    assert.ok(initialState);
115    assert.equal(initialState.mode, "running");
116    assert.ok(initialState.updatedAt >= 1_000_000_000_000);
117
118    await db.prepare("DELETE FROM system_state WHERE state_key = 'automation'").run();
119    await repository.ensureAutomationState("paused");
120
121    const recreatedState = await repository.getAutomationState();
122    assert.ok(recreatedState);
123    assert.equal(recreatedState.mode, "paused");
124    assert.ok(recreatedState.updatedAt >= 1_000_000_000_000);
125  } finally {
126    db.close();
127  }
128});
129
130test("SqliteD1Database supports task log queries through D1ControlPlaneRepository", async () => {
131  const db = new SqliteD1Database(":memory:", {
132    schemaSql: CONTROL_PLANE_SCHEMA_SQL
133  });
134  const repository = new D1ControlPlaneRepository(db);
135
136  try {
137    await repository.ensureAutomationState();
138    await repository.heartbeatController({
139      controllerId: "mini-main",
140      heartbeatAt: 100,
141      host: "mini",
142      priority: 100,
143      role: "primary",
144      startedAt: 100,
145      status: "alive",
146      version: "1.0.0"
147    });
148    await repository.upsertWorker({
149      workerId: "worker-shell-1",
150      controllerId: "mini-main",
151      host: "mini",
152      workerType: "shell",
153      status: "idle",
154      maxParallelism: 1,
155      currentLoad: 0,
156      lastHeartbeatAt: 100,
157      capabilitiesJson: null,
158      metadataJson: null
159    });
160    await repository.insertTask({
161      acceptanceJson: null,
162      assignedControllerId: "mini-main",
163      baseRef: "main",
164      branchName: null,
165      constraintsJson: null,
166      createdAt: 100,
167      currentStepIndex: 0,
168      errorText: null,
169      finishedAt: null,
170      goal: "verify sqlite adapter",
171      metadataJson: null,
172      plannerProvider: null,
173      planningStrategy: null,
174      priority: 50,
175      repo: "/tmp/repo",
176      resultJson: null,
177      resultSummary: null,
178      source: "test",
179      startedAt: 100,
180      status: "running",
181      targetHost: "mini",
182      taskId: "task_demo",
183      taskType: "shell",
184      title: "demo",
185      updatedAt: 100
186    });
187    await repository.insertTaskStep({
188      stepId: "step_demo",
189      taskId: "task_demo",
190      stepIndex: 0,
191      stepName: "run demo",
192      stepKind: "shell",
193      status: "running",
194      assignedWorkerId: "worker-shell-1",
195      assignedControllerId: "mini-main",
196      timeoutSec: 60,
197      retryLimit: 0,
198      retryCount: 0,
199      leaseExpiresAt: 130,
200      inputJson: null,
201      outputJson: null,
202      summary: null,
203      errorText: null,
204      createdAt: 100,
205      updatedAt: 100,
206      startedAt: 100,
207      finishedAt: null
208    });
209    await repository.insertTaskRun({
210      runId: "run_demo",
211      taskId: "task_demo",
212      stepId: "step_demo",
213      workerId: "worker-shell-1",
214      controllerId: "mini-main",
215      host: "mini",
216      pid: 1234,
217      status: "running",
218      leaseExpiresAt: 130,
219      heartbeatAt: 100,
220      logDir: "/tmp/run-demo",
221      stdoutPath: null,
222      stderrPath: null,
223      workerLogPath: null,
224      checkpointSeq: 0,
225      exitCode: null,
226      resultJson: null,
227      errorText: null,
228      createdAt: 100,
229      startedAt: 100,
230      finishedAt: null
231    });
232    await repository.appendTaskLog({
233      taskId: "task_demo",
234      stepId: "step_demo",
235      runId: "run_demo",
236      seq: 1,
237      stream: "stdout",
238      level: "info",
239      message: "first",
240      createdAt: 100
241    });
242    await repository.appendTaskLog({
243      taskId: "task_demo",
244      stepId: "step_demo",
245      runId: "run_demo",
246      seq: 2,
247      stream: "stdout",
248      level: "info",
249      message: "second",
250      createdAt: 101
251    });
252    await repository.appendTaskLog({
253      taskId: "task_demo",
254      stepId: "step_demo",
255      runId: "run_demo",
256      seq: 3,
257      stream: "stdout",
258      level: "info",
259      message: "third",
260      createdAt: 102
261    });
262
263    const logs = await repository.listTaskLogs("task_demo", {
264      limit: 2
265    });
266
267    assert.equal(logs.length, 2);
268    assert.deepEqual(
269      logs.map((entry) => entry.message),
270      ["second", "third"]
271    );
272  } finally {
273    db.close();
274  }
275});
276
277test("D1ControlPlaneRepository upserts and lists browser login state metadata", async () => {
278  const db = new SqliteD1Database(":memory:", {
279    schemaSql: CONTROL_PLANE_SCHEMA_SQL
280  });
281  const repository = new D1ControlPlaneRepository(db);
282
283  try {
284    await repository.upsertBrowserLoginState({
285      platform: "claude",
286      host: "mini",
287      browser: "firefox",
288      clientId: "firefox-claude",
289      account: "ops@example.com",
290      credentialFingerprint: "fp-1",
291      capturedAt: 100,
292      lastSeenAt: 120,
293      status: "fresh"
294    });
295    await repository.upsertBrowserEndpointMetadata({
296      platform: "claude",
297      clientId: "firefox-claude",
298      account: "ops@example.com",
299      endpoints: ["/api/organizations", "/api/organizations", " /api/bootstrap "],
300      updatedAt: 125,
301      lastVerifiedAt: 126
302    });
303
304    const state = await repository.getBrowserLoginState({
305      platform: "claude",
306      clientId: "firefox-claude",
307      account: "ops@example.com"
308    });
309    const endpoints = await repository.getBrowserEndpointMetadata({
310      platform: "claude",
311      clientId: "firefox-claude",
312      account: "ops@example.com"
313    });
314    const listedStates = await repository.listBrowserLoginStates({
315      platform: "claude",
316      host: "mini",
317      browser: "firefox",
318      account: "ops@example.com",
319      status: "fresh"
320    });
321    const listedEndpoints = await repository.listBrowserEndpointMetadata({
322      platform: "claude",
323      account: "ops@example.com"
324    });
325
326    assert.deepEqual(state, {
327      platform: "claude",
328      host: "mini",
329      browser: "firefox",
330      clientId: "firefox-claude",
331      account: "ops@example.com",
332      credentialFingerprint: "fp-1",
333      capturedAt: 100,
334      lastSeenAt: 120,
335      status: "fresh"
336    });
337    assert.deepEqual(endpoints, {
338      platform: "claude",
339      clientId: "firefox-claude",
340      account: "ops@example.com",
341      endpoints: ["/api/bootstrap", "/api/organizations"],
342      updatedAt: 125,
343      lastVerifiedAt: 126
344    });
345    assert.equal(listedStates.length, 1);
346    assert.deepEqual(listedStates[0], state);
347    assert.equal(listedEndpoints.length, 1);
348    assert.deepEqual(listedEndpoints[0], endpoints);
349  } finally {
350    db.close();
351  }
352});
353
354test("browser login state upsert preserves capturedAt for unchanged fingerprints", async () => {
355  const db = new SqliteD1Database(":memory:", {
356    schemaSql: CONTROL_PLANE_SCHEMA_SQL
357  });
358  const repository = new D1ControlPlaneRepository(db);
359  const key = {
360    platform: "chatgpt",
361    clientId: "firefox-chatgpt",
362    account: "agent@example.com"
363  };
364
365  try {
366    await repository.upsertBrowserLoginState({
367      ...key,
368      host: "mini",
369      browser: "firefox",
370      credentialFingerprint: "fp-1",
371      capturedAt: 100,
372      lastSeenAt: 100,
373      status: "fresh"
374    });
375    await repository.upsertBrowserLoginState({
376      ...key,
377      host: "mini",
378      browser: "firefox",
379      credentialFingerprint: "fp-1",
380      capturedAt: 140,
381      lastSeenAt: 160,
382      status: "fresh"
383    });
384
385    const unchangedFingerprint = await repository.getBrowserLoginState(key);
386
387    assert.deepEqual(unchangedFingerprint, {
388      ...key,
389      host: "mini",
390      browser: "firefox",
391      credentialFingerprint: "fp-1",
392      capturedAt: 100,
393      lastSeenAt: 160,
394      status: "fresh"
395    });
396
397    await repository.upsertBrowserLoginState({
398      ...key,
399      host: "mini",
400      browser: "firefox",
401      credentialFingerprint: "fp-2",
402      capturedAt: 170,
403      lastSeenAt: 180,
404      status: "fresh"
405    });
406
407    const rotatedFingerprint = await repository.getBrowserLoginState(key);
408
409    assert.deepEqual(rotatedFingerprint, {
410      ...key,
411      host: "mini",
412      browser: "firefox",
413      credentialFingerprint: "fp-2",
414      capturedAt: 170,
415      lastSeenAt: 180,
416      status: "fresh"
417    });
418  } finally {
419    db.close();
420  }
421});
422
423test("browser login state status transitions from fresh to stale to lost", async () => {
424  const db = new SqliteD1Database(":memory:", {
425    schemaSql: CONTROL_PLANE_SCHEMA_SQL
426  });
427  const repository = new D1ControlPlaneRepository(db);
428  const key = {
429    platform: "gemini",
430    clientId: "firefox-gemini",
431    account: "ops@example.com"
432  };
433
434  try {
435    await repository.upsertBrowserLoginState({
436      ...key,
437      host: "macbook",
438      browser: "firefox",
439      credentialFingerprint: "fp-gemini",
440      capturedAt: 200,
441      lastSeenAt: 210,
442      status: "fresh"
443    });
444
445    const staleCount = await repository.markBrowserLoginStatesStale(210);
446    const staleRecord = await repository.getBrowserLoginState(key);
447    const lostCount = await repository.markBrowserLoginStatesLost(220);
448    const lostRecord = await repository.getBrowserLoginState(key);
449
450    assert.equal(staleCount, 1);
451    assert.equal(staleRecord?.status, "stale");
452    assert.equal(lostCount, 1);
453    assert.equal(lostRecord?.status, "lost");
454    assert.deepEqual(BROWSER_LOGIN_STATE_STATUS_VALUES, ["fresh", "stale", "lost"]);
455  } finally {
456    db.close();
457  }
458});
459
460test("D1ControlPlaneRepository bounds BAA dedupe tables and journal with stable oldest-first pruning", async () => {
461  const db = new SqliteD1Database(":memory:", {
462    schemaSql: CONTROL_PLANE_SCHEMA_SQL
463  });
464  const repository = new D1ControlPlaneRepository(db, {
465    baaExecutionJournalLimit: 2,
466    baaInstructionDedupeLimit: 2,
467    baaMessageDedupeLimit: 2
468  });
469
470  try {
471    await repository.putBaaMessageDedupe({
472      assistantMessageId: "msg-1",
473      conversationId: null,
474      createdAt: 100,
475      dedupeKey: "message-1",
476      observedAt: null,
477      platform: "chatgpt"
478    });
479    await repository.putBaaMessageDedupe({
480      assistantMessageId: "msg-2",
481      conversationId: null,
482      createdAt: 101,
483      dedupeKey: "message-2",
484      observedAt: null,
485      platform: "chatgpt"
486    });
487    await repository.putBaaMessageDedupe({
488      assistantMessageId: "msg-3",
489      conversationId: null,
490      createdAt: 102,
491      dedupeKey: "message-3",
492      observedAt: null,
493      platform: "chatgpt"
494    });
495
496    assert.equal(await repository.hasBaaMessageDedupe("message-1"), false);
497    assert.equal(await repository.hasBaaMessageDedupe("message-2"), true);
498    assert.equal(await repository.hasBaaMessageDedupe("message-3"), true);
499
500    await repository.putBaaInstructionDedupe({
501      assistantMessageId: "msg-1",
502      conversationId: null,
503      createdAt: 100,
504      dedupeKey: "inst-1",
505      instructionId: "instruction-1",
506      platform: "chatgpt",
507      target: "conductor",
508      tool: "describe"
509    });
510    await repository.putBaaInstructionDedupe({
511      assistantMessageId: "msg-2",
512      conversationId: null,
513      createdAt: 101,
514      dedupeKey: "inst-2",
515      instructionId: "instruction-2",
516      platform: "chatgpt",
517      target: "conductor",
518      tool: "status"
519    });
520    await repository.putBaaInstructionDedupe({
521      assistantMessageId: "msg-3",
522      conversationId: null,
523      createdAt: 102,
524      dedupeKey: "inst-3",
525      instructionId: "instruction-3",
526      platform: "chatgpt",
527      target: "conductor",
528      tool: "exec"
529    });
530
531    assert.equal(await repository.hasBaaInstructionDedupe("inst-1"), false);
532    assert.equal(await repository.hasBaaInstructionDedupe("inst-2"), true);
533    assert.equal(await repository.hasBaaInstructionDedupe("inst-3"), true);
534
535    await repository.appendBaaExecutionJournal({
536      assistantMessageId: "msg-1",
537      conversationId: null,
538      ingestedAt: 200,
539      kind: "ingest",
540      messageDedupeKey: "message-1",
541      observedAt: null,
542      platform: "chatgpt",
543      source: "browser.final_message",
544      status: "executed",
545      summaryJson: JSON.stringify({
546        assistant_message_id: "msg-1",
547        status: "executed"
548      })
549    });
550    await repository.appendBaaExecutionJournal({
551      assistantMessageId: "msg-2",
552      conversationId: null,
553      ingestedAt: 200,
554      kind: "ingest",
555      messageDedupeKey: "message-2",
556      observedAt: null,
557      platform: "chatgpt",
558      source: "browser.final_message",
559      status: "duplicate_message",
560      summaryJson: JSON.stringify({
561        assistant_message_id: "msg-2",
562        status: "duplicate_message"
563      })
564    });
565    await repository.appendBaaExecutionJournal({
566      assistantMessageId: "msg-3",
567      conversationId: null,
568      ingestedAt: 200,
569      kind: "ingest",
570      messageDedupeKey: "message-3",
571      observedAt: null,
572      platform: "chatgpt",
573      source: "browser.final_message",
574      status: "failed",
575      summaryJson: JSON.stringify({
576        assistant_message_id: "msg-3",
577        status: "failed"
578      })
579    });
580
581    const journal = await repository.listBaaExecutionJournal({
582      kind: "ingest",
583      limit: 10
584    });
585
586    assert.equal(journal.length, 2);
587    assert.deepEqual(
588      journal.map((entry) => entry.assistantMessageId),
589      ["msg-3", "msg-2"]
590    );
591    assert.deepEqual(
592      journal.map((entry) => entry.status),
593      ["failed", "duplicate_message"]
594    );
595  } finally {
596    db.close();
597  }
598});