Evandro Fonseca Junior
d1d04001d0
Fork of upstream llblab/[email protected].0 (SHA afe09c5) repackaged as a drop-in extension for can1357/oh-my-pi (omp >= 16.5.2). Changes vs upstream: - Rename package: @llblab/pi-telegram -> @evandrodevbr/oh-my-pi-telegram - Bump version: 0.22.0 -> 0.22.0-evandro.1 - Update repository/homepage/bugs URLs to evandrodevbr - Add omp field in package.json (displayName + compatibleWith) - README: prepend 'Oh My Pi (omp) Compatibility' section with install commands, verified-components table, upstream lineage, and rationale - CHANGELOG: add '0.22.0-evandro.1' entry under Unreleased - docs/: rename @llblab/pi-telegram imports to @evandrodevbr/oh-my-pi-telegram - tests/public-api.test.ts: rename package import paths Gates re-run on the fork: - npm run typecheck -> tsc --noEmit (success) - npm test -> 1229 pass / 0 fail / 1 platform-only skip - npm run audit -> found 0 vulnerabilities - npm run pack:check -> evandrodevbr-oh-my-pi-telegram-0.22.0-evandro.1.tgz Verified: - omp install --link ./pi-telegram -> Linked OK - omp plugin list -> @evandrodevbr/[email protected].0-evandro.1 Upstream lineage: badlogic/pi-telegram -> llblab/pi-telegram (0.22.0) -> evandrodevbr/oh-my-pi-telegram (this fork)
1209 lines
35 KiB
TypeScript
1209 lines
35 KiB
TypeScript
/**
|
|
* Regression tests for Telegram synchronization helpers
|
|
* Covers demand-driven reconciliation triggers and pure sync-slice state transitions
|
|
*/
|
|
|
|
import assert from "node:assert/strict";
|
|
import { mkdtemp, rm } from "node:fs/promises";
|
|
import { tmpdir } from "node:os";
|
|
import { join } from "node:path";
|
|
import test from "node:test";
|
|
|
|
import {
|
|
TELEGRAM_SYNC_SLICES,
|
|
createTelegramLeaderHealthRuntime,
|
|
createTelegramManualThreadDisconnectHandler,
|
|
createTelegramProvisioningActivityRuntime,
|
|
createTelegramSyncStateRuntime,
|
|
createTelegramTopicLifecycleSyncHandler,
|
|
createUnknownTelegramSyncState,
|
|
ensureTelegramLeaderThreadBinding,
|
|
markTelegramSyncSliceFresh,
|
|
markTelegramSyncSliceSuspect,
|
|
recoverStaleTelegramTopicApiError,
|
|
shouldReconcileTelegramSync,
|
|
} from "../lib/sync.ts";
|
|
import { createTelegramTopicTargetStore } from "../lib/threads.ts";
|
|
|
|
test("Telegram sync state runtime owns transitions and nested provisioning activity", () => {
|
|
const syncState = createTelegramSyncStateRuntime();
|
|
syncState.markConfigChange("profile-change");
|
|
assert.equal(syncState.getState()["bot-identity"]?.status, "fresh");
|
|
syncState.markSliceFresh("target-bindings", {
|
|
nowMs: 10,
|
|
action: "binding-restored",
|
|
});
|
|
assert.deepEqual(syncState.getState()["target-bindings"], {
|
|
status: "fresh",
|
|
updatedAtMs: 10,
|
|
lastReconcileAction: "binding-restored",
|
|
});
|
|
|
|
const provisioning = createTelegramProvisioningActivityRuntime();
|
|
provisioning.start();
|
|
provisioning.start();
|
|
provisioning.end();
|
|
assert.equal(provisioning.isActive(), true);
|
|
provisioning.end();
|
|
provisioning.end();
|
|
assert.equal(provisioning.isActive(), false);
|
|
});
|
|
|
|
test("Telegram sync reconciliation is demand-driven, not per ordinary action", () => {
|
|
for (const trigger of [
|
|
"startup",
|
|
"reload",
|
|
"topic-lifecycle",
|
|
"stale-api-error",
|
|
"setup-change",
|
|
"pairing-change",
|
|
"follower-register",
|
|
"follower-prune",
|
|
"status-request",
|
|
"leader-health-tick",
|
|
] as const) {
|
|
assert.equal(shouldReconcileTelegramSync(trigger), true);
|
|
}
|
|
|
|
assert.equal(shouldReconcileTelegramSync("ordinary-message"), false);
|
|
assert.equal(shouldReconcileTelegramSync("ordinary-send"), false);
|
|
});
|
|
|
|
function createTopicStore(
|
|
records: Array<{
|
|
profileKey?: string;
|
|
target: { chatId: number; threadId: number };
|
|
status?: "active" | "offline" | "stale" | "pending" | "starting" | "failed";
|
|
createdAtMs?: number;
|
|
updatedAtMs?: number;
|
|
}> = [],
|
|
) {
|
|
const normalizedRecords = records.map((record, index) => ({
|
|
profileKey: record.profileKey ?? `topic:${index}`,
|
|
target: record.target,
|
|
status: record.status ?? "active",
|
|
createdAtMs: record.createdAtMs ?? 1000,
|
|
updatedAtMs: record.updatedAtMs ?? 1000,
|
|
}));
|
|
return {
|
|
persisted: false,
|
|
records: normalizedRecords,
|
|
async load() {},
|
|
list() {
|
|
return this.records;
|
|
},
|
|
listReservations() {
|
|
return [];
|
|
},
|
|
listPendingProvisions() {
|
|
return [];
|
|
},
|
|
markStaleByTarget(target: { chatId: number; threadId?: number }) {
|
|
const record = this.records.find(
|
|
(item) =>
|
|
item.target.chatId === target.chatId &&
|
|
item.target.threadId === target.threadId,
|
|
);
|
|
if (!record) return false;
|
|
record.status = "stale";
|
|
return true;
|
|
},
|
|
removePendingProvision() {
|
|
return false;
|
|
},
|
|
markActiveByTarget(target: { chatId: number; threadId?: number }) {
|
|
const record = this.records.find(
|
|
(item) =>
|
|
item.target.chatId === target.chatId &&
|
|
item.target.threadId === target.threadId,
|
|
);
|
|
if (!record) return false;
|
|
record.status = "active";
|
|
return true;
|
|
},
|
|
async persist() {
|
|
this.persisted = true;
|
|
},
|
|
};
|
|
}
|
|
|
|
test("Telegram sync recovers stale topic API errors outside the entrypoint", async () => {
|
|
const store = createTopicStore([
|
|
{ target: { chatId: 7, threadId: 42 }, status: "active" },
|
|
]);
|
|
let state = createUnknownTelegramSyncState();
|
|
const events: Array<{
|
|
category: string;
|
|
phase?: unknown;
|
|
threadId?: unknown;
|
|
}> = [];
|
|
|
|
const recovered = await recoverStaleTelegramTopicApiError(
|
|
{ chat_id: 7, message_thread_id: 42 },
|
|
new Error("Telegram API sendMessage failed: TOPIC_ID_INVALID"),
|
|
{
|
|
topicTargetStore: store,
|
|
getNowMs: () => 1234,
|
|
getSyncState: () => state,
|
|
setSyncState: (nextState) => {
|
|
state = nextState;
|
|
},
|
|
recordEvent: (category, _message, details) => {
|
|
events.push({
|
|
category,
|
|
phase: details?.phase,
|
|
threadId: details?.threadId,
|
|
});
|
|
},
|
|
},
|
|
);
|
|
|
|
assert.equal(recovered, true);
|
|
assert.equal(store.records[0]?.status, "stale");
|
|
assert.equal(store.persisted, true);
|
|
assert.equal(state["topic-state"]?.status, "suspect");
|
|
assert.equal(
|
|
state["transport-health"]?.lastReconcileAction,
|
|
"topic-target-stale",
|
|
);
|
|
assert.deepEqual(events, [
|
|
{ category: "bus", phase: "topic-target-stale", threadId: 42 },
|
|
]);
|
|
});
|
|
|
|
test("Telegram sync ignores non-stale topic API errors", async () => {
|
|
const store = createTopicStore([
|
|
{ target: { chatId: 7, threadId: 42 }, status: "active" },
|
|
]);
|
|
let state = createUnknownTelegramSyncState();
|
|
|
|
const recovered = await recoverStaleTelegramTopicApiError(
|
|
{ chat_id: 7, message_thread_id: 42 },
|
|
new Error("network down"),
|
|
{
|
|
topicTargetStore: store,
|
|
getSyncState: () => state,
|
|
setSyncState: (nextState) => {
|
|
state = nextState;
|
|
},
|
|
recordEvent: () => undefined,
|
|
},
|
|
);
|
|
|
|
assert.equal(recovered, false);
|
|
assert.equal(store.records[0]?.status, "active");
|
|
assert.equal(store.persisted, false);
|
|
});
|
|
|
|
test("Telegram leader health runtime refreshes sync slices outside the entrypoint", async () => {
|
|
let state = createUnknownTelegramSyncState();
|
|
let calls = 0;
|
|
const runtime = createTelegramLeaderHealthRuntime({
|
|
intervalMs: 1,
|
|
getNowMs: () => 1234,
|
|
callGetMe: async () => {
|
|
calls += 1;
|
|
},
|
|
getSyncState: () => state,
|
|
setSyncState: (nextState) => {
|
|
state = nextState;
|
|
},
|
|
recordEvent: () => undefined,
|
|
});
|
|
runtime.start();
|
|
await new Promise((resolve) => setTimeout(resolve, 10));
|
|
runtime.stop();
|
|
|
|
assert.equal(calls > 0, true);
|
|
assert.equal(state["transport-health"]?.status, "fresh");
|
|
assert.equal(state["bot-identity"]?.status, "fresh");
|
|
assert.equal(
|
|
state["transport-health"]?.lastReconcileAction,
|
|
"leader-health-tick",
|
|
);
|
|
});
|
|
|
|
test("Telegram leader health runtime marks transport suspect on probe failure", async () => {
|
|
let state = createUnknownTelegramSyncState();
|
|
const events: Array<{ category: string; phase?: unknown }> = [];
|
|
const runtime = createTelegramLeaderHealthRuntime({
|
|
intervalMs: 1,
|
|
getNowMs: () => 1234,
|
|
callGetMe: async () => {
|
|
throw new Error("boom");
|
|
},
|
|
getSyncState: () => state,
|
|
setSyncState: (nextState) => {
|
|
state = nextState;
|
|
},
|
|
recordEvent: (category, _message, details) => {
|
|
events.push({ category, phase: details?.phase });
|
|
},
|
|
});
|
|
runtime.start();
|
|
await new Promise((resolve) => setTimeout(resolve, 10));
|
|
runtime.stop();
|
|
|
|
assert.equal(state["transport-health"]?.status, "suspect");
|
|
assert.equal(
|
|
state["transport-health"]?.lastReconcileAction,
|
|
"leader-health-tick",
|
|
);
|
|
assert.deepEqual(events.at(-1), {
|
|
category: "telegram",
|
|
phase: "leader-health-tick",
|
|
});
|
|
});
|
|
|
|
test("Telegram sync model includes every local assumption slice", () => {
|
|
assert.deepEqual(TELEGRAM_SYNC_SLICES, [
|
|
"bot-identity",
|
|
"bot-capabilities",
|
|
"pairing",
|
|
"allowed-user",
|
|
"topic-capability",
|
|
"topic-state",
|
|
"target-bindings",
|
|
"reservations",
|
|
"transport-health",
|
|
]);
|
|
assert.deepEqual(createUnknownTelegramSyncState()["reservations"], {
|
|
status: "unknown",
|
|
});
|
|
});
|
|
|
|
test("Telegram sync slice state records suspect and fresh transitions", () => {
|
|
const suspect = markTelegramSyncSliceSuspect({}, "topic-state", {
|
|
nowMs: 1000,
|
|
reason: "topic closed",
|
|
action: "lifecycle-closed",
|
|
});
|
|
assert.deepEqual(suspect["topic-state"], {
|
|
status: "suspect",
|
|
suspectAtMs: 1000,
|
|
reason: "topic closed",
|
|
lastReconcileAction: "lifecycle-closed",
|
|
});
|
|
|
|
const fresh = markTelegramSyncSliceFresh(suspect, "topic-state", {
|
|
nowMs: 2000,
|
|
action: "startup-probe",
|
|
});
|
|
assert.deepEqual(fresh["topic-state"], {
|
|
status: "fresh",
|
|
updatedAtMs: 2000,
|
|
lastReconcileAction: "startup-probe",
|
|
});
|
|
});
|
|
|
|
test("Leader thread sync creates a visible leader thread when no binding exists", async () => {
|
|
const dir = await mkdtemp(join(tmpdir(), "pi-telegram-leader-sync-"));
|
|
const store = createTelegramTopicTargetStore({
|
|
path: join(dir, "telegram-targets.json"),
|
|
getNowMs: () => 2000,
|
|
});
|
|
const calls: Array<{ method: string; body: Record<string, unknown> }> = [];
|
|
try {
|
|
const result = await ensureTelegramLeaderThreadBinding({
|
|
getAllowedUserId: () => 7,
|
|
instanceId: "leader-a",
|
|
cwd: "/repo",
|
|
topicTargetStore: store,
|
|
async callApi<TResponse>(method: string, body: Record<string, unknown>) {
|
|
calls.push({ method, body });
|
|
return { message_thread_id: 11 } as TResponse;
|
|
},
|
|
recordEvent() {},
|
|
});
|
|
|
|
assert.deepEqual(result, {
|
|
target: { chatId: 7, threadId: 11 },
|
|
slot: "A",
|
|
threadName: "Atlas",
|
|
reused: false,
|
|
});
|
|
assert.deepEqual(calls, [
|
|
{ method: "createForumTopic", body: { chat_id: 7, name: "Atlas" } },
|
|
]);
|
|
assert.deepEqual(store.getByProfileKey("cwd:/repo")?.target, {
|
|
chatId: 7,
|
|
threadId: 11,
|
|
});
|
|
} finally {
|
|
await rm(dir, { force: true, recursive: true });
|
|
}
|
|
});
|
|
|
|
test("Leader thread sync reuses same-profile topic across reload", async () => {
|
|
const dir = await mkdtemp(join(tmpdir(), "pi-telegram-leader-sync-prev-"));
|
|
const store = createTelegramTopicTargetStore({
|
|
path: join(dir, "telegram-targets.json"),
|
|
getNowMs: () => 2000,
|
|
});
|
|
const calls: Array<{ method: string; body: Record<string, unknown> }> = [];
|
|
const events: Array<Record<string, unknown>> = [];
|
|
try {
|
|
store.upsert({
|
|
profileKey: "cwd:/repo",
|
|
target: { chatId: 7, threadId: 10 },
|
|
status: "active",
|
|
createdAtMs: 1000,
|
|
updatedAtMs: 1000,
|
|
threadName: "🧭 Axial",
|
|
instanceId: "previous-instance",
|
|
slot: "A",
|
|
});
|
|
await store.persist();
|
|
|
|
const result = await ensureTelegramLeaderThreadBinding({
|
|
getAllowedUserId: () => 7,
|
|
instanceId: "leader-a",
|
|
cwd: "/repo",
|
|
topicTargetStore: store,
|
|
async callApi<TResponse>(method: string, body: Record<string, unknown>) {
|
|
calls.push({ method, body });
|
|
return { message_thread_id: 11 } as TResponse;
|
|
},
|
|
recordEvent(_category, _message, details) {
|
|
if (details) events.push(details);
|
|
},
|
|
});
|
|
|
|
assert.deepEqual(result, {
|
|
target: { chatId: 7, threadId: 10 },
|
|
slot: "A",
|
|
threadName: "🧭 Axial",
|
|
reused: true,
|
|
});
|
|
assert.deepEqual(calls, []);
|
|
const record = store.getByProfileKey("cwd:/repo");
|
|
assert.equal(record?.target.threadId, 10);
|
|
assert.equal(record?.instanceId, "leader-a");
|
|
assert.deepEqual(record?.owner, {
|
|
kind: "leader",
|
|
cwd: "/repo",
|
|
instanceId: "leader-a",
|
|
});
|
|
assert.equal(record?.threadName, "🧭 Axial");
|
|
assert.equal(store.listReservations().length, 0);
|
|
assert.equal(
|
|
events.some(
|
|
(event) => event.phase === "leader-topic-same-profile-preserve",
|
|
),
|
|
true,
|
|
);
|
|
} finally {
|
|
await rm(dir, { force: true, recursive: true });
|
|
}
|
|
});
|
|
|
|
test("Leader thread sync rejects reused binding after load loses epoch", async () => {
|
|
const dir = await mkdtemp(join(tmpdir(), "pi-telegram-leader-reuse-epoch-"));
|
|
const store = createTelegramTopicTargetStore({
|
|
path: join(dir, "state.json"),
|
|
getNowMs: () => 2000,
|
|
});
|
|
let leaderEpoch: number | undefined = 1;
|
|
try {
|
|
store.upsert({
|
|
profileKey: "cwd:/repo",
|
|
target: { chatId: 7, threadId: 10 },
|
|
status: "active",
|
|
createdAtMs: 1000,
|
|
updatedAtMs: 1000,
|
|
instanceId: "leader-a",
|
|
slot: "A",
|
|
});
|
|
await store.persist();
|
|
const epochLosingStore = {
|
|
...store,
|
|
async load() {
|
|
await store.load();
|
|
leaderEpoch = undefined;
|
|
},
|
|
};
|
|
|
|
await assert.rejects(
|
|
ensureTelegramLeaderThreadBinding({
|
|
getAllowedUserId: () => 7,
|
|
instanceId: "leader-a",
|
|
cwd: "/repo",
|
|
getCurrentLeaderEpoch: () => leaderEpoch,
|
|
topicTargetStore: epochLosingStore,
|
|
async callApi<TResponse>() {
|
|
return {} as TResponse;
|
|
},
|
|
recordEvent() {},
|
|
}),
|
|
/lost ownership \(after-load\)/,
|
|
);
|
|
} finally {
|
|
await rm(dir, { force: true, recursive: true });
|
|
}
|
|
});
|
|
|
|
test("Leader thread sync reuses same-process legacy leader topic across reload", async () => {
|
|
const dir = await mkdtemp(
|
|
join(tmpdir(), "pi-telegram-leader-sync-same-pid-"),
|
|
);
|
|
const store = createTelegramTopicTargetStore({
|
|
path: join(dir, "telegram-targets.json"),
|
|
getNowMs: () => 2000,
|
|
});
|
|
const calls: Array<{ method: string; body: Record<string, unknown> }> = [];
|
|
const events: Array<Record<string, unknown>> = [];
|
|
try {
|
|
store.upsert({
|
|
profileKey: "leader:3824869:1782087343414",
|
|
target: { chatId: 7, threadId: 10 },
|
|
status: "active",
|
|
createdAtMs: 1000,
|
|
updatedAtMs: 1000,
|
|
threadName: "Nexus",
|
|
instanceId: "3824869:1782087343414",
|
|
slot: "N",
|
|
});
|
|
await store.persist();
|
|
|
|
const result = await ensureTelegramLeaderThreadBinding({
|
|
getAllowedUserId: () => 7,
|
|
instanceId: "3824869:1782088512458",
|
|
cwd: "/repo",
|
|
topicTargetStore: store,
|
|
async callApi<TResponse>(method: string, body: Record<string, unknown>) {
|
|
calls.push({ method, body });
|
|
return { message_thread_id: 11 } as TResponse;
|
|
},
|
|
recordEvent(_category, _message, details) {
|
|
if (details) events.push(details);
|
|
},
|
|
});
|
|
|
|
assert.deepEqual(result, {
|
|
target: { chatId: 7, threadId: 10 },
|
|
slot: "N",
|
|
threadName: "Nexus",
|
|
reused: true,
|
|
});
|
|
assert.deepEqual(calls, []);
|
|
const record = store.getByProfileKey("cwd:/repo");
|
|
assert.equal(record?.target.threadId, 10);
|
|
assert.equal(record?.instanceId, "3824869:1782088512458");
|
|
assert.equal(record?.threadName, "Nexus");
|
|
assert.equal(
|
|
store.getByProfileKey("leader:3824869:1782087343414"),
|
|
undefined,
|
|
);
|
|
assert.equal(store.listReservations().length, 0);
|
|
assert.equal(
|
|
events.some(
|
|
(event) => event.phase === "leader-topic-same-process-preserve",
|
|
),
|
|
true,
|
|
);
|
|
} finally {
|
|
await rm(dir, { force: true, recursive: true });
|
|
}
|
|
});
|
|
|
|
test("Leader thread sync does not visibly probe same-profile topic during reload", async () => {
|
|
const dir = await mkdtemp(join(tmpdir(), "pi-telegram-leader-sync-defer-"));
|
|
const store = createTelegramTopicTargetStore({
|
|
path: join(dir, "telegram-targets.json"),
|
|
getNowMs: () => 2000,
|
|
});
|
|
const calls: Array<{ method: string; body: Record<string, unknown> }> = [];
|
|
try {
|
|
store.upsert({
|
|
profileKey: "cwd:/repo",
|
|
target: { chatId: 7, threadId: 10 },
|
|
status: "active",
|
|
createdAtMs: 1000,
|
|
updatedAtMs: 1000,
|
|
instanceId: "previous-instance",
|
|
slot: "A",
|
|
});
|
|
await store.persist();
|
|
|
|
const result = await ensureTelegramLeaderThreadBinding({
|
|
getAllowedUserId: () => 7,
|
|
instanceId: "leader-a",
|
|
cwd: "/repo",
|
|
topicTargetStore: store,
|
|
async callApi<TResponse>(method: string, body: Record<string, unknown>) {
|
|
calls.push({ method, body });
|
|
return { message_thread_id: 11 } as TResponse;
|
|
},
|
|
recordEvent() {},
|
|
});
|
|
|
|
assert.deepEqual(result, {
|
|
target: { chatId: 7, threadId: 10 },
|
|
slot: "A",
|
|
threadName: "Atlas",
|
|
reused: true,
|
|
});
|
|
assert.deepEqual(calls, []);
|
|
assert.equal(store.getByProfileKey("cwd:/repo")?.target.threadId, 10);
|
|
assert.equal(store.listReservations().length, 0);
|
|
} finally {
|
|
await rm(dir, { force: true, recursive: true });
|
|
}
|
|
});
|
|
|
|
test("Leader thread sync proactively deletes known reservations on startup", async () => {
|
|
const dir = await mkdtemp(
|
|
join(tmpdir(), "pi-telegram-leader-reservation-cleanup-"),
|
|
);
|
|
const store = createTelegramTopicTargetStore({
|
|
path: join(dir, "state.json"),
|
|
getNowMs: () => 2000,
|
|
});
|
|
const calls: Array<{ method: string; body: Record<string, unknown> }> = [];
|
|
try {
|
|
store.reserveThread({
|
|
target: { chatId: 7, threadId: 9 },
|
|
slot: "A",
|
|
reason: "previous-process-still-probes-alive",
|
|
createdAtMs: 1000,
|
|
updatedAtMs: 1000,
|
|
expiresAtMs: 10_000,
|
|
});
|
|
await store.persist();
|
|
|
|
await ensureTelegramLeaderThreadBinding({
|
|
getAllowedUserId: () => 7,
|
|
instanceId: "leader-a",
|
|
cwd: "/repo",
|
|
topicTargetStore: store,
|
|
async callApi<TResponse>(method: string, body: Record<string, unknown>) {
|
|
calls.push({ method, body });
|
|
if (method === "sendChatAction" && body.message_thread_id === 9) {
|
|
throw new Error(
|
|
"Telegram API sendChatAction failed: HTTP 400: Bad Request: message thread not found",
|
|
);
|
|
}
|
|
return { message_thread_id: 10 } as TResponse;
|
|
},
|
|
recordEvent() {},
|
|
});
|
|
|
|
assert.deepEqual(
|
|
calls.slice(0, 3).map((call) => call.method),
|
|
["closeForumTopic", "deleteForumTopic", "createForumTopic"],
|
|
);
|
|
assert.deepEqual(
|
|
store.listReservations().map((reservation) => reservation.target),
|
|
[{ chatId: 7, threadId: 9 }],
|
|
);
|
|
} finally {
|
|
await rm(dir, { force: true, recursive: true });
|
|
}
|
|
});
|
|
|
|
test("Leader thread sync preserves follower records from previous instances", async () => {
|
|
const dir = await mkdtemp(
|
|
join(tmpdir(), "pi-telegram-leader-preserve-followers-"),
|
|
);
|
|
const store = createTelegramTopicTargetStore({
|
|
path: join(dir, "state.json"),
|
|
getNowMs: () => 2000,
|
|
});
|
|
const calls: Array<{ method: string; body: Record<string, unknown> }> = [];
|
|
try {
|
|
store.upsert({
|
|
profileKey: "manual:follower-a",
|
|
owner: { kind: "manual-follower", instanceId: "follower-a" },
|
|
target: { chatId: 7, threadId: 10 },
|
|
status: "active",
|
|
createdAtMs: 1000,
|
|
updatedAtMs: 1000,
|
|
instanceId: "follower-a",
|
|
slot: "T",
|
|
});
|
|
await store.persist();
|
|
|
|
await ensureTelegramLeaderThreadBinding({
|
|
getAllowedUserId: () => 7,
|
|
instanceId: "leader-a",
|
|
cwd: "/repo",
|
|
topicTargetStore: store,
|
|
async callApi<TResponse>(method: string, body: Record<string, unknown>) {
|
|
calls.push({ method, body });
|
|
return { message_thread_id: 11 } as TResponse;
|
|
},
|
|
recordEvent() {},
|
|
});
|
|
|
|
assert.deepEqual(store.getByProfileKey("manual:follower-a")?.target, {
|
|
chatId: 7,
|
|
threadId: 10,
|
|
});
|
|
assert.equal(store.getByProfileKey("manual:follower-a")?.status, "active");
|
|
assert.equal(
|
|
calls.some(
|
|
(call) =>
|
|
(call.method === "closeForumTopic" ||
|
|
call.method === "deleteForumTopic") &&
|
|
call.body.message_thread_id === 10,
|
|
),
|
|
false,
|
|
);
|
|
} finally {
|
|
await rm(dir, { force: true, recursive: true });
|
|
}
|
|
});
|
|
|
|
test("Leader thread sync cleans previous leader records regardless of profile key", async () => {
|
|
const dir = await mkdtemp(
|
|
join(tmpdir(), "pi-telegram-leader-cross-profile-"),
|
|
);
|
|
const store = createTelegramTopicTargetStore({
|
|
path: join(dir, "state.json"),
|
|
getNowMs: () => 2000,
|
|
});
|
|
try {
|
|
store.upsert({
|
|
profileKey: "leader:previous-id",
|
|
target: { chatId: 7, threadId: 10 },
|
|
status: "active",
|
|
createdAtMs: 1000,
|
|
updatedAtMs: 1000,
|
|
instanceId: "previous-instance",
|
|
slot: "B",
|
|
});
|
|
await store.persist();
|
|
|
|
await ensureTelegramLeaderThreadBinding({
|
|
getAllowedUserId: () => 7,
|
|
instanceId: "leader-a",
|
|
cwd: "/repo",
|
|
topicTargetStore: store,
|
|
async callApi<TResponse>() {
|
|
return { message_thread_id: 11 } as TResponse;
|
|
},
|
|
recordEvent() {},
|
|
});
|
|
|
|
assert.equal(store.getByProfileKey("leader:previous-id"), undefined);
|
|
const [reservation] = store.listReservations();
|
|
assert.deepEqual(reservation?.target, { chatId: 7, threadId: 10 });
|
|
assert.equal(reservation?.slot, "B");
|
|
assert.equal(
|
|
reservation?.reason,
|
|
"previous-process-cleaned-without-visible-probe",
|
|
);
|
|
} finally {
|
|
await rm(dir, { force: true, recursive: true });
|
|
}
|
|
});
|
|
|
|
test("Leader thread sync aborts local previous-leader cleanup after ownership loss", async () => {
|
|
const dir = await mkdtemp(
|
|
join(tmpdir(), "pi-telegram-leader-cleanup-epoch-loss-"),
|
|
);
|
|
const store = createTelegramTopicTargetStore({
|
|
path: join(dir, "state.json"),
|
|
getNowMs: () => 2000,
|
|
});
|
|
let currentEpoch: number | undefined = 1;
|
|
const calls: string[] = [];
|
|
try {
|
|
store.upsert({
|
|
profileKey: "leader:previous-id",
|
|
target: { chatId: 7, threadId: 10 },
|
|
status: "active",
|
|
createdAtMs: 1000,
|
|
updatedAtMs: 1000,
|
|
instanceId: "previous-instance",
|
|
slot: "B",
|
|
});
|
|
await store.persist();
|
|
|
|
await assert.rejects(
|
|
ensureTelegramLeaderThreadBinding({
|
|
getAllowedUserId: () => 7,
|
|
instanceId: "leader-a",
|
|
cwd: "/repo",
|
|
topicTargetStore: store,
|
|
getCurrentLeaderEpoch: () => currentEpoch,
|
|
async callApi<TResponse>(method: string) {
|
|
calls.push(method);
|
|
currentEpoch = undefined;
|
|
return {} as TResponse;
|
|
},
|
|
recordEvent() {},
|
|
}),
|
|
/ownership changed during topic reconciliation/,
|
|
);
|
|
|
|
assert.deepEqual(calls, ["closeForumTopic"]);
|
|
assert.equal(store.getByProfileKey("leader:previous-id")?.status, "active");
|
|
assert.deepEqual(store.listReservations(), []);
|
|
} finally {
|
|
await rm(dir, { force: true, recursive: true });
|
|
}
|
|
});
|
|
|
|
test("Leader thread sync gets next monotonic slot after D on reload", async () => {
|
|
const dir = await mkdtemp(join(tmpdir(), "pi-telegram-leader-sync-d-to-e-"));
|
|
const store = createTelegramTopicTargetStore({
|
|
path: join(dir, "state.json"),
|
|
getNowMs: () => 2000,
|
|
});
|
|
try {
|
|
store.upsert({
|
|
profileKey: "leader:old-a",
|
|
target: { chatId: 7, threadId: 10 },
|
|
status: "active",
|
|
createdAtMs: 1000,
|
|
updatedAtMs: 1000,
|
|
instanceId: "previous-instance",
|
|
slot: "D",
|
|
});
|
|
await store.persist();
|
|
|
|
await ensureTelegramLeaderThreadBinding({
|
|
getAllowedUserId: () => 7,
|
|
instanceId: "leader-a",
|
|
cwd: "/repo",
|
|
topicTargetStore: store,
|
|
async callApi<TResponse>() {
|
|
return { message_thread_id: 11 } as TResponse;
|
|
},
|
|
recordEvent() {},
|
|
});
|
|
|
|
const [reservation] = store.listReservations();
|
|
assert.equal(reservation?.slot, "D");
|
|
assert.equal(
|
|
reservation?.reason,
|
|
"previous-process-cleaned-without-visible-probe",
|
|
);
|
|
const newRecord = store.getByProfileKey("cwd:/repo");
|
|
assert.equal(newRecord?.slot, "E");
|
|
} finally {
|
|
await rm(dir, { force: true, recursive: true });
|
|
}
|
|
});
|
|
|
|
test("Leader thread sync can force-refresh unnamed stale-prone leader bindings", async () => {
|
|
const dir = await mkdtemp(join(tmpdir(), "pi-telegram-leader-sync-refresh-"));
|
|
const store = createTelegramTopicTargetStore({
|
|
path: join(dir, "telegram-targets.json"),
|
|
getNowMs: () => 2000,
|
|
});
|
|
const calls: Array<{ method: string; body: Record<string, unknown> }> = [];
|
|
try {
|
|
store.upsert({
|
|
profileKey: "cwd:/repo",
|
|
target: { chatId: 7, threadId: 10 },
|
|
status: "active",
|
|
createdAtMs: 1000,
|
|
updatedAtMs: 1000,
|
|
instanceId: "leader-a",
|
|
slot: "A",
|
|
});
|
|
await store.persist();
|
|
|
|
const result = await ensureTelegramLeaderThreadBinding({
|
|
getAllowedUserId: () => 7,
|
|
instanceId: "leader-a",
|
|
cwd: "/repo",
|
|
forceFreshUnnamed: true,
|
|
topicTargetStore: store,
|
|
async callApi<TResponse>(method: string, body: Record<string, unknown>) {
|
|
calls.push({ method, body });
|
|
return { message_thread_id: 11 } as TResponse;
|
|
},
|
|
recordEvent() {},
|
|
});
|
|
|
|
assert.deepEqual(result, {
|
|
target: { chatId: 7, threadId: 11 },
|
|
slot: "A",
|
|
threadName: "Atlas",
|
|
reused: false,
|
|
});
|
|
assert.deepEqual(calls, [
|
|
{ method: "createForumTopic", body: { chat_id: 7, name: "Atlas" } },
|
|
{
|
|
method: "closeForumTopic",
|
|
body: { chat_id: 7, message_thread_id: 10 },
|
|
},
|
|
]);
|
|
assert.equal(store.getByProfileKey("cwd:/repo")?.target.threadId, 11);
|
|
} finally {
|
|
await rm(dir, { force: true, recursive: true });
|
|
}
|
|
});
|
|
|
|
test("Leader thread sync force-refresh preserves promoted follower bindings", async () => {
|
|
const dir = await mkdtemp(
|
|
join(tmpdir(), "pi-telegram-promoted-follower-refresh-"),
|
|
);
|
|
const store = createTelegramTopicTargetStore({
|
|
path: join(dir, "state.json"),
|
|
getNowMs: () => 2000,
|
|
});
|
|
const calls: Array<{ method: string; body: Record<string, unknown> }> = [];
|
|
try {
|
|
store.upsert({
|
|
profileKey: "manual:follower-b",
|
|
owner: { kind: "manual-follower", instanceId: "follower-b" },
|
|
target: { chatId: 7, threadId: 12 },
|
|
status: "active",
|
|
createdAtMs: 1000,
|
|
updatedAtMs: 1000,
|
|
instanceId: "follower-b",
|
|
slot: "B",
|
|
});
|
|
await store.persist();
|
|
|
|
const result = await ensureTelegramLeaderThreadBinding({
|
|
getAllowedUserId: () => 7,
|
|
instanceId: "follower-b",
|
|
cwd: "/repo",
|
|
forceFreshUnnamed: true,
|
|
topicTargetStore: store,
|
|
async callApi<TResponse>(method: string, body: Record<string, unknown>) {
|
|
calls.push({ method, body });
|
|
return { message_thread_id: 99 } as TResponse;
|
|
},
|
|
recordEvent() {},
|
|
});
|
|
|
|
assert.deepEqual(result, {
|
|
target: { chatId: 7, threadId: 12 },
|
|
slot: "B",
|
|
reused: true,
|
|
});
|
|
assert.deepEqual(calls, []);
|
|
assert.equal(
|
|
store.getActiveByInstanceId("follower-b")?.target.threadId,
|
|
12,
|
|
);
|
|
} finally {
|
|
await rm(dir, { force: true, recursive: true });
|
|
}
|
|
});
|
|
|
|
test("Topic lifecycle sync marks known topics stale or active", async () => {
|
|
const store = createTopicStore([
|
|
{ target: { chatId: 7, threadId: 42 }, status: "active" },
|
|
]);
|
|
const events: Array<Record<string, unknown>> = [];
|
|
const handler = createTelegramTopicLifecycleSyncHandler({
|
|
topicTargetStore: store,
|
|
isBusEnabled: () => true,
|
|
async callApi() {
|
|
throw new Error("unexpected API call");
|
|
},
|
|
recordEvent(_category, _message, details) {
|
|
events.push(details ?? {});
|
|
},
|
|
});
|
|
|
|
await handler({
|
|
kind: "closed",
|
|
target: { chatId: 7, threadId: 42 },
|
|
message: {},
|
|
});
|
|
assert.equal(store.records[0]?.status, "stale");
|
|
assert.equal(store.persisted, true);
|
|
assert.equal(events[0]?.phase, "topic-lifecycle");
|
|
assert.equal(events[0]?.changed, true);
|
|
|
|
store.persisted = false;
|
|
await handler({
|
|
kind: "reopened",
|
|
target: { chatId: 7, threadId: 42 },
|
|
message: {},
|
|
});
|
|
assert.equal(store.records[0]?.status, "active");
|
|
assert.equal(store.persisted, true);
|
|
});
|
|
|
|
test("Topic lifecycle sync loads persisted bindings before deleting created topics", async () => {
|
|
const calls: Array<{ method: string; body: unknown }> = [];
|
|
let loaded = false;
|
|
const store = {
|
|
persisted: false,
|
|
records: [] as Array<{
|
|
profileKey: string;
|
|
target: { chatId: number; threadId: number };
|
|
status: "active";
|
|
createdAtMs: number;
|
|
updatedAtMs: number;
|
|
}>,
|
|
async load() {
|
|
loaded = true;
|
|
this.records = [
|
|
{
|
|
profileKey: "manual:follower",
|
|
target: { chatId: 7, threadId: 42 },
|
|
status: "active",
|
|
createdAtMs: 1000,
|
|
updatedAtMs: 1000,
|
|
},
|
|
];
|
|
},
|
|
list() {
|
|
return this.records;
|
|
},
|
|
listReservations() {
|
|
return [];
|
|
},
|
|
listPendingProvisions() {
|
|
return [];
|
|
},
|
|
markStaleByTarget() {
|
|
return false;
|
|
},
|
|
markActiveByTarget() {
|
|
return false;
|
|
},
|
|
removePendingProvision() {
|
|
return false;
|
|
},
|
|
async persist() {
|
|
this.persisted = true;
|
|
},
|
|
};
|
|
const handler = createTelegramTopicLifecycleSyncHandler({
|
|
topicTargetStore: store,
|
|
isBusEnabled: () => true,
|
|
async callApi<TResponse>(method: string, body: Record<string, unknown>) {
|
|
calls.push({ method, body });
|
|
return {} as TResponse;
|
|
},
|
|
});
|
|
|
|
await handler({
|
|
kind: "created",
|
|
target: { chatId: 7, threadId: 42 },
|
|
message: {},
|
|
});
|
|
|
|
assert.equal(loaded, true);
|
|
assert.deepEqual(calls, []);
|
|
assert.equal(store.persisted, false);
|
|
});
|
|
|
|
test("Topic lifecycle sync observes unknown created topics without deleting them", async () => {
|
|
const store = createTopicStore();
|
|
const calls: Array<{ method: string; body: unknown }> = [];
|
|
const events: Array<Record<string, unknown>> = [];
|
|
const handler = createTelegramTopicLifecycleSyncHandler({
|
|
topicTargetStore: store,
|
|
isBusEnabled: () => true,
|
|
async callApi<TResponse>(method: string, body: Record<string, unknown>) {
|
|
calls.push({ method, body });
|
|
return {} as TResponse;
|
|
},
|
|
recordEvent(_category, _message, details) {
|
|
events.push(details ?? {});
|
|
},
|
|
});
|
|
|
|
await handler({
|
|
kind: "created",
|
|
target: { chatId: 7, threadId: 42 },
|
|
message: {},
|
|
});
|
|
assert.deepEqual(calls, []);
|
|
assert.equal(store.persisted, false);
|
|
assert.equal(events[0]?.phase, "topic-lifecycle-unknown-created-observed");
|
|
|
|
const classicCalls: Array<{ method: string; body: unknown }> = [];
|
|
store.persisted = false;
|
|
const classicHandler = createTelegramTopicLifecycleSyncHandler({
|
|
topicTargetStore: store,
|
|
isBusEnabled: () => false,
|
|
async callApi<TResponse>(method: string, body: Record<string, unknown>) {
|
|
classicCalls.push({ method, body });
|
|
return {} as TResponse;
|
|
},
|
|
});
|
|
await classicHandler({
|
|
kind: "created",
|
|
target: { chatId: 7, threadId: 43 },
|
|
message: {},
|
|
});
|
|
assert.deepEqual(classicCalls, []);
|
|
assert.equal(store.persisted, false);
|
|
});
|
|
|
|
test("Topic lifecycle sync does not delete unknown created topics during provisioning", async () => {
|
|
const store = createTopicStore();
|
|
const calls: Array<{ method: string; body: unknown }> = [];
|
|
const events: Array<Record<string, unknown>> = [];
|
|
const handler = createTelegramTopicLifecycleSyncHandler({
|
|
topicTargetStore: store,
|
|
isBusEnabled: () => true,
|
|
isTopicProvisioningActive: () => true,
|
|
async callApi<TResponse>(method: string, body: Record<string, unknown>) {
|
|
calls.push({ method, body });
|
|
return {} as TResponse;
|
|
},
|
|
recordEvent(_category, _message, details) {
|
|
events.push(details ?? {});
|
|
},
|
|
});
|
|
|
|
await handler({
|
|
kind: "created",
|
|
target: { chatId: 7, threadId: 44 },
|
|
message: {},
|
|
});
|
|
|
|
assert.deepEqual(calls, []);
|
|
assert.equal(store.persisted, false);
|
|
assert.equal(events[0]?.phase, "topic-lifecycle-provisioning-skip");
|
|
});
|
|
|
|
test("Manual follower disconnect preserves its Telegram thread binding", async () => {
|
|
const calls: unknown[] = [];
|
|
let offlineCalls = 0;
|
|
let persisted = 0;
|
|
let syncState = createUnknownTelegramSyncState();
|
|
const disconnect = createTelegramManualThreadDisconnectHandler({
|
|
instanceId: "follower-runtime:1",
|
|
getCurrentThreadRecord: () => ({
|
|
owner: { kind: "manual-follower" },
|
|
instanceId: "follower-runtime:1",
|
|
target: { chatId: 7, threadId: 42 },
|
|
}),
|
|
topicTargetStore: {
|
|
markOfflineByInstanceId: () => {
|
|
offlineCalls += 1;
|
|
return 1;
|
|
},
|
|
persist: async () => {
|
|
persisted += 1;
|
|
},
|
|
},
|
|
callApi: async <TResponse>(method: string, body: Record<string, unknown>) => {
|
|
calls.push({ method, body });
|
|
return { ok: true } as TResponse;
|
|
},
|
|
getLeaderTarget: () => undefined,
|
|
clearLeaderTarget: () => undefined,
|
|
getSyncState: () => syncState,
|
|
setSyncState: (state) => {
|
|
syncState = state;
|
|
},
|
|
stopPolling: async () => "stopped",
|
|
recordRuntimeEvent: () => undefined,
|
|
getNowMs: () => 2000,
|
|
});
|
|
|
|
assert.equal(await disconnect(), "stopped");
|
|
assert.deepEqual(calls, []);
|
|
assert.equal(offlineCalls, 0);
|
|
assert.equal(persisted, 0);
|
|
});
|
|
|
|
test("Manual leader disconnect stops local teardown after epoch loss", async () => {
|
|
const trace: string[] = [];
|
|
let currentEpoch: number | undefined = 1;
|
|
let syncState = createUnknownTelegramSyncState();
|
|
const disconnect = createTelegramManualThreadDisconnectHandler({
|
|
instanceId: "leader-runtime:1",
|
|
getCurrentThreadRecord: () => ({
|
|
owner: { kind: "leader" },
|
|
instanceId: "leader-runtime:1",
|
|
target: { chatId: 7, threadId: 42 },
|
|
}),
|
|
topicTargetStore: {
|
|
markOfflineByInstanceId: () => {
|
|
trace.push("mark-offline");
|
|
return 1;
|
|
},
|
|
persist: async () => {
|
|
trace.push("persist");
|
|
},
|
|
},
|
|
callApi: async <TResponse>(method: string) => {
|
|
trace.push(method);
|
|
currentEpoch = undefined;
|
|
return { ok: true } as TResponse;
|
|
},
|
|
getCurrentLeaderEpoch: () => currentEpoch,
|
|
getLeaderTarget: () => {
|
|
trace.push("get-leader-target");
|
|
return { chatId: 7, threadId: 42 };
|
|
},
|
|
clearLeaderTarget: () => {
|
|
trace.push("clear-leader-target");
|
|
},
|
|
getSyncState: () => syncState,
|
|
setSyncState: (state) => {
|
|
trace.push("set-sync-state");
|
|
syncState = state;
|
|
},
|
|
stopPolling: async () => {
|
|
trace.push("stop-polling");
|
|
return "stopped";
|
|
},
|
|
recordRuntimeEvent: () => undefined,
|
|
});
|
|
|
|
assert.equal(await disconnect(), "stopped");
|
|
assert.deepEqual(trace, ["closeForumTopic", "stop-polling"]);
|
|
});
|
|
|
|
test("Manual leader disconnect rechecks epoch after zero-change offline mutation", async () => {
|
|
const trace: string[] = [];
|
|
let currentEpoch: number | undefined = 1;
|
|
let syncState = createUnknownTelegramSyncState();
|
|
const disconnect = createTelegramManualThreadDisconnectHandler({
|
|
instanceId: "leader-runtime:1",
|
|
getCurrentThreadRecord: () => ({
|
|
owner: { kind: "leader" },
|
|
instanceId: "leader-runtime:1",
|
|
target: { chatId: 7, threadId: 42 },
|
|
}),
|
|
topicTargetStore: {
|
|
markOfflineByInstanceId: () => {
|
|
trace.push("mark-offline");
|
|
currentEpoch = undefined;
|
|
return 0;
|
|
},
|
|
persist: async () => {
|
|
trace.push("persist");
|
|
},
|
|
},
|
|
callApi: async <TResponse>(method: string) => {
|
|
trace.push(method);
|
|
return { ok: true } as TResponse;
|
|
},
|
|
getCurrentLeaderEpoch: () => currentEpoch,
|
|
getLeaderTarget: () => {
|
|
trace.push("get-leader-target");
|
|
return { chatId: 7, threadId: 42 };
|
|
},
|
|
clearLeaderTarget: () => {
|
|
trace.push("clear-leader-target");
|
|
},
|
|
getSyncState: () => syncState,
|
|
setSyncState: (state) => {
|
|
trace.push("set-sync-state");
|
|
syncState = state;
|
|
},
|
|
stopPolling: async () => {
|
|
trace.push("stop-polling");
|
|
return "stopped";
|
|
},
|
|
recordRuntimeEvent: () => undefined,
|
|
});
|
|
|
|
assert.equal(await disconnect(), "stopped");
|
|
assert.deepEqual(trace, [
|
|
"closeForumTopic",
|
|
"deleteForumTopic",
|
|
"mark-offline",
|
|
"stop-polling",
|
|
]);
|
|
});
|