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
3 changes: 2 additions & 1 deletion src/agent-device/AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,8 @@ Front (`createFront`, prefix `agent-device`):
reply "ended by simlock".
7. Release → forwarded; on success `lease.release`; map marks it ended by `agent`.
8. Command on a lease → `Renewals.begin`: renew every TTL/3 until the body ends,
capped at `lease.maxTtlMs`.
capped at `lease.maxTtlMs`. A reply naming `endedLeases` (in `data`, `error.details`, or a
batch's `data`) → `front/ended.ts` ends each simlock lease this token holds, as a release does.

Host (`AgentDeviceHost.forward`): one proxy per daemon generation. Tags the request,
and `authenticate` sets tenant = requester id, overwriting the client's. Rewrites
Expand Down
58 changes: 56 additions & 2 deletions src/agent-device/edge.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1197,7 +1197,7 @@ describe("the agent-device front, through the whole worker daemon", () => {
await expect.poll(async () => (await leases(rig)).length).toBe(0);
});

it("release, and release after `close` (which the daemon answers `released: false`), go to the agent-device daemon first, then release the simlock lease, and the device is reclaimed and free", async () => {
it("release goes to the agent-device daemon first, then frees the simlock lease and the device; a release after `close`, which already freed the simlock lease, reaches the daemon with no lease held and is answered `released: false`", async () => {
const rig = await setUp();
let heldAtDaemonRelease: readonly string[] = [];
rig.fake.onRelease = async () => {
Expand All @@ -1216,16 +1216,70 @@ describe("the agent-device front, through the whole worker daemon", () => {

await granted(rig, 2);
await command(rig, "close", 2);
await expect.poll(async () => leases(rig)).toEqual([]);
const afterClose = await envelopeOf(await release(rig, 2));

expect(afterClose.result).toMatchObject({ ok: true, released: false });
expect(heldAtDaemonRelease).toEqual([hex(2)]);
expect(heldAtDaemonRelease).toEqual([]);
expect(await leases(rig)).toEqual([]);
await expect
.poll(async () => (await devices(rig)).map((device) => device.state))
.toEqual(["ready"]);
});

const ended = [{ leaseId: hex(1), outcome: "released" }];
it.each([
[
"a `close` that succeeds, in `data`",
{ result: { data: { endedLeases: ended } } },
(rig: Rig) => command(rig, "close", 1),
],
[
"a `close` that fails, in `error.details`",
{ error: { data: { details: { endedLeases: ended } } } },
(rig: Rig) => {
rig.fake.failNextCommand({ code: "COMMAND_FAILED", message: "close failed" });
return command(rig, "close", 1);
},
],
[
"a `batch` with a close step, in `data` beside `results`",
{ result: { data: { endedLeases: ended, results: [] } } },
(rig: Rig) => batch(rig, 1, [{ command: "snapshot" }, { command: "close" }]),
],
] as const)(
"%s: a command that ends the agent-device lease and names it in `endedLeases` releases the simlock lease and frees the device though the CLI sends no release, and a later heartbeat answers `LEASE_NOT_FOUND`",
async (_case, reply, send) => {
const rig = await setUp();
await granted(rig, 1);

expect(await envelopeOf(await send(rig))).toMatchObject(reply);
await expect.poll(async () => listed(rig)).toEqual([]);
await expect
.poll(async () => (await devices(rig)).map((device) => device.state))
.toEqual(["ready"]);
expectLeaseGone(await envelopeOf(await heartbeat(rig, 1)), "LEASE_NOT_FOUND");
},
);

it("a streamed command reply whose last frame names `endedLeases` releases the simlock lease though the CLI sends no release, and the CLI gets the body unchanged", async () => {
const rig = await setUp();
await granted(rig, 1);
const chunks = [
'{"progress":1}\n',
`${JSON.stringify({ result: { data: { endedLeases: ended } } })}\n`,
];
rig.fake.streamNextCommand(chunks);

const streamed = await command(rig, "close", 1);

expect(await streamed.text()).toBe(chunks.join(""));
await expect.poll(async () => listed(rig)).toEqual([]);
await expect
.poll(async () => (await devices(rig)).map((device) => device.state))
.toEqual(["ready"]);
});

it("a release the agent-device daemon fails keeps the simlock lease and answers the daemon's error", async () => {
const rig = await setUp();
await granted(rig, 1);
Expand Down
58 changes: 58 additions & 0 deletions src/agent-device/front/ended.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,58 @@
import type { TokenIdentity } from "../../http/token-store.js";
import { leaseIdOf } from "../shared/rpc.js";
import { asRecord } from "../shared/selectors.js";
import type { FrontContext } from "./context.js";
import { endSimlockLease } from "./release.js";

function idsIn(holder: unknown): string[] {
const list = asRecord(holder)?.["endedLeases"];
if (!Array.isArray(list)) return [];
const ids: string[] = [];
for (const item of list) {
const id = leaseIdOf(asRecord(item)?.["leaseId"]);
if (id !== undefined) ids.push(id);
}
return ids;
}

// why: agent-device names a lease it ended (a `close`) in `data` on success, in `error.details` on failure, which the JSON-RPC error carries as `error.data.details`; a streamed reply is one JSON frame per line, and any frame may carry it.
function idsInFrame(body: unknown): string[] {
const root = asRecord(body);
const error = asRecord(root?.["error"]);
return [
...idsIn(asRecord(asRecord(root?.["result"])?.["data"])),
...idsIn(asRecord(asRecord(error?.["data"])?.["details"])),
];
}

function parsed(text: string): unknown {
try {
return JSON.parse(text);
} catch {
return undefined;
}
}

function endedLeaseIds(text: string): string[] {
const whole = parsed(text);
if (whole !== undefined) return idsInFrame(whole);
return text.split("\n").flatMap((line) => idsInFrame(parsed(line)));
}

// why: the daemon no longer sends a release after it ends a lease, so the front ends the simlock lease from the command's reply; a copy is read so the CLI's reply goes out untouched.
export function releaseEndedLeases(
context: FrontContext,
identity: TokenIdentity,
response: Response,
): void {
if (!(response.headers.get("content-type") ?? "").includes("json")) return;
void response
.clone()
.text()
.then(async (text) => {
for (const leaseId of new Set(endedLeaseIds(text))) {
await endSimlockLease(context, identity, leaseId);
}
})
.catch(() => undefined);
}
5 changes: 4 additions & 1 deletion src/agent-device/front/front.ts
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ import { relay, type FrontContext } from "./context.js";
import { anyUnreachable, devicesThroughGateway, filterDevices } from "./devices-filter.js";
import { heartbeat } from "./heartbeat.js";
import type { LinkStatus } from "./host-link.js";
import { releaseEndedLeases } from "./ended.js";
import { release } from "./release.js";
import { checkRequest } from "./request-rules.js";

Expand Down Expand Up @@ -166,7 +167,9 @@ async function runCommand(
end();
throw error;
}
return isUnreachableAnswer(response) ? response : untilBodyEnds(response, end);
if (isUnreachableAnswer(response)) return response;
releaseEndedLeases(context, identity, response);
return untilBodyEnds(response, end);
}

// why: a command counts as running until its body has gone out, since the stream is what the CLI waits on.
Expand Down
10 changes: 9 additions & 1 deletion src/agent-device/front/release.ts
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,15 @@ export async function release(
return unreachableReply(reply, call, "agent-device did not answer the release");
}
if (leaseId === undefined || reply.reading.kind !== "success") return relay(reply);
await endSimlockLease(context, identity, leaseId);
return relay(reply);
}

export async function endSimlockLease(
context: FrontContext,
identity: TokenIdentity,
leaseId: string,
): Promise<void> {
const entry = context.map.get(leaseId);
if (entry === undefined || entry.identity.requesterId !== identity.requesterId) {
// why: a lease the map does not hold for this token (as after a daemon restart) is freed by simlock's own ownership check, which refuses another token's lease.
Expand All @@ -28,5 +37,4 @@ export async function release(
}
context.renewals.cancelIfEnded(context, leaseId);
}
return relay(reply);
}
36 changes: 29 additions & 7 deletions src/agent-device/testing.ts
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,6 @@ interface FakeAgentDeviceLease {
readonly deviceKey: unknown;
};
readonly ttlMs: unknown;
ended: boolean;
}

interface FakeFailure {
Expand Down Expand Up @@ -513,7 +512,6 @@ export class FakeAgentDevice implements AgentDeviceUpstream {
this.#replyDevice = undefined;
this.leases.set(leaseId, {
address,
ended: false,
leaseId,
scope,
tenant,
Expand Down Expand Up @@ -603,7 +601,7 @@ export class FakeAgentDevice implements AgentDeviceUpstream {
});
await Promise.race([Promise.all(running.map((command) => command.ended)), aborted]);
}
return envelope(id, { ok: true, released: !lease.ended });
return envelope(id, { ok: true, released: true });
}

// fallow-ignore-next-line complexity -- one fake mirroring agent-device's command handling.
Expand Down Expand Up @@ -650,9 +648,17 @@ export class FakeAgentDevice implements AgentDeviceUpstream {
{ headers: { "content-type": "application/x-ndjson" } },
);
}
const endedLeases = this.#endOnClose(command, params, leaseId);
const failure = this.#failCommand;
this.#failCommand = undefined;
if (failure !== undefined) return errorEnvelope(id, failure);
if (failure !== undefined) {
return errorEnvelope(
id,
endedLeases.length === 0
? failure
: { ...failure, details: { ...failure.details, endedLeases } },
);
}
const status = this.#failCommandStatus;
this.#failCommandStatus = undefined;
if (status !== undefined) return new Response(status.body, { status: status.status });
Expand All @@ -661,10 +667,26 @@ export class FakeAgentDevice implements AgentDeviceUpstream {
if ("body" in inventory) return new Response(inventory.body, { status: 200 });
return envelope(id, { data: { devices: inventory.list }, ok: true });
}
if (command === "close" && typeof leaseId === "string") {
const lease = this.leases.get(leaseId);
if (lease !== undefined) lease.ended = true;
if (endedLeases.length > 0) {
const data = command === "batch" ? { endedLeases, results: [] } : { endedLeases };
return envelope(id, { command, data, ok: true });
}
return envelope(id, { command, ok: true });
}

// why: agent-device ends the lease a `close`, or a `batch` with a close step, closes with, and names it in the reply's `endedLeases` instead of waiting for a release.
#endOnClose(
command: string,
params: Record<string, unknown>,
leaseId: unknown,
): readonly { readonly leaseId: string; readonly outcome: "released" }[] {
const flags = (params["flags"] ?? {}) as Record<string, unknown>;
const steps = Array.isArray(flags["batchSteps"]) ? (flags["batchSteps"] as unknown[]) : [];
const closes =
command === "close" ||
(command === "batch" &&
steps.some((step) => (step as { readonly command?: unknown }).command === "close"));
if (!closes || typeof leaseId !== "string" || !this.leases.delete(leaseId)) return [];
return [{ leaseId, outcome: "released" }];
}
}
Loading