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
23 changes: 0 additions & 23 deletions .dependency-cruiser-known-cycles.json
Original file line number Diff line number Diff line change
Expand Up @@ -22,29 +22,6 @@
}
]
},
{
"type": "cycle",
"from": "src/daemon",
"to": "src/agent-device",
"rule": {
"severity": "error",
"name": "no-cycle-between-components"
},
"cycle": [
{
"name": "src/agent-device",
"dependencyTypes": []
},
{
"name": "src/http",
"dependencyTypes": []
},
{
"name": "src/daemon",
"dependencyTypes": []
}
]
},
{
"type": "cycle",
"from": "src/daemon",
Expand Down
7 changes: 2 additions & 5 deletions .dependency-cruiser.mjs
Original file line number Diff line number Diff line change
Expand Up @@ -172,13 +172,10 @@ export default {
{
name: "main-loads-modules-lazily",
comment:
"main.ts imports src/agent-device only by dynamic import or for types, so a daemon that has the module turned off loads none of its code",
"main.ts imports nothing from src/agent-device, of any type: a module reaches a daemon only through the module host",
severity: "error",
from: { path: "^src/daemon/main\\.ts$" },
to: {
dependencyTypesNot: ["type-only", "type-import", "dynamic-import"],
path: "^src/agent-device/",
},
to: { path: "^src/agent-device/" },
},
direction(
"not-modules",
Expand Down
8 changes: 5 additions & 3 deletions src/agent-device/AGENTS.md
Original file line number Diff line number Diff line change
@@ -1,9 +1,9 @@
# Agent-device edge

Does: runs one agent-device daemon per worker and serves the agent-device CLI at `/v1/agent-device/*`, one simlock lease per agent-device lease.
Entry: `index.ts` → `createGatewayAgentDeviceFrontParts` (gateway), `createAgentDeviceFront`, `disposeAgentDeviceEdge`, `cancelWaitingAllocates`; the worker half is `worker.ts` (`createAgentDeviceEdgeParts`, builds `AgentDeviceSupervisor`), loaded by the module host.
Entry: `index.ts` → `agentDeviceModule`. The front parts live in `front/edge-parts.ts` (`createGatewayAgentDeviceFrontParts`, `createAgentDeviceFront`, `disposeAgentDeviceEdge`, `cancelWaitingAllocates`). The gateway half is `gateway.ts` (`createGateway`, builds the `FleetPort` from the gateway context); the worker half is `worker.ts` (`createAgentDeviceEdgeParts`, builds `AgentDeviceSupervisor`). The module host loads both.
Uses: ports, http (session, errors, token identity), contract types, core and leasing types. Never imports: gateway, drivers, `fs`, `child_process`.
Imported by: the module registry (`definition.ts` only); `daemon/main.ts` loads `index.ts` with a dynamic `import()` on a gateway that has it on (tests may use `testing.ts`). Pinned in `lint-rules.test.ts`.
Imported by: the module registry (`index.ts` only); `daemon/main.ts` imports nothing from it (tests may use `testing.ts`). Pinned in `lint-rules.test.ts`.
Off unless `capabilities.agent-device.enabled` (false). Off: nothing mounted, no daemon.

## Files
Expand All @@ -27,6 +27,8 @@ Off unless `capabilities.agent-device.enabled` (false). Off: nothing mounted, no
| `front/devices-filter.ts`, `front/token-devices.ts` | `devices` cut to the token's leases |
| `front/host-link.ts` | `HostLink` port; `InProcessHostLink` (worker) |
| `front/relay-host-link.ts` | `RelayHostLink` (gateway); `FleetPort` |
| `front/edge-parts.ts` | Front parts, dispose and cancel steps, `createAgentDeviceFront` |
| `gateway.ts` | Gateway side: `FleetPort` from the gateway context, the mount, `beforeDrain` |
| `shared/` | Paths, routes, RPC reader, replies, compatibility, daemon env, release scope |
| `testing.ts` | `FakeAgentDevice` (in-memory daemon), `FakeAgentDeviceHealth` |

Expand Down Expand Up @@ -130,7 +132,7 @@ renewals and release scopes: memory only.

- The simlock lease is the only clock; an agent-device lease never outlives it.
- A pid is signalled only after the `--simlock-home` marker matches.
- The edge names no gateway type; the gateway reaches it through `FleetPort`.
- The edge names no gateway type; the gateway side builds `FleetPort` from its gateway context.
- A daemon stop cancels waiting allocates (`cancelWaitingAllocates`) and stops renewals.
- Tests: `edge.test.ts`, `gateway-front.test.ts`, `front/*.test.ts`, `host/*.test.ts`.

Expand Down
1 change: 1 addition & 0 deletions src/agent-device/definition.ts
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ export const agentDeviceModule: OptionalModuleDefinition<AgentDeviceModuleConfig
leaseTtlCapMs: integerInRange(MIN_AGENT_DEVICE_LEASE_TTL_MS, Number.MAX_SAFE_INTEGER),
}),
},
loadGateway: () => import("./gateway.js"),
loadWorker: () => import("./worker.js"),
name: "agent-device",
};
2 changes: 1 addition & 1 deletion src/agent-device/dispose.test.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import { describe, expect, it } from "vitest";
import { FakeClock } from "../ports/clock.js";
import { disposeAgentDeviceEdge, type AgentDeviceFrontParts } from "./index.js";
import { disposeAgentDeviceEdge, type AgentDeviceFrontParts } from "./front/edge-parts.js";
import { FrontMap } from "./front/front-map.js";
import { Renewals } from "./front/renewals.js";
import type { HostLink } from "./front/host-link.js";
Expand Down
55 changes: 55 additions & 0 deletions src/agent-device/front/edge-parts.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,55 @@
import type { HttpDispatch } from "../../http/dispatcher-session.js";
import type { HttpMount } from "../../http/app.js";
import type { Clock, Logger } from "../../ports/index.js";
import { FrontMap } from "./front-map.js";
import { createFront } from "./front.js";
import type { HostLink } from "./host-link.js";
import { RelayHostLink, type FleetPort } from "./relay-host-link.js";
import { Renewals } from "./renewals.js";

export type { FleetPort, FleetWorker } from "./relay-host-link.js";

export interface AgentDeviceFrontParts {
readonly link: HostLink;
readonly map: FrontMap;
readonly renewals: Renewals;
}

export function createGatewayAgentDeviceFrontParts(
fleet: FleetPort,
clock: Clock,
leaseMaxTtlMs: number,
): AgentDeviceFrontParts {
return {
link: new RelayHostLink({ clock, fleet, leaseMaxTtlMs }),
map: new FrontMap(clock),
renewals: new Renewals(clock),
};
}

// why: a stop ends the gateway's sockets, so the calls that end then must send no renewal that races leasing teardown.
export function disposeAgentDeviceEdge(parts: AgentDeviceFrontParts): void {
parts.renewals.dispose();
parts.link.dispose?.();
}

// why: a stop refuses every dispatch, so waiting allocates' lease requests are cancelled through leasing directly, before the gateway stops; otherwise they stay open across a restart.
export async function cancelWaitingAllocates(
parts: AgentDeviceFrontParts,
cancel: (requesterId: string) => Promise<unknown>,
): Promise<void> {
await Promise.allSettled(parts.map.waitingIds().map((id) => cancel(id)));
}

export function createAgentDeviceFront(
parts: AgentDeviceFrontParts,
deps: {
readonly dispatch: HttpDispatch;
readonly clock: Clock;
readonly logger: Logger;
readonly leaseTtlCapMs?: number;
readonly maxRunMs: number;
},
): HttpMount {
return createFront({ ...deps, link: parts.link, map: parts.map, renewals: parts.renewals });
}
2 changes: 1 addition & 1 deletion src/agent-device/front/relay-host-link.ts
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ export interface FleetLeaseFacts {
};
}

// why: the edge imports nothing from the gateway, so the gateway's parts reach the link only through this port, built where the gateway is composed.
// why: the edge imports nothing from the gateway, so the gateway's parts reach the link only through this port, built in `gateway.ts` from the gateway context.
export interface FleetPort {
readonly instanceId: string;
// why: when this gateway process started, which opens the window in which a lease it cannot place may still come back.
Expand Down
76 changes: 76 additions & 0 deletions src/agent-device/gateway.test.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,76 @@
import { describe, expect, it } from "vitest";

import type { WorkerView } from "../modules/index.js";
import { fakeGatewayContext } from "../modules/testing.js";
import { agentDeviceModule } from "./definition.js";

const noWorker = "no worker connected to this gateway can serve agent-device";

function viewOf(overrides: Record<string, unknown> = {}): WorkerView {
return {
capabilities: [{ name: "agent-device", state: "available" }],
catalog: [],
catalogReadAt: 1_000,
connection: "connected",
drained: false,
id: "w-1",
label: "w1",
lease: { maxTtlMs: 600_000 },
...overrides,
} as unknown as WorkerView;
}

async function healthThroughGateway(served: readonly WorkerView[]) {
const { ctx, relayed } = await fakeGatewayContext(
"agent-device",
agentDeviceModule.config.defaults,
{
relay: () => Response.json({ ok: true, service: "agent-device-proxy" }),
serving: served,
},
);
const loaded = await agentDeviceModule.loadGateway?.();
const module = loaded?.createGateway(ctx);
const handler = module?.mount?.({ dispatch: () => Promise.reject(new Error("unused")) });
const answer = await handler?.(new Request("http://gateway.test/v1/agent-device/health"), {
requesterId: "tok_one",
role: "agent",
});
return { body: await answer?.text(), relayed };
}

describe("agent-device's gateway side", () => {
it("a worker whose view has no lease.maxTtlMs is not served by agent-device's fleet port, while the same view with it is", async () => {
const served = await healthThroughGateway([viewOf()]);
expect(served.relayed.map((call) => call.workerId)).toEqual(["w-1"]);
expect(served.body).toContain("agent-device-proxy");

const unknown = await healthThroughGateway([viewOf({ lease: undefined })]);

expect(unknown.relayed).toEqual([]);
expect(unknown.body).toContain(noWorker);
});

it("a worker whose catalog is unread, and a drained one, are not offered to a new allocate, while the same worker read and not drained is", async () => {
expect((await healthThroughGateway([viewOf()])).relayed).toHaveLength(1);

const unread = await healthThroughGateway([viewOf({ catalogReadAt: undefined })]);
const drained = await healthThroughGateway([viewOf({ drained: true })]);

expect([unread.relayed, drained.relayed]).toEqual([[], []]);
expect([unread.body, drained.body]).toEqual([
expect.stringContaining(noWorker),
expect.stringContaining(noWorker),
]);
});

it("agent-device's definition has a gateway side that creates a mount and a beforeDrain hook", async () => {
const { ctx } = await fakeGatewayContext("agent-device", agentDeviceModule.config.defaults);
const loaded = await agentDeviceModule.loadGateway?.();

const module = loaded?.createGateway(ctx);

expect(module?.mount).toBeTypeOf("function");
expect(module?.hooks?.beforeDrain).toBeTypeOf("function");
});
});
155 changes: 155 additions & 0 deletions src/agent-device/gateway.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,155 @@
import type { HttpDispatch } from "../http/dispatcher-session.js";
import type { TokenIdentity } from "../http/token-store.js";
import type { FleetLease, GatewayContext, GatewayModule, WorkerView } from "../modules/index.js";
import type { AgentDeviceModuleConfig } from "./definition.js";
import {
cancelWaitingAllocates,
createAgentDeviceFront,
createGatewayAgentDeviceFrontParts,
disposeAgentDeviceEdge,
type FleetPort,
type FleetWorker,
} from "./front/edge-parts.js";

type Context = GatewayContext<AgentDeviceModuleConfig>;
type FleetLeaseFacts = ReturnType<FleetPort["leasesOf"]>[number];

// why: the context lists workers whose agent-device is available; the rest of what makes one fit for agent-device is judged here.
function serving(view: WorkerView, withDrained: boolean): FleetWorker | undefined {
const maxTtlMs = view.lease?.maxTtlMs;
if (
(view.drained && !withDrained) ||
view.catalogReadAt === undefined ||
maxTtlMs === undefined
) {
return undefined;
}
const platforms = (view.catalog ?? []).flatMap((entry) =>
entry.platform === "ios" || entry.platform === "android" ? [entry.platform] : [],
);
return { id: view.id, leaseMaxTtlMs: maxTtlMs, platforms };
}

function fleetPort(ctx: Context): FleetPort {
return {
capableWorkers: () =>
ctx.serving().flatMap((view) => {
const worker = serving(view, false);
return worker === undefined ? [] : [worker];
}),
holder: (workerId) => {
const view = ctx.serving().find((candidate) => candidate.id === workerId);
return view === undefined ? undefined : serving(view, true);
},
instanceId: ctx.instanceId,
leasesOf: (ownerId) =>
ctx.leases
.all()
.filter((entry) => entry.ownerId === ownerId)
.map((entry) => leaseFacts(ctx, entry)),
leaseOf: (leaseId) => {
const entry = ctx.leases.resolve(leaseId);
return entry === undefined ? undefined : { ownerId: entry.ownerId, workerId: entry.workerId };
},
onLeaseRemoved: (listener) => {
ctx.leases.onRemoved((entry) =>
listener({
leaseId: entry.gatewayLeaseId,
ownerId: entry.ownerId,
workerId: entry.workerId,
}),
);
},
reach: (workerId) => {
const view = ctx.worker(workerId);
return view === undefined
? undefined
: {
connected: view.connection === "connected",
refreshed: view.catalogReadAt !== undefined,
};
},
// why: looked up at each call, so a test that replaces the relay's method sits between the link and the worker.
relay: (workerId, request, tenant, leaseId) => ctx.relay(workerId, request, tenant, leaseId),
startedAt: ctx.startedAt,
};
}

// why: the worker's view lists the lease, its granted device and the model's class in its catalog; a lease missing any of them is not offered for adoption.
function listedLease(
ctx: Context,
view: WorkerView | undefined,
entry: FleetLease,
): FleetLeaseFacts["listed"] {
const record = view?.leases?.find((lease) => lease.id === entry.workerLeaseId);
if (view === undefined || record === undefined) return undefined;
const device = ctx
.grantedDevices()
.find((granted) => granted.workerId === entry.workerId && granted.id === record.deviceId);
if (device === undefined) return undefined;
return describeDevice(view, record, device);
}

function describeDevice(
view: WorkerView,
record: { readonly ttlMs: number; readonly ttlDeadline: number },
device: ReturnType<Context["grantedDevices"]>[number],
): FleetLeaseFacts["listed"] {
const { platform, model } = device.spec;
if (device.address === undefined || (platform !== "ios" && platform !== "android")) {
return undefined;
}
const deviceClass = view.catalog?.find((catalog) => catalog.platform === platform)?.modelClasses[
model
];
return {
address: device.address,
model,
platform,
ttlDeadline: record.ttlDeadline,
ttlMs: record.ttlMs,
...(deviceClass === undefined ? {} : { deviceClass }),
};
}

function leaseFacts(ctx: Context, entry: FleetLease): FleetLeaseFacts {
const view = ctx.worker(entry.workerId);
const listed = listedLease(ctx, view, entry);
return {
connected: view?.connection === "connected",
leaseId: entry.gatewayLeaseId,
refreshed: view?.catalogReadAt !== undefined,
requesterId: entry.requesterId,
workerId: entry.workerId,
...(listed === undefined ? {} : { listed }),
};
}

// fallow-ignore-next-line unused-export -- reached through the namespace of definition.ts's dynamic import.
export function createGateway(ctx: Context): GatewayModule {
const parts = createGatewayAgentDeviceFrontParts(
fleetPort(ctx),
ctx.clock,
ctx.daemonConfig.lease.maxTtlMs,
);
return {
hooks: {
beforeDrain: async () => {
disposeAgentDeviceEdge(parts);
await cancelWaitingAllocates(parts, (id) => ctx.cancelPending(id));
},
},
mount: (deps) => {
const front = createAgentDeviceFront(parts, {
clock: ctx.clock,
dispatch: deps.dispatch as HttpDispatch,
logger: ctx.logger,
maxRunMs: ctx.daemonConfig.lease.maxTtlMs,
...(ctx.config.leaseTtlCapMs === undefined
? {}
: { leaseTtlCapMs: ctx.config.leaseTtlCapMs }),
});
return (request, identity) => front.handle(request, identity as TokenIdentity);
},
};
}
Loading
Loading