Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
13 changes: 13 additions & 0 deletions packages/command-registry/src/timeout-policy.ts
Original file line number Diff line number Diff line change
Expand Up @@ -103,6 +103,19 @@ export function resolveCommandRequestTimeoutMs(
);
}

/**
* How long the daemon may work on a request before its client envelope expires, less the margin
* that lets the daemon's own result win that race. For a command with an enforced daemon deadline
* this is that deadline.
*/
export function resolveCommandDaemonBudgetMs(
policy: CommandTimeoutPolicy,
input: RequestTimeoutInput,
): number | undefined {
const envelopeMs = resolveCommandRequestTimeoutMs(policy, input);
return envelopeMs === undefined ? undefined : envelopeMs - REQUEST_TIMEOUT_BUDGET_MARGIN_MS;
}

/** Resolves an enforced daemon request deadline, when the command declares one. */
export function resolveCommandDaemonTimeoutMs(
policy: CommandTimeoutPolicy,
Expand Down
6 changes: 6 additions & 0 deletions packages/contracts/src/application-lifecycle-runtime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,12 @@ export type ApplicationLifecycleExecution = Readonly<{
* means the platform's default boot wait; `prepare` derives its own deadline from `timeoutMs`.
*/
startupDeadlineAtMs?: number;
/**
* Absolute time by which the request must have its result: its enforced daemon deadline, or its
* admission plus what its client envelope leaves the daemon. Absent for a request no transport
* admitted.
*/
requestDeadlineAtMs?: number;
requestId?: string;
logPath?: string;
traceLogPath?: string;
Expand Down
19 changes: 18 additions & 1 deletion packages/daemon/src/core/daemon-command-registry.ts
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,12 @@ import {
resolveCommandDeviceClaimPolicy,
resolveCommandTimeoutPolicy,
} from '@agent-device/command-registry/registry';
import { resolveCommandDaemonTimeoutMs } from '@agent-device/command-registry/timeout-policy';
import {
resolveCommandDaemonBudgetMs,
resolveCommandDaemonTimeoutMs,
} from '@agent-device/command-registry/timeout-policy';
import type { RefFrameEffect } from '@agent-device/contracts/replay';
import type { DaemonWireRequest } from '@agent-device/contracts/command';
import type { DaemonRequest } from './daemon-request.ts';

export type { DaemonCommandDescriptor, DaemonCommandRoute, SessionCommandKind };
Expand All @@ -37,6 +41,19 @@ export function resolveDaemonRequestTimeoutMs(req: DaemonRequest): number | unde
return resolveCommandDaemonTimeoutMs(resolveCommandTimeoutPolicy(req.command), req);
}

/**
* When a request admitted now must have its result: the client envelope starts as the transport
* admits the request, and the daemon keeps the envelope's result margin. `undefined` for a command
* whose client waits without an envelope.
*/
export function resolveDaemonRequestResultDeadlineMs(
req: DaemonWireRequest,
admittedAtMs = Date.now(),
): number | undefined {
const budgetMs = resolveCommandDaemonBudgetMs(resolveCommandTimeoutPolicy(req.command), req);
return budgetMs === undefined ? undefined : admittedAtMs + budgetMs;
}

export function getDaemonCommandRoute(command: string): DaemonCommandRoute {
return getDaemonCommandDescriptor(command)?.route ?? 'generic';
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,8 @@ import { test } from 'vitest';
import assert from 'node:assert/strict';
import { getRequestSignal } from '@agent-device/host-kit/request';
import { closeLoopbackServer, skipWhenLoopbackUnavailable } from '@agent-device/test-kit';
import { postRpcCommand, startRpcServer } from './http-server-rpc.fixtures.ts';
import { postRpcCommand, postRpcRaw, startRpcServer } from './http-server-rpc.fixtures.ts';
import { applicationLifecycleExecutionFromRequest } from '../../sessions/application-lifecycle-execution.ts';

const TOKEN = 'lifetime-token';

Expand Down Expand Up @@ -33,3 +34,31 @@ test('an rpc request registers its tracking id while in flight and releases it o
await closeLoopbackServer(server);
}
});

test('an admitted open records a result deadline of its client envelope less the daemon result margin', async (t) => {
if (await skipWhenLoopbackUnavailable(t)) return;
let deadlineAtMs: number | undefined;
const { port, server } = await startRpcServer(TOKEN, async (request) => {
deadlineAtMs = applicationLifecycleExecutionFromRequest(
request,
'/tmp/request.log',
).requestDeadlineAtMs;
return { ok: true, data: {} } satisfies DaemonResponse;
});
try {
const sentAtMs = Date.now();
const requestId = 'req-open-result-deadline';
const body = JSON.stringify({
jsonrpc: '2.0',
id: requestId,
method: 'agent_device.command',
params: { command: 'open', positionals: ['com.example.app'], meta: { requestId } },
});
assert.equal((await postRpcRaw(port, TOKEN, body)).status, 200);
const answeredAtMs = Date.now();
assert.ok(deadlineAtMs !== undefined, 'the open must carry a deadline');
assert.ok(deadlineAtMs >= sentAtMs + 60_000 && deadlineAtMs <= answeredAtMs + 60_000);
} finally {
await closeLoopbackServer(server);
}
});
5 changes: 4 additions & 1 deletion packages/daemon/src/server/http/http-server.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ import { normalizeError } from '@agent-device/kernel/errors';
import { LEASE_BACKENDS } from '@agent-device/kernel/contracts';
import { isRecord } from '@agent-device/kernel/record';
import type { DaemonInvokeFn, DaemonRequest } from '../../core/daemon-request.ts';
import { resolveDaemonRequestResultDeadlineMs } from '../../core/daemon-command-registry.ts';
import { emitDiagnostic } from '@agent-device/host-kit/diagnostics';
import {
clearRequestAbortRegistration,
Expand Down Expand Up @@ -472,7 +473,9 @@ function beginRpcRequestLifetime(params: {
return {
admit(daemonRequest) {
requestId = resolveRequestTrackingId(daemonRequest.meta?.requestId, rpcId);
abortRegistration = registerRequestAbort(requestId);
abortRegistration = registerRequestAbort(requestId, undefined, {
resultDeadlineMs: resolveDaemonRequestResultDeadlineMs(daemonRequest),
});
return {
...daemonRequest,
meta: { ...daemonRequest.meta, requestId },
Expand Down
94 changes: 92 additions & 2 deletions packages/daemon/src/server/http/transport.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,12 @@ import http from 'node:http';
import net from 'node:net';
import { test } from 'vitest';
import { createSocketServer, listenHttpServer, listenNetServer } from './transport.ts';
import { getRequestSignal } from '@agent-device/host-kit/request';
import {
clearRequestAbortRegistration,
getRequestSignal,
registerRequestAbort,
} from '@agent-device/host-kit/request';
import { applicationLifecycleExecutionFromRequest } from '../../sessions/application-lifecycle-execution.ts';
import { closeLoopbackServer, skipWhenLoopbackUnavailable } from '@agent-device/test-kit';

test('Provider-backed integration daemon socket transport frames requests and normalizes malformed input', async (t) => {
Expand Down Expand Up @@ -160,6 +165,90 @@ test('socket disconnect cancels only the disconnected connection in-flight reque
}
});

/** The deadline an `open` running under `requestId` would carry. */
function openDeadlineAtMs(requestId: string | undefined): number | undefined {
return applicationLifecycleExecutionFromRequest(
{
token: 'provider-scenario-token',
session: 'default',
command: 'open',
positionals: ['com.example.app'],
meta: { requestId },
},
'/tmp/request.log',
).requestDeadlineAtMs;
}

test('a socket-admitted open --timeout moves its deadline with the envelope it widens', async (t) => {
if (await skipWhenLoopbackUnavailable(t)) return;
let deadlineAtMs: number | undefined;
const server = createSocketServer(async (req): Promise<DaemonResponse> => {
deadlineAtMs = applicationLifecycleExecutionFromRequest(
req,
'/tmp/request.log',
).requestDeadlineAtMs;
return { ok: true, data: {} };
});
try {
const client = await connectSocket(await listenNetServer(server));
const sentAtMs = Date.now();
await writeSocketRequests(
client,
[
JSON.stringify({
token: 'provider-scenario-token',
session: 'default',
command: 'open',
positionals: ['com.example.app'],
flags: { timeoutMs: 120_000 },
meta: { requestId: 'socket-open-widened-deadline' },
}),
],
1,
);
const answeredAtMs = Date.now();
client.destroy();
assert.ok(deadlineAtMs !== undefined, 'the open must carry a deadline');
assert.ok(deadlineAtMs >= sentAtMs + 120_000 && deadlineAtMs <= answeredAtMs + 120_000);
} finally {
await closeLoopbackServer(server);
}
});

test('an attempt nested under a test request without a client envelope has no deadline', async (t) => {
if (await skipWhenLoopbackUnavailable(t)) return;
let attemptDeadlineAtMs: number | undefined = Number.NaN;
const server = createSocketServer(async (req): Promise<DaemonResponse> => {
const registration = registerRequestAbort('socket-test-attempt', req.meta?.requestId);
try {
attemptDeadlineAtMs = openDeadlineAtMs('socket-test-attempt');
} finally {
clearRequestAbortRegistration(registration);
}
return { ok: true, data: {} };
});
try {
const client = await connectSocket(await listenNetServer(server));
await writeSocketRequests(
client,
[
JSON.stringify({
token: 'provider-scenario-token',
session: 'default',
command: 'test',
positionals: ['flows'],
meta: { requestId: 'socket-test-suite' },
}),
],
1,
);
client.destroy();
assert.equal(attemptDeadlineAtMs, undefined);
} finally {
await closeLoopbackServer(server);
}
});

async function waitForAbort(signal: AbortSignal): Promise<void> {
if (signal.aborted) return;
await new Promise<void>((resolve) => {
Expand All @@ -181,6 +270,7 @@ async function connectSocket(port: number): Promise<net.Socket> {
async function writeSocketRequests(
socket: net.Socket,
lines: string[],
expectedResponses = 2,
): Promise<Array<{ ok?: boolean; data?: any; error?: any }>> {
return await new Promise((resolve, reject) => {
const responses: Array<{ ok?: boolean; data?: any; error?: any }> = [];
Expand All @@ -195,7 +285,7 @@ async function writeSocketRequests(
if (line) {
responses.push(JSON.parse(line));
}
if (responses.length === 2) {
if (responses.length === expectedResponses) {
resolve(responses);
return;
}
Expand Down
5 changes: 4 additions & 1 deletion packages/daemon/src/server/http/transport.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import { AsyncResource } from 'node:async_hooks';
import net from 'node:net';
import type { Server as HttpServer } from 'node:http';
import type { DaemonInvokeFn, DaemonRequest } from '../../core/daemon-request.ts';
import { resolveDaemonRequestResultDeadlineMs } from '../../core/daemon-command-registry.ts';
import {
clearRequestAbortRegistration,
isRequestCanceled,
Expand Down Expand Up @@ -76,7 +77,9 @@ export function createSocketServer(handleRequest: DaemonInvokeFn): DaemonServer
...req.meta,
requestId: requestIdForCleanup,
};
requestAbortRegistration = registerRequestAbort(requestIdForCleanup);
requestAbortRegistration = registerRequestAbort(requestIdForCleanup, undefined, {
resultDeadlineMs: resolveDaemonRequestResultDeadlineMs(req),
});
activeRequestIds.add(requestIdForCleanup);
if (isRequestCanceled(requestIdForCleanup)) {
throw createRequestCanceledError();
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,39 @@
import { afterEach, expect, test, vi } from 'vitest';
import { withRequestDeadline } from '@agent-device/host-kit/request';
import type { ApplicationLifecycleExecution } from '@agent-device/contracts/application-lifecycle-runtime';
import { applicationLifecycleExecutionFromRequest } from './application-lifecycle-execution.ts';

const NOW_MS = 1_000_000;

afterEach(() => {
vi.useRealTimers();
});

function openExecution(requestId: string | undefined): ApplicationLifecycleExecution {
return applicationLifecycleExecutionFromRequest(
{
token: 'token',
session: 'default',
command: 'open',
positionals: ['com.example.app'],
meta: { requestId },
},
'/tmp/request.log',
);
}

test('an enforced request deadline is the deadline', async () => {
vi.useFakeTimers({ toFake: ['Date'] });
vi.setSystemTime(NOW_MS);
let execution: ApplicationLifecycleExecution | undefined;
await withRequestDeadline('open-in-bounded-batch', 20_000, async () => {
execution = openExecution('open-in-bounded-batch');
return { ok: true, data: {} };
});
expect(execution?.requestDeadlineAtMs).toBe(NOW_MS + 20_000);
});

test('a request no transport admitted has no deadline', () => {
expect(openExecution(undefined).requestDeadlineAtMs).toBeUndefined();
expect(openExecution('never-admitted').requestDeadlineAtMs).toBeUndefined();
});
14 changes: 14 additions & 0 deletions packages/daemon/src/sessions/application-lifecycle-execution.ts
Original file line number Diff line number Diff line change
@@ -1,3 +1,7 @@
import {
getRequestExecutionDeadlineMs,
getRequestResultDeadlineMs,
} from '@agent-device/host-kit/request';
import { resolvePlannedOperations } from '../core/execution-plan.ts';
import type { ApplicationLifecycleExecution } from '@agent-device/contracts/application-lifecycle-runtime';
import { resolveRunnerLogicalLeaseContext } from '../core/lease-context.ts';
Expand All @@ -10,6 +14,7 @@ export function applicationLifecycleExecutionFromRequest(
traceLogPath?: string,
): ApplicationLifecycleExecution {
return {
requestDeadlineAtMs: requestDeadlineAtMs(req),
requestId: req.meta?.requestId,
logPath,
traceLogPath,
Expand All @@ -25,3 +30,12 @@ export function applicationLifecycleExecutionFromRequest(
plannedOperations: resolvePlannedOperations(req.internal?.executionPlan),
};
}

/**
* An enforced daemon deadline is the request's deadline. Otherwise it is the result deadline the
* transport that admitted the request (or its parent) set, ahead of the client's envelope.
*/
function requestDeadlineAtMs(req: DaemonRequest): number | undefined {
const requestId = req.meta?.requestId;
return getRequestExecutionDeadlineMs(requestId) ?? getRequestResultDeadlineMs(requestId);
}
24 changes: 20 additions & 4 deletions packages/host-kit/src/internal/request-cancel.ts
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,15 @@ type RequestCancelState = {
requestAbortControllers: Map<string, RequestAbortController>;
};

type RequestAbortController = AbortController & { executionDeadlineMs?: number };
type RequestAbortController = AbortController & {
executionDeadlineMs?: number;
/**
* The daemon's own result deadline, which the transport that admitted the request set ahead of
* the client's: the end of the client envelope less the daemon's result margin. An attempt
* registered under a parent inherits the parent's, because it runs inside the parent's envelope.
*/
resultDeadlineMs?: number;
};

const REQUEST_CANCEL_SLOT = Symbol('requestCancel');
const requestCancelState = (): RequestCancelState =>
Expand Down Expand Up @@ -80,6 +88,7 @@ function evictOldestSetEntries<V>(set: Set<V>): void {
export function registerRequestAbort(
requestId: string | undefined,
parentRequestId?: string,
admission?: Readonly<{ resultDeadlineMs?: number }>,
): RequestAbortRegistration | undefined {
if (!requestId) return undefined;
if (requestAbortControllers().has(requestId)) {
Expand All @@ -90,9 +99,11 @@ export function registerRequestAbort(
}
evictOldestEntries(requestAbortControllers());
const controller: RequestAbortController = new AbortController();
controller.executionDeadlineMs = parentRequestId
? requestAbortControllers().get(parentRequestId)?.executionDeadlineMs
: undefined;
const parent = parentRequestId ? requestAbortControllers().get(parentRequestId) : undefined;
controller.executionDeadlineMs = parent?.executionDeadlineMs;
controller.resultDeadlineMs = parentRequestId
? parent?.resultDeadlineMs
: admission?.resultDeadlineMs;
requestAbortControllers().set(requestId, controller);
if (canceledRequestIds().has(requestId)) {
controller.abort(createRequestCanceledError());
Expand Down Expand Up @@ -154,6 +165,11 @@ export function getRequestExecutionDeadlineMs(requestId: string | undefined): nu
return requestId ? requestAbortControllers().get(requestId)?.executionDeadlineMs : undefined;
}

/** The daemon's result deadline the admitting transport set for the request, if it set one. */
export function getRequestResultDeadlineMs(requestId: string | undefined): number | undefined {
return requestId ? requestAbortControllers().get(requestId)?.resultDeadlineMs : undefined;
}

export function throwIfRequestCanceled(requestId: string | undefined): void {
getRequestSignal(requestId)?.throwIfAborted();
if (isRequestCanceled(requestId)) {
Expand Down
1 change: 1 addition & 0 deletions packages/host-kit/src/request.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ export {
clearRequestCanceled,
getRequestSignal,
getRequestExecutionDeadlineMs,
getRequestResultDeadlineMs,
isRequestCanceled,
markRequestCanceled,
registerRequestAbort,
Expand Down
Loading
Loading