Skip to content

fix(agent): floor heartbeat attempt timeouts #4028

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kwakayama merged 5 commits into from
Aug 24, 2026
Merged

fix(agent): floor heartbeat attempt timeouts #4028

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
34857e5
fix(agent): floor heartbeat attempt timeouts
kwakayama Aug 23, 2026
0c79f37
test(agent): make heartbeat timing checks deterministic
kwakayama Aug 23, 2026
d258d50
test(agent): allow bounded heartbeat escalation
kwakayama Aug 23, 2026
90e0ed9
test(agent): harden heartbeat timeout regressions
kwakayama Aug 24, 2026
5ad3762
fix(agent): share in-flight lifecycle heartbeat
kwakayama Aug 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/api-reference/veryfront/agent.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -930,7 +930,7 @@ Input delivered to a hosted agent-service detached execution callback.
| `recoverConversationRunAppendFailure` | Recover conversation run append failure helper. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/conversation/durable.ts#L346) |
| `recoverConversationRunCursorMismatch` | Recover conversation run cursor mismatch helper. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/conversation/durable.ts#L247) |
| `registerAgent` | Registers agent. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/composition/composition.ts#L233) |
| `resolveAgentServiceRegistrationInput` | Input payload for resolve agent service registration. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/service/registration.ts#L311) |
| `resolveAgentServiceRegistrationInput` | Input payload for resolve agent service registration. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/service/registration.ts#L313) |
| `resolveConversationHostedStreamErrorState` | State for resolve conversation hosted stream error. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/conversation/hosted-terminal.ts#L69) |
| `resolveConversationHostedTerminalState` | State for resolve conversation hosted terminal. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/conversation/hosted-terminal.ts#L44) |
| `resolveConversationRunTargets` | Resolves conversation run targets. | [source](https://github.com/veryfront/veryfront-code/blob/main/src/agent/conversation/durable-contracts.ts#L34) |
Expand Down
258 changes: 157 additions & 101 deletions src/agent/service/registration.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ import "#veryfront/schemas/_test-setup.ts";
import { assert, assertEquals, assertRejects } from "#veryfront/testing/assert.ts";
import { describe, it } from "#veryfront/testing/bdd.ts";
import { waitFor } from "#veryfront/testing";
import { FakeTime } from "#std/testing/time";
import { NETWORK_ERROR } from "#veryfront/errors";
import {
type AgentServiceRegistrationLogger,
Expand Down Expand Up @@ -290,6 +291,25 @@ function lifecycleOptions(
};
}

function abortAwareDelayedJsonResponse(
body: unknown,
delayMs: number,
signal?: AbortSignal | null,
): Promise<Response> {
return new Promise<Response>((resolve, reject) => {
const timer = setTimeout(() => {
signal?.removeEventListener("abort", onAbort);
resolve(jsonResponse(body));
}, delayMs);
function onAbort() {
clearTimeout(timer);
reject(signal?.reason ?? new DOMException("Aborted", "AbortError"));
}
if (signal?.aborted) onAbort();
else signal?.addEventListener("abort", onAbort, { once: true });
});
}

describe("agent/agent-service-registration heartbeat retry", () => {
it("retries a transient 500 so a one-second blip never counts as a failure", async () => {
const script = scriptedHeartbeatFetch([500, 200]);
Expand Down Expand Up @@ -460,10 +480,11 @@ describe("agent/agent-service-registration heartbeat retry", () => {
);
});

it("times out a permanently hung heartbeat and escalates in bounded time", async () => {
it("times out a permanently hung heartbeat and escalates with fake time", async () => {
using time = new FakeTime();
const intervalMs = 20;
let heartbeatRequests = 0;
let inFlight = 0;
const activeSignals = new Set<AbortSignal>();
let maxConcurrent = 0;
const log = recordingLogger();

Expand All @@ -472,122 +493,169 @@ describe("agent/agent-service-registration heartbeat retry", () => {
return Promise.resolve(jsonResponse(serviceResponse));
}
heartbeatRequests++;
inFlight++;
maxConcurrent = Math.max(maxConcurrent, inFlight);
const signal = init?.signal;
assert(signal, "heartbeat requests must carry an abort signal");
activeSignals.add(signal);
maxConcurrent = Math.max(maxConcurrent, activeSignals.size);
return new Promise<Response>((_resolve, reject) => {
const abort = () => {
inFlight--;
reject(signal?.reason ?? new DOMException("Aborted", "AbortError"));
activeSignals.delete(signal);
reject(signal.reason ?? new DOMException("Aborted", "AbortError"));
};
if (signal?.aborted) abort();
else signal?.addEventListener("abort", abort, { once: true });
if (signal.aborted) abort();
else signal.addEventListener("abort", abort, { once: true });
});
};

const lifecycle = await createAgentServiceRegistrationLifecycle(
lifecycleOptions(fetch, { heartbeatIntervalMs: intervalMs, logger: log.logger }),
);

const escalationBudgetMs = 1_500;
const startedAt = Date.now();
try {
await waitFor(() => log.errors.length > 0, {
timeout: escalationBudgetMs,
interval: 10,
message: "hung heartbeat attempts never reached persistent-failure escalation",
await time.tickAsync(intervalMs);
for (let step = 0; step < 20 && log.errors.length === 0; step++) {
await time.tickAsync(5_010);
await time.tickAsync(0);
}
assertEquals(heartbeatRequests, 9, "three failed ticks must exhaust three attempts each");
assertEquals(
log.errors[0]?.metadata?.consecutiveFailures,
3,
"a timeout must count as one failed tick after its retries are exhausted",
);
assertEquals(maxConcurrent, 1, "timeouts must preserve the in-flight guard");

await time.tickAsync(intervalMs);
assertEquals(activeSignals.size, 1, "the next heartbeat must be in flight before teardown");
lifecycle.stop();
await time.tickAsync(0);
assertEquals(activeSignals.size, 0, "stop() must abort the hung heartbeat request");
});

it("floors a short heartbeat attempt timeout at five seconds", async () => {
using time = new FakeTime();
let aborts = 0;
const fetch: typeof globalThis.fetch = (input, init) => {
if (!input.toString().endsWith("/heartbeat")) {
return Promise.resolve(jsonResponse(serviceResponse));
}
const signal = init?.signal;
assert(signal, "heartbeat requests must carry an abort signal");
return new Promise<Response>((_resolve, reject) => {
signal.addEventListener("abort", () => {
aborts++;
reject(signal.reason);
}, { once: true });
});
assert(
Date.now() - startedAt < escalationBudgetMs,
"hung heartbeat escalation exceeded its bounded test budget",
);
assertEquals(
log.errors[0]?.metadata?.consecutiveFailures,
3,
"a timeout must count as one failed tick after its retries are exhausted",
);
assert(
heartbeatRequests >= 9,
"three failed ticks must each exhaust the three-attempt retry policy",
);
assertEquals(maxConcurrent, 1, "timeouts must preserve the in-flight guard");
} finally {
lifecycle.stop();
await waitFor(() => inFlight === 0, {
timeout: 1_000,
interval: 10,
message: "stop() did not abort the hung heartbeat request",
};
const lifecycle = await createAgentServiceRegistrationLifecycle(
lifecycleOptions(fetch, { heartbeatIntervalMs: 100 }),
);
const pending = lifecycle.heartbeat().catch(() => undefined);
await time.tickAsync(0);

await time.tickAsync(4_999);
assertEquals(aborts, 0, "a short interval must not abort before the five-second floor");
await time.tickAsync(1);
assertEquals(aborts, 1, "the first attempt must abort at the five-second floor");

lifecycle.stop();
await time.tickAsync(0);
await pending;
});

it("keeps a configured heartbeat attempt timeout above five seconds", async () => {
using time = new FakeTime();
let aborts = 0;
const fetch: typeof globalThis.fetch = (input, init) => {
if (!input.toString().endsWith("/heartbeat")) {
return Promise.resolve(jsonResponse(serviceResponse));
}
const signal = init?.signal;
assert(signal, "heartbeat requests must carry an abort signal");
return new Promise<Response>((_resolve, reject) => {
signal.addEventListener("abort", () => {
aborts++;
reject(signal.reason);
}, { once: true });
});
}
};
const lifecycle = await createAgentServiceRegistrationLifecycle(
lifecycleOptions(fetch, { heartbeatIntervalMs: 30_000 }),
);
const pending = lifecycle.heartbeat().catch(() => undefined);
await time.tickAsync(0);

await time.tickAsync(29_999);
assertEquals(aborts, 0, "a higher configured interval must remain the attempt timeout");
await time.tickAsync(1);
assertEquals(aborts, 1, "the first attempt must abort at the configured higher interval");

lifecycle.stop();
await time.tickAsync(0);
await pending;
});

it("leaves healthy and intermittently slow heartbeats alone", async () => {
// The deadline is one interval, so a heartbeat that answers inside its own
// interval must never be cut off. This is the false-positive guard: an
// eager timeout would escalate a merely slow control plane.
// The slow answer sits at 40% of the deadline, far enough inside it that a
// loaded runner cannot push it over, and far enough outside a quarter of
// the interval that an over-eager deadline would still be caught.
const intervalMs = 500;
const slowLatencyMs = 200;
let heartbeatRequests = 0;
it("lets a known-valid slow heartbeat finish with a short configured interval", async () => {
// The control plane has been seen answering in about 2.7s while degraded.
// A short heartbeat interval must not turn that known-valid latency into a
// persistent-failure escalation.
const intervalMs = 100;
const slowLatencyMs = 3_000;
let completedHeartbeats = 0;
let inFlight = 0;
const log = recordingLogger();

const fetch: typeof globalThis.fetch = (input, init) => {
if (!input.toString().endsWith("/heartbeat")) {
return Promise.resolve(jsonResponse(serviceResponse));
}
heartbeatRequests++;
// Alternate a fast answer with one that eats most of the interval.
const latencyMs = heartbeatRequests % 2 === 0 ? slowLatencyMs : 5;
const signal = init?.signal;
inFlight++;
return new Promise<Response>((resolve, reject) => {
// Honour the abort the way a real fetch does, so a deadline that fires
// early actually shows up here instead of being answered late anyway.
const timer = setTimeout(() => {
signal?.removeEventListener("abort", onAbort);
return abortAwareDelayedJsonResponse(serviceResponse, slowLatencyMs, signal).then(
(response) => {
completedHeartbeats++;
inFlight--;
resolve(jsonResponse(serviceResponse));
}, latencyMs);
function onAbort() {
clearTimeout(timer);
return response;
},
(error) => {
inFlight--;
reject(signal?.reason ?? new DOMException("Aborted", "AbortError"));
}
if (signal?.aborted) onAbort();
else signal?.addEventListener("abort", onAbort, { once: true });
});
throw error;
},
);
};

const lifecycle = await createAgentServiceRegistrationLifecycle(
lifecycleOptions(fetch, { heartbeatIntervalMs: intervalMs, logger: log.logger }),
);

await waitFor(() => heartbeatRequests >= 4, {
timeout: 10_000,
interval: 10,
message: "the heartbeat never issued enough requests to observe the deadline",
});
lifecycle.stop();
await waitFor(() => inFlight === 0, {
timeout: 10_000,
interval: 10,
message: "in-flight heartbeat requests never settled after stop()",
});

assertEquals(
log.errors.length,
0,
"a heartbeat answering inside its interval must never escalate",
);
assertEquals(
log.warnings.length,
0,
`a heartbeat answering inside its interval must not retry or skip, saw ` +
`${log.warnings.map((entry) => entry.message).join(", ")}`,
);
try {
await waitFor(() => completedHeartbeats > 0 || log.errors.length > 0, {
timeout: 10_000,
interval: 10,
message: "the slow heartbeat neither completed nor escalated",
});
assertEquals(completedHeartbeats, 1, "a known-valid slow heartbeat must complete");
assertEquals(
log.warnings.filter((entry) =>
entry.message === "Agent service heartbeat tick skipped, previous tick still running"
).length,
1,
"one slow successful heartbeat must log at most one skip warning",
);
assertEquals(
log.warnings.filter((entry) => entry.message.includes("retrying")).length,
0,
"a slow success must not be retried",
);
assertEquals(log.errors.length, 0, "a slow success must not escalate");
} finally {
lifecycle.stop();
await waitFor(() => inFlight === 0, {
timeout: 1_000,
interval: 10,
message: "stop() did not abort the slow heartbeat request",
});
}
});

it("lets a slow-but-successful heartbeat finish at the production interval", async () => {
Expand All @@ -609,20 +677,7 @@ describe("agent/agent-service-registration heartbeat retry", () => {
}
heartbeatRequests++;
const signal = init?.signal;
return new Promise<Response>((resolve, reject) => {
// Raw timer on purpose: this is the double's simulated latency, on the
// same unscaled clock as the deadline it is being measured against.
const timer = setTimeout(() => {
signal?.removeEventListener("abort", onAbort);
resolve(jsonResponse(serviceResponse));
}, slowLatencyMs);
function onAbort() {
clearTimeout(timer);
reject(signal?.reason ?? new DOMException("Aborted", "AbortError"));
}
if (signal?.aborted) onAbort();
else signal?.addEventListener("abort", onAbort, { once: true });
});
return abortAwareDelayedJsonResponse(serviceResponse, slowLatencyMs, signal);
};

const lifecycle = await createAgentServiceRegistrationLifecycle(
Expand All @@ -632,7 +687,8 @@ describe("agent/agent-service-registration heartbeat retry", () => {
// Driven directly rather than through the interval: this is about the
// deadline on one attempt, not about scheduling. Nothing here asserts on
// escalation, because escalation is counted in the interval tick and a 30s
// interval never fires inside this test. The 500ms test above covers that.
// interval never fires inside this test. The short-interval test above
// covers that.
await lifecycle.heartbeat();
lifecycle.stop();

Expand Down
Loading
Loading