mirror of
https://github.com/block/buzz.git
synced 2026-11-10 03:11:40 +00:00
5d8ede446f
## What
One-line default change: `DEFAULT_AGENT_PARALLELISM` 24 → 10
(`desktop/src-tauri/src/managed_agents/types.rs`), requested by Tyler in
[buzz-generic-acp-harnesses](buzz://message?channel=d6d84fdc-47fe-4e41-acf5-a08a4a013e6b&id=a047ead3a34ba804c376bf79f29e355d304162a88c7b2a7121e092547c572022).
## Why
Each unit of parallelism is a **full agent subprocess** per agent ×
community pair, spawned eagerly and serially at create/start before the
harness connects to the relay:
- At 24, a desktop with a few dozen agent×community pairs carries >1,000
potential idle subprocesses.
- Slow-starting harnesses (OpenClaw measured at 4–10s per worker) take
minutes to create, and messages sent during that window can be
permanently missed (the startup watermark is captured only after eager
pool init).
- Parallelism is per-channel concurrency — a channel never uses more
than one worker — so 10 still covers concurrent-channel fan-out for
typical agents.
Context: verified RCA + live 1-worker/3-mention experiment in the same
thread. Demand-driven pool growth (seed 1, grow on contention up to the
cap) is agreed as a follow-up; this is the interim risk reduction.
## Scope / blast radius
- Existing agent records keep their persisted `parallelism` value — only
new records and unset defaults change.
- Per-agent override unchanged (1..=32 via agent editor /
`BUZZ_ACP_AGENTS`).
- Readers of the constant (`team_snapshot`, `personas/snapshot/import`,
`commands/agents` mint path) all take the new default transitively; no
other code pins 24.
- `activeAgentTurnsStore.test.mjs` comments updated so the 24-turn
eviction case is framed against the harness cap (32) instead of the
default — coverage unchanged, no assertion touched.
## Verification
At 63bb0c1c4, same shell: desktop Rust `cargo test --lib` 1693/0 (14
ignored) · JS suite 3523/3523 · tsc clean · biome clean on touched file
· cargo fmt clean.
Signed-off-by: Tyler Longwell <tlongwell@block.xyz>
Co-authored-by: npub1qyvc0c5kl4gqv2fd97fsk46tu378sqgy35vc83rvgfwne90sel7s0ed67d <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz>
Co-authored-by: Tyler Longwell <tlongwell@block.xyz>
2021 lines
64 KiB
JavaScript
2021 lines
64 KiB
JavaScript
import assert from "node:assert/strict";
|
|
import { describe, it, beforeEach, afterEach, mock } from "node:test";
|
|
|
|
import {
|
|
syncAgentTurnsFromEvents,
|
|
syncActiveAgentTurnsFromObserver,
|
|
getActiveTurnsForAgent,
|
|
getActiveTurnsByChannel,
|
|
resetActiveAgentTurnsStore,
|
|
subscribeActiveAgentTurns,
|
|
saveActiveAgentTurnsForCommunity,
|
|
restoreActiveAgentTurnsForCommunity,
|
|
clearSavedCommunitySnapshot,
|
|
clearActiveTurnsForAgent,
|
|
} from "./activeAgentTurnsStore.ts";
|
|
import {
|
|
injectObserverEventsForE2E,
|
|
getAgentObserverSnapshot,
|
|
resetAgentObserverStore,
|
|
} from "./observerRelayStore.ts";
|
|
import { formatElapsed } from "./ui/agentSessionUtils.ts";
|
|
|
|
const AGENT =
|
|
"abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234abcd1234";
|
|
const AGENT_2 =
|
|
"dcba4321dcba4321dcba4321dcba4321dcba4321dcba4321dcba4321dcba4321";
|
|
|
|
/** Channel-id Set view of the summary array — keeps legacy assertions terse. */
|
|
function channelIdsOf(turns) {
|
|
return new Set(turns.map((t) => t.channelId));
|
|
}
|
|
|
|
function makeEvent(overrides) {
|
|
return {
|
|
seq: 1,
|
|
timestamp: "2024-01-01T00:00:00Z",
|
|
kind: "turn_started",
|
|
agentIndex: 0,
|
|
channelId: "chan-1",
|
|
sessionId: "sess-1",
|
|
turnId: "turn-1",
|
|
payload: null,
|
|
...overrides,
|
|
};
|
|
}
|
|
|
|
describe("activeAgentTurnsStore", () => {
|
|
beforeEach(() => {
|
|
resetActiveAgentTurnsStore();
|
|
});
|
|
|
|
describe("seq filtering", () => {
|
|
it("processes events with increasing seq", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
]);
|
|
const channels = channelIdsOf(getActiveTurnsForAgent(AGENT));
|
|
assert.equal(channels.size, 1);
|
|
assert.ok(channels.has("c1"));
|
|
});
|
|
|
|
it("skips events at or below the watermark", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 5, turnId: "t1", channelId: "c1" }),
|
|
]);
|
|
// Try to process an older event — should be ignored
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 3, turnId: "t2", channelId: "c2" }),
|
|
]);
|
|
const channels = channelIdsOf(getActiveTurnsForAgent(AGENT));
|
|
assert.equal(channels.size, 1);
|
|
assert.ok(channels.has("c1"));
|
|
assert.ok(!channels.has("c2"));
|
|
});
|
|
|
|
it("skips duplicate seq", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
]);
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t2", channelId: "c2" }),
|
|
]);
|
|
const channels = channelIdsOf(getActiveTurnsForAgent(AGENT));
|
|
assert.equal(channels.size, 1);
|
|
assert.ok(channels.has("c1"));
|
|
});
|
|
});
|
|
|
|
describe("seq restart detection", () => {
|
|
it("processes post-restart events whose timestamp climbs past the watermark", () => {
|
|
// Process events up to seq 50.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 50,
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:00Z",
|
|
}),
|
|
]);
|
|
assert.equal(getActiveTurnsForAgent(AGENT).length, 1);
|
|
|
|
// Agent restarts — seq resets to 1, but wall-clock timestamp keeps
|
|
// climbing. The composite watermark accepts it on timestamp alone.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "t2",
|
|
channelId: "c2",
|
|
timestamp: "2024-01-01T00:01:00Z",
|
|
}),
|
|
]);
|
|
const channels = channelIdsOf(getActiveTurnsForAgent(AGENT));
|
|
assert.ok(channels.has("c2"), "post-restart event should be processed");
|
|
});
|
|
|
|
it("processes subsequent events after restart", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 100,
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:00Z",
|
|
}),
|
|
]);
|
|
|
|
// Restart: seq goes 1, 2, 3 with climbing timestamps.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "t2",
|
|
channelId: "c2",
|
|
timestamp: "2024-01-01T00:01:00Z",
|
|
}),
|
|
makeEvent({
|
|
seq: 2,
|
|
turnId: "t3",
|
|
channelId: "c3",
|
|
timestamp: "2024-01-01T00:01:01Z",
|
|
}),
|
|
makeEvent({
|
|
seq: 3,
|
|
kind: "turn_completed",
|
|
turnId: "t2",
|
|
channelId: "c2",
|
|
timestamp: "2024-01-01T00:01:02Z",
|
|
}),
|
|
]);
|
|
const channels = channelIdsOf(getActiveTurnsForAgent(AGENT));
|
|
// t1 still active (not ended), t2 ended, t3 still active.
|
|
assert.ok(channels.has("c1"));
|
|
assert.ok(!channels.has("c2"));
|
|
assert.ok(channels.has("c3"));
|
|
});
|
|
});
|
|
|
|
describe("eviction at MAX_TURNS_PER_AGENT", () => {
|
|
// Mirrors the store's private cap: the harness's hard upper bound on
|
|
// parallel agent subprocesses (`--agents` accepts 1..=32).
|
|
const CAP = 32;
|
|
/** Desktop's former default harness parallelism; still a legitimate
|
|
* per-agent turn count (any value up to the harness cap of 32 is valid,
|
|
* regardless of DEFAULT_AGENT_PARALLELISM). */
|
|
const DEFAULT_PARALLELISM = 24;
|
|
const EPOCH = Date.parse("2024-01-01T00:00:00Z");
|
|
const at = (ms) => new Date(EPOCH + ms).toISOString();
|
|
|
|
/** One turn_started per channel, minute apart so start order is unambiguous. */
|
|
function startTurns(count, firstSeq = 1) {
|
|
const events = [];
|
|
for (let i = 1; i <= count; i++) {
|
|
events.push(
|
|
makeEvent({
|
|
seq: firstSeq + i - 1,
|
|
turnId: `t${i}`,
|
|
channelId: `c${i}`,
|
|
timestamp: at(i * 60_000),
|
|
}),
|
|
);
|
|
}
|
|
return events;
|
|
}
|
|
|
|
it("evicts oldest turn when exceeding the cap", () => {
|
|
syncAgentTurnsFromEvents(AGENT, startTurns(CAP + 1));
|
|
const channels = channelIdsOf(getActiveTurnsForAgent(AGENT));
|
|
// c1 (oldest) evicted to make room for the 33rd turn.
|
|
assert.equal(channels.size, CAP);
|
|
assert.ok(!channels.has("c1"), "oldest turn should be evicted");
|
|
assert.ok(channels.has("c2"));
|
|
assert.ok(channels.has(`c${CAP + 1}`));
|
|
});
|
|
|
|
it("tracks every turn of a high-parallelism agent working in 24 channels", () => {
|
|
// Parallelism up to the harness cap (32) is user-configurable, so a
|
|
// single agent can legitimately run 24 concurrent turns. Every one
|
|
// must keep its working badge.
|
|
syncAgentTurnsFromEvents(AGENT, startTurns(DEFAULT_PARALLELISM));
|
|
const channels = channelIdsOf(getActiveTurnsForAgent(AGENT));
|
|
assert.equal(
|
|
channels.size,
|
|
DEFAULT_PARALLELISM,
|
|
"all 24 concurrently-worked channels must be tracked",
|
|
);
|
|
for (let i = 1; i <= DEFAULT_PARALLELISM; i++) {
|
|
assert.ok(channels.has(`c${i}`), `c${i} must be tracked`);
|
|
}
|
|
});
|
|
|
|
it("keeps the tracked channel set stable as liveness arrives for every turn", () => {
|
|
// The flicker: with the cap below real parallelism, turns above it are
|
|
// evicted while still alive, and their 10s turn_liveness frames land on
|
|
// resurrectTurn — which evicts one of the survivors to make room. The set
|
|
// then rotates forever. Under a cap at the harness maximum, liveness for
|
|
// any live turn is a plain refresh and the set never moves.
|
|
const TURNS = 6;
|
|
syncAgentTurnsFromEvents(AGENT, startTurns(TURNS));
|
|
const expected = channelIdsOf(getActiveTurnsForAgent(AGENT));
|
|
|
|
// Liveness for the two earliest-started turns — the first to be evicted
|
|
// under the old cap, hence the first to trigger a resurrection swap.
|
|
for (const [i, turnId] of ["t1", "t2"].entries()) {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: TURNS + i + 1,
|
|
kind: "turn_liveness",
|
|
turnId,
|
|
channelId: turnId.replace("t", "c"),
|
|
timestamp: at((TURNS + i + 1) * 60_000),
|
|
}),
|
|
]);
|
|
assert.deepEqual(
|
|
[...channelIdsOf(getActiveTurnsForAgent(AGENT))].sort(),
|
|
[...expected].sort(),
|
|
`liveness for ${turnId} must not change the tracked channel set`,
|
|
);
|
|
}
|
|
assert.equal(
|
|
expected.size,
|
|
TURNS,
|
|
"all six live turns must be tracked, not a rotating subset",
|
|
);
|
|
});
|
|
});
|
|
|
|
describe("channel aggregation", () => {
|
|
it("collapses active turns by channel across agents", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "agent-1-early",
|
|
channelId: "shared",
|
|
timestamp: "2024-01-01T00:00:00Z",
|
|
}),
|
|
makeEvent({
|
|
seq: 2,
|
|
turnId: "agent-1-late",
|
|
channelId: "shared",
|
|
timestamp: "2024-01-01T00:01:00Z",
|
|
}),
|
|
]);
|
|
syncAgentTurnsFromEvents(AGENT_2, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "agent-2",
|
|
channelId: "shared",
|
|
timestamp: "2024-01-01T00:02:00Z",
|
|
}),
|
|
]);
|
|
|
|
const summaries = getActiveTurnsByChannel();
|
|
assert.deepEqual(
|
|
summaries.map(({ channelId, agentCount, agentPubkeys }) => ({
|
|
channelId,
|
|
agentCount,
|
|
agentPubkeys,
|
|
})),
|
|
[
|
|
{
|
|
channelId: "shared",
|
|
agentCount: 2,
|
|
agentPubkeys: [AGENT, AGENT_2],
|
|
},
|
|
],
|
|
);
|
|
assert.equal(
|
|
summaries[0].anchorAt,
|
|
getActiveTurnsForAgent(AGENT)[0].anchorAt,
|
|
);
|
|
});
|
|
|
|
it("removes a channel summary when the last active turn ends", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
makeEvent({
|
|
seq: 2,
|
|
kind: "turn_completed",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
}),
|
|
]);
|
|
|
|
assert.deepEqual(getActiveTurnsByChannel(), []);
|
|
});
|
|
});
|
|
|
|
describe("endTurn turnId-vs-channelId fallback", () => {
|
|
it("ends turn by turnId when provided", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
makeEvent({
|
|
seq: 2,
|
|
kind: "turn_completed",
|
|
turnId: "t1",
|
|
channelId: null,
|
|
}),
|
|
]);
|
|
assert.equal(getActiveTurnsForAgent(AGENT).length, 0);
|
|
});
|
|
|
|
it("falls back to channelId when turnId is null", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
makeEvent({
|
|
seq: 2,
|
|
kind: "turn_completed",
|
|
turnId: null,
|
|
channelId: "c1",
|
|
}),
|
|
]);
|
|
assert.equal(getActiveTurnsForAgent(AGENT).length, 0);
|
|
});
|
|
|
|
it("does nothing when both turnId and channelId are null", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
makeEvent({
|
|
seq: 2,
|
|
kind: "turn_completed",
|
|
turnId: null,
|
|
channelId: null,
|
|
}),
|
|
]);
|
|
// Turn should still be active — no way to identify which to end
|
|
assert.equal(getActiveTurnsForAgent(AGENT).length, 1);
|
|
});
|
|
|
|
it("channelId fallback removes only one matching turn", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
makeEvent({ seq: 2, turnId: "t2", channelId: "c1" }),
|
|
makeEvent({
|
|
seq: 3,
|
|
kind: "turn_completed",
|
|
turnId: null,
|
|
channelId: "c1",
|
|
}),
|
|
]);
|
|
// Only one of the two turns in c1 should be removed
|
|
const channels = channelIdsOf(getActiveTurnsForAgent(AGENT));
|
|
assert.equal(channels.size, 1);
|
|
assert.ok(channels.has("c1"));
|
|
});
|
|
|
|
it("agent_panic with an explicit turnId removes only that turn", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
makeEvent({ seq: 2, turnId: "t2", channelId: "c1" }),
|
|
makeEvent({
|
|
seq: 3,
|
|
kind: "agent_panic",
|
|
turnId: "t2",
|
|
channelId: "c1",
|
|
}),
|
|
]);
|
|
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
1,
|
|
"the explicit panic must preserve the other live turn",
|
|
);
|
|
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 4,
|
|
kind: "turn_completed",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
}),
|
|
]);
|
|
assert.equal(getActiveTurnsForAgent(AGENT).length, 0);
|
|
});
|
|
|
|
it("turn_error with an explicit turnId removes only that turn", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
makeEvent({ seq: 2, turnId: "t2", channelId: "c1" }),
|
|
makeEvent({
|
|
seq: 3,
|
|
kind: "turn_error",
|
|
turnId: "t2",
|
|
channelId: "c1",
|
|
}),
|
|
]);
|
|
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
1,
|
|
"the explicit error must preserve the other live turn",
|
|
);
|
|
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 4,
|
|
kind: "turn_completed",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
}),
|
|
]);
|
|
assert.equal(getActiveTurnsForAgent(AGENT).length, 0);
|
|
});
|
|
});
|
|
|
|
describe("listener notifications", () => {
|
|
it("notifies on turn_started", () => {
|
|
let called = 0;
|
|
const unsub = subscribeActiveAgentTurns(() => {
|
|
called++;
|
|
});
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
]);
|
|
assert.ok(called > 0);
|
|
unsub();
|
|
});
|
|
|
|
it("notifies on turn_completed", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
]);
|
|
let called = 0;
|
|
const unsub = subscribeActiveAgentTurns(() => {
|
|
called++;
|
|
});
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 2, kind: "turn_completed", turnId: "t1" }),
|
|
]);
|
|
assert.ok(called > 0);
|
|
unsub();
|
|
});
|
|
});
|
|
|
|
describe("replay idempotency", () => {
|
|
it("replaying the same buffer produces no additional state change or notifications", () => {
|
|
const buffer = [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:00Z",
|
|
}),
|
|
makeEvent({
|
|
seq: 2,
|
|
turnId: "t2",
|
|
channelId: "c2",
|
|
timestamp: "2024-01-01T00:00:01Z",
|
|
}),
|
|
];
|
|
|
|
// Initial pass.
|
|
syncAgentTurnsFromEvents(AGENT, buffer);
|
|
const afterFirst = getActiveTurnsForAgent(AGENT);
|
|
assert.equal(afterFirst.length, 2);
|
|
|
|
// Subscribe, then replay the identical buffer.
|
|
let notified = 0;
|
|
const unsub = subscribeActiveAgentTurns(() => {
|
|
notified++;
|
|
});
|
|
syncAgentTurnsFromEvents(AGENT, buffer);
|
|
unsub();
|
|
|
|
assert.equal(notified, 0, "replay must not notify listeners");
|
|
const afterReplay = getActiveTurnsForAgent(AGENT);
|
|
assert.equal(
|
|
afterReplay,
|
|
afterFirst,
|
|
"replay must not change turn state (stable reference)",
|
|
);
|
|
});
|
|
|
|
it("post-restart replay does not reprocess seen events or resurrect evicted turns", () => {
|
|
// Start a turn, then complete it (turn evicted).
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:00Z",
|
|
}),
|
|
makeEvent({
|
|
seq: 2,
|
|
kind: "turn_completed",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:01Z",
|
|
}),
|
|
]);
|
|
assert.equal(getActiveTurnsForAgent(AGENT).length, 0);
|
|
|
|
// Agent restarts. The harness replays its buffer with seq reset to 1,
|
|
// but the original event timestamps (older than the watermark) are
|
|
// unchanged. The start event must NOT resurrect the evicted turn.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:00Z",
|
|
}),
|
|
makeEvent({
|
|
seq: 2,
|
|
kind: "turn_completed",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:01Z",
|
|
}),
|
|
]);
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
0,
|
|
"stale replayed start must not resurrect an evicted turn",
|
|
);
|
|
});
|
|
});
|
|
|
|
describe("replayed eviction safety", () => {
|
|
it("replayed stale turn_error with null turnId does not kill the live turn", () => {
|
|
// A turn errors out (harness emits turn_error with a null turnId), then a
|
|
// fresh turn starts in the same channel.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:00Z",
|
|
}),
|
|
makeEvent({
|
|
seq: 2,
|
|
kind: "turn_error",
|
|
turnId: null,
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:01Z",
|
|
}),
|
|
makeEvent({
|
|
seq: 3,
|
|
turnId: "t2",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:02Z",
|
|
}),
|
|
]);
|
|
assert.equal(getActiveTurnsForAgent(AGENT).length, 1);
|
|
|
|
// The full buffer is replayed on the next observer event. The stale
|
|
// turn_error (below the watermark) must NOT re-run its channel-match
|
|
// fallback and delete the live turn t2.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:00Z",
|
|
}),
|
|
makeEvent({
|
|
seq: 2,
|
|
kind: "turn_error",
|
|
turnId: null,
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:01Z",
|
|
}),
|
|
makeEvent({
|
|
seq: 3,
|
|
turnId: "t2",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:02Z",
|
|
}),
|
|
]);
|
|
const channels = channelIdsOf(getActiveTurnsForAgent(AGENT));
|
|
assert.equal(
|
|
channels.size,
|
|
1,
|
|
"replayed stale turn_error must not delete the live turn",
|
|
);
|
|
assert.ok(channels.has("c1"));
|
|
});
|
|
|
|
it("replaying evictions fires no spurious listener notifications", () => {
|
|
const buffer = [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:00Z",
|
|
}),
|
|
makeEvent({
|
|
seq: 2,
|
|
kind: "turn_error",
|
|
turnId: null,
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:01Z",
|
|
}),
|
|
makeEvent({
|
|
seq: 3,
|
|
kind: "agent_panic",
|
|
turnId: null,
|
|
channelId: "c2",
|
|
timestamp: "2024-01-01T00:00:02Z",
|
|
}),
|
|
];
|
|
|
|
// Initial pass processes the buffer.
|
|
syncAgentTurnsFromEvents(AGENT, buffer);
|
|
|
|
// Subscribe, then replay the identical buffer. Every event is below the
|
|
// watermark, so the replay must be a complete no-op.
|
|
let notified = 0;
|
|
const unsub = subscribeActiveAgentTurns(() => {
|
|
notified++;
|
|
});
|
|
syncAgentTurnsFromEvents(AGENT, buffer);
|
|
unsub();
|
|
|
|
assert.equal(notified, 0, "replayed evictions must not notify listeners");
|
|
});
|
|
});
|
|
|
|
describe("getActiveTurnsForAgent", () => {
|
|
it("returns empty array for null/undefined pubkey", () => {
|
|
assert.equal(getActiveTurnsForAgent(null).length, 0);
|
|
assert.equal(getActiveTurnsForAgent(undefined).length, 0);
|
|
});
|
|
|
|
it("returns stable reference when unchanged", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
]);
|
|
const ref1 = getActiveTurnsForAgent(AGENT);
|
|
const ref2 = getActiveTurnsForAgent(AGENT);
|
|
assert.equal(ref1, ref2, "should return cached array reference");
|
|
});
|
|
|
|
it("anchors a turn to its skew-corrected start, not the local insert clock", () => {
|
|
// The badge anchor must reflect the agent's true start translated into
|
|
// desktop time (startedAt + clock offset), so a turn whose event arrives
|
|
// with a stale timestamp does NOT reset to ~Date.now(). With a single
|
|
// event the offset is exactly Date.now() - startedAt, so the anchor lands
|
|
// on Date.now() here — the regression coverage for skew lives below.
|
|
const before = Date.now();
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: "2000-01-01T00:00:00Z",
|
|
}),
|
|
]);
|
|
const after = Date.now();
|
|
const [summary] = getActiveTurnsForAgent(AGENT);
|
|
assert.equal(summary.channelId, "c1");
|
|
assert.ok(
|
|
summary.anchorAt >= before && summary.anchorAt <= after,
|
|
"anchorAt must be the skew-corrected start, here equal to the local clock",
|
|
);
|
|
});
|
|
|
|
it("gives two turns with different startedAt different anchors (no lockstep)", () => {
|
|
// The lockstep bug: turns processed in the same JS tick were all anchored
|
|
// to one shared Date.now(), so their elapsed counters ticked in unison.
|
|
// Anchoring to startedAt + offset makes distinct agent-host starts produce
|
|
// distinct anchors. A single sampleClockOffset minimum is shared, so the
|
|
// anchor difference equals the startedAt difference.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "t1",
|
|
channelId: "c-early",
|
|
timestamp: "2024-01-01T00:00:00Z",
|
|
}),
|
|
makeEvent({
|
|
seq: 2,
|
|
turnId: "t2",
|
|
channelId: "c-late",
|
|
timestamp: "2024-01-01T00:05:00Z",
|
|
}),
|
|
]);
|
|
const byChannel = new Map(
|
|
getActiveTurnsForAgent(AGENT).map((s) => [s.channelId, s.anchorAt]),
|
|
);
|
|
assert.notEqual(
|
|
byChannel.get("c-early"),
|
|
byChannel.get("c-late"),
|
|
"distinct startedAt must yield distinct anchors",
|
|
);
|
|
assert.equal(
|
|
byChannel.get("c-late") - byChannel.get("c-early"),
|
|
5 * 60_000,
|
|
"anchor spacing must equal the agent-host start spacing",
|
|
);
|
|
});
|
|
|
|
it("collapses two turns in one channel to the earliest anchor", () => {
|
|
// Same agent-host start timestamp, distinct turns (seq bumped so the
|
|
// second passes the watermark). Identical timestamps mean the offset does
|
|
// not move, so the surfaced anchor is stable and the earliest wins.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:00Z",
|
|
}),
|
|
]);
|
|
const firstAnchor = getActiveTurnsForAgent(AGENT)[0].anchorAt;
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 2,
|
|
turnId: "t2",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:00Z",
|
|
}),
|
|
]);
|
|
const summaries = getActiveTurnsForAgent(AGENT);
|
|
assert.equal(summaries.length, 1, "same channel collapses to one entry");
|
|
assert.equal(
|
|
summaries[0].anchorAt,
|
|
firstAnchor,
|
|
"earliest start's anchor must be surfaced",
|
|
);
|
|
});
|
|
|
|
it("advances to the surviving turn's anchor after the earliest ends", () => {
|
|
// Two turns in one channel; the array must be rebuilt from the LIVE map
|
|
// on every mutation, so ending the earliest-started turn must surface the
|
|
// survivor's (later) anchor — not a stale cached minimum.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "t-early",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:00Z",
|
|
}),
|
|
makeEvent({
|
|
seq: 2,
|
|
turnId: "t-later",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:02:00Z",
|
|
}),
|
|
]);
|
|
const tEarly = getActiveTurnsForAgent(AGENT)[0].anchorAt;
|
|
|
|
// End the earliest turn by its turnId. Reuse t-later's timestamp (seq
|
|
// bumped to pass the watermark) so the offset does not tighten and the
|
|
// surviving anchor's advance is exactly the 2-minute start gap.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 3,
|
|
kind: "turn_completed",
|
|
turnId: "t-early",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:02:00Z",
|
|
}),
|
|
]);
|
|
const [survivor] = getActiveTurnsForAgent(AGENT);
|
|
assert.equal(survivor.channelId, "c1");
|
|
assert.equal(
|
|
survivor.anchorAt - tEarly,
|
|
2 * 60_000,
|
|
"surfaced anchor must advance to the surviving turn after eviction",
|
|
);
|
|
});
|
|
|
|
it("sorts summaries by channelId", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c-zebra" }),
|
|
makeEvent({ seq: 2, turnId: "t2", channelId: "c-alpha" }),
|
|
]);
|
|
const ids = getActiveTurnsForAgent(AGENT).map((s) => s.channelId);
|
|
assert.deepEqual(ids, ["c-alpha", "c-zebra"]);
|
|
});
|
|
});
|
|
|
|
describe("turn_liveness prune backstop", () => {
|
|
// The prune sweep runs on an internal setInterval keyed off Date.now();
|
|
// faking both lets us drive the 25s bound deterministically. The fixed
|
|
// epoch is the clock floor — event timestamps below anchor lastActivityAt
|
|
// to it, so elapsed time is exactly what mock.timers.tick advances.
|
|
const EPOCH = Date.parse("2024-01-01T00:00:00Z");
|
|
const at = (ms) => new Date(EPOCH + ms).toISOString();
|
|
// Mirrors the store's private timing constants. Keep these consumer-level
|
|
// tests deterministic without exporting implementation details.
|
|
const PRUNE_INTERVAL_MS = 5_000;
|
|
const PRUNE_PAUSE_MAX_MS = 3 * 60_000;
|
|
|
|
let unsubscribe;
|
|
|
|
beforeEach(() => {
|
|
mock.timers.enable({ apis: ["setInterval", "Date"], now: EPOCH });
|
|
// Subscribing starts the prune interval under the faked clock.
|
|
unsubscribe = subscribeActiveAgentTurns(() => {});
|
|
});
|
|
|
|
afterEach(() => {
|
|
unsubscribe();
|
|
mock.timers.reset();
|
|
});
|
|
|
|
it("keeps a turn alive when turn_liveness refreshes before the bound", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
|
|
]);
|
|
|
|
// Refresh activity at 20s — under the 25s bound — then advance to 40s.
|
|
// Without the refresh the turn would have been pruned by 25s; the
|
|
// liveness ping resets lastActivityAt so it survives.
|
|
mock.timers.tick(20_000);
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 2,
|
|
kind: "turn_liveness",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: at(20_000),
|
|
}),
|
|
]);
|
|
mock.timers.tick(20_000);
|
|
|
|
const channels = channelIdsOf(getActiveTurnsForAgent(AGENT));
|
|
assert.ok(
|
|
channels.has("c1"),
|
|
"liveness within the bound must defer the prune",
|
|
);
|
|
});
|
|
|
|
it("prunes a stale turn at the bound when its tracked sibling stays fresh", () => {
|
|
// A same-agent tracked sibling keeps the agent's max lastActivityAt
|
|
// fresh, so the stale turn is genuinely dead and must still prune at 25s.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "dead",
|
|
channelId: "c1",
|
|
timestamp: at(0),
|
|
}),
|
|
makeEvent({
|
|
seq: 2,
|
|
turnId: "live",
|
|
channelId: "c2",
|
|
timestamp: at(0),
|
|
}),
|
|
]);
|
|
assert.equal(getActiveTurnsForAgent(AGENT).length, 2);
|
|
|
|
// Keep the live turn fresh across the dead turn's bound: ping every 10s.
|
|
for (let t = 10_000; t <= 30_000; t += 10_000) {
|
|
mock.timers.tick(10_000);
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 2 + t / 10_000,
|
|
kind: "turn_liveness",
|
|
turnId: "live",
|
|
channelId: "c2",
|
|
timestamp: at(t),
|
|
}),
|
|
]);
|
|
}
|
|
|
|
const channels = channelIdsOf(getActiveTurnsForAgent(AGENT));
|
|
assert.ok(!channels.has("c1"), "the dead turn must prune at the bound");
|
|
assert.ok(channels.has("c2"), "the live sibling must survive");
|
|
});
|
|
|
|
it("pauses an agent's stale turns while another agent stays fresh", () => {
|
|
// Agent B keeps reporting tracked-turn activity, but that must not cause
|
|
// agent A's fully stale stream to prune at 25s.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "stale",
|
|
channelId: "a",
|
|
timestamp: at(0),
|
|
}),
|
|
]);
|
|
syncAgentTurnsFromEvents(AGENT_2, [
|
|
makeEvent({ seq: 1, turnId: "live", channelId: "b", timestamp: at(0) }),
|
|
]);
|
|
|
|
for (let t = 10_000; t <= 30_000; t += 10_000) {
|
|
mock.timers.tick(10_000);
|
|
syncAgentTurnsFromEvents(AGENT_2, [
|
|
makeEvent({
|
|
seq: t / 10_000 + 1,
|
|
kind: "turn_liveness",
|
|
turnId: "live",
|
|
channelId: "b",
|
|
timestamp: at(t),
|
|
}),
|
|
]);
|
|
}
|
|
|
|
assert.ok(
|
|
channelIdsOf(getActiveTurnsForAgent(AGENT)).has("a"),
|
|
"a fully stale agent must stay visible while another agent remains fresh",
|
|
);
|
|
|
|
// Keep B's tracked turn fresh until A passes the bounded 3-minute pause.
|
|
for (let t = 40_000; t <= PRUNE_PAUSE_MAX_MS; t += 10_000) {
|
|
mock.timers.tick(10_000);
|
|
syncAgentTurnsFromEvents(AGENT_2, [
|
|
makeEvent({
|
|
seq: t / 10_000 + 1,
|
|
kind: "turn_liveness",
|
|
turnId: "live",
|
|
channelId: "b",
|
|
timestamp: at(t),
|
|
}),
|
|
]);
|
|
}
|
|
mock.timers.tick(PRUNE_INTERVAL_MS);
|
|
|
|
assert.ok(
|
|
!channelIdsOf(getActiveTurnsForAgent(AGENT)).has("a"),
|
|
"the stale agent must prune after the 3-minute pause cap",
|
|
);
|
|
assert.ok(
|
|
channelIdsOf(getActiveTurnsForAgent(AGENT_2)).has("b"),
|
|
"the fresh agent must stay active",
|
|
);
|
|
});
|
|
|
|
it("prunes a lone silent turn after the 3-minute pause cap", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
|
|
]);
|
|
|
|
mock.timers.tick(PRUNE_PAUSE_MAX_MS);
|
|
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
0,
|
|
"the pause backstop must clear a dead lone agent at the 3-minute cap",
|
|
);
|
|
});
|
|
|
|
it("treats a turn_liveness with a null turnId as a no-op", () => {
|
|
// A null-turnId liveness must refresh NOTHING. With a live sibling
|
|
// keeping the max fresh, the dead turn still prunes at the bound — so if
|
|
// the null ping wrongly refreshed the dead turn it would survive here.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "dead",
|
|
channelId: "c1",
|
|
timestamp: at(0),
|
|
}),
|
|
makeEvent({
|
|
seq: 2,
|
|
turnId: "live",
|
|
channelId: "c2",
|
|
timestamp: at(0),
|
|
}),
|
|
]);
|
|
|
|
// A null-turnId liveness for the dead turn must not refresh it. Keep the
|
|
// live sibling pinging so the pause never engages.
|
|
assert.doesNotThrow(() => {
|
|
for (let t = 10_000; t <= 30_000; t += 10_000) {
|
|
mock.timers.tick(10_000);
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 100 + t,
|
|
kind: "turn_liveness",
|
|
turnId: null,
|
|
channelId: "c1",
|
|
timestamp: at(t),
|
|
}),
|
|
makeEvent({
|
|
seq: 200 + t,
|
|
kind: "turn_liveness",
|
|
turnId: "live",
|
|
channelId: "c2",
|
|
timestamp: at(t),
|
|
}),
|
|
]);
|
|
}
|
|
});
|
|
|
|
const channels = channelIdsOf(getActiveTurnsForAgent(AGENT));
|
|
assert.ok(
|
|
!channels.has("c1"),
|
|
"a null-turnId liveness must not refresh the dead turn, so it prunes",
|
|
);
|
|
assert.ok(channels.has("c2"), "the live sibling must survive");
|
|
});
|
|
});
|
|
|
|
describe("skew-corrected elapsed (real-time arrival)", () => {
|
|
// The clock offset estimate (running minimum of Date.now() - event time)
|
|
// is only meaningful when events arrive at distinct real times — exactly
|
|
// how the harness streams them. Faking Date lets us advance the desktop
|
|
// clock between events so an earlier event calibrates the offset before the
|
|
// measured turn starts. The fixed epoch is the desktop clock floor.
|
|
const EPOCH = Date.parse("2024-06-01T00:00:00Z");
|
|
|
|
beforeEach(() => {
|
|
mock.timers.enable({ apis: ["Date"], now: EPOCH });
|
|
});
|
|
|
|
afterEach(() => {
|
|
mock.timers.reset();
|
|
});
|
|
|
|
/** Agent-host clock = desktop clock + skew, as an ISO timestamp. */
|
|
const agentTs = (desktopMs, skew) =>
|
|
new Date(desktopMs + skew).toISOString();
|
|
|
|
it("shows a large elapsed for a turn that started well in the past", () => {
|
|
// Clocks synced (skew 0). An early event at the true present calibrates
|
|
// offset ≈ 0. Five true minutes pass. Then the desktop first observes a
|
|
// turn whose start timestamp is that 5-minutes-ago instant — the badge
|
|
// must read ~5 minutes, not reset to 0s on first sight.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
kind: "turn_liveness",
|
|
turnId: "warm",
|
|
channelId: "c0",
|
|
timestamp: agentTs(EPOCH - 1_000, 0),
|
|
}),
|
|
]);
|
|
mock.timers.tick(5 * 60_000); // 5 true minutes elapse
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 2,
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: agentTs(EPOCH, 0), // started 5 minutes ago
|
|
}),
|
|
]);
|
|
const summary = getActiveTurnsForAgent(AGENT).find(
|
|
(s) => s.channelId === "c1",
|
|
);
|
|
assert.equal(
|
|
Date.now() - summary.anchorAt,
|
|
5 * 60_000 - 1_000,
|
|
"a 5-minute-old turn must show ~5 minutes elapsed, not 0s",
|
|
);
|
|
});
|
|
|
|
it("corrects for agent-host clock skew so elapsed tracks true duration", () => {
|
|
// Agent host is 1 hour AHEAD of the desktop. A liveness event received at
|
|
// the true present (desktop EPOCH) carries a timestamp an hour in the
|
|
// future, calibrating offset ≈ -1h. The turn then starts 30s later in
|
|
// true time; its future-stamped start, corrected by the offset, anchors
|
|
// to the true start — without correction elapsed would be deeply negative.
|
|
const SKEW = 60 * 60_000;
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
kind: "turn_liveness",
|
|
turnId: "warm",
|
|
channelId: "c0",
|
|
timestamp: agentTs(EPOCH, SKEW),
|
|
}),
|
|
]);
|
|
mock.timers.tick(30_000); // 30s of true time passes
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 2,
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: agentTs(EPOCH + 30_000, SKEW),
|
|
}),
|
|
]);
|
|
const summary = getActiveTurnsForAgent(AGENT).find(
|
|
(s) => s.channelId === "c1",
|
|
);
|
|
assert.equal(
|
|
Date.now() - summary.anchorAt,
|
|
0,
|
|
"a just-started turn under heavy skew must read ~0s, not a negative/huge value",
|
|
);
|
|
|
|
// Let the turn run 45s; elapsed must track that true duration exactly.
|
|
mock.timers.tick(45_000);
|
|
const stillRunning = getActiveTurnsForAgent(AGENT).find(
|
|
(s) => s.channelId === "c1",
|
|
);
|
|
assert.equal(
|
|
Date.now() - stillRunning.anchorAt,
|
|
45_000,
|
|
"skew-corrected elapsed must track true duration as the clock advances",
|
|
);
|
|
});
|
|
|
|
it("retroactively corrects a live turn's anchor when the offset tightens", () => {
|
|
// The design's load-bearing invariant: anchors are derived at READ time,
|
|
// so a later, tighter offset must shift an ALREADY-LIVE turn earlier.
|
|
// The turn first goes live under a loose offset (its start arrives with a
|
|
// +5s processing delay → offset +5000), then a delay-free liveness sample
|
|
// tightens the running minimum to 0. The live turn's surfaced anchor must
|
|
// move earlier by exactly that 5000ms delta. A regression that froze
|
|
// anchorAt at startTurn would leave the anchor at its loose value and
|
|
// fail this assertion.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: agentTs(EPOCH - 5_000, 0), // observed 5s after its start
|
|
}),
|
|
]);
|
|
const looseAnchor = getActiveTurnsForAgent(AGENT).find(
|
|
(s) => s.channelId === "c1",
|
|
).anchorAt;
|
|
|
|
mock.timers.tick(1_000); // 1s of true time so the liveness arrives later
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 2,
|
|
kind: "turn_liveness",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: agentTs(EPOCH + 1_000, 0), // delay-free → offset tightens to 0
|
|
}),
|
|
]);
|
|
const tightAnchor = getActiveTurnsForAgent(AGENT).find(
|
|
(s) => s.channelId === "c1",
|
|
).anchorAt;
|
|
|
|
assert.equal(
|
|
tightAnchor - looseAnchor,
|
|
-5_000,
|
|
"a tighter offset must shift the live turn's read-time anchor earlier by the tightening delta",
|
|
);
|
|
});
|
|
});
|
|
|
|
describe("resurrection after a prune (A) gated by completion (C)", () => {
|
|
const EPOCH = Date.parse("2024-01-01T00:00:00Z");
|
|
const at = (ms) => new Date(EPOCH + ms).toISOString();
|
|
const PRUNE_INTERVAL_MS = 5_000;
|
|
|
|
let unsubscribe;
|
|
|
|
beforeEach(() => {
|
|
mock.timers.enable({ apis: ["setInterval", "Date"], now: EPOCH });
|
|
unsubscribe = subscribeActiveAgentTurns(() => {});
|
|
});
|
|
|
|
afterEach(() => {
|
|
unsubscribe();
|
|
mock.timers.reset();
|
|
});
|
|
|
|
it("resurrects a pruned turn at startedAt from the first recovered ACP frame", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
|
|
]);
|
|
|
|
// A fresh tracked sibling permits pruning stale t1 at the normal bound.
|
|
mock.timers.tick(30_000);
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 2,
|
|
turnId: "t2",
|
|
channelId: "c2",
|
|
timestamp: at(30_000),
|
|
}),
|
|
]);
|
|
mock.timers.tick(PRUNE_INTERVAL_MS);
|
|
assert.ok(
|
|
!channelIdsOf(getActiveTurnsForAgent(AGENT)).has("c1"),
|
|
"the stale turn must prune before its activity recovers",
|
|
);
|
|
|
|
mock.timers.tick(10_000);
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 3,
|
|
kind: "acp_read",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: at(45_000),
|
|
startedAt: at(0),
|
|
}),
|
|
]);
|
|
|
|
const resurrected = getActiveTurnsForAgent(AGENT).find(
|
|
(turn) => turn.channelId === "c1",
|
|
);
|
|
assert.equal(
|
|
Date.now() - resurrected.anchorAt,
|
|
45_000,
|
|
"the first recovered ACP frame must preserve elapsed time from the original turn start",
|
|
);
|
|
});
|
|
|
|
it("preserves a valid Unix-epoch startedAt envelope timestamp", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
kind: "turn_liveness",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: at(0),
|
|
startedAt: "1970-01-01T00:00:00.000Z",
|
|
}),
|
|
]);
|
|
|
|
const [resurrected] = getActiveTurnsForAgent(AGENT);
|
|
assert.equal(
|
|
resurrected.anchorAt,
|
|
0,
|
|
"a valid zero timestamp must not be replaced with the recovery time",
|
|
);
|
|
});
|
|
|
|
it("falls back to the frame timestamp when startedAt is absent or invalid", () => {
|
|
for (const invalidStartedAt of [
|
|
undefined,
|
|
null,
|
|
"not-a-timestamp",
|
|
"future",
|
|
]) {
|
|
resetActiveAgentTurnsStore();
|
|
const startedAtMs = Date.now();
|
|
const iso = (offset) => new Date(startedAtMs + offset).toISOString();
|
|
const startedAt =
|
|
invalidStartedAt === "future" ? iso(46_000) : invalidStartedAt;
|
|
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: iso(0),
|
|
}),
|
|
]);
|
|
mock.timers.tick(30_000);
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 2,
|
|
turnId: "t2",
|
|
channelId: "c2",
|
|
timestamp: iso(30_000),
|
|
}),
|
|
]);
|
|
mock.timers.tick(PRUNE_INTERVAL_MS);
|
|
mock.timers.tick(10_000);
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 3,
|
|
kind: "turn_liveness",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: iso(45_000),
|
|
startedAt,
|
|
}),
|
|
]);
|
|
|
|
const resurrected = getActiveTurnsForAgent(AGENT).find(
|
|
(turn) => turn.channelId === "c1",
|
|
);
|
|
assert.equal(
|
|
Date.now() - resurrected.anchorAt,
|
|
0,
|
|
"old or malformed frames must retain the existing recovery anchor",
|
|
);
|
|
}
|
|
});
|
|
|
|
it("does NOT resurrect a turn whose liveness is older than its completion", () => {
|
|
// Bound-proving (stale side): a turn completes, then a liveness frame
|
|
// arrives carrying a timestamp BEFORE the completion. It must not revive.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
|
|
makeEvent({
|
|
seq: 2,
|
|
kind: "turn_completed",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: at(10_000),
|
|
}),
|
|
]);
|
|
assert.equal(getActiveTurnsForAgent(AGENT).length, 0);
|
|
|
|
// A liveness stamped at 5s (before the 10s completion) but delivered with
|
|
// a later seq so it clears the watermark on seq. It is stale relative to
|
|
// the completion and must NOT resurrect.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 3,
|
|
kind: "turn_liveness",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: at(5_000),
|
|
}),
|
|
]);
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
0,
|
|
"a liveness older than the recorded completion must not resurrect the turn",
|
|
);
|
|
});
|
|
|
|
it("DOES resurrect a turn whose liveness is strictly newer than its completion", () => {
|
|
// Bound-proving (live side): the same completed turn, but a liveness frame
|
|
// strictly NEWER than the completion (a genuine restart of the same id)
|
|
// must revive — the completion only blocks stale frames, not new work.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
|
|
makeEvent({
|
|
seq: 2,
|
|
kind: "turn_completed",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: at(10_000),
|
|
}),
|
|
]);
|
|
assert.equal(getActiveTurnsForAgent(AGENT).length, 0);
|
|
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 3,
|
|
kind: "turn_liveness",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: at(20_000),
|
|
}),
|
|
]);
|
|
assert.ok(
|
|
channelIdsOf(getActiveTurnsForAgent(AGENT)).has("c1"),
|
|
"a liveness strictly newer than the completion must resurrect the turn",
|
|
);
|
|
});
|
|
|
|
it("does NOT resurrect from a liveness frame with no channelId", () => {
|
|
// A pruned turn cannot be rebuilt without a channelId to anchor the badge.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
|
|
makeEvent({
|
|
seq: 2,
|
|
turnId: "t2",
|
|
channelId: "c2",
|
|
timestamp: at(0),
|
|
}),
|
|
]);
|
|
// Drop t1 by ending it, then send a channelId-less liveness for it.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 3,
|
|
kind: "turn_completed",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: at(5_000),
|
|
}),
|
|
makeEvent({
|
|
seq: 4,
|
|
kind: "turn_liveness",
|
|
turnId: "t1",
|
|
channelId: null,
|
|
timestamp: at(10_000),
|
|
}),
|
|
]);
|
|
assert.ok(
|
|
!channelIdsOf(getActiveTurnsForAgent(AGENT)).has("c1"),
|
|
"a channelId-less liveness cannot resurrect a badge",
|
|
);
|
|
});
|
|
|
|
it("clears completion tombstones on reset so a later turn can run", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
|
|
makeEvent({
|
|
seq: 2,
|
|
kind: "turn_completed",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: at(10_000),
|
|
}),
|
|
]);
|
|
|
|
resetActiveAgentTurnsStore();
|
|
|
|
// After reset, an OLD-stamped liveness for the same id must resurrect,
|
|
// proving the tombstone (which would otherwise block it) was cleared.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
kind: "turn_liveness",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: at(1_000),
|
|
}),
|
|
]);
|
|
assert.ok(
|
|
channelIdsOf(getActiveTurnsForAgent(AGENT)).has("c1"),
|
|
"reset must clear terminal tombstones so they do not leak across reset",
|
|
);
|
|
});
|
|
|
|
it("evicts the oldest tombstone once past the cap so the map stays bounded", () => {
|
|
// The tombstone map is capped at MAX_TERMINAL_TOMBSTONES (MAX_TURNS_PER_AGENT
|
|
// * 4 = 128). Complete 130 distinct turns so eviction fires twice, dropping
|
|
// the two oldest by insertion order (t0, t1). Probe via the ONE behavior a
|
|
// tombstone gates that a strictly-newer frame cannot mask: an
|
|
// EQUAL-timestamp liveness (frameAt == terminalAt). All completions share
|
|
// timestamp T with rising seq, so the probe clears the per-agent watermark
|
|
// on the seq tiebreak (compareObserverEvents is timestamp-primary,
|
|
// seq-secondary) yet stays equal to the recorded terminal — reaching
|
|
// resurrectTurn's tombstone check rather than being shadowed by the
|
|
// watermark.
|
|
const CAP = 128;
|
|
const TOTAL = CAP + 2;
|
|
const T = at(0);
|
|
const completions = [];
|
|
for (let i = 0; i < TOTAL; i++) {
|
|
completions.push(
|
|
makeEvent({
|
|
seq: i + 1,
|
|
kind: "turn_completed",
|
|
turnId: `t${i}`,
|
|
channelId: `c${i}`,
|
|
timestamp: T,
|
|
}),
|
|
);
|
|
}
|
|
syncAgentTurnsFromEvents(AGENT, completions);
|
|
|
|
// A surviving tombstone (t2, third-completed) still blocks an
|
|
// equal-timestamp liveness — proves the tombstone is present and doing
|
|
// the work the watermark cannot.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: TOTAL + 1,
|
|
kind: "turn_liveness",
|
|
turnId: "t2",
|
|
channelId: "c2",
|
|
timestamp: T,
|
|
}),
|
|
]);
|
|
assert.ok(
|
|
!channelIdsOf(getActiveTurnsForAgent(AGENT)).has("c2"),
|
|
"a surviving tombstone must still block an equal-timestamp liveness",
|
|
);
|
|
|
|
// The oldest tombstone (t0) was evicted, so the same equal-timestamp
|
|
// liveness now resurrects — proving the cap fired AND evicted the
|
|
// oldest-by-insertion entry, not an arbitrary one.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: TOTAL + 2,
|
|
kind: "turn_liveness",
|
|
turnId: "t0",
|
|
channelId: "c0",
|
|
timestamp: T,
|
|
}),
|
|
]);
|
|
assert.ok(
|
|
channelIdsOf(getActiveTurnsForAgent(AGENT)).has("c0"),
|
|
"the oldest tombstone must be evicted once past the cap",
|
|
);
|
|
});
|
|
});
|
|
});
|
|
|
|
/**
|
|
* Regression: raw observer ingestion and derived active-turn liveness must
|
|
* stay in sync. The channel activity path previously mounted only the observer
|
|
* bridge, so a raw `turn_completed` could be visible in the activity feed
|
|
* while the derived liveness indicator kept spinning. This drives events
|
|
* through the real observer store (appendAgentEvent path) and the same sync
|
|
* the bridge hook runs, asserting derived liveness clears with the raw feed.
|
|
*/
|
|
describe("observer → active-turns bridge sync", () => {
|
|
const bridgeAgents = [{ pubkey: AGENT, status: "deployed" }];
|
|
|
|
beforeEach(() => {
|
|
resetActiveAgentTurnsStore();
|
|
resetAgentObserverStore();
|
|
});
|
|
|
|
afterEach(() => {
|
|
resetAgentObserverStore();
|
|
});
|
|
|
|
it("clears derived liveness when raw turn_completed arrives", () => {
|
|
injectObserverEventsForE2E(AGENT, [
|
|
makeEvent({ seq: 1, kind: "turn_started" }),
|
|
]);
|
|
syncActiveAgentTurnsFromObserver(bridgeAgents);
|
|
assert.ok(
|
|
channelIdsOf(getActiveTurnsForAgent(AGENT)).has("chan-1"),
|
|
"turn_started must surface an active turn",
|
|
);
|
|
|
|
injectObserverEventsForE2E(AGENT, [
|
|
makeEvent({
|
|
seq: 2,
|
|
kind: "turn_completed",
|
|
timestamp: "2024-01-01T00:00:05Z",
|
|
}),
|
|
]);
|
|
syncActiveAgentTurnsFromObserver(bridgeAgents);
|
|
|
|
const rawEvents = getAgentObserverSnapshot(AGENT, true).events;
|
|
assert.equal(
|
|
rawEvents.at(-1)?.kind,
|
|
"turn_completed",
|
|
"raw feed must contain the completion event",
|
|
);
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
0,
|
|
"derived liveness must clear when the raw feed shows turn_completed",
|
|
);
|
|
});
|
|
|
|
it("skips agents that are neither running nor deployed", () => {
|
|
injectObserverEventsForE2E(AGENT, [
|
|
makeEvent({ seq: 1, kind: "turn_started" }),
|
|
]);
|
|
syncActiveAgentTurnsFromObserver([{ pubkey: AGENT, status: "stopped" }]);
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
0,
|
|
"inactive agents must not populate the active-turns store",
|
|
);
|
|
});
|
|
});
|
|
|
|
describe("formatElapsed", () => {
|
|
it("renders sub-10s as whole seconds", () => {
|
|
assert.equal(formatElapsed(0), "0s");
|
|
assert.equal(formatElapsed(4_900), "4s");
|
|
});
|
|
|
|
it("renders sub-minute as whole seconds", () => {
|
|
assert.equal(formatElapsed(59_000), "59s");
|
|
});
|
|
|
|
it("rolls into minutes at exactly 60s", () => {
|
|
assert.equal(formatElapsed(60_000), "1m 0s");
|
|
});
|
|
|
|
it("renders minutes and seconds", () => {
|
|
assert.equal(formatElapsed(83_000), "1m 23s");
|
|
});
|
|
|
|
it("rolls 59m 59s cleanly into 1h 0m 0s at 3600s", () => {
|
|
assert.equal(formatElapsed(3_599_000), "59m 59s");
|
|
assert.equal(formatElapsed(3_600_000), "1h 0m 0s");
|
|
});
|
|
|
|
it("clamps negative input to 0s", () => {
|
|
assert.equal(formatElapsed(-5_000), "0s");
|
|
});
|
|
});
|
|
|
|
describe("community-switch save / restore", () => {
|
|
beforeEach(() => {
|
|
resetActiveAgentTurnsStore();
|
|
});
|
|
|
|
it("restores original startedAt timestamps across a round-trip", () => {
|
|
// Simulate a turn active in community A with a known start timestamp.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:00Z",
|
|
}),
|
|
]);
|
|
const anchorBefore = getActiveTurnsForAgent(AGENT)[0].anchorAt;
|
|
|
|
// Switch away (save A, reset, apply B).
|
|
saveActiveAgentTurnsForCommunity("ws-a");
|
|
resetActiveAgentTurnsStore();
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
0,
|
|
"store must be empty after reset",
|
|
);
|
|
|
|
// Switch back (restore A).
|
|
restoreActiveAgentTurnsForCommunity("ws-a");
|
|
const turns = getActiveTurnsForAgent(AGENT);
|
|
assert.equal(turns.length, 1, "restored turn must reappear");
|
|
assert.equal(turns[0].channelId, "c1");
|
|
assert.equal(
|
|
turns[0].anchorAt,
|
|
anchorBefore,
|
|
"anchorAt must equal the pre-switch value — startedAt and offset are preserved",
|
|
);
|
|
});
|
|
|
|
it("no-op restore when no snapshot exists for the community", () => {
|
|
// Populate the store so we can verify nothing changes.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
]);
|
|
let notified = 0;
|
|
const unsub = subscribeActiveAgentTurns(() => {
|
|
notified++;
|
|
});
|
|
restoreActiveAgentTurnsForCommunity("ws-unknown");
|
|
unsub();
|
|
assert.equal(notified, 0, "no-op restore must not notify listeners");
|
|
// Store should be unchanged — still has the turn we set up above.
|
|
assert.equal(getActiveTurnsForAgent(AGENT).length, 1);
|
|
});
|
|
|
|
it("empty store discards a prior snapshot rather than saving an empty one", () => {
|
|
// First round-trip: save something.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
]);
|
|
saveActiveAgentTurnsForCommunity("ws-a");
|
|
|
|
// Second pass: the turns ended while we were away, so the store is empty
|
|
// when we try to save again.
|
|
resetActiveAgentTurnsStore();
|
|
saveActiveAgentTurnsForCommunity("ws-a"); // empty store → delete snapshot
|
|
|
|
// Restore must be a no-op now.
|
|
restoreActiveAgentTurnsForCommunity("ws-a");
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
0,
|
|
"restoring after an empty save must yield no turns",
|
|
);
|
|
});
|
|
|
|
it("snapshot is consumed on restore — a second restore is a no-op", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
]);
|
|
saveActiveAgentTurnsForCommunity("ws-a");
|
|
resetActiveAgentTurnsStore();
|
|
|
|
restoreActiveAgentTurnsForCommunity("ws-a");
|
|
assert.equal(getActiveTurnsForAgent(AGENT).length, 1, "first restore ok");
|
|
|
|
// Second restore — snapshot was consumed.
|
|
resetActiveAgentTurnsStore();
|
|
restoreActiveAgentTurnsForCommunity("ws-a");
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
0,
|
|
"second restore must be no-op — snapshot is consumed on first use",
|
|
);
|
|
});
|
|
|
|
it("watermark is preserved — events already processed before save are not reprocessed", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 5,
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:00Z",
|
|
}),
|
|
]);
|
|
saveActiveAgentTurnsForCommunity("ws-a");
|
|
resetActiveAgentTurnsStore();
|
|
restoreActiveAgentTurnsForCommunity("ws-a");
|
|
|
|
// Replaying an event at or below the watermark must be a no-op.
|
|
let notified = 0;
|
|
const unsub = subscribeActiveAgentTurns(() => {
|
|
notified++;
|
|
});
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 3,
|
|
turnId: "t2",
|
|
channelId: "c2",
|
|
timestamp: "2024-01-01T00:00:00Z",
|
|
}),
|
|
]);
|
|
unsub();
|
|
|
|
assert.equal(notified, 0, "stale event after restore must not notify");
|
|
const channels = new Set(
|
|
getActiveTurnsForAgent(AGENT).map((s) => s.channelId),
|
|
);
|
|
assert.ok(!channels.has("c2"), "stale event must not add a turn");
|
|
});
|
|
|
|
it("terminal tombstones are preserved — a stale liveness cannot resurrect a completed turn", () => {
|
|
// Complete a turn before saving.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 1,
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:00Z",
|
|
}),
|
|
makeEvent({
|
|
seq: 2,
|
|
kind: "turn_completed",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:10Z",
|
|
}),
|
|
]);
|
|
saveActiveAgentTurnsForCommunity("ws-a");
|
|
resetActiveAgentTurnsStore();
|
|
restoreActiveAgentTurnsForCommunity("ws-a");
|
|
|
|
// A liveness stamped before the completion must not resurrect t1.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 3,
|
|
kind: "turn_liveness",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: "2024-01-01T00:00:05Z",
|
|
}),
|
|
]);
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
0,
|
|
"terminal tombstone must survive save/restore — stale liveness must not resurrect",
|
|
);
|
|
});
|
|
|
|
it("notifies listeners after restore so UI re-renders with recovered turns", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
]);
|
|
saveActiveAgentTurnsForCommunity("ws-a");
|
|
resetActiveAgentTurnsStore();
|
|
|
|
let notified = 0;
|
|
const unsub = subscribeActiveAgentTurns(() => {
|
|
notified++;
|
|
});
|
|
restoreActiveAgentTurnsForCommunity("ws-a");
|
|
unsub();
|
|
|
|
assert.ok(notified > 0, "restore must notify listeners");
|
|
});
|
|
|
|
it("multiple communities maintain independent snapshots", () => {
|
|
// Community A: agent in c1.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
]);
|
|
saveActiveAgentTurnsForCommunity("ws-a");
|
|
|
|
// Community B: different agent/channel.
|
|
resetActiveAgentTurnsStore();
|
|
syncAgentTurnsFromEvents(AGENT_2, [
|
|
makeEvent({ seq: 1, turnId: "t2", channelId: "c2" }),
|
|
]);
|
|
saveActiveAgentTurnsForCommunity("ws-b");
|
|
|
|
// Switch to A — only A's turns must appear.
|
|
resetActiveAgentTurnsStore();
|
|
restoreActiveAgentTurnsForCommunity("ws-a");
|
|
const aChannels = new Set(
|
|
getActiveTurnsForAgent(AGENT).map((s) => s.channelId),
|
|
);
|
|
assert.ok(aChannels.has("c1"), "ws-a must restore c1");
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT_2).length,
|
|
0,
|
|
"ws-b turns must not appear in ws-a restore",
|
|
);
|
|
|
|
// Switch to B — only B's turns.
|
|
resetActiveAgentTurnsStore();
|
|
restoreActiveAgentTurnsForCommunity("ws-b");
|
|
const bChannels = new Set(
|
|
getActiveTurnsForAgent(AGENT_2).map((s) => s.channelId),
|
|
);
|
|
assert.ok(bChannels.has("c2"), "ws-b must restore c2");
|
|
});
|
|
|
|
it("clearSavedCommunitySnapshot discards the snapshot so restore is a no-op", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
]);
|
|
saveActiveAgentTurnsForCommunity("ws-a");
|
|
|
|
// Simulate community deletion: GC the snapshot before switching back.
|
|
clearSavedCommunitySnapshot("ws-a");
|
|
|
|
resetActiveAgentTurnsStore();
|
|
restoreActiveAgentTurnsForCommunity("ws-a");
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
0,
|
|
"restore must be no-op after clearSavedCommunitySnapshot",
|
|
);
|
|
});
|
|
});
|
|
|
|
describe("clearActiveTurnsForAgent", () => {
|
|
const EPOCH = Date.parse("2024-01-01T00:00:00Z");
|
|
const at = (ms) => new Date(EPOCH + ms).toISOString();
|
|
|
|
beforeEach(() => {
|
|
resetActiveAgentTurnsStore();
|
|
});
|
|
|
|
it("clear removes the agent turns and notifies subscribers; other agents untouched", () => {
|
|
// Give AGENT two turns and AGENT_2 one turn.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1" }),
|
|
makeEvent({ seq: 2, turnId: "t2", channelId: "c2" }),
|
|
]);
|
|
syncAgentTurnsFromEvents(AGENT_2, [
|
|
makeEvent({ seq: 1, turnId: "t3", channelId: "c3" }),
|
|
]);
|
|
|
|
let notified = 0;
|
|
const unsub = subscribeActiveAgentTurns(() => {
|
|
notified++;
|
|
});
|
|
clearActiveTurnsForAgent(AGENT);
|
|
unsub();
|
|
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
0,
|
|
"cleared agent must have no turns",
|
|
);
|
|
assert.equal(notified, 1, "must notify listeners exactly once");
|
|
|
|
// AGENT_2 is unaffected.
|
|
const a2channels = channelIdsOf(getActiveTurnsForAgent(AGENT_2));
|
|
assert.ok(a2channels.has("c3"), "other agent's turns must survive clear");
|
|
});
|
|
|
|
it("full-buffer replay after clear is a no-op (watermark preserved — badge stays gone)", () => {
|
|
// Process initial events to set the watermark at seq 2.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
|
|
makeEvent({
|
|
seq: 2,
|
|
turnId: "t2",
|
|
channelId: "c2",
|
|
timestamp: at(1_000),
|
|
}),
|
|
]);
|
|
clearActiveTurnsForAgent(AGENT);
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
0,
|
|
"should be empty after clear",
|
|
);
|
|
|
|
// Replay the identical buffer — every event is at or below the watermark.
|
|
let notified = 0;
|
|
const unsub = subscribeActiveAgentTurns(() => {
|
|
notified++;
|
|
});
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
|
|
makeEvent({
|
|
seq: 2,
|
|
turnId: "t2",
|
|
channelId: "c2",
|
|
timestamp: at(1_000),
|
|
}),
|
|
]);
|
|
unsub();
|
|
|
|
assert.equal(
|
|
notified,
|
|
0,
|
|
"replay must not notify — watermark must be preserved",
|
|
);
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
0,
|
|
"badge must stay gone",
|
|
);
|
|
});
|
|
|
|
it("late turn_liveness frame with timestamp ≤ clear time does not resurrect (tombstone)", () => {
|
|
mock.timers.enable({ apis: ["Date"], now: EPOCH });
|
|
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
|
|
]);
|
|
|
|
// Clear at EPOCH (t=0 in agent-host clock).
|
|
clearActiveTurnsForAgent(AGENT);
|
|
assert.equal(getActiveTurnsForAgent(AGENT).length, 0);
|
|
|
|
// A liveness frame whose timestamp is at or before the clear time must not
|
|
// resurrect the badge (tombstone blocks it). Advance seq past the
|
|
// watermark by using a higher seq than the initial turn_started.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 2,
|
|
kind: "turn_liveness",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: at(0), // equal to clear time — must NOT resurrect
|
|
}),
|
|
]);
|
|
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
0,
|
|
"liveness at or before clear time must not resurrect the cleared turn",
|
|
);
|
|
|
|
mock.timers.reset();
|
|
});
|
|
|
|
it("new turn_started after clear (restart picked up new work) is tracked normally", () => {
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
|
|
]);
|
|
clearActiveTurnsForAgent(AGENT);
|
|
assert.equal(getActiveTurnsForAgent(AGENT).length, 0);
|
|
|
|
// A genuinely new turn arrives after the clear with a later timestamp and
|
|
// a new turnId — it must be tracked normally.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 2,
|
|
turnId: "t2",
|
|
channelId: "c1",
|
|
timestamp: at(5_000), // strictly newer than the cleared turn's timestamp
|
|
}),
|
|
]);
|
|
|
|
const turns = getActiveTurnsForAgent(AGENT);
|
|
assert.equal(turns.length, 1, "new turn after clear must be tracked");
|
|
assert.ok(channelIdsOf(turns).has("c1"), "new turn must surface c1");
|
|
});
|
|
|
|
it("badge is gone when stop succeeds even if start subsequently fails (stop-boundary clear)", () => {
|
|
// Arrange: agent has an active turn.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
|
|
]);
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
1,
|
|
"turn must be active before stop",
|
|
);
|
|
|
|
// Act: simulate what onStopped does — clear at the stop-success boundary,
|
|
// before start is called. Start fails (not called here).
|
|
clearActiveTurnsForAgent(AGENT);
|
|
|
|
// Assert: the badge is gone regardless of what happens to start.
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
0,
|
|
"badge must clear at stop-success boundary, not waiting for start to resolve",
|
|
);
|
|
});
|
|
|
|
it("new frame arriving while start is pending does not resurrect the cleared badge (tombstone boundary)", () => {
|
|
// Simulate: agent was active, stop succeeded and clear ran (onStopped
|
|
// fired), start is now in-flight. A stale liveness frame for the OLD
|
|
// turn arrives on the wire during the start-pending window. It must NOT
|
|
// resurrect the badge — the clear tombstoned it.
|
|
mock.timers.enable({ apis: ["Date"], now: EPOCH });
|
|
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({ seq: 1, turnId: "t1", channelId: "c1", timestamp: at(0) }),
|
|
]);
|
|
|
|
// onStopped fires: clear at stop boundary (agent-host clock = EPOCH).
|
|
clearActiveTurnsForAgent(AGENT);
|
|
|
|
// Stale liveness for t1 arrives with timestamp ≤ clear time (on-wire
|
|
// frame from before the kill). Must be blocked by the tombstone.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 2,
|
|
kind: "turn_liveness",
|
|
turnId: "t1",
|
|
channelId: "c1",
|
|
timestamp: at(0),
|
|
}),
|
|
]);
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
0,
|
|
"stale liveness during start-pending must not resurrect the cleared badge",
|
|
);
|
|
|
|
// Genuine new turn from the restarted agent arrives later with a new id
|
|
// and strictly newer timestamp — must be tracked normally.
|
|
syncAgentTurnsFromEvents(AGENT, [
|
|
makeEvent({
|
|
seq: 3,
|
|
turnId: "t2-new",
|
|
channelId: "c1",
|
|
timestamp: at(3_000),
|
|
}),
|
|
]);
|
|
assert.equal(
|
|
getActiveTurnsForAgent(AGENT).length,
|
|
1,
|
|
"genuine new turn from restarted agent must be tracked",
|
|
);
|
|
|
|
mock.timers.reset();
|
|
});
|
|
});
|