From 6346d9bedb4d67be29ce1993c037e59b28ea1608 Mon Sep 17 00:00:00 2001 From: rain Date: Thu, 29 Oct 2026 06:58:06 +0800 Subject: [PATCH] Host: count and journal a traced unmount's abort-path outcome A traced normal unmount that ends a stalled, already-detached connection carries LayerFS's `unmount` telemetry section (layerfs #501, tokenaissance/layerfs#2): `stage`, `abort`, `bytes` or `errno`, `detached`, `loops_exited`, `loops_joined`, `active`, `retained` and `waited_ms`. The Host now reads it from the trace line's `daemon` telemetry before that telemetry is stripped, counts it by stage and abort disposition, and journals each outcome to `/unmounts.jsonl`. Layers: Entities `packages/model/src/unmount.js` (the value; no imports, no I/O); Use Case `packages/model/src/unmount-count.js` (records through the journal port, counts read from it); Interface Adapter `packages/host/src/unmount-journal.js` (the JSONL file); the Host's trace collection in `collector.js` feeds them, `index.js` composes them, and the count surfaces as the `layerfsTraces` service's `unmountsOf(project)`. No metrics server, OTel or event bus: the journal is the one sink and `unmountsOf` its one reader. A plain unmount carries no section and records nothing. Verification, LAYERFS_SOURCE at ../layerfs-main (9a03429adc, not the pinned fabefdf87a: that is the LayerFS checkout available here): - pnpm test: host, agent wrappers, bindings-shared, mock (10 scenarios), cli, playback, initlimit, historycache, model trace, model unmount, checks: PASS (120 s alarm), twice. - node packages/model/tests/unmount.mjs and packages/host/tests/host.mjs: PASS, three runs each. - The real SDK trace writer of the #501/#2 branch (api/agent/src/trace.js) was driven for a success, a retained stop, a short write, a failed write and a plain unmount; every line parses exactly, including the section's key order and bytes/errno placement. - pnpm run build: the configured build needs a deepseek-harness checkout with tsdown (DSH_SOURCE); none is here. The client bundle is unchanged. - pnpm run check:layerfs: refused, the available LayerFS checkout is at 9a03429adc, not the pinned fabefdf87a. Tests deleted or skipped: none. --- README.md | 17 +++++++ package.json | 2 +- packages/host/src/collector.js | 28 +++++++++- packages/host/src/index.js | 15 ++++-- packages/host/src/unmount-journal.js | 35 +++++++++++++ packages/host/tests/host.mjs | 41 ++++++++++++++- packages/model/src/unmount-count.js | 63 +++++++++++++++++++++++ packages/model/src/unmount.js | 60 ++++++++++++++++++++++ packages/model/tests/unmount.mjs | 76 ++++++++++++++++++++++++++++ 9 files changed, 329 insertions(+), 8 deletions(-) create mode 100644 packages/host/src/unmount-journal.js create mode 100644 packages/model/src/unmount-count.js create mode 100644 packages/model/src/unmount.js create mode 100644 packages/model/tests/unmount.mjs diff --git a/README.md b/README.md index b13f8d9..71e71ca 100644 --- a/README.md +++ b/README.md @@ -481,6 +481,22 @@ A bash call in a session bound to a project gets the project, the SDK and a trac LayerFS reads only its own `LAYERFS_*` names, and dsh lets a plugin set only `DSH_*` names in a bash call. Each wrapper therefore loads `packages/host/src/agent/env.js` first, which sets every unset `LAYERFS_` from `DSH_LAYERFS_` in the agent's process, then exposes LayerFS's own entry unchanged (the same exports; the same arguments, output and exit codes for the CLI). A `LAYERFS_` already set is kept. - **Records.** Each finished SDK call becomes a `plugin:layerfs/call` record in the calling agent's own session. The root session gets one `plugin:layerfs/project` record. - **Journal.** Finished lines are also journaled per root session under the state folder, because trace files live in temporary storage. +- **Unmount outcomes.** A normal unmount that has to end a stalled, already-detached connection on LayerFS's abort path reports it beside the ordinary reply, as the `unmount` telemetry section of its `Observed` envelope ([layerfs #501](https://github.com/Ephemeral-AI-Lab/layerfs/pull/501), [tokenaissance/layerfs#2](https://github.com/tokenaissance/layerfs/pull/2)). The SDK trace line carries that telemetry under `daemon`; the Host reads the section before telemetry is stripped from the records, journals it to `/unmounts.jsonl`, and counts it by stage and abort disposition. A plain unmount carries no section and adds nothing. The section's stable keys are: + + | Key | Meaning | + | --- | --- | + | `stage` | the boundary the unmount was at: `Join` (the stalled drain ended by the abort), `Abort` (the one write was short or failed), or `Revoke`/`Close` (a later step stopped with the abort recorded) | + | `abort` | the connection's single abort write: `Written`, `Short` or `Failed` | + | `bytes` | a `Short` write's accepted byte count (present only then) | + | `errno` | a `Failed` write's original errno (present only then) | + | `detached` | kernel detach was already known when the abort was made | + | `loops_exited` | receive loops exited, at the observation | + | `loops_joined` | receive loops joined, at the observation | + | `active` | request owner count at the observation | + | `retained` | request owner count retained at the observation | + | `waited_ms` | milliseconds from the stalled drain to the end of the abort path | + + The Host's journal record adds `at`, `project`, `agent`, `id` and `reply` (`Unmounted` on success, or the refusal's code: `Failed` for a retained stop) beside those; the file is append-only, never pruned, and gets one record per abort path, not one per unmount. The count surfaces as the `layerfsTraces` service's `unmountsOf(project)` answer, `{ total, byStage, byAbort, waitedMs: { total, max }, records }`, counted from the journal, so a restarted Host answers the same. Its layers are the Entities value `packages/model/src/unmount.js`, the Use Case (journal port and counting) `packages/model/src/unmount-count.js`, the Interface Adapter `packages/host/src/unmount-journal.js`, and the trace collection that feeds them `packages/host/src/collector.js` (wired in `packages/host/src/index.js`). No metrics server, OTel or event bus is added: the journal is the one sink and `unmountsOf` its one reader. Only the structured trace section is read: the daemon's `LAYERFS_DAEMON_UNMOUNT_ABORT` stderr line is not, and only a Linux daemon takes the abort path. - **Projects.** Each project is attached to at most one workspace folder, or detached; a folder can have several. Init imports a source folder (the workspace folder unless another is chosen) and attaches the new project to the workspace folder. A session uses the folder's only (or last used) project unless it picks another or none. The choice is fixed at the first bash call that uses it. - **Agent names.** Subagents are named by the label their parent gave them. - **Web route.** `/layerfs` sits behind dsh's own connection checks: @@ -585,6 +601,7 @@ Run the preset's sessions in Workspace Write mode; subagents inherit it. `traceDir` must be writable by the agents. Other platforms keep the default traces in the OS temp area. - Everything else follows LayerFS's platform limits: Store calls need Linux and a local ext4, xfs or btrfs filesystem, which on macOS the worker's container provides. +- The unmount outcome count covers the trace lines the Host read while it ran: a line already in an agent's trace file when the Host starts is history and is not journaled. Records are keyed by project name, as the trace-line journal is, so a removed project re-initialized under the same name counts the older ones. - Windows Live uses WSL2 and its Linux Docker Engine through the Windows adapter; native Windows Store execution remains unsupported. Sim's local `/bin/sh` executor is not ported to Windows. macOS execution was not rerun for this change. diff --git a/package.json b/package.json index c546f8c..b92773c 100644 --- a/package.json +++ b/package.json @@ -11,7 +11,7 @@ "preinstall": "node tools/layerfs-source.mjs link", "link:layerfs": "node tools/layerfs-source.mjs link", "check:layerfs": "node tools/layerfs-source.mjs check", - "test": "node tools/layerfs-source.mjs link && node packages/host/tests/host.mjs && node packages/host/tests/agent.mjs && node packages/host/tests/bindings-shared.mjs && node packages/mock/tests/mock.mjs && node packages/model/tests/cli.mjs && node packages/client/tests/playback.mjs && node packages/client/tests/initlimit.mjs && node packages/client/tests/historycache.mjs && node packages/model/tests/trace.mjs && node tools/checks/tests/checks.mjs", + "test": "node tools/layerfs-source.mjs link && node packages/host/tests/host.mjs && node packages/host/tests/agent.mjs && node packages/host/tests/bindings-shared.mjs && node packages/mock/tests/mock.mjs && node packages/model/tests/cli.mjs && node packages/client/tests/playback.mjs && node packages/client/tests/initlimit.mjs && node packages/client/tests/historycache.mjs && node packages/model/tests/trace.mjs && node packages/model/tests/unmount.mjs && node tools/checks/tests/checks.mjs", "build": "node tools/build-client.mjs", "test:windows": "node packages/host/tests/windows-cli.mjs && node packages/host/tests/host.mjs && node packages/host/tests/agent.mjs && node packages/host/tests/bindings-shared.mjs", "test:scenarios": "node packages/mock/tests/mock.mjs && node packages/mock/tests/windows-shell.mjs && node packages/mock/tests/fixture-checkout.mjs && node packages/mock/tests/progress-failures.mjs && node packages/client/tests/scenario-projects.mjs && node packages/mock/tests/runtime-image.mjs" diff --git a/packages/host/src/collector.js b/packages/host/src/collector.js index bde4f91..7fec1c3 100644 --- a/packages/host/src/collector.js +++ b/packages/host/src/collector.js @@ -3,7 +3,10 @@ // per root session for the views, and writes each finished call to the // calling agent's session as a plugin:layerfs/call record. Finished lines are // also journaled per root session under the state folder, since trace files -// live in temporary storage. +// live in temporary storage. A traced unmount's abort-path telemetry is read +// before that telemetry is stripped: its outcome is journaled and counted by +// the use case (packages/model/src/unmount-count.js) through the Host's +// adapter (unmount-journal.js). import { appendFileSync, closeSync, existsSync, mkdirSync, openSync, readFileSync, readSync, readdirSync, statSync } from "node:fs"; import { join } from "node:path"; import { appendCall } from "./records.js"; @@ -21,11 +24,13 @@ export class Collector { * @param {object} ctx plugin context (ctx.sessions) * @param {object} spec resolved Host settings * @param {import("./binding.js").BindingStore} bindings + * @param {import("@dsh-layerfs/model/unmount-count.js").UnmountOutcomes} outcomes the unmount outcome use case */ - constructor(ctx, spec, bindings) { + constructor(ctx, spec, bindings, outcomes) { this.ctx = ctx; this.spec = spec; this.bindings = bindings; + this.outcomes = outcomes; /** sessionId -> { file, offset, rest, project } */ this.tails = new Map(); /** rootId -> { lines: [], seen: Set, running: Map, loaded } */ @@ -137,12 +142,31 @@ export class Collector { if (!this.keep(view, own)) return; if (!live) return; this.journal(agent.root, own); + this.recordUnmount({ project, agent: agent.name, at: Date.parse(line.start) + (line.ms || 0), line }); const record = { agent: { id: sessionId, name: agent.name }, project, line: stripped(line) }; if (!this.pending.has(sessionId)) this.pending.set(sessionId, []); this.pending.get(sessionId).push(record); this.flush(sessionId); } + /** Journals and counts one line's unmount outcome, when it carries one; a journal failure is warned, never the call's. */ + recordUnmount(seen) { + try { + this.outcomes.record({ ...seen, at: Number.isFinite(seen.at) ? seen.at : null }); + } catch (error) { + this.ctx.logger?.warn?.(`layerfs: the unmount outcome could not be journaled: ${error.message}`); + } + } + + /** + * A project's unmount outcomes: counts by stage and abort disposition, the + * waited time, and the journaled records (the use case's aggregate). + * @param {string} project + */ + unmountsOf(project) { + return this.outcomes.summaryOf(project); + } + /** Appends an agent's waiting records once its Session is live. */ flush(sessionId) { const records = this.pending.get(sessionId), session = records && this.ctx.sessions.get(sessionId); diff --git a/packages/host/src/index.js b/packages/host/src/index.js index c60907f..e04e686 100644 --- a/packages/host/src/index.js +++ b/packages/host/src/index.js @@ -11,12 +11,16 @@ // ctx.systemPrompt each agent's runtime context names its conversation's // LayerFS project (where a system prompt is assembled) // layerfsTraces a service giving one bash call's trace lines (the -// scripted model's live expect check) +// scripted model's live expect check) and a project's +// unmount outcomes (counts by stage and abort disposition, +// and the journaled records) // LayerFS itself is @layerfs/runtime's (the worker on macOS and Windows, Init, // removal, Sandbox upkeep), opened once, and agents' clients are // @layerfs/agent's (#428), through the Host's wrappers (agent/). +import { UnmountOutcomes } from "@dsh-layerfs/model/unmount-count.js"; import { BindingStore } from "./binding.js"; import { Collector } from "./collector.js"; +import { UnmountJournal } from "./unmount-journal.js"; import z from "@deepseek-ai/schemastery"; import { openRuntime, resolveConfig } from "./config.js"; import { contributor, projectContext, shimOnPath } from "./env.js"; @@ -60,12 +64,17 @@ export function apply(ctx, config) { if (spec.worker && spec.bundle && spec.image) start(); // a call of the Host's or an agent's that finds the worker down or gone starts it again, one start // at a time (the runtime's); the failed call keeps its own answer - const collector = new Collector(ctx, spec, bindings); + // a traced unmount's abort path: counted by the use case, journaled under the state folder + const outcomes = new UnmountOutcomes(new UnmountJournal(spec.stateDir)); + const collector = new Collector(ctx, spec, bindings, outcomes); collector.onUnreachable = spec.worker ? start : null; ctx.effect(() => ctx.shellEnv.register(contributor(ctx, spec, bindings, collector)), "layerfs-host: bash environment"); ctx.effect(() => shimOnPath(), "layerfs-host: layerfs on PATH"); ctx.effect(() => collector.start(), "layerfs-host: trace collection"); - ctx.provide("layerfsTraces", { callLines: (sessionId, step) => collector.callLines(sessionId, step) }); + ctx.provide("layerfsTraces", { + callLines: (sessionId, step) => collector.callLines(sessionId, step), + unmountsOf: (project) => collector.unmountsOf(project), + }); ctx.on("session/event", (session, event) => collector.observe(session, event)); const trees = new TreeReader(spec); ctx.effect(() => () => trees.dispose(), "layerfs-host: tree clients"); diff --git a/packages/host/src/unmount-journal.js b/packages/host/src/unmount-journal.js new file mode 100644 index 0000000..b83e65e --- /dev/null +++ b/packages/host/src/unmount-journal.js @@ -0,0 +1,35 @@ +// The unmount journal: one JSON line per normal unmount that took LayerFS's +// abort path, under the state folder (`/unmounts.jsonl`), so its +// outcome outlives the temporary trace files and the Host. This is the +// Interface Adapter behind the use case's port +// (packages/model/src/unmount-count.js); the Entities value it stores is +// packages/model/src/unmount.js. Append-only; a torn last line is skipped +// when read, as the trace-line journal (collector.js) does. +import { appendFileSync, existsSync, mkdirSync, readFileSync } from "node:fs"; +import { dirname, join } from "node:path"; + +export class UnmountJournal { + /** @param {string} stateDir the Host's state folder (config.js: stateDir) */ + constructor(stateDir) { + this.file = join(stateDir, "unmounts.jsonl"); + } + + /** Appends one record; a failed write is thrown to the caller that records it. */ + append(record) { + mkdirSync(dirname(this.file), { recursive: true }); + appendFileSync(this.file, `${JSON.stringify(record)}\n`); + } + + /** The records of one project, oldest first; other projects' and a torn last line are skipped. */ + records(project) { + if (!existsSync(this.file)) return []; + const out = []; + for (const raw of readFileSync(this.file, "utf8").split("\n")) { + if (!raw) continue; + let record; + try { record = JSON.parse(raw); } catch { continue; } // a torn last line + if (record && record.project === project) out.push(record); + } + return out; + } +} diff --git a/packages/host/tests/host.mjs b/packages/host/tests/host.mjs index bfae629..880f03b 100644 --- a/packages/host/tests/host.mjs +++ b/packages/host/tests/host.mjs @@ -17,6 +17,8 @@ export async function resolve(specifier, context, next) { const { resolveConfig, runtimeState } = await import("../src/config.js"); const { BindingStore } = await import("../src/binding.js"); const { Collector } = await import("../src/collector.js"); +const { UnmountJournal } = await import("../src/unmount-journal.js"); +const { UnmountOutcomes } = await import("@dsh-layerfs/model/unmount-count.js"); const { BIN, contributor, projectContext, shimOnPath } = await import("../src/env.js"); const { spawnSync } = await import("node:child_process"); const { registerRoutes } = await import("../src/route.js"); @@ -44,8 +46,9 @@ function host() { const live = new Map(); const ctx = { sessions: { get: (id) => live.get(id) } }; const bindings = new BindingStore(spec.stateDir); - const collector = new Collector(ctx, spec, bindings); - return { spec, live, ctx, bindings, collector, env: contributor(ctx, spec, bindings, collector) }; + const outcomes = new UnmountOutcomes(new UnmountJournal(spec.stateDir)); + const collector = new Collector(ctx, spec, bindings, outcomes); + return { spec, live, ctx, bindings, collector, outcomes, env: contributor(ctx, spec, bindings, collector) }; } const trace = (id, method, phase, extra = {}) => JSON.stringify({ layerfs_trace: 1, id, step: "call_1", method, args: {}, start: "2026-10-16T00:00:00.000Z", phase, ...(phase === "end" ? { ms: 1, ok: true, daemon: { t: 1 }, scope: "x" } : {}), ...extra }) + "\n"; @@ -153,6 +156,40 @@ try { rmSync(h.spec.traceDir, { recursive: true, force: true }); assert.deepEqual(host().collector.snapshot("session-root").lines.map((l) => l.method), ["workspace.mount", "workspace.exec"]); + // A traced unmount's abort path (layerfs #501, tokenaissance/layerfs#2): the + // `unmount` section of its daemon telemetry is counted by stage and abort + // disposition and journaled under the state folder, on success and on a + // retained stop; a plain unmount has no section and adds nothing. + { + const unmounted = session("session-unmount", FOLDER), traced = host(); + traced.live.set("child-unmount", unmounted); + traced.bindings.setAgent("child-unmount", "session-unmount", "tracer"); + const file = traced.collector.track({ id: "child-unmount", root: "session-unmount", cwd: FOLDER }, { name: "website" }); + const section = (fields) => [{ timing: [] }, { timing: [], unmount: fields }]; + const unmount = (id, extra) => trace(id, "workspace.unmount", "end", { sandbox: "1", path: "/workspaces/1", ...extra }); + const at = Date.parse("2026-10-16T00:00:00.000Z") + 1; + appendFileSync(file, unmount("u-1", { kind: "Unmounted", daemon: section({ stage: "Join", abort: "Written", detached: true, loops_exited: 3, loops_joined: 3, active: 0, retained: 0, waited_ms: 5000 }) })); + appendFileSync(file, unmount("u-2", { ok: false, code: "Failed", message: "Retained: TeardownCustody { stage: Revoke }", daemon: section({ stage: "Revoke", abort: "Short", bytes: 0, detached: true, loops_exited: 3, loops_joined: 2, active: 1, retained: 1, waited_ms: 4000 }) })); + appendFileSync(file, unmount("u-3", { ok: false, code: "Failed", message: "Retained: TeardownCustody { stage: Abort }", daemon: section({ stage: "Abort", abort: "Failed", errno: "EIO", detached: true, loops_exited: 3, loops_joined: 2, active: 0, retained: 1, waited_ms: 4000 }) })); + traced.collector.poll(); + const outcomes = traced.collector.unmountsOf("website"); + assert.deepEqual([outcomes.total, outcomes.byStage, outcomes.byAbort, outcomes.waitedMs], [3, { Join: 1, Abort: 1, Revoke: 1, Close: 0 }, { Written: 1, Short: 1, Failed: 1 }, { total: 13_000, max: 5_000 }]); + assert.deepEqual(outcomes.records[0], { at, project: "website", agent: "tracer", id: "u-1", stage: "Join", abort: "Written", detached: true, loopsExited: 3, loopsJoined: 3, active: 0, retained: 0, waitedMs: 5000, reply: "Unmounted" }); + assert.deepEqual(outcomes.records.map((r) => [r.id, r.reply, r.stage, r.abort, "bytes" in r ? r.bytes : null, "errno" in r ? r.errno : null, r.loopsJoined, r.active, r.retained]), + [["u-1", "Unmounted", "Join", "Written", null, null, 3, 0, 0], ["u-2", "Failed", "Revoke", "Short", 0, null, 2, 1, 1], ["u-3", "Failed", "Abort", "Failed", null, "EIO", 2, 0, 1]]); + // a plain unmount's telemetry (timing only), and a call with no section at all + appendFileSync(file, unmount("u-4", { kind: "Unmounted", daemon: [{ timing: [] }] })); + appendFileSync(file, trace("u-5", "workspace.exec", "end", { sandbox: "1", path: "/workspaces/1", exitCode: 0 })); + traced.collector.poll(); + assert.equal(traced.collector.unmountsOf("website").total, 3, "a plain unmount records nothing"); + // the journal is one JSON line per outcome, and a restarted Host reads the aggregate from it + const journaled = readFileSync(join(traced.spec.stateDir, "unmounts.jsonl"), "utf8").trim().split("\n").map((raw) => JSON.parse(raw)); + assert.deepEqual(journaled.map((r) => [r.id, r.project, r.agent, r.stage, r.abort]), [["u-1", "website", "tracer", "Join", "Written"], ["u-2", "website", "tracer", "Revoke", "Short"], ["u-3", "website", "tracer", "Abort", "Failed"]]); + assert.deepEqual(host().collector.unmountsOf("website"), outcomes); + assert.deepEqual([traced.collector.unmountsOf("other").total, traced.collector.unmountsOf("other").byStage.Close], [0, 0]); + assert.ok(unmounted.records.every((r) => !("daemon" in r.data.line)), "no daemon telemetry in records"); + } + // Windows traces live under the writable workspace. Keep their location // across renaming and a Host restart, including a line not yet journaled. { diff --git a/packages/model/src/unmount-count.js b/packages/model/src/unmount-count.js new file mode 100644 index 0000000..4027e66 --- /dev/null +++ b/packages/model/src/unmount-count.js @@ -0,0 +1,63 @@ +// The Use Case: an unmount outcome the Host observes is journaled through +// the port, and the outcomes a project journaled are counted by stage and +// abort disposition. The port is an interface the Host's adapter implements +// (packages/host/src/unmount-journal.js, the state folder's unmounts.jsonl); +// this module holds no I/O, and it depends only on the Entities value +// (unmount.js). A plain unmount has no section and is not recorded. +import { ABORTS, STAGES, outcomeOf } from "./unmount.js"; + +/** + * Counts of journaled outcomes: by stage and abort disposition, with the + * waited time's total and longest. A value outside the stable sets is counted + * under its own key, so a later daemon's stage is not lost. + * @param {object[]} records journal records + */ +export function countOf(records) { + const counts = { + total: 0, + byStage: Object.fromEntries(STAGES.map((stage) => [stage, 0])), + byAbort: Object.fromEntries(ABORTS.map((abort) => [abort, 0])), + waitedMs: { total: 0, max: 0 }, + }; + for (const record of records) { + if (!record) continue; + counts.total += 1; + counts.byStage[record.stage] = (counts.byStage[record.stage] ?? 0) + 1; + counts.byAbort[record.abort] = (counts.byAbort[record.abort] ?? 0) + 1; + const waited = Number.isSafeInteger(record.waitedMs) ? record.waitedMs : 0; + counts.waitedMs.total += waited; + counts.waitedMs.max = Math.max(counts.waitedMs.max, waited); + } + return counts; +} + +/** + * The unmount outcomes of traced unmounts: each is journaled through the + * port, and a project's aggregate is counted from the journal, so the counts + * outlive the Host and always match the journal. + */ +export class UnmountOutcomes { + /** @param {{ append(record: object): void, records(project: string): object[] }} journal the journal port */ + constructor(journal) { + this.journal = journal; + } + + /** + * Records one finished trace line's unmount outcome, when it has one. + * @param {{ project: string|null, agent: string|null, at: number|null, line: object }} seen + * @returns {object|null} the journaled record, or null for a line with no section + */ + record({ project, agent, at, line }) { + const outcome = outcomeOf(line); + if (!outcome) return null; + const record = { at: at ?? null, project: project ?? null, agent: agent ?? null, id: line?.id ?? null, ...outcome }; + this.journal.append(record); + return record; + } + + /** A project's counts and its journaled records, oldest first. */ + summaryOf(project) { + const records = this.journal.records(project); + return { ...countOf(records), records }; + } +} diff --git a/packages/model/src/unmount.js b/packages/model/src/unmount.js new file mode 100644 index 0000000..b94b03c --- /dev/null +++ b/packages/model/src/unmount.js @@ -0,0 +1,60 @@ +// The normal unmount's abort-path observation (layerfs #501, +// tokenaissance/layerfs#2) as an Entities value: the `unmount` telemetry +// section a daemon puts in a traced unmount's `Observed` reply, which the SDK +// trace line carries under `daemon`. An unmount that never stalled carries no +// section, and no other call carries one. +// +// daemon: [{ timing: [...], unmount: { +// stage: "Join" | "Abort" | "Revoke" | "Close", +// abort: "Written" | "Short" | "Failed", +// bytes: , errno: , +// detached, loops_exited, loops_joined, active, retained, waited_ms } }] +// +// Plain data with no I/O: the use case counts it (unmount-count.js) and the +// Host's adapter journals it (packages/host/src/unmount-journal.js). + +/** The boundaries the daemon reports; any other value is still counted. */ +export const STAGES = Object.freeze(["Join", "Abort", "Revoke", "Close"]); +/** The abort write's dispositions; any other value is still counted. */ +export const ABORTS = Object.freeze(["Written", "Short", "Failed"]); +/** The section's stable keys, in the daemon's order; `bytes` or `errno` is present, never both. */ +export const KEYS = Object.freeze(["stage", "abort", "bytes", "errno", "detached", "loops_exited", "loops_joined", "active", "retained", "waited_ms"]); + +/** The abort-path section of a trace line's daemon telemetry, or null. */ +export function sectionOf(line) { + const telemetry = Array.isArray(line?.daemon) ? line.daemon : []; + for (const report of telemetry) { + const section = report?.unmount; + if (section && typeof section === "object") return section; + } + return null; +} + +// a counter the daemon sends as a nonnegative integer; anything else is no count +const count = (value) => (Number.isSafeInteger(value) && value >= 0 ? value : 0); + +/** + * The outcome a traced unmount reports: the section's stable facts, plus the + * reply that carried it (`Unmounted` on success, or the refusal's code, which + * a retained stop answers as `Failed`), and nothing else. Null for a line + * with no section, or a section with no string `stage` or `abort`. + * @param {object} line an SDK trace line (its end phase) + */ +export function outcomeOf(line) { + const section = sectionOf(line); + if (!section || typeof section.stage !== "string" || typeof section.abort !== "string") return null; + return { + stage: section.stage, + abort: section.abort, + // a Short write owns its byte count; a Failed write owns its errno; never both + ...(section.abort === "Short" ? { bytes: count(section.bytes) } : {}), + ...(section.abort === "Failed" ? { errno: typeof section.errno === "string" ? section.errno : "" } : {}), + detached: section.detached === true, + loopsExited: count(section.loops_exited), + loopsJoined: count(section.loops_joined), + active: count(section.active), + retained: count(section.retained), + waitedMs: count(section.waited_ms), + reply: line?.ok === true ? (typeof line.kind === "string" ? line.kind : "Unmounted") : (typeof line.code === "string" ? line.code : "Failed"), + }; +} diff --git a/packages/model/tests/unmount.mjs b/packages/model/tests/unmount.mjs new file mode 100644 index 0000000..dd756e4 --- /dev/null +++ b/packages/model/tests/unmount.mjs @@ -0,0 +1,76 @@ +// The normal unmount's abort-path value and its counting: the `unmount` +// telemetry section of a traced unmount (layerfs #501, +// tokenaissance/layerfs#2), parsed as an Entities value, journaled through the +// use case's port and counted by stage and abort disposition. A plain unmount +// carries no section, so it records nothing. +// node packages/model/tests/unmount.mjs +import assert from "node:assert/strict"; +import { ABORTS, KEYS, STAGES, outcomeOf, sectionOf } from "../src/unmount.js"; +import { UnmountOutcomes, countOf } from "../src/unmount-count.js"; + +const START = "2026-10-29T00:00:00.000Z"; +const trace = (section, extra = {}) => ({ + layerfs_trace: 1, id: "u-1", method: "workspace.unmount", args: {}, start: START, ms: 5000, phase: "end", + ok: true, kind: "Unmounted", scope: "ab", daemon: [{ timing: [] }, ...(section ? [{ timing: [], unmount: section }] : [])], ...extra, +}); +const written = { stage: "Join", abort: "Written", detached: true, loops_exited: 3, loops_joined: 3, active: 0, retained: 0, waited_ms: 5000 }; + +// The section's stable key set, as the daemon writes it, and where it sits. +assert.deepEqual(KEYS, ["stage", "abort", "bytes", "errno", "detached", "loops_exited", "loops_joined", "active", "retained", "waited_ms"]); +assert.deepEqual(STAGES, ["Join", "Abort", "Revoke", "Close"]); +assert.deepEqual(ABORTS, ["Written", "Short", "Failed"]); +assert.deepEqual(sectionOf(trace(written)), written); +assert.equal(sectionOf(trace(null)), null); +assert.equal(sectionOf({ layerfs_trace: 1, daemon: { t: 1 } }), null, "daemon is a list of telemetry maps"); +assert.equal(sectionOf({}), null); + +// Success: `Unmounted`, an abort write that was accepted whole, no bytes or errno. +assert.deepEqual(outcomeOf(trace(written)), { + stage: "Join", abort: "Written", detached: true, loopsExited: 3, loopsJoined: 3, active: 0, retained: 0, waitedMs: 5000, reply: "Unmounted", +}); +// A short write owns its byte count; a failed write owns its errno; never both. +assert.deepEqual(outcomeOf(trace({ ...written, stage: "Revoke", abort: "Short", bytes: 0 })), { + stage: "Revoke", abort: "Short", bytes: 0, detached: true, loopsExited: 3, loopsJoined: 3, active: 0, retained: 0, waitedMs: 5000, reply: "Unmounted", +}); +assert.deepEqual(outcomeOf(trace({ ...written, stage: "Abort", abort: "Failed", errno: "EIO" })), { + stage: "Abort", abort: "Failed", errno: "EIO", detached: true, loopsExited: 3, loopsJoined: 3, active: 0, retained: 0, waitedMs: 5000, reply: "Unmounted", +}); +// A retained stop keeps the section and answers a refusal; the reply's code is kept. +assert.deepEqual(outcomeOf(trace({ ...written, stage: "Close", detached: false, retained: 2 }, { ok: false, kind: undefined, code: "Failed", message: "Retained: TeardownCustody { stage: Close }" })), { + stage: "Close", abort: "Written", detached: false, loopsExited: 3, loopsJoined: 3, active: 0, retained: 2, waitedMs: 5000, reply: "Failed", +}); +// A section outside the stable sets is kept and counted under its own key, not dropped. +assert.equal(outcomeOf(trace({ ...written, stage: "Detach" })).stage, "Detach"); +// A plain unmount has no section at all. +assert.equal(outcomeOf(trace(null)), null); +assert.equal(outcomeOf({ layerfs_trace: 1, method: "workspace.unmount", ok: true, kind: "Unmounted" }), null); +// A section that lost its strings is not an outcome. +assert.equal(outcomeOf(trace({ abort: "Written" })), null); +assert.equal(outcomeOf(trace({ stage: 7, abort: "Written" })), null); + +// Counting: zeros first, then by stage and disposition, with the waited time's total and longest. +assert.deepEqual(countOf([]), { total: 0, byStage: { Join: 0, Abort: 0, Revoke: 0, Close: 0 }, byAbort: { Written: 0, Short: 0, Failed: 0 }, waitedMs: { total: 0, max: 0 } }); +const records = [ + { stage: "Join", abort: "Written", waitedMs: 5000 }, + { stage: "Join", abort: "Short", waitedMs: 3000 }, + { stage: "Revoke", abort: "Failed", waitedMs: 4000 }, + { stage: "Detach", abort: "Written", waitedMs: 0 }, +]; +assert.deepEqual(countOf(records), { + total: 4, byStage: { Join: 2, Abort: 0, Revoke: 1, Close: 0, Detach: 1 }, byAbort: { Written: 2, Short: 1, Failed: 1 }, waitedMs: { total: 12_000, max: 5000 }, +}); + +// The use case records through its journal port only when the line has a section. +const appended = []; +const outcomes = new UnmountOutcomes({ append: (record) => appended.push(record), records: (project) => appended.filter((record) => record.project === project) }); +assert.equal(outcomes.record({ project: "p", agent: "main", at: 1, line: trace(null) }), null); +assert.equal(outcomes.record({ project: "p", agent: "main", at: 1, line: trace(written) }).stage, "Join"); +outcomes.record({ project: "p", agent: "main", at: 2, line: trace({ ...written, stage: "Revoke", abort: "Short", bytes: 0 }, { ok: false, code: "Failed" }) }); +outcomes.record({ project: "q", agent: "main", at: 3, line: trace({ ...written, stage: "Close", abort: "Failed", errno: "EIO" }) }); +assert.deepEqual(appended.map((record) => [record.project, record.id, record.stage, record.abort]), [["p", "u-1", "Join", "Written"], ["p", "u-1", "Revoke", "Short"], ["q", "u-1", "Close", "Failed"]]); +assert.deepEqual(outcomes.summaryOf("p"), { + total: 2, byStage: { Join: 1, Abort: 0, Revoke: 1, Close: 0 }, byAbort: { Written: 1, Short: 1, Failed: 0 }, waitedMs: { total: 10_000, max: 5000 }, + records: [appended[0], appended[1]], +}); +assert.deepEqual([outcomes.summaryOf("other").total, outcomes.summaryOf("other").records], [0, []]); +console.log("PASS model unmount: the abort-path value, its counts and its journal port");