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
Original file line number Diff line number Diff line change
Expand Up @@ -60,7 +60,6 @@ export function daemonRuntimeDepsFixture(
return async (input) => {
const resolved = typeof overrides === 'function' ? overrides(input) : overrides;
return Object.freeze({
runtimeStates: [],
ownedProcessRecords: createOwnedProcessRecordStore({
stateDir: input.baseDir,
sessionsDir: input.sessionsDir,
Expand Down
9 changes: 2 additions & 7 deletions packages/daemon/src/core/interactor-resolution.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { defineRuntimeSlots, type RuntimeStateHandle } from '@agent-device/host-kit/runtime-slots';
import { runtimeSlot } from '@agent-device/host-kit/runtime-slots';
import type { Interactor, RunnerContext } from '@agent-device/contracts/interactor-types';
import type { DeviceInfo } from '@agent-device/kernel/device';
import { AppError } from '@agent-device/kernel/errors';
Expand Down Expand Up @@ -30,17 +30,12 @@ const NO_INTERACTOR_RESOLUTION: InteractorResolution = {

type InteractorResolutionState = { resolution: InteractorResolution };

const interactorResolutionRuntime = defineRuntimeSlots();
const INTERACTOR_RESOLUTION_SLOT = Symbol('interactorResolution');
const interactorResolutionState = (): InteractorResolutionState =>
interactorResolutionRuntime.slot(INTERACTOR_RESOLUTION_SLOT, (): InteractorResolutionState => ({
runtimeSlot(INTERACTOR_RESOLUTION_SLOT, (): InteractorResolutionState => ({
resolution: NO_INTERACTOR_RESOLUTION,
}));

/** The resolution one daemon runtime owns; the runtime installs its composition's into it. */
export const createInteractorResolutionRuntimeState = (): RuntimeStateHandle =>
interactorResolutionRuntime.createHandle();

/**
* Root composition's installation point, beside the provider-device admission that shares the
* same request-scoped provider scope: one change names both. The interactor graph stays behind
Expand Down
9 changes: 2 additions & 7 deletions packages/daemon/src/core/provider-device-admission.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { defineRuntimeSlots, type RuntimeStateHandle } from '@agent-device/host-kit/runtime-slots';
import { runtimeSlot } from '@agent-device/host-kit/runtime-slots';
import type { DeviceInfo } from '@agent-device/kernel/device';

/**
Expand All @@ -21,17 +21,12 @@ const NO_PROVIDER_DEVICE_ADMISSION: ProviderDeviceAdmission = {

type ProviderAdmissionState = { admission: ProviderDeviceAdmission };

const providerAdmissionRuntime = defineRuntimeSlots();
const PROVIDER_ADMISSION_SLOT = Symbol('providerDeviceAdmission');
const providerAdmissionState = (): ProviderAdmissionState =>
providerAdmissionRuntime.slot(PROVIDER_ADMISSION_SLOT, (): ProviderAdmissionState => ({
runtimeSlot(PROVIDER_ADMISSION_SLOT, (): ProviderAdmissionState => ({
admission: NO_PROVIDER_DEVICE_ADMISSION,
}));

/** The admission one daemon runtime owns; the runtime installs its composition's into it. */
export const createProviderDeviceAdmissionRuntimeState = (): RuntimeStateHandle =>
providerAdmissionRuntime.createHandle();

/**
* Root composition's installation point. The provider runtime scope is ambient per request, so
* this is the one place the daemon names where the fact comes from, and the daemon never reads
Expand Down
9 changes: 3 additions & 6 deletions packages/daemon/src/resources/artifact-tracking.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,8 @@ import fs from 'node:fs';
import path from 'node:path';
import { expect, test } from 'vitest';
import { mkdtempForTestSync } from '@agent-device/test-kit';
import {
createArtifactTrackingRuntimeState,
trackDownloadableArtifact,
trackUploadedArtifact,
} from './artifact-tracking.ts';
import { createRuntimeStateHandle } from '@agent-device/host-kit/runtime-slots';
import { trackDownloadableArtifact, trackUploadedArtifact } from './artifact-tracking.ts';

test('disposing a runtime removes the temporary files it still tracks', () => {
const root = mkdtempForTestSync('agent-device-artifact-dispose-');
Expand All @@ -15,7 +12,7 @@ test('disposing a runtime removes the temporary files it still tracks', () => {
fs.writeFileSync(artifactPath, 'artifact');
fs.mkdirSync(uploadDir);
fs.writeFileSync(path.join(uploadDir, 'payload'), 'payload');
const runtime = createArtifactTrackingRuntimeState();
const runtime = createRuntimeStateHandle();

runtime.run(() => {
trackDownloadableArtifact({
Expand Down
9 changes: 2 additions & 7 deletions packages/daemon/src/resources/artifact-tracking.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { defineRuntimeSlots, type RuntimeStateHandle } from '@agent-device/host-kit/runtime-slots';
import { runtimeSlot } from '@agent-device/host-kit/runtime-slots';
import crypto from 'node:crypto';
import fs from 'node:fs';
import os from 'node:os';
Expand Down Expand Up @@ -50,10 +50,9 @@ type ArtifactTrackingState = {
uploads: Map<string, UploadEntry>;
};

const artifactTrackingRuntime = defineRuntimeSlots();
const ARTIFACT_TRACKING_SLOT = Symbol('artifactTracking');
const artifactTracking = (): ArtifactTrackingState =>
artifactTrackingRuntime.slot(
runtimeSlot(
ARTIFACT_TRACKING_SLOT,
(): ArtifactTrackingState => ({ artifacts: new Map(), uploads: new Map() }),
(state) => {
Expand All @@ -71,10 +70,6 @@ const artifactTracking = (): ArtifactTrackingState =>
const pendingArtifacts = () => artifactTracking().artifacts;
const pendingUploads = () => artifactTracking().uploads;

/** The artifacts and uploads one daemon runtime tracks; disposing it stops their expiry timers. */
export const createArtifactTrackingRuntimeState = (): RuntimeStateHandle =>
artifactTrackingRuntime.createHandle();

/** What a finished request registers for each artifact it hands back to the client. */
export type DownloadableArtifactRegistration = {
artifactPath: string;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@ import { setTimeout as delay } from 'node:timers/promises';
import { lifecycleDeviceRuntimeGateway } from '../../__tests__/support/test-device-runtime-gateway.ts';
import { expect, test, vi } from 'vitest';
import { createTestDeviceInventoryGateways } from '@agent-device/device-selection/inventory-gateways-fixtures';
import type { RuntimeStateHandle } from '@agent-device/host-kit/runtime-slots';
import { runtimeSlot } from '@agent-device/host-kit/runtime-slots';
import type { ProviderDeviceRuntime } from '@agent-device/contracts/device';
import { resolveDaemonPaths } from '@agent-device/daemon-contracts/daemon-resolution';
import { createTestHostEnvironment, mkdtempForTestSync } from '@agent-device/test-kit';
Expand Down Expand Up @@ -178,22 +178,27 @@ test('unfinished platform owner shutdown is bounded, reports pending and retains
});
const clearLock = vi.fn(async () => {});
const pluginShutdown = vi.fn(async () => []);
// The slot lives on the runtime's one entered store, so its disposer runs exactly when that
// store is disposed (#3794): it stands in for the per-runtime state the close must not drop
// while work is unfinished.
const disposeState = vi.fn();
const state: RuntimeStateHandle = { run: (fn) => fn(), enter: () => {}, dispose: disposeState };
const heldSlot = Symbol('bounded-close-held-state');
const runtime = await startDaemonRuntime({
host: hostFor(stateDir),
...silent,
shutdownBudgets: { ownerMs: 20 },
composeDeps: daemonRuntimeDepsFixture({
deviceRuntimeGateway: {
...lifecycleDeviceRuntimeGateway,
applicationLifecycle: applicationLifecycleResourceFixture(),
shutdown: () => ownerClosed,
},
ownerLifecycle: platformOwnerLifecycleFixture({ clearDaemonLockConfiguration: clearLock }),
shutdownProviderRuntimes: pluginShutdown,
runtimeStates: [state],
}),
composeDeps: async (input) => {
runtimeSlot(heldSlot, () => ({}), disposeState);
return await daemonRuntimeDepsFixture({
deviceRuntimeGateway: {
...lifecycleDeviceRuntimeGateway,
applicationLifecycle: applicationLifecycleResourceFixture(),
shutdown: () => ownerClosed,
},
ownerLifecycle: platformOwnerLifecycleFixture({ clearDaemonLockConfiguration: clearLock }),
shutdownProviderRuntimes: pluginShutdown,
})(input);
},
});
const closing = runtime.close();
try {
Expand All @@ -215,3 +220,25 @@ test('unfinished platform owner shutdown is bounded, reports pending and retains
await closing;
}
});

test('a clean close disposes the state composition declared in the entered handle', async () => {
const stateDir = mkdtempForTestSync('agent-device-daemon-bounded-close-clean-');
const disposeState = vi.fn();
const heldSlot = Symbol('bounded-close-clean-state');
const runtime = await startDaemonRuntime({
host: hostFor(stateDir),
...silent,
composeDeps: async (input) => {
// The slot is declared through the store the runtime entered, so it only exists when
// composition runs inside the handle. The disposer running once at close proves both
// halves: composition entered the handle, and the close disposed that state.
runtimeSlot(heldSlot, () => ({}), disposeState);
return await daemonRuntimeDepsFixture({})(input);
},
});

const report = await runtime.close();

expect(report.platformOwners).toBe('closed');
expect(disposeState).toHaveBeenCalledOnce();
});
34 changes: 19 additions & 15 deletions packages/daemon/src/server/http/daemon-runtime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -42,9 +42,7 @@ import { finalizeDaemonSessionApplicationLifecycle } from '../application-lifecy
import { runtimeHintValues } from '../../sessions/session-runtime.ts';
import {
closeDaemonServers,
createRuntimeContext,
teardownDaemonSessionForShutdown,
type RuntimeContext,
type WritableOutput,
createPluginClose,
createStartupRollback,
Expand All @@ -71,7 +69,10 @@ import { createPerfCaptureAdmissionLedger } from '@agent-device/capture-kit/perf
import { createScreenRecordingAdmissionLedger } from '@agent-device/capture-kit/screen-recording-admission-ledger';
import { withCommandHost } from '@agent-device/host-kit/command';
import { cancelAllRequests } from '@agent-device/host-kit/request';
import type { RuntimeStateHandle } from '@agent-device/host-kit/runtime-slots';
import {
createRuntimeStateHandle,
type RuntimeStateHandle,
} from '@agent-device/host-kit/runtime-slots';
import { emitDiagnostic, type ResourceDiagnostic } from '@agent-device/host-kit/diagnostics';
import type { HostEnvironment } from '@agent-device/kernel/source-value';
import {
Expand Down Expand Up @@ -134,8 +135,6 @@ const DAEMON_RECORDING_REAP_TERM_TIMEOUT_MS = 5_000;
* same factory, so the production assembly is the one every runtime test boots.
*/
export type DaemonRuntimeDeps = Readonly<{
/** State the composition's platform modules keep per daemon runtime (the Apple runner's). */
runtimeStates: readonly RuntimeStateHandle[];
ownedProcessRecords: OwnedProcessRecordStore;
hostDiagnostics: HostDiagnostics;
ownerLifecycle: PlatformOwnerLifecycle;
Expand Down Expand Up @@ -254,17 +253,23 @@ export function daemonStartupExitCode(error: unknown): number {
return typeof exitCode === 'number' ? exitCode : 1;
}

/**
* Starts one daemon runtime: the single entered runtime-state store, the command host, then the
* composition and servers built inside both. Every read of per-runtime state (requests, close,
* timers the startup spawns) runs within this handle, so two runtimes in one process never share
* it and no module exposes a store of its own (#3794).
*/
export async function startDaemonRuntime(
options: DaemonRuntimeOptions,
): Promise<DaemonRuntimeController> {
const runtimeContext = createRuntimeContext();
return await runtimeContext.run(
const runtimeState = createRuntimeStateHandle();
return await runtimeState.run(
async () =>
await withCommandHost({ host: options.host, requestEnv: undefined }, async () => {
const rollback = createStartupRollback();
rollback.onRollback(() => runtimeContext.dispose());
rollback.onRollback(() => runtimeState.dispose());
try {
return await startDaemonRuntimeOnHost(options, rollback, runtimeContext);
return await startDaemonRuntimeOnHost(options, rollback, runtimeState);
} catch (error) {
await rollback.run();
throw error;
Expand All @@ -276,7 +281,7 @@ export async function startDaemonRuntime(
async function startDaemonRuntimeOnHost(
options: DaemonRuntimeOptions,
rollback: StartupRollback,
runtimeContext: RuntimeContext,
runtimeState: RuntimeStateHandle,
): Promise<DaemonRuntimeController> {
const { host } = options;
const { env } = host;
Expand Down Expand Up @@ -316,7 +321,6 @@ async function startDaemonRuntimeOnHost(
}),
openWebSessionNames: () => openWebSessionNames(sessionStore),
});
runtimeContext.extend(deps.runtimeStates);
const appLogAdmissionLedger = createAppLogAdmissionLedger();
const audioProbeAdmissionLedger = createAudioProbeAdmissionLedger();
const perfCaptureAdmissionLedger = createPerfCaptureAdmissionLedger();
Expand Down Expand Up @@ -561,7 +565,7 @@ async function startDaemonRuntimeOnHost(
const handleRequest: DaemonInvokeFn = lifecycleAdmission.admit(async (req) => {
if (shuttingDown) return daemonShuttingDownResponse();
idleReap.cancel();
const dispatch = runtimeContext.run(() => dispatchRequest(req));
const dispatch = runtimeState.run(() => dispatchRequest(req));
inFlightRequests.add(dispatch);
try {
return await dispatch;
Expand Down Expand Up @@ -911,7 +915,7 @@ async function startDaemonRuntimeOnHost(
try {
await deps.ownerLifecycle.clearDaemonLockConfiguration();
} catch {}
runtimeContext.dispose();
runtimeState.dispose();
}
}
};
Expand All @@ -928,7 +932,7 @@ async function startDaemonRuntimeOnHost(
closed.catch(() => {});
const close = (closeOptions: { cause?: unknown } = {}): Promise<ShutdownReport> => {
if (!closing) {
closing = runtimeContext.run(() =>
closing = runtimeState.run(() =>
withCommandHost(
{ host, requestEnv: undefined },
async () => await closeOnHost(closeOptions),
Expand All @@ -945,7 +949,7 @@ async function startDaemonRuntimeOnHost(
httpPort,
close,
closed,
run: runtimeContext.run,
run: runtimeState.run,
socketPort,
token,
};
Expand Down
43 changes: 0 additions & 43 deletions packages/daemon/src/server/http/server-shutdown.ts
Original file line number Diff line number Diff line change
Expand Up @@ -5,17 +5,10 @@ import type {
import { normalizeError, type NormalizedError } from '@agent-device/kernel/errors';
import type { DaemonServer } from './transport.ts';
import { sleep } from '@agent-device/host-kit/retry';
import { createRequestCancelRuntimeState } from '@agent-device/host-kit/request';
import type { RuntimeStateHandle } from '@agent-device/host-kit/runtime-slots';
import type { RuntimeHintValues } from '@agent-device/contracts/application-lifecycle-runtime';
import { SessionStore } from '../../core/session-store.ts';
import type { SessionRef, SessionState } from '../../core/session-state.ts';
import type { PlatformResourceCleanup } from '../../core/platform-resource-cleanup.ts';
import { createProviderDeviceAdmissionRuntimeState } from '../../core/provider-device-admission.ts';
import { createInteractorResolutionRuntimeState } from '../../core/interactor-resolution.ts';
import { createArtifactTrackingRuntimeState } from '../../resources/artifact-tracking.ts';
import { createResumableUploadRuntimeState } from '../resumable-upload.ts';
import { createMaterializedPathRuntimeState } from '../../sessions/materialized-path-registry.ts';
import { stopSessionAppLog, teardownSessionResources } from '../../sessions/session-teardown.ts';
import { resolveDaemonSessionTeardownTimeoutMs } from '../../sessions/session-teardown-budget.ts';
import { runtimeHintValues } from '../../sessions/session-runtime.ts';
Expand Down Expand Up @@ -372,39 +365,3 @@ export async function teardownDaemonSessionForShutdown(params: {
sessionStore.retire(ref);
return teardownSucceeded;
}

export type RuntimeContext = Readonly<{
run<T>(fn: () => T): T;
/** Adds state the composition contributed and binds it for the rest of the start. */
extend(handles: readonly RuntimeStateHandle[]): void;
dispose(): void;
}>;

/** The per-runtime state of every module that keeps some, in the order it is entered. */
function createDaemonRuntimeStates(): RuntimeStateHandle[] {
return [
createRequestCancelRuntimeState(),
createProviderDeviceAdmissionRuntimeState(),
createInteractorResolutionRuntimeState(),
createArtifactTrackingRuntimeState(),
createResumableUploadRuntimeState(),
createMaterializedPathRuntimeState(),
];
}

/** The context one daemon runtime enters: every module's per-runtime state, composed in order. */
export function createRuntimeContext(): RuntimeContext {
const states = createDaemonRuntimeStates();
return {
run: (fn) => states.reduceRight((next, state) => () => state.run(next), fn)(),
extend: (handles) => {
for (const handle of handles) {
states.push(handle);
handle.enter();
}
},
dispose: () => {
for (const state of states) state.dispose();
},
};
}
9 changes: 2 additions & 7 deletions packages/daemon/src/server/resumable-upload.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { defineRuntimeSlots, type RuntimeStateHandle } from '@agent-device/host-kit/runtime-slots';
import { runtimeSlot } from '@agent-device/host-kit/runtime-slots';
import crypto from 'node:crypto';
import fs from 'node:fs';
import path from 'node:path';
Expand Down Expand Up @@ -26,10 +26,9 @@ type ResumableUploadState = {
byKey: Map<string, string>;
};

const resumableUploadRuntime = defineRuntimeSlots();
const RESUMABLE_UPLOAD_SLOT = Symbol('resumableUploads');
const resumableUploads = (): ResumableUploadState =>
resumableUploadRuntime.slot(
runtimeSlot(
RESUMABLE_UPLOAD_SLOT,
(): ResumableUploadState => ({ byId: new Map(), byKey: new Map() }),
(state) => {
Expand All @@ -46,10 +45,6 @@ const resumableUploads = (): ResumableUploadState =>
const RESUMABLE_UPLOADS_BY_ID = () => resumableUploads().byId;
const RESUMABLE_UPLOADS_BY_KEY = () => resumableUploads().byKey;

/** The resumable uploads one daemon runtime holds; disposing it stops their expiry timers. */
export const createResumableUploadRuntimeState = (): RuntimeStateHandle =>
resumableUploadRuntime.createHandle();

export type BeginResumableUploadOptions = {
baseUrl: string;
tokenHeaders: Record<string, string>;
Expand Down
Loading
Loading