wehub-resource-sync
529 lines
18 KiB
TypeScript
529 lines
18 KiB
TypeScript
import { describe, expect, test } from "bun:test"
|
|
import { spawnSync } from "node:child_process"
|
|
import { mkdtempSync, writeFileSync, readFileSync } from "node:fs"
|
|
import { tmpdir } from "node:os"
|
|
import path from "node:path"
|
|
|
|
const SCRIPT = path.join(
|
|
__dirname,
|
|
"../skills/ce-sweep/scripts/sweep-state.py",
|
|
)
|
|
|
|
function run(
|
|
cwd: string,
|
|
...args: string[]
|
|
): { code: number; stdout: string; stderr: string } {
|
|
const r = spawnSync("python3", [SCRIPT, ...args], { cwd, encoding: "utf8" })
|
|
return {
|
|
code: r.status ?? -1,
|
|
stdout: r.stdout ?? "",
|
|
stderr: r.stderr ?? "",
|
|
}
|
|
}
|
|
|
|
function tmp(): string {
|
|
return mkdtempSync(path.join(tmpdir(), "sweep-state-"))
|
|
}
|
|
function statePath(dir: string): string {
|
|
return path.join(dir, "state.yml")
|
|
}
|
|
function status(stdout: string): string {
|
|
return stdout.split("\n")[0].trim()
|
|
}
|
|
function payload(stdout: string): any {
|
|
const nl = stdout.indexOf("\n")
|
|
if (nl === -1) return undefined
|
|
const rest = stdout.slice(nl + 1).trim()
|
|
return rest ? JSON.parse(rest) : undefined
|
|
}
|
|
|
|
const NOW = "2026-07-02T12:00:00+00:00"
|
|
|
|
function acquire(dir: string, s: string, writer = "w1", now = NOW) {
|
|
return run(dir, "lease-acquire", "--state", s, "--writer", writer, "--now", now)
|
|
}
|
|
function upsert(
|
|
dir: string,
|
|
s: string,
|
|
id: string,
|
|
source: string,
|
|
item: object,
|
|
writer = "w1",
|
|
now = NOW,
|
|
) {
|
|
return run(
|
|
dir,
|
|
"upsert-item",
|
|
"--state",
|
|
s,
|
|
"--id",
|
|
id,
|
|
"--source",
|
|
source,
|
|
"--writer",
|
|
writer,
|
|
"--now",
|
|
now,
|
|
"--json",
|
|
JSON.stringify(item),
|
|
)
|
|
}
|
|
function read(dir: string, s: string) {
|
|
return payload(run(dir, "read", "--state", s).stdout)
|
|
}
|
|
function cursorAdvance(
|
|
dir: string,
|
|
s: string,
|
|
source: string,
|
|
to: string,
|
|
pastItem: string,
|
|
writer = "w1",
|
|
now = NOW,
|
|
) {
|
|
return run(
|
|
dir,
|
|
"cursor-advance",
|
|
"--state",
|
|
s,
|
|
"--source",
|
|
source,
|
|
"--to",
|
|
to,
|
|
"--past-item",
|
|
pastItem,
|
|
"--writer",
|
|
writer,
|
|
"--now",
|
|
now,
|
|
)
|
|
}
|
|
|
|
describe("sweep-state engine — core round-trip and lifecycle", () => {
|
|
test("upsert then read round-trips the item", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
expect(status(acquire(dir, s).stdout)).toBe("OK")
|
|
expect(
|
|
status(
|
|
upsert(dir, s, "100.5", "slack:C1", {
|
|
status: "ingested",
|
|
body: "hello world",
|
|
}).stdout,
|
|
),
|
|
).toBe("OK")
|
|
const rd = run(dir, "read", "--state", s)
|
|
expect(status(rd.stdout)).toBe("OK")
|
|
const state = payload(rd.stdout)
|
|
// Items are stored under a source-scoped composite key so ids never collide
|
|
// across sources; the item stays self-describing via source + id.
|
|
expect(state.items["slack:C1:100.5"].status).toBe("ingested")
|
|
expect(state.items["slack:C1:100.5"].body).toBe("hello world")
|
|
expect(state.items["slack:C1:100.5"].source).toBe("slack:C1")
|
|
expect(state.items["slack:C1:100.5"].id).toBe("100.5")
|
|
})
|
|
|
|
test("the same id under two sources does not collide", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
acquire(dir, s)
|
|
upsert(dir, s, "123", "slack-alpha", { status: "ingested", body: "from slack" })
|
|
upsert(dir, s, "123", "gh-issues", { status: "ingested", body: "from github" })
|
|
const items = read(dir, s).items
|
|
expect(items["slack-alpha:123"].body).toBe("from slack")
|
|
expect(items["gh-issues:123"].body).toBe("from github")
|
|
})
|
|
|
|
test("read on an absent file is NO-STATE", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
const r = run(dir, "read", "--state", s)
|
|
expect(r.code).toBe(0)
|
|
expect(status(r.stdout)).toBe("NO-STATE")
|
|
})
|
|
|
|
test("id-keyed merge preserves unknown fields (incl. nested dict + list)", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
acquire(dir, s)
|
|
upsert(dir, s, "i1", "src1", {
|
|
status: "ingested",
|
|
keep: "me",
|
|
weird: { nested: { deep: [1, 2, 3] } },
|
|
})
|
|
// A later upsert replaces only `status`; every other field must survive.
|
|
upsert(dir, s, "i1", "src1", { status: "analyzed" })
|
|
const item = read(dir, s).items["src1:i1"]
|
|
expect(item.status).toBe("analyzed")
|
|
expect(item.keep).toBe("me")
|
|
expect(item.weird).toEqual({ nested: { deep: [1, 2, 3] } })
|
|
})
|
|
|
|
test("an unknown status value is preserved, never dropped", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
acquire(dir, s)
|
|
upsert(dir, s, "i1", "src1", { status: "some_future_status" })
|
|
expect(read(dir, s).items["src1:i1"].status).toBe("some_future_status")
|
|
})
|
|
})
|
|
|
|
describe("sweep-state engine — cursors", () => {
|
|
test("cursor-advance advances, then refuses regression and unknown item", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
acquire(dir, s)
|
|
upsert(dir, s, "i1", "src1", {})
|
|
expect(status(cursorAdvance(dir, s, "src1", "200", "i1").stdout)).toBe("OK")
|
|
expect(payload(run(dir, "cursor-get", "--state", s, "--source", "src1").stdout).cursor).toBe("200")
|
|
// regression
|
|
expect(status(cursorAdvance(dir, s, "src1", "100", "i1").stdout)).toBe("REFUSED")
|
|
// unknown item id
|
|
expect(status(cursorAdvance(dir, s, "src1", "300", "nope").stdout)).toBe("REFUSED")
|
|
// cursor unchanged after both refusals
|
|
expect(payload(run(dir, "cursor-get", "--state", s, "--source", "src1").stdout).cursor).toBe("200")
|
|
})
|
|
|
|
test("cursor-advance compares pure-digit cursors numerically, not lexically", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
acquire(dir, s)
|
|
upsert(dir, s, "i1", "src1", {})
|
|
// '9' -> '10' must advance: lexical compare would wrongly REFUSE it.
|
|
cursorAdvance(dir, s, "src1", "9", "i1")
|
|
expect(status(cursorAdvance(dir, s, "src1", "10", "i1").stdout)).toBe("OK")
|
|
expect(payload(run(dir, "cursor-get", "--state", s, "--source", "src1").stdout).cursor).toBe("10")
|
|
})
|
|
|
|
test("two sequential runs against the same state never double-advance a cursor", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
// Run 1: acquire, ingest, advance, release.
|
|
acquire(dir, s, "w1")
|
|
upsert(dir, s, "i1", "src1", { status: "acknowledged" }, "w1")
|
|
cursorAdvance(dir, s, "src1", "200", "i1", "w1")
|
|
run(dir, "lease-release", "--state", s, "--writer", "w1")
|
|
// Run 2: fresh writer, re-reads state, advances forward only.
|
|
acquire(dir, s, "w2", "2026-07-02T13:00:00+00:00")
|
|
upsert(dir, s, "i2", "src1", { status: "acknowledged" }, "w2", "2026-07-02T13:00:00+00:00")
|
|
cursorAdvance(dir, s, "src1", "300", "i2", "w2", "2026-07-02T13:00:00+00:00")
|
|
// A stale re-advance from run 2 to run 1's value is refused, not doubled.
|
|
expect(status(cursorAdvance(dir, s, "src1", "200", "i2", "w2", "2026-07-02T13:00:00+00:00").stdout)).toBe("REFUSED")
|
|
expect(payload(run(dir, "cursor-get", "--state", s, "--source", "src1").stdout).cursor).toBe("300")
|
|
})
|
|
|
|
test("cursor-advance to the SAME cursor is idempotently allowed", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
acquire(dir, s)
|
|
upsert(dir, s, "i1", "src1", {})
|
|
cursorAdvance(dir, s, "src1", "200", "i1")
|
|
expect(status(cursorAdvance(dir, s, "src1", "200", "i1").stdout)).toBe("OK")
|
|
})
|
|
|
|
test("cursor-get returns null for an unknown source", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
acquire(dir, s)
|
|
expect(payload(run(dir, "cursor-get", "--state", s, "--source", "nope").stdout).cursor).toBeNull()
|
|
})
|
|
})
|
|
|
|
describe("sweep-state engine — lease", () => {
|
|
test("lease-acquire returns LOCKED against a live lease held by another writer", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
expect(status(acquire(dir, s, "w1").stdout)).toBe("OK")
|
|
const r = acquire(dir, s, "w2", "2026-07-02T12:30:00+00:00")
|
|
expect(status(r.stdout)).toBe("LOCKED")
|
|
})
|
|
|
|
test("lease-acquire reclaims a stale lease and reports the previous writer", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
acquire(dir, s, "w1", NOW) // stamped 12:00, ttl 60
|
|
const r = acquire(dir, s, "w2", "2026-07-02T14:00:00+00:00") // +2h → stale
|
|
expect(status(r.stdout)).toBe("STALE-RECLAIMED")
|
|
const pay = payload(r.stdout)
|
|
expect(pay.previous_writer).toBe("w1")
|
|
expect(pay.previous_timestamp).toBe(NOW)
|
|
expect(read(dir, s).lease.writer).toBe("w2")
|
|
})
|
|
|
|
test("re-entrant acquire by the same writer succeeds and re-stamps", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
acquire(dir, s, "w1", NOW)
|
|
const LATER = "2026-07-02T12:20:00+00:00"
|
|
expect(status(acquire(dir, s, "w1", LATER).stdout)).toBe("OK")
|
|
expect(read(dir, s).lease.timestamp).toBe(LATER)
|
|
})
|
|
|
|
test("upsert re-stamps the owning writer's lease timestamp", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
acquire(dir, s, "w1", NOW)
|
|
const LATER = "2026-07-02T12:40:00+00:00"
|
|
upsert(dir, s, "i1", "src1", { status: "ingested" }, "w1", LATER)
|
|
const lease = read(dir, s).lease
|
|
expect(lease.writer).toBe("w1")
|
|
expect(lease.timestamp).toBe(LATER)
|
|
})
|
|
|
|
test("a mutating call by a non-owner is LEASE-LOST and leaves the file unchanged", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
acquire(dir, s, "w1")
|
|
upsert(dir, s, "i1", "src1", { status: "ingested" })
|
|
const before = readFileSync(s, "utf8")
|
|
const r = upsert(dir, s, "i1", "src1", { status: "HACKED" }, "intruder")
|
|
expect(status(r.stdout)).toBe("LEASE-LOST")
|
|
expect(readFileSync(s, "utf8")).toBe(before)
|
|
})
|
|
|
|
test("lease-release clears own lease (OK), refuses another's (LEASE-LOST, no write)", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
acquire(dir, s, "w1")
|
|
const before = readFileSync(s, "utf8")
|
|
expect(status(run(dir, "lease-release", "--state", s, "--writer", "w2").stdout)).toBe("LEASE-LOST")
|
|
expect(readFileSync(s, "utf8")).toBe(before)
|
|
expect(status(run(dir, "lease-release", "--state", s, "--writer", "w1").stdout)).toBe("OK")
|
|
expect(read(dir, s).lease).toBeUndefined()
|
|
// freed → another writer can now take it
|
|
expect(status(acquire(dir, s, "w2").stdout)).toBe("OK")
|
|
})
|
|
|
|
test("lease-release on an absent file is OK (nothing to release)", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
expect(status(run(dir, "lease-release", "--state", s, "--writer", "w1").stdout)).toBe("OK")
|
|
})
|
|
})
|
|
|
|
describe("sweep-state engine — sensitivity redaction", () => {
|
|
test("upsert on a sensitive item drops body and quote, keeps other fields", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
acquire(dir, s)
|
|
upsert(dir, s, "i1", "src1", {
|
|
sensitive: true,
|
|
body: "secret",
|
|
quote: "hush",
|
|
status: "ingested",
|
|
title: "ok to keep",
|
|
})
|
|
const item = read(dir, s).items["src1:i1"]
|
|
expect(item.body).toBeUndefined()
|
|
expect(item.quote).toBeUndefined()
|
|
expect(item.sensitive).toBe(true)
|
|
expect(item.title).toBe("ok to keep")
|
|
})
|
|
|
|
test("a source flagged sensitive redacts body/quote for its items", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
// Seed a valid state whose source carries a config-derived sensitive flag.
|
|
writeFileSync(
|
|
s,
|
|
[
|
|
"schema_version: 1",
|
|
"lease:",
|
|
' writer: "w1"',
|
|
` timestamp: "${NOW}"`,
|
|
" ttl_minutes: 60",
|
|
"sources:",
|
|
' "src1":',
|
|
" sensitive: true",
|
|
"",
|
|
].join("\n"),
|
|
)
|
|
upsert(dir, s, "i1", "src1", {
|
|
status: "ingested",
|
|
body: "secret",
|
|
quote: "hush",
|
|
title: "keep",
|
|
})
|
|
const item = read(dir, s).items["src1:i1"]
|
|
expect(item.body).toBeUndefined()
|
|
expect(item.quote).toBeUndefined()
|
|
expect(item.title).toBe("keep")
|
|
})
|
|
})
|
|
|
|
describe("sweep-state engine — validate (closed-item evidence rule)", () => {
|
|
test("validate downgrades a closed item lacking evidence to fix_pending", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
acquire(dir, s)
|
|
upsert(dir, s, "i1", "src1", { status: "closed" })
|
|
const v = run(dir, "validate", "--state", s)
|
|
expect(status(v.stdout)).toBe("OK")
|
|
expect(payload(v.stdout).downgraded).toContain("src1:i1")
|
|
expect(read(dir, s).items["src1:i1"].status).toBe("fix_pending")
|
|
})
|
|
|
|
test("validate leaves a closed item WITH full evidence intact", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
acquire(dir, s)
|
|
upsert(dir, s, "i1", "src1", {
|
|
status: "closed",
|
|
fix_ref: "PR#42",
|
|
verified_merge_sha: "abc123",
|
|
verified_at: NOW,
|
|
})
|
|
const v = run(dir, "validate", "--state", s)
|
|
expect(payload(v.stdout).downgraded).toEqual([])
|
|
expect(read(dir, s).items["src1:i1"].status).toBe("closed")
|
|
})
|
|
})
|
|
|
|
describe("sweep-state engine — run-record", () => {
|
|
test("run-record records completed then failed outcomes", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
acquire(dir, s, "w1")
|
|
expect(
|
|
status(
|
|
run(dir, "run-record", "--state", s, "--writer", "w1", "--outcome", "completed", "--counts", '{"ingested":3}', "--timestamp", NOW).stdout,
|
|
),
|
|
).toBe("OK")
|
|
let lr = read(dir, s).last_run
|
|
expect(lr.outcome).toBe("completed")
|
|
expect(lr.counts).toEqual({ ingested: 3 })
|
|
expect(lr.timestamp).toBe(NOW)
|
|
|
|
const T2 = "2026-07-02T13:00:00+00:00"
|
|
expect(
|
|
status(
|
|
run(dir, "run-record", "--state", s, "--writer", "w1", "--outcome", "failed", "--counts", "{}", "--timestamp", T2).stdout,
|
|
),
|
|
).toBe("OK")
|
|
lr = read(dir, s).last_run
|
|
expect(lr.outcome).toBe("failed")
|
|
expect(lr.timestamp).toBe(T2)
|
|
})
|
|
|
|
test("run-record is lease-agnostic (aborted-locked can still be recorded)", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
acquire(dir, s, "w1") // w1 holds the lease
|
|
const r = run(dir, "run-record", "--state", s, "--writer", "w2", "--outcome", "aborted-locked", "--counts", "{}", "--timestamp", NOW)
|
|
expect(status(r.stdout)).toBe("OK")
|
|
// w1's lease is untouched
|
|
expect(read(dir, s).lease.writer).toBe("w1")
|
|
expect(read(dir, s).last_run.outcome).toBe("aborted-locked")
|
|
})
|
|
})
|
|
|
|
describe("sweep-state engine — import-legacy", () => {
|
|
test("maps Cora-style channel cursors and best-effort items", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
const legacy = path.join(dir, "legacy.json")
|
|
writeFileSync(
|
|
legacy,
|
|
JSON.stringify({
|
|
channels: {
|
|
C111: { last_processed_ts: "1699999999.000100" },
|
|
C222: { last_processed_ts: "1700000000.000200" },
|
|
C333: { note: "no ts here" }, // skipped: no cursor
|
|
},
|
|
items: {
|
|
"C111:1699999999.000100": { status: "acknowledged", channel: "C111" },
|
|
},
|
|
}),
|
|
)
|
|
const r = run(dir, "import-legacy", "--state", s, "--file", legacy)
|
|
expect(status(r.stdout)).toBe("OK")
|
|
const pay = payload(r.stdout)
|
|
expect(pay.cursors_imported).toBe(2)
|
|
expect(pay.items_imported).toBe(1)
|
|
const state = read(dir, s)
|
|
expect(state.sources.C111.cursor).toBe("1699999999.000100")
|
|
expect(state.sources.C222.cursor).toBe("1700000000.000200")
|
|
expect(state.sources.C333).toBeUndefined()
|
|
expect(state.items["C111:1699999999.000100"].source).toBe("C111")
|
|
expect(state.items["C111:1699999999.000100"].status).toBe("acknowledged")
|
|
})
|
|
|
|
test("import-legacy never fails on a garbage file", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
const legacy = path.join(dir, "legacy.json")
|
|
writeFileSync(legacy, "\x00\x01 not a legacy file at all")
|
|
const r = run(dir, "import-legacy", "--state", s, "--file", legacy)
|
|
expect(r.code).toBe(0)
|
|
expect(status(r.stdout)).toBe("OK")
|
|
expect(payload(r.stdout)).toEqual({ cursors_imported: 0, items_imported: 0 })
|
|
})
|
|
|
|
test("import-legacy --source-map lands cursors under the configured source id", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
const legacy = path.join(dir, "legacy.json")
|
|
writeFileSync(
|
|
legacy,
|
|
JSON.stringify({ channels: { C0AQ: { last_processed_ts: "1700000000.000100" } } }),
|
|
)
|
|
run(dir, "import-legacy", "--state", s, "--file", legacy, "--source-map", '{"C0AQ":"slack-alpha"}')
|
|
const state = read(dir, s)
|
|
// The connector reads cursor-get --source slack-alpha; the cursor must be there,
|
|
// not orphaned under the bare legacy channel id.
|
|
expect(state.sources["slack-alpha"].cursor).toBe("1700000000.000100")
|
|
expect(state.sources.C0AQ).toBeUndefined()
|
|
})
|
|
|
|
test("import-legacy never regresses an already-advanced cursor", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
acquire(dir, s)
|
|
upsert(dir, s, "i1", "C111", {})
|
|
// C111 has already swept forward past the legacy value.
|
|
cursorAdvance(dir, s, "C111", "1700000005.000000", "i1")
|
|
const legacy = path.join(dir, "legacy.json")
|
|
writeFileSync(
|
|
legacy,
|
|
JSON.stringify({ channels: { C111: { last_processed_ts: "1699999999.000100" } } }),
|
|
)
|
|
run(dir, "import-legacy", "--state", s, "--file", legacy)
|
|
// A re-import must not rewind the cursor and re-ingest everything since.
|
|
expect(read(dir, s).sources.C111.cursor).toBe("1700000005.000000")
|
|
})
|
|
})
|
|
|
|
describe("sweep-state engine — robustness", () => {
|
|
test("corrupt state file yields CORRUPT and exit 0 (never a traceback)", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
writeFileSync(s, "not yaml at all\n")
|
|
const r = run(dir, "read", "--state", s)
|
|
expect(r.code).toBe(0)
|
|
expect(status(r.stdout)).toBe("CORRUPT")
|
|
})
|
|
|
|
test("a mutating command on a corrupt file is CORRUPT and does not write", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
writeFileSync(s, "garbage-with-no-colon\n")
|
|
const before = readFileSync(s, "utf8")
|
|
const r = upsert(dir, s, "i1", "src1", { status: "ingested" })
|
|
expect(r.code).toBe(0)
|
|
expect(status(r.stdout)).toBe("CORRUPT")
|
|
expect(readFileSync(s, "utf8")).toBe(before)
|
|
})
|
|
|
|
test("upsert on an absent state (no lease) is LEASE-LOST", () => {
|
|
const dir = tmp()
|
|
const s = statePath(dir)
|
|
const r = upsert(dir, s, "i1", "src1", { status: "ingested" })
|
|
expect(status(r.stdout)).toBe("LEASE-LOST")
|
|
})
|
|
|
|
test("a bad subcommand exits non-zero (CLI misuse, not an operational path)", () => {
|
|
const dir = tmp()
|
|
expect(run(dir, "frobnicate").code).not.toBe(0)
|
|
expect(run(dir).code).not.toBe(0)
|
|
})
|
|
})
|