diff --git a/architecture.md b/architecture.md index 097526722..046eb3c28 100644 --- a/architecture.md +++ b/architecture.md @@ -57,6 +57,20 @@ prove that the earlier effect happened. These distinctions govern Start with the properties every execution shares: predictable behavior, scoped ownership and composition across independently loaded package copies. +### Internal persistence during prerelease + +XMD is in prerelease. Each kind of internally stored data has exactly one +supported format: the current format. We change that format in place. We do not +preserve earlier formats or add version branches, migrations, upgrade paths, or +compatibility adapters for earlier development builds. This prohibition includes +manually invoked paths. + +Readers validate the complete current declaration and refuse incomplete, +obsolete or extra fields. This applies to session ownership as well as workflow +storage and other internal records. A retained schema label is not a promise to +accept another shape. Third-party protocol versions and public artifact +contracts keep their separately stated requirements. + ### Security model Installed plugins, providers, middleware and TypeScript components are trusted @@ -81,7 +95,7 @@ receive. Missing fields and invalid values in a JSON response are relevant; deliberately malicious JavaScript getters require executable code covered by the trusted-extension assumption. -Saved history is checked for format, supported version and consistency with +Saved history is checked for its current format and consistency with the intended run and effects. These checks protect continuation from corrupted or mismatched records. XMD makes no authenticity guarantee against someone who can deliberately rewrite the underlying local database and construct a @@ -3533,6 +3547,13 @@ agent session right now, across processes. It is passed directly into the provid needs it and is deliberately not contextual: a decision document middleware could replace is not one. Acquisition never waits. +The same host also builds an ownership-administration view for the CLI alone. +Inspection reads one exact digest's record bytes without probing a lock or +contacting a provider. Explicit force compares their SHA-256 under the same +occupancy reservation and persistent sidecar lock used by ordinary acquisition. +A live owner stays busy. This capability never enters provider dependencies or +an authored component. + #### Construction routes A construction route records how one logical agent session was first constructed — @@ -3586,13 +3607,30 @@ provider-returned, which is the only compatibility inference. Quiescence is the one-use acknowledgement an ownership body gives. It does not mean "I finished" — it means nothing this owner started can still act on the session, which is -why a handoff that could not release its handle never gives it. +why a handoff that could not release its handle never gives it. Ordinary release +writes idle only after this acknowledgement. Pending host I/O settles before +locks, file handles or occupancy are released, including calls that finish +after cancellation. #### Unproven release A recovery tombstone is an ownership record left active because its owner never proved it stopped. A crash releases the kernel lock and not this, and no pid, elapsed time, -released lock or empty transcript clears it. +released lock or empty transcript clears it. An operator can explicitly accept +that uncertainty with `xmd session recover --key-digest D --expect R --force`. +The exact byte revision must still match under the lock. Forced release records +backend quiescence unproved and interrupted outcome unknown; it proves neither +that prior work finished nor that a backend stopped. + +Ownership has one current unversioned record with release state and a nullable +latest forced-release marker. Force writes the marker; later active and genuinely +quiesced writes preserve it. Obsolete shapes refuse under the +[prerelease persistence policy](#internal-persistence-during-prerelease). Publication uses a fresh +exclusive staging file, whole writes, file sync, atomic replacement and directory +sync. Cancellation before replacement removes only that attempt's staging; +a replacement already in flight settles and is synced while the lock is held. +A post-replacement failure reports publication unknown and permits no automatic +repeat. Routes, provider identities and journals are separate and unchanged. #### Provider partitions diff --git a/packages/cli/src/cli.ts b/packages/cli/src/cli.ts index 676a83eb9..47bafa60f 100755 --- a/packages/cli/src/cli.ts +++ b/packages/cli/src/cli.ts @@ -1,3 +1,4 @@ +import { runSessionOwnership } from "./session-ownership.ts"; import { runSidekickCommand, SIDEKICK_HELP } from "./sidekick-acp/command.ts"; import { SidekickTransport } from "./sidekick-acp/protocol.ts"; import { isAbsolute, resolve } from "node:path"; @@ -572,6 +573,7 @@ const xmd = program({ agent: { ...agentConfig, description: AGENT_DESCRIPTION }, upgrade: { ...upgradeConfig, description: UPGRADE_DESCRIPTION }, repl: { ...replConfig, description: REPL_DESCRIPTION }, + session: { description: "Inspect or explicitly reclaim session ownership.", ...object({}) }, sidekick: { description: "Attach to or passively read one live Sidekick.", ...object({}) }, "test-agent": testAgentConfig, workflow: workflowConfig, @@ -2539,6 +2541,7 @@ const COMMAND_NAMES = [ "agent", "repl", "sidekick", + "session", "test-agent", "workflow", ]; @@ -3467,6 +3470,13 @@ export function* runXmd( // restore. replHost?: ReplHostAssembly, ): Operation { + if (args[0] === "session") { + const code = yield* runSessionOwnership(args.slice(1), sessions?.ownershipAdministration); + if (code !== 0) { + yield* exit(code); + } + return; + } if (args[0] === "sidekick") { try { yield* runSidekickCommand(args.slice(1)); diff --git a/packages/cli/src/session-coordinator.ts b/packages/cli/src/session-coordinator.ts index a090b9023..6f913e418 100644 --- a/packages/cli/src/session-coordinator.ts +++ b/packages/cli/src/session-coordinator.ts @@ -14,11 +14,12 @@ */ import { homedir } from "node:os"; import { join } from "node:path"; -import { - createDenoAgentSessionCoordinator, - createDenoExecutableObserver, +import { createDenoAgentSessionHost, createDenoExecutableObserver } from "@executablemd/runtime"; +import type { + AgentSessionCoordinator, + AgentSessionOwnershipAdministration, + ExecutableObserver, } from "@executablemd/runtime"; -import type { AgentSessionCoordinator, ExecutableObserver } from "@executablemd/runtime"; import { ADVERTISED_CLIENT_NATIVE_ATTACHMENT, ADVERTISED_NATIVE_LAUNCH, @@ -42,6 +43,7 @@ export function sessionCoordinatorRoot(): string { */ export interface MachineSessionAssembly { coordinator?: AgentSessionCoordinator; + ownershipAdministration?: AgentSessionOwnershipAdministration; routeStore?: AgentSessionRouteStore; executableObserver?: ExecutableObserver; advertiseNativeLaunch: readonly string[]; @@ -50,7 +52,7 @@ export interface MachineSessionAssembly { /** This host's session coordinator, or nothing when it cannot provide one. */ export function useSessionCoordinator(): AgentSessionCoordinator | undefined { - return createDenoAgentSessionCoordinator(sessionCoordinatorRoot()); + return createDenoAgentSessionHost(sessionCoordinatorRoot())?.coordinator; } /** @@ -78,12 +80,15 @@ export function useExecutableObserver(): ExecutableObserver | undefined { * proven against the installed CLI. */ export function useMachineSessions(): MachineSessionAssembly { + const host = createDenoAgentSessionHost(sessionCoordinatorRoot()); + const routeStore = useSessionRouteStore(); + const executableObserver = useExecutableObserver(); return { - ...(useSessionCoordinator() === undefined ? {} : { coordinator: useSessionCoordinator() }), - ...(useSessionRouteStore() === undefined ? {} : { routeStore: useSessionRouteStore() }), - ...(useExecutableObserver() === undefined - ? {} - : { executableObserver: useExecutableObserver() }), + ...(host + ? { coordinator: host.coordinator, ownershipAdministration: host.ownershipAdministration } + : {}), + ...(routeStore ? { routeStore } : {}), + ...(executableObserver ? { executableObserver } : {}), advertiseNativeLaunch: ADVERTISED_NATIVE_LAUNCH, advertiseClientNativeAttachment: ADVERTISED_CLIENT_NATIVE_ATTACHMENT, }; diff --git a/packages/cli/src/session-ownership.ts b/packages/cli/src/session-ownership.ts new file mode 100644 index 000000000..07a8dea0e --- /dev/null +++ b/packages/cli/src/session-ownership.ts @@ -0,0 +1,196 @@ +import { Err, Ok, type Result, type Operation } from "effection"; +import { agentSessionKeyDigest, AgentSessionOwnershipError } from "@executablemd/runtime"; +import type { + AgentSessionOwnershipAdministration, + AgentSessionOwnershipFailureCode, +} from "@executablemd/runtime"; + +export const SESSION_HELP = `Inspect or explicitly reclaim one session ownership record. + +xmd session ownership --key-digest D [--json] +xmd session ownership --provider P --agent-command A --session-key K [--json] +xmd session recover --key-digest D --expect R --force [--json] +xmd session --help +xmd session ownership --help +xmd session recover --help + +D and R are lowercase SHA-256 digests. A is the exact resolved configured +agent command, not an alias. Inspection reads one record without probing its lock. +--force accepts backend quiescence unproved and interrupted outcome UNKNOWN. +A live lock holder stays busy. These commands send no provider requests. +`; + +type Command = + | { kind: "help" } + | { kind: "ownership"; keyDigest: string; json: boolean } + | { kind: "recover"; keyDigest: string; expectedRevision: string; force: true; json: boolean }; +const digest = /^[0-9a-f]{64}$/; + +function parse(args: string[]): Result { + if ( + (args.length === 1 && args[0] === "--help") || + (args.length === 2 && ["ownership", "recover"].includes(args[0]!) && args[1] === "--help") + ) { + return Ok({ kind: "help" }); + } + const kind = args[0]; + if (kind !== "ownership" && kind !== "recover") { + return Err(new Error("expected session ownership, recover or --help")); + } + const values = new Map(); + for (let i = 1; i < args.length; i++) { + const token = args[i]!; + const equal = token.indexOf("="); + const name = equal < 0 ? token : token.slice(0, equal); + if ( + ![ + "--key-digest", + "--provider", + "--agent-command", + "--session-key", + "--expect", + "--force", + "--json", + ].includes(name) + ) { + return Err(new Error("unknown session option or positional argument")); + } + if (values.has(name)) { + return Err(new Error(`duplicate session option ${name}`)); + } + if (name === "--force" || name === "--json") { + if (equal >= 0) { + return Err(new Error(`${name} accepts no value`)); + } + values.set(name, "true"); + } else { + const value = equal < 0 ? args[++i] : token.slice(equal + 1); + if (value === undefined || value.length === 0 || value.startsWith("--")) { + return Err(new Error(`${name} requires a nonempty value`)); + } + values.set(name, value); + } + } + const direct = values.get("--key-digest"); + const provider = values.get("--provider"); + const agent = values.get("--agent-command"); + const sessionKey = values.get("--session-key"); + const triple = provider !== undefined || agent !== undefined || sessionKey !== undefined; + if (direct !== undefined && (!digest.test(direct) || triple)) { + return Err( + new Error( + "use one lowercase --key-digest or the complete provider/agent-command/session-key triple", + ), + ); + } + if (kind === "recover" && triple) { + return Err(new Error("recovery accepts only --key-digest")); + } + let keyDigest: string; + if (direct !== undefined) { + keyDigest = direct; + } else if (provider && agent && sessionKey) { + keyDigest = agentSessionKeyDigest({ provider, agent, sessionKey }); + } else { + return Err( + new Error("supply --key-digest or the complete provider/agent-command/session-key triple"), + ); + } + const json = values.has("--json"); + if (kind === "ownership") { + if (values.has("--expect") || values.has("--force")) { + return Err(new Error("inspection accepts neither --expect nor --force")); + } + return Ok({ kind, keyDigest, json }); + } + const expectedRevision = values.get("--expect"); + if (!expectedRevision || !digest.test(expectedRevision) || !values.has("--force")) { + return Err(new Error("recovery requires lowercase --expect and explicit --force")); + } + return Ok({ kind, keyDigest, expectedRevision, force: true, json }); +} + +function failure( + code: AgentSessionOwnershipFailureCode, + message: string, + json: boolean, + keyDigest?: string, +): number { + if (json) { + console.error( + JSON.stringify({ + schema: "agent-session-ownership-error", + code, + message, + ...(keyDigest ? { keyDigest } : {}), + ...(code === "publication-unknown" ? { applied: "unknown" } : {}), + }), + ); + } else { + console.error(`xmd session: ${message}`); + } + return 1; +} + +export function* runSessionOwnership( + args: string[], + administration?: AgentSessionOwnershipAdministration, +): Operation { + const command = parse(args); + if (!command.ok) { + return failure("record-invalid", command.error.message, args.includes("--json")); + } + const input = command.value; + if (input.kind === "help") { + console.log(SESSION_HELP); + return 0; + } + if (!administration) { + const error = new AgentSessionOwnershipError("unsupported-capability"); + return failure(error.code, error.message, input.json, input.keyDigest); + } + try { + if (input.kind === "ownership") { + const result = yield* administration.inspectOwnership(input.keyDigest); + if (!result.ok) { + const error = + result.error instanceof AgentSessionOwnershipError + ? result.error + : new AgentSessionOwnershipError("io-failure"); + return failure(error.code, error.message, input.json, input.keyDigest); + } + const value = result.value; + console.log( + input.json + ? JSON.stringify({ schema: "agent-session-ownership-inspection", ...value }) + : `Session ${value.keyDigest}\nState: ${value.record.state}\nOwner: ${value.record.ownerKind} ${value.record.operationId}\nRecord revision: ${value.recordRevision}`, + ); + } else { + const result = yield* administration.forceRecoverOwnership({ + keyDigest: input.keyDigest, + expectedRevision: input.expectedRevision, + force: true, + }); + if (!result.ok) { + const error = + result.error instanceof AgentSessionOwnershipError + ? result.error + : new AgentSessionOwnershipError("io-failure"); + return failure(error.code, error.message, input.json, input.keyDigest); + } + const value = result.value; + console.log( + input.json + ? JSON.stringify({ schema: "agent-session-force-recovery", ...value }) + : `Session ${value.keyDigest}\nForced reclaim: ${value.recoveryId}\nBackend quiescence: unproved\nInterrupted outcome: UNKNOWN\nProvider requests: none\nRecord revision: ${value.recordRevision}`, + ); + } + return 0; + } catch (cause) { + const error = + cause instanceof AgentSessionOwnershipError + ? cause + : new AgentSessionOwnershipError("io-failure"); + return failure(error.code, error.message, input.json, input.keyDigest); + } +} diff --git a/packages/cli/tests/agent-session-coordinator.test.ts b/packages/cli/tests/agent-session-coordinator.test.ts index 6afe36852..bf852d4fa 100644 --- a/packages/cli/tests/agent-session-coordinator.test.ts +++ b/packages/cli/tests/agent-session-coordinator.test.ts @@ -573,3 +573,92 @@ describe("Tier HC — host session ownership", () => { expect(useExecutableObserver() === undefined).toBe(!onDeno()); }); }); + +import { + parseAgentSessionOwnership, + serializeAgentSessionOwnership, + agentSessionKeyDigest, +} from "@executablemd/runtime"; +import { hostAcpDependencies, resolvePlanWriterStack } from "../src/agent-stack.ts"; + +describe("Current ownership format and administration assembly", () => { + it("HC13: one closed current shape retains force facts and refuses format selection", function* () { + const keyDigest = agentSessionKeyDigest({ + provider: "acpx", + agent: "fixture", + sessionKey: "fixture", + }); + const current = { + keyDigest, + state: "idle", + ownerKind: "session", + operationId: "old", + release: "quiesced", + lastForcedRecovery: null, + }; + const marker = { + recoveryId: "00000000-0000-4000-8000-000000000001", + abandonedOwnerKind: "prompt", + abandonedOperationId: "old", + expectedRevision: "a".repeat(64), + backendQuiescenceProved: false, + interruptedOutcome: "unknown", + }; + const forced = { + ...current, + ownerKind: "prompt", + release: "forced", + lastForcedRecovery: marker, + }; + for (const value of [current, forced]) { + const parsed = parseAgentSessionOwnership(value); + expect(parsed).toEqual(value); + if (!parsed) { + throw new Error("valid fixture refused"); + } + expect(JSON.parse(serializeAgentSessionOwnership(parsed))).not.toHaveProperty("schema"); + expect( + parseAgentSessionOwnership(JSON.parse(serializeAgentSessionOwnership(parsed))), + ).toEqual(value); + } + const { release: _release, lastForcedRecovery: _last, ...incomplete } = current; + for (const invalid of [ + incomplete, + { schema: "agent-session-ownership.v1", ...incomplete }, + { schema: "agent-session-ownership.v2", ...forced }, + { schema: "agent-session-ownership.v3", ...forced }, + { ...current, extra: true }, + { ...forced, lastForcedRecovery: null }, + { ...forced, lastForcedRecovery: { ...marker, extra: true } }, + { ...forced, lastForcedRecovery: { ...marker, backendQuiescenceProved: true } }, + { ...forced, state: "active" }, + { ...forced, release: "none" }, + { ...forced, operationId: "different" }, + { ...forced, lastForcedRecovery: { ...marker, recoveryId: "not-a-uuid" } }, + { ...forced, keyDigest: "BAD" }, + ]) { + expect(parseAgentSessionOwnership(invalid)).toBe(undefined); + } + for (const value of [ + { ...forced, state: "active", release: "none", operationId: "next" }, + { ...forced, release: "quiesced", operationId: "next" }, + { ...current, state: "active", release: "none" }, + ]) { + expect(parseAgentSessionOwnership(value)).not.toBe(undefined); + } + }); + it("HC14: administration stays in host assembly and never crosses the provider whitelist", function* () { + const assembly = useMachineSessions(); + expect(assembly.ownershipAdministration === undefined).toBe(!onDeno()); + const stack = yield* resolvePlanWriterStack( + { agentProvider: "acpx", defaultAgent: "claude" }, + assembly, + ); + if (!stack.ok) { + throw stack.error; + } + const dependencies = hostAcpDependencies(stack.value); + expect(Reflect.has(dependencies, "ownershipAdministration")).toBe(false); + expect(dependencies.coordinator).toBe(assembly.coordinator); + }); +}); diff --git a/packages/cli/tests/cli-help.test.ts b/packages/cli/tests/cli-help.test.ts index 145667e34..9747d35ea 100644 --- a/packages/cli/tests/cli-help.test.ts +++ b/packages/cli/tests/cli-help.test.ts @@ -477,3 +477,30 @@ describe("Sidekick attachment help", () => { } }); }); + +describe("Session ownership help", () => { + it("describes exact-target inspection and explicit force without initializing a session", function* () { + expect((yield* runCli(["--help"]).expect()).stdout).toContain("session"); + for (const args of [ + ["session", "--help"], + ["session", "ownership", "--help"], + ["session", "recover", "--help"], + ]) { + const result = yield* runCli(args).expect(); + for (const word of [ + "--key-digest", + "--provider", + "--agent-command", + "--session-key", + "--expect", + "--force", + "--json", + "unproved", + "UNKNOWN", + "no provider requests", + ]) { + expect(result.stdout).toContain(word); + } + } + }); +}); diff --git a/packages/cli/tests/session-ownership-cli.test.ts b/packages/cli/tests/session-ownership-cli.test.ts new file mode 100644 index 000000000..ca9372cb8 --- /dev/null +++ b/packages/cli/tests/session-ownership-cli.test.ts @@ -0,0 +1,336 @@ +import { describe, it } from "@executablemd/test-support/bdd"; +import { expect } from "@executablemd/test-support/expect"; +import { useTempDirectory } from "@executablemd/test-support/temp"; +import { cliRuntime, runCli } from "@executablemd/test-support/launch"; +import { ensureDir, writeTextFile, readTextFile } from "@effectionx/fs"; +import { chmod, readdir, stat } from "node:fs/promises"; +import { join } from "node:path"; +import { createHash } from "node:crypto"; +import { createContext, ensure, scoped, until, Ok, Err, type Operation } from "effection"; +import { + agentSessionKeyDigest, + AgentSessionOwnershipError, + serializeAgentSessionOwnership, + parseAgentSessionOwnership, +} from "@executablemd/runtime"; +import type { + AgentSessionOwnershipAdministration, + AgentSessionOwnershipRecord, +} from "@executablemd/runtime"; +import { runXmd } from "../src/cli.ts"; +import { SOURCE_UPGRADE } from "./support/upgrade-assembly.ts"; + +const D = agentSessionKeyDigest({ + provider: "acpx", + agent: "poison-command", + sessionKey: "fixture-only", +}); +const R = "a".repeat(64); +const record: AgentSessionOwnershipRecord = { + release: "none", + lastForcedRecovery: null, + keyDigest: D, + state: "active", + ownerKind: "prompt", + operationId: "fixture-abandoned", +}; +const ExitContext = createContext<(value: { status: number }) => Operation>("exit"); + +function invoke(args: string[], administration?: AgentSessionOwnershipAdministration) { + return scoped(function* () { + let status = 0; + let touched = 0; + const stdout: string[] = []; + const stderr: string[] = []; + const log = console.log; + const warn = console.error; + yield* ensure(() => { + console.log = log; + console.error = warn; + }); + console.log = (text) => { + stdout.push(String(text)); + }; + console.error = (text) => { + stderr.push(String(text)); + }; + yield* ExitContext.set(function* (value) { + status = value.status; + }); + function* forbidden(): Operation { + touched++; + throw new Error("session reached an unrelated host capability"); + } + yield* runXmd( + args, + forbidden, + SOURCE_UPGRADE, + forbidden, + forbidden, + forbidden, + { + ownershipAdministration: administration, + advertiseNativeLaunch: [], + advertiseClientNativeAttachment: [], + get coordinator(): never { + touched++; + throw new Error("provider assembly reached"); + }, + }, + { select: forbidden, install: forbidden }, + ); + return { status, stdout, stderr, touched }; + }); +} + +describe("Session ownership command", () => { + it("SO1: runXmd routes exact capabilities before stdin, document/plugin, service, workflow, REPL or provider selection", function* () { + const calls: unknown[] = []; + const admin: AgentSessionOwnershipAdministration = { + *inspectOwnership(keyDigest) { + calls.push(["inspect", keyDigest]); + return Ok({ keyDigest, recordRevision: R, record }); + }, + *forceRecoverOwnership(request) { + calls.push(["force", request]); + return Ok({ + keyDigest: request.keyDigest, + previousRecordRevision: request.expectedRevision, + recordRevision: R, + recoveryId: "00000000-0000-4000-8000-000000000001", + release: "forced", + backendQuiescenceProved: false, + interruptedOutcome: "unknown", + providerRequestSent: false, + }); + }, + }; + const inspect = yield* invoke( + [ + "session", + "ownership", + "--provider=acpx", + "--agent-command", + "poison-command", + "--session-key=fixture-only", + "--json", + ], + admin, + ); + expect(inspect.status).toBe(0); + expect(inspect.touched).toBe(0); + expect(calls).toEqual([["inspect", D]]); + expect(JSON.parse(inspect.stdout[0]!)).toEqual({ + schema: "agent-session-ownership-inspection", + keyDigest: D, + recordRevision: R, + record, + }); + const forced = yield* invoke( + ["session", "recover", `--key-digest=${D}`, `--expect=${R}`, "--force", "--json"], + admin, + ); + expect(forced.status).toBe(0); + expect(forced.touched).toBe(0); + expect(calls[1]).toEqual(["force", { keyDigest: D, expectedRevision: R, force: true }]); + expect(JSON.parse(forced.stdout[0]!)).toMatchObject({ + schema: "agent-session-force-recovery", + release: "forced", + backendQuiescenceProved: false, + interruptedOutcome: "unknown", + providerRequestSent: false, + }); + for (const args of [["--help"], ["ownership", "--help"], ["recover", "--help"]]) { + const result = yield* invoke(["session", ...args]); + expect(result.status).toBe(0); + expect(result.touched).toBe(0); + expect(result.stdout.join()).toContain("UNKNOWN"); + } + }); + + it("SO2: malformed lines refuse before capability access; absence is unsupported", function* () { + let calls = 0; + const admin: AgentSessionOwnershipAdministration = { + *inspectOwnership() { + calls++; + throw new Error("grammar touched administration"); + }, + *forceRecoverOwnership() { + calls++; + throw new Error("grammar touched administration"); + }, + }; + for (const args of [ + [], + ["ownership"], + ["ownership", "--key-digest", "ABC"], + ["ownership", "--key-digest", D, "--key-digest", D], + ["ownership", "--key-digest", D, "--provider", "acpx"], + ["ownership", "--provider", "acpx", "--session-key", "K"], + ["ownership", "--key-digest", D, "--force"], + ["ownership", "--key-digest", D, "--expect", R], + ["ownership", "--key-digest", D, "--root", "/fixture"], + ["ownership", "--key-digest", D, "extra"], + ["ownership", "--key-digest="], + ["ownership", "--provider=", "--agent-command=A", "--session-key=K"], + ["ownership", "--key-digest", D, "--json=true"], + ["recover", "--key-digest", D, "--expect", R], + ["recover", "--key-digest", D, "--force"], + ["recover", "--key-digest", D, "--expect", R, "--force=true"], + ["recover", "--provider=P", "--agent-command=A", "--session-key=K", "--expect", R, "--force"], + ["ownership", "--plugin=poison.ts", "--json"], + ["ownership", "--key-digest", D, "--json", "--json"], + ["recover", "--key-digest", D, "--expect", R.toUpperCase(), "--force"], + ]) { + const result = yield* invoke(["session", ...args], admin); + expect([args, result.status, result.touched]).toEqual([args, 1, 0]); + } + expect(calls).toBe(0); + const unsupported = yield* invoke(["session", "ownership", "--key-digest", D, "--json"]); + expect(JSON.parse(unsupported.stderr[0]!)).toEqual({ + schema: "agent-session-ownership-error", + code: "unsupported-capability", + message: "this host cannot administer session ownership", + keyDigest: D, + }); + const uncertain = yield* invoke( + ["session", "recover", "--key-digest", D, "--expect", R, "--force", "--json"], + { + ...admin, + *forceRecoverOwnership() { + return Err( + new AgentSessionOwnershipError("publication-unknown", { + cause: new Error("private fixture path"), + }), + ); + }, + }, + ); + expect(JSON.parse(uncertain.stderr[0]!)).toMatchObject({ + code: "publication-unknown", + applied: "unknown", + }); + expect(uncertain.stderr.join()).not.toContain("private fixture path"); + }); + + it("SO3: the actual command inspects and forces only a seeded isolated HOME, with no provider or document effects", function* () { + const home = yield* useTempDirectory("xmd-ownership-home-"); + const work = yield* useTempDirectory("xmd-ownership-work-"); + const ownership = join(home, ".acpx", "xmd-native-sessions", "v1", "ownership"); + yield* ensureDir(ownership); + yield* until(chmod(ownership, 0o700)); + const path = join(ownership, D + ".json"); + const text = serializeAgentSessionOwnership(record); + yield* writeTextFile(path, text); + yield* until(chmod(path, 0o600)); + const revision = createHash("sha256").update(text).digest("hex"); + const poison = join(work, "poison.cjs"); + const contact = join(work, "provider-contact"); + const effect = join(work, "document-effect"); + yield* writeTextFile( + poison, + `#!/usr/bin/env node\nrequire('node:fs').writeFileSync(${JSON.stringify(contact)},'contacted');process.exit(1);\n`, + ); + yield* until(chmod(poison, 0o755)); + yield* writeTextFile( + join(home, ".acpx", "config.json"), + JSON.stringify({ agents: { poison: { command: poison } } }), + ); + yield* writeTextFile( + join(work, "poison.ts"), + `import { writeFileSync } from "node:fs"; writeFileSync(${JSON.stringify(effect)}, "unexpected plugin effect");`, + ); + yield* writeTextFile( + join(work, "session"), + `# Poison document\nunexpected\n`, + ); + const options = { + cwd: work, + env: { HOME: home, XDG_DATA_HOME: join(home, "share"), DEFAULT_AGENT_NAME: "poison" }, + inheritEnv: false, + }; + const inspected = yield* runCli( + ["session", "ownership", "--key-digest", D, "--json"], + options, + ).join(); + if (cliRuntime() !== "deno") { + expect(inspected.code).toBe(1); + expect(inspected.stderr).toContain("unsupported-capability"); + const recovered = yield* runCli( + ["session", "recover", "--key-digest", D, "--expect", revision, "--force", "--json"], + options, + ).join(); + expect(recovered.code).toBe(1); + expect(recovered.stderr).toContain("unsupported-capability"); + expect(yield* readTextFile(path)).toBe(text); + } else { + expect(inspected.code).toBe(0); + expect(JSON.parse(inspected.stdout)).toEqual({ + schema: "agent-session-ownership-inspection", + keyDigest: D, + recordRevision: revision, + record, + }); + expect(yield* readTextFile(path)).toBe(text); + expect((yield* until(stat(path))).mode & 0o777).toBe(0o600); + for (const args of [ + ["--expect", revision], + ["--expect", "0".repeat(64), "--force"], + ]) { + const result = yield* runCli( + ["session", "recover", "--key-digest", D, ...args, "--json"], + options, + ).join(); + expect(result.code).toBe(1); + expect(yield* readTextFile(path)).toBe(text); + } + const force = yield* runCli( + ["session", "recover", "--key-digest", D, "--expect", revision, "--force", "--json"], + options, + ).join(); + expect(force.code).toBe(0); + const receipt = JSON.parse(force.stdout); + expect(Object.keys(receipt).sort()).toEqual( + [ + "schema", + "keyDigest", + "previousRecordRevision", + "recordRevision", + "recoveryId", + "release", + "backendQuiescenceProved", + "interruptedOutcome", + "providerRequestSent", + ].sort(), + ); + expect(receipt).toMatchObject({ + schema: "agent-session-force-recovery", + previousRecordRevision: revision, + release: "forced", + backendQuiescenceProved: false, + interruptedOutcome: "unknown", + providerRequestSent: false, + }); + const bytes = yield* readTextFile(path); + expect(receipt.recordRevision).toBe(createHash("sha256").update(bytes).digest("hex")); + expect(parseAgentSessionOwnership(JSON.parse(bytes))).toMatchObject({ + state: "idle", + release: "forced", + lastForcedRecovery: { recoveryId: receipt.recoveryId }, + }); + } + const plugin = yield* runCli( + ["session", "ownership", "--key-digest", D, "--plugin", join(work, "poison.ts")], + options, + ).join(); + expect(plugin.code).toBe(1); + expect((yield* until(readdir(work))).sort()).toEqual(["poison.cjs", "poison.ts", "session"]); + expect(yield* until(readdir(join(home, ".acpx")))).toEqual([ + "config.json", + "xmd-native-sessions", + ]); + const dirs = yield* until(readdir(join(home, ".acpx", "xmd-native-sessions", "v1"))); + expect(dirs.includes("routes")).toBe(false); + expect((yield* until(readdir(home))).includes(".xmd")).toBe(false); + }); +}); diff --git a/packages/runtime/agent-session-coordinator.ts b/packages/runtime/agent-session-coordinator.ts index 9e8d0afe3..02c73603e 100644 --- a/packages/runtime/agent-session-coordinator.ts +++ b/packages/runtime/agent-session-coordinator.ts @@ -62,6 +62,13 @@ export class AgentSessionBusy extends Error { */ export class AgentSessionRecoveryRequired extends Error { override name = "AgentSessionRecoveryRequired"; + readonly keyDigest?: string; + readonly recordRevision?: string; + constructor(message: string, metadata?: { keyDigest: string; recordRevision: string }) { + super(message); + this.keyDigest = metadata?.keyDigest; + this.recordRevision = metadata?.recordRevision; + } } export interface AgentSessionCoordinator { @@ -94,12 +101,13 @@ export function agentSessionKeyDigest(key: AgentSessionKey): string { } /** The exact durable record one owner writes. No other member is accepted. */ -export interface AgentSessionOwnershipRecordV1 { - schema: "agent-session-ownership.v1"; +export interface AgentSessionOwnershipRecord { keyDigest: string; state: "active" | "idle"; ownerKind: AgentSessionOwnerKind; operationId: string; + release: "none" | "quiesced" | "forced"; + lastForcedRecovery: AgentSessionForcedRecovery | null; } const OWNER_KINDS: readonly AgentSessionOwnerKind[] = ["session", "prompt", "native-launch"]; @@ -108,52 +116,155 @@ function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } -/** - * Read an ownership record strictly. - * - * An unknown schema, a missing field, or a member this build cannot account for - * describes state it must not act on — so it is refused rather than read - * partially, and never repaired. - */ +export interface AgentSessionForcedRecovery { + recoveryId: string; + abandonedOwnerKind: AgentSessionOwnerKind; + abandonedOperationId: string; + expectedRevision: string; + backendQuiescenceProved: false; + interruptedOutcome: "unknown"; +} + +export interface AgentSessionOwnershipInspection { + keyDigest: string; + recordRevision: string; + record: AgentSessionOwnershipRecord; +} +export interface AgentSessionForceRecoveryReceipt { + keyDigest: string; + previousRecordRevision: string; + recordRevision: string; + recoveryId: string; + release: "forced"; + backendQuiescenceProved: false; + interruptedOutcome: "unknown"; + providerRequestSent: false; +} +export interface AgentSessionOwnershipAdministration { + inspectOwnership(keyDigest: string): Operation>; + forceRecoverOwnership(request: { + keyDigest: string; + expectedRevision: string; + force: true; + }): Operation>; +} +export interface AgentSessionHost { + coordinator: AgentSessionCoordinator; + ownershipAdministration: AgentSessionOwnershipAdministration; +} +export type AgentSessionOwnershipFailureCode = + | "busy" + | "stale-expectation" + | "not-found" + | "not-recovery-required" + | "record-invalid" + | "unsupported-capability" + | "publication-unknown" + | "io-failure"; +const FAILURE_MESSAGES: Record = { + busy: "another live owner holds this session", + "stale-expectation": "the ownership record changed; this expectation authorizes no recovery", + "not-found": "no ownership record exists for this digest", + "not-recovery-required": "this record does not require recovery", + "record-invalid": "the ownership record or request is invalid", + "unsupported-capability": "this host cannot administer session ownership", + "publication-unknown": "ownership publication is UNKNOWN; do not repeat recovery automatically", + "io-failure": "ownership IO failed", +}; +export class AgentSessionOwnershipError extends Error { + override name = "AgentSessionOwnershipError"; + constructor( + readonly code: AgentSessionOwnershipFailureCode, + options?: ErrorOptions, + ) { + super(FAILURE_MESSAGES[code], options); + } +} + +/** Read only the complete current ownership shape; incomplete or extra fields refuse. */ export function parseAgentSessionOwnership( value: unknown, -): AgentSessionOwnershipRecordV1 | undefined { - if (!isRecord(value) || value.schema !== "agent-session-ownership.v1") { +): AgentSessionOwnershipRecord | undefined { + if ( + !isRecord(value) || + Object.keys(value).sort().join() !== + ["keyDigest", "state", "ownerKind", "operationId", "release", "lastForcedRecovery"] + .sort() + .join() + ) { return undefined; } - const allowed = ["schema", "keyDigest", "state", "ownerKind", "operationId"]; - for (const member of Object.keys(value)) { - if (!allowed.includes(member)) { - return undefined; - } - } - const { keyDigest, state, operationId } = value; - if (typeof keyDigest !== "string" || !/^[0-9a-f]{64}$/.test(keyDigest)) { + const { keyDigest, state, operationId, release } = value; + const ownerKind = OWNER_KINDS.find((kind) => kind === value.ownerKind); + if ( + typeof keyDigest !== "string" || + !/^[0-9a-f]{64}$/.test(keyDigest) || + (state !== "active" && state !== "idle") || + typeof operationId !== "string" || + operationId.length === 0 || + !ownerKind || + (release !== "none" && release !== "quiesced" && release !== "forced") || + (state === "active") !== (release === "none") + ) { return undefined; } - if (state !== "active" && state !== "idle") { + const lastForcedRecovery = + value.lastForcedRecovery === null ? null : parseForcedRecovery(value.lastForcedRecovery); + if ( + lastForcedRecovery === undefined || + (release === "forced" && + (lastForcedRecovery === null || + ownerKind !== lastForcedRecovery.abandonedOwnerKind || + operationId !== lastForcedRecovery.abandonedOperationId)) + ) { return undefined; } - if (typeof operationId !== "string" || operationId.length === 0) { + return { keyDigest, state, ownerKind, operationId, release, lastForcedRecovery }; +} + +function parseForcedRecovery(last: unknown): AgentSessionForcedRecovery | undefined { + if ( + !isRecord(last) || + Object.keys(last).sort().join() !== + [ + "recoveryId", + "abandonedOwnerKind", + "abandonedOperationId", + "expectedRevision", + "backendQuiescenceProved", + "interruptedOutcome", + ] + .sort() + .join() + ) { return undefined; } - const ownerKind = OWNER_KINDS.find((kind) => kind === value.ownerKind); - if (!ownerKind) { + const abandonedOwnerKind = OWNER_KINDS.find((kind) => kind === last.abandonedOwnerKind); + if ( + !abandonedOwnerKind || + typeof last.abandonedOperationId !== "string" || + last.abandonedOperationId.length === 0 || + typeof last.recoveryId !== "string" || + !/^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/.test( + last.recoveryId, + ) || + typeof last.expectedRevision !== "string" || + !/^[0-9a-f]{64}$/.test(last.expectedRevision) || + last.backendQuiescenceProved !== false || + last.interruptedOutcome !== "unknown" + ) { return undefined; } - return { schema: "agent-session-ownership.v1", keyDigest, state, ownerKind, operationId }; -} - -export function serializeAgentSessionOwnership(record: AgentSessionOwnershipRecordV1): string { - return `${JSON.stringify( - { - schema: record.schema, - keyDigest: record.keyDigest, - state: record.state, - ownerKind: record.ownerKind, - operationId: record.operationId, - }, - null, - 2, - )}\n`; + return { + recoveryId: last.recoveryId, + abandonedOwnerKind, + abandonedOperationId: last.abandonedOperationId, + expectedRevision: last.expectedRevision, + backendQuiescenceProved: false, + interruptedOutcome: "unknown", + }; +} + +export function serializeAgentSessionOwnership(record: AgentSessionOwnershipRecord): string { + return `${JSON.stringify({ keyDigest: record.keyDigest, state: record.state, ownerKind: record.ownerKind, operationId: record.operationId, release: record.release, lastForcedRecovery: record.lastForcedRecovery }, null, 2)}\n`; } diff --git a/packages/runtime/deno-agent-session-coordinator.ts b/packages/runtime/deno-agent-session-coordinator.ts index 13ea24d95..f154457c9 100644 --- a/packages/runtime/deno-agent-session-coordinator.ts +++ b/packages/runtime/deno-agent-session-coordinator.ts @@ -1,313 +1,526 @@ -/** - * The Deno implementation of session ownership. - * - * Constructed at a runtime-named boundary and handed to a provider by the host - * that built it. Shared modules never reach for it, and never ask what runtime - * they are on. - * - * Exclusivity is the kernel's, and durability is the filesystem's, because the - * two questions are different. The advisory lock answers "is another live - * process in this session right now", which a crash correctly ends. The - * ownership record answers "did the last owner prove it stopped", which a crash - * must *not* end — so the record is written active before any provider work and - * replaced with idle only after the body acknowledges quiescence. A process - * that dies in between releases the lock and leaves the active record, and the - * next owner refuses rather than inferring safety from a pid, an elapsed time, - * an empty transcript, or the lock being free. - * - * A process-local occupancy table sits in front of both. Advisory locks are - * per-process on some hosts, so two provider scopes in one process could - * otherwise both believe they hold the same session. - */ - -import { ensure, Err, Ok, type Operation, scoped, until } from "effection"; -import type { Result } from "effection"; +import { ensure, Err, Ok, scoped, until, type Operation, type Result } from "effection"; +import { createHash, randomUUID } from "node:crypto"; import { join } from "node:path"; import { AgentSessionBusy, - agentSessionKeyDigest, AgentSessionRecoveryRequired, + AgentSessionOwnershipError, + agentSessionKeyDigest, parseAgentSessionOwnership, serializeAgentSessionOwnership, } from "./agent-session-coordinator.ts"; import type { AgentSessionCoordinator, - AgentSessionKey, - AgentSessionOwner, - AgentSessionOwnership, - AgentSessionOwnershipRecordV1, + AgentSessionForceRecoveryReceipt, + AgentSessionHost, + AgentSessionOwnershipInspection, + AgentSessionOwnershipRecord, } from "./agent-session-coordinator.ts"; -type HostCall = (...args: unknown[]) => unknown; - -/** One of the host's methods, bound to it, or nothing when it has none. */ -function callable(host: object, name: string): HostCall | undefined { - const member: unknown = Reflect.get(host, name); - if (typeof member !== "function") { - return undefined; - } - return (...args) => Reflect.apply(member, host, args); +export interface AgentSessionHostFile { + read(bytes: Uint8Array): Promise; + write(bytes: Uint8Array): Promise; + tryLock(exclusive: boolean): Promise; + unlock(): Promise; + sync(): Promise; + close(): void; } -/** - * The filesystem surface this adapter needs, read off the host rather than - * imported, so the module stays loadable where it is never constructed. - */ -interface CoordinatorHost { +export interface AgentSessionHostIO { mkdir(path: string, options: { recursive: boolean; mode: number }): Promise; - open(path: string, options: Record): Promise; - readTextFile(path: string): Promise; - writeTextFile(path: string, data: string, options: { mode: number }): Promise; + open(path: string, options: Record): Promise; rename(from: string, to: string): Promise; remove(path: string): Promise; } -interface HostFile { - tryLock(exclusive: boolean): Promise; - unlock(): Promise; - sync(): Promise; - close(): void; +type HostCall = (...args: unknown[]) => unknown; +function callable(host: object, name: string): HostCall | undefined { + const member: unknown = Reflect.get(host, name); + if (typeof member !== "function") { + return undefined; + } + return (...args) => Reflect.apply(member, host, args); } -function hostFile(value: unknown): HostFile | undefined { +function file(value: unknown): AgentSessionHostFile { if (typeof value !== "object" || value === null) { - return undefined; + throw new Error("host returned no file"); } + const read = callable(value, "read"); + const write = callable(value, "write"); const tryLock = callable(value, "tryLock"); const unlock = callable(value, "unlock"); const sync = callable(value, "sync"); const close = callable(value, "close"); - if (!tryLock || !unlock || !sync || !close) { - return undefined; + if (!read || !write || !tryLock || !unlock || !sync || !close) { + throw new Error("host lacks ownership file operations"); } return { - tryLock: (exclusive) => Promise.resolve(tryLock(exclusive)), - unlock: () => Promise.resolve(unlock()), - sync: () => Promise.resolve(sync()), + read: (bytes) => + Promise.resolve(read(bytes)).then((n) => { + if (n === null || typeof n === "number") { + return n; + } + throw new Error("invalid host read"); + }), + write: (bytes) => + Promise.resolve(write(bytes)).then((n) => { + if (typeof n === "number") { + return n; + } + throw new Error("invalid host write"); + }), + tryLock: (exclusive) => Promise.resolve(tryLock(exclusive)).then((v) => v === true), + unlock: () => Promise.resolve(unlock()).then(() => {}), + sync: () => Promise.resolve(sync()).then(() => {}), close: () => { close(); }, }; } -function coordinatorHost(): CoordinatorHost | undefined { - if (!Reflect.has(globalThis, "Deno")) { - return undefined; - } +/** Native IO is exported only here so host tests can gate the real operations. */ +export function createDenoAgentSessionHostIO(): AgentSessionHostIO | undefined { const host: unknown = Reflect.get(globalThis, "Deno"); if (typeof host !== "object" || host === null) { return undefined; } - const names = ["mkdir", "open", "readTextFile", "writeTextFile", "rename", "remove"] as const; - const calls: Record = {}; - for (const name of names) { - const call = callable(host, name); - if (!call) { - return undefined; - } - calls[name] = call; + const mkdir = callable(host, "mkdir"); + const open = callable(host, "open"); + const rename = callable(host, "rename"); + const remove = callable(host, "remove"); + if (!mkdir || !open || !rename || !remove) { + return undefined; } return { - mkdir: (path, options) => Promise.resolve(calls.mkdir!(path, options)), - open: (path, options) => Promise.resolve(calls.open!(path, options)), - readTextFile: (path) => - Promise.resolve(calls.readTextFile!(path)).then((value: unknown) => String(value)), - writeTextFile: (path, data, options) => - Promise.resolve(calls.writeTextFile!(path, data, options)), - rename: (from, to) => Promise.resolve(calls.rename!(from, to)), - remove: (path) => Promise.resolve(calls.remove!(path)), + mkdir: (path, options) => Promise.resolve(mkdir(path, options)), + open: (path, options) => Promise.resolve(open(path, options)).then(file), + rename: (from, to) => Promise.resolve(rename(from, to)), + remove: (path) => Promise.resolve(remove(path)), }; } -/** Whether this host can coordinate agent sessions at all. */ export function hasDenoAgentSessionCoordinator(): boolean { - return coordinatorHost() !== undefined; + return createDenoAgentSessionHostIO() !== undefined; } -function isMissing(cause: unknown): boolean { +function missing(error: unknown): boolean { return ( - (typeof cause === "object" && cause !== null && Reflect.get(cause, "code") === "ENOENT") || - (cause instanceof Error && cause.name === "NotFound") + error instanceof Error && (error.name === "NotFound" || Reflect.get(error, "code") === "ENOENT") ); } -/** - * Build the coordinator rooted at `root`. - * - * Returns nothing on a host with no such filesystem: a host that cannot answer - * the ownership question installs no coordinator, and every advertised - * provider-returned operation refuses. - */ -export function createDenoAgentSessionCoordinator( - root: string, -): AgentSessionCoordinator | undefined { - const found = coordinatorHost(); - if (!found) { - return undefined; - } - const host: CoordinatorHost = found; - // Sibling provider scopes in one process contend here before they contend in - // the kernel, because an advisory lock does not always separate them. - const occupied = new Set(); +function error(code: ConstructorParameters[0], cause?: unknown) { + return new AgentSessionOwnershipError(code, { cause }); +} - function* readOwnership(path: string): Operation { - const text = yield* until( - host.readTextFile(path).catch((cause: unknown) => { - if (isMissing(cause)) { - return undefined; +/** Settlement stays inside the caller's resource lifetime, including late answers. */ +function pending(start: () => Promise, adopt: (value: T) => void = () => {}): Operation { + return scoped(function* () { + let call: Promise> | undefined; + let consumed = false; + yield* ensure(function* () { + if (!consumed && call) { + const result = yield* until(call); + consumed = true; + if (!result.ok) { + throw result.error; } - throw cause; - }), + adopt(result.value); + } + }); + call = start().then( + (value) => Ok(value), + (cause: unknown) => + Err(cause instanceof Error ? cause : new Error("host IO failed", { cause })), ); - if (text === undefined) { - return undefined; - } - let parsed: unknown; - try { - parsed = JSON.parse(text); - } catch { - throw new AgentSessionRecoveryRequired( - "the retained agent session ownership record is not readable", - ); - } - const record = parseAgentSessionOwnership(parsed); - if (!record) { - throw new AgentSessionRecoveryRequired( - "the retained agent session ownership record is not valid", - ); + const result = yield* until(call); + consumed = true; + if (!result.ok) { + throw result.error; } - return record; - } + adopt(result.value); + return result.value; + }); +} - /** - * Replace the ownership record durably. - * - * Written whole, flushed, renamed over the destination, and the directory - * flushed behind it — so a reader sees the old record or the new one, and a - * crash cannot leave a half-written state that parses as something weaker. - */ - function* publish( - directory: string, - destination: string, - record: AgentSessionOwnershipRecordV1, - ): Operation { - const staging = `${destination}.${record.operationId}.staging`; - yield* until( - host.writeTextFile(staging, serializeAgentSessionOwnership(record), { mode: 0o600 }), - ); - const staged = hostFile(yield* until(host.open(staging, { read: true }))); - if (staged) { - yield* until(staged.sync()); - staged.close(); - } - yield* until(host.rename(staging, destination)); - const opened = hostFile(yield* until(host.open(directory, { read: true }))); - if (opened) { - yield* until(opened.sync()); - opened.close(); - } - } +const MAX_OWNERSHIP_RECORD_BYTES = 65536; +const digestPattern = /^[0-9a-f]{64}$/; +const revision = (bytes: Uint8Array) => createHash("sha256").update(bytes).digest("hex"); - return { - coordinate( - key: AgentSessionKey, - owner: AgentSessionOwner, - body: (ownership: AgentSessionOwnership) => Operation, - ): Operation> { - // Scoped, not a resource: ownership has to end when the body ends. A - // resource acquired in the caller's scope would hold the lock and the - // active record until *that* scope closed, which is a different — and - // much longer — lifetime than the work it protects. - return scoped(function* (): Operation> { - const digest = agentSessionKeyDigest(key); - const leases = join(root, "leases"); - const ownership = join(root, "ownership"); - yield* until(host.mkdir(leases, { recursive: true, mode: 0o700 })); - yield* until(host.mkdir(ownership, { recursive: true, mode: 0o700 })); - const record = join(ownership, `${digest}.json`); +export function createDenoAgentSessionHost(root: string): AgentSessionHost | undefined { + const io = createDenoAgentSessionHostIO(); + return io ? createAgentSessionHostFromIO(root, io) : undefined; +} + +export function createDenoAgentSessionCoordinator( + root: string, +): AgentSessionCoordinator | undefined { + return createDenoAgentSessionHost(root)?.coordinator; +} - if (occupied.has(digest)) { - return Err(busy(key)); +export function createAgentSessionHostFromIO( + root: string, + io: AgentSessionHostIO, +): AgentSessionHost { + const occupied = new Set(); + const leases = join(root, "leases"); + const ownership = join(root, "ownership"); + const recordPath = (digest: string) => join(ownership, `${digest}.json`); + + function read(digest: string) { + return scoped(function* () { + let handle: AgentSessionHostFile | undefined; + yield* ensure(() => { + handle?.close(); + }); + try { + yield* pending( + () => io.open(recordPath(digest), { read: true }), + (value) => { + handle = value; + }, + ); + } catch (cause) { + if (missing(cause)) { + throw error("not-found"); } - occupied.add(digest); - yield* ensure(() => { - occupied.delete(digest); - }); + throw error("io-failure", cause); + } + if (!handle) { + throw error("io-failure"); + } + const opened = handle; + const bytes = new Uint8Array(MAX_OWNERSHIP_RECORD_BYTES + 1); + let length = 0; + while (length < bytes.length) { + const n = yield* pending(() => opened.read(bytes.subarray(length))); + if (n === null) { + break; + } + if (n <= 0 || n > bytes.length - length) { + throw error("record-invalid"); + } + length += n; + } + if (length > MAX_OWNERSHIP_RECORD_BYTES) { + throw error("record-invalid"); + } + const raw = bytes.subarray(0, length); + let record: AgentSessionOwnershipRecord | undefined; + try { + record = parseAgentSessionOwnership( + JSON.parse(new TextDecoder("utf-8", { fatal: true }).decode(raw)), + ); + } catch (cause) { + throw error("record-invalid", cause); + } + if (!record || record.keyDigest !== digest) { + throw error("record-invalid"); + } + return { keyDigest: digest, recordRevision: revision(raw), record }; + }); + } - // Opened for the ownership scope and never unlinked: deleting it would - // let a second process create a fresh file and lock that instead. - const opened = hostFile( - yield* until( - host.open(join(leases, `${digest}.lease`), { - read: true, - write: true, - create: true, - mode: 0o600, - }), - ), + function publish(digest: string, record: AgentSessionOwnershipRecord) { + return scoped(function* () { + const bytes = new TextEncoder().encode(serializeAgentSessionOwnership(record)); + if (bytes.length > MAX_OWNERSHIP_RECORD_BYTES) { + throw error("record-invalid"); + } + const destination = recordPath(digest); + const staging = `${destination}.${randomUUID()}.staging`; + let staged: AgentSessionHostFile | undefined; + let directory: AgentSessionHostFile | undefined; + let created = false; + let renameStarted = false; + let replaced = false; + let synced = false; + let syncStarted = false; + let firstFailure: unknown; + yield* ensure(function* () { + try { + staged?.close(); + staged = undefined; + if (replaced && !synced && !syncStarted) { + if (!directory) { + yield* pending( + () => io.open(ownership, { read: true }), + (value) => { + directory = value; + }, + ); + } + if (!directory) { + throw error("publication-unknown"); + } + const opened = directory; + syncStarted = true; + yield* pending( + () => opened.sync(), + () => { + synced = true; + }, + ); + } + } catch (cause) { + if (firstFailure === undefined) { + throw error(renameStarted ? "publication-unknown" : "io-failure", cause); + } + } finally { + directory?.close(); + } + if (created && !replaced) { + try { + yield* pending(() => io.remove(staging)); + } catch (cause) { + if (!missing(cause) && firstFailure === undefined) { + throw error("io-failure", cause); + } + } + } + }); + try { + yield* pending( + () => io.open(staging, { write: true, createNew: true, mode: 0o600 }), + (value) => { + staged = value; + created = true; + }, ); - if (!opened) { - throw new Error("this host opened a file with no advisory-lock operations"); + if (!staged) { + throw error("io-failure"); } - let locked = false; - yield* ensure(function* () { - if (locked) { - yield* until(opened.unlock()); + const opened = staged; + let offset = 0; + while (offset < bytes.length) { + const n = yield* pending(() => opened.write(bytes.subarray(offset))); + if (n <= 0 || n > bytes.length - offset) { + throw error("io-failure"); } - opened.close(); - }); - locked = (yield* until(opened.tryLock(true))) === true; - if (!locked) { - return Err(busy(key)); + offset += n; } - - const retained = yield* readOwnership(record); - if (retained?.state === "active") { - // The lock is free and the record is not. Whoever held it last never - // said it had stopped, and nothing here can say it for them. - return Err( - new AgentSessionRecoveryRequired( - `session "${key.sessionKey}" was left owned by an XMD process that did not ` + - `finish. Nothing here can prove that owner stopped, so it stays owned until ` + - `it is recovered deliberately.`, - ), - ); + yield* pending(() => opened.sync()); + opened.close(); + staged = undefined; + renameStarted = true; + yield* pending( + () => io.rename(staging, destination), + () => { + replaced = true; + }, + ); + yield* pending( + () => io.open(ownership, { read: true }), + (value) => { + directory = value; + }, + ); + if (!directory) { + throw error("publication-unknown"); } + const dir = directory; + syncStarted = true; + yield* pending( + () => dir.sync(), + () => { + synced = true; + }, + ); + } catch (cause) { + firstFailure = cause; + throw error(renameStarted ? "publication-unknown" : "io-failure", cause); + } + return revision(bytes); + }); + } - const active: AgentSessionOwnershipRecordV1 = { - schema: "agent-session-ownership.v1", - keyDigest: digest, - state: "active", - ownerKind: owner.kind, - operationId: owner.operationId, - }; - yield* publish(ownership, record, active); + function exclusive(digest: string, body: () => Operation): Operation { + return scoped(function* () { + let claimed = false; + let handle: AgentSessionHostFile | undefined; + let locked = false; + yield* ensure(function* () { + try { + if (handle && locked) { + const opened = handle; + yield* pending(() => opened.unlock()); + } + } finally { + handle?.close(); + if (claimed) { + occupied.delete(digest); + } + } + }); + if (occupied.has(digest)) { + throw error("busy"); + } + occupied.add(digest); + claimed = true; + yield* pending(() => io.mkdir(leases, { recursive: true, mode: 0o700 })); + yield* pending(() => io.mkdir(ownership, { recursive: true, mode: 0o700 })); + yield* pending( + () => + io.open(join(leases, `${digest}.lease`), { + read: true, + write: true, + create: true, + mode: 0o600, + }), + (value) => { + handle = value; + }, + ); + if (!handle) { + throw error("io-failure"); + } + const opened = handle; + yield* pending( + () => opened.tryLock(true), + (value) => { + locked = value; + }, + ); + if (!locked) { + throw error("busy"); + } + return yield* body(); + }); + } - let quiesced = false; - // Registered before the body runs, so an idle record is published on - // every ordinary exit — and on none of the others. - yield* ensure(function* () { - if (!quiesced) { - return; + return { + coordinator: { + coordinate(key, owner, body) { + return scoped(function* () { + const digest = agentSessionKeyDigest(key); + try { + return yield* exclusive(digest, function* () { + let retained: AgentSessionOwnershipInspection | undefined; + try { + retained = yield* read(digest); + } catch (cause) { + if (!(cause instanceof AgentSessionOwnershipError) || cause.code !== "not-found") { + if ( + cause instanceof AgentSessionOwnershipError && + cause.code === "record-invalid" + ) { + throw new AgentSessionRecoveryRequired( + "the retained ownership record is invalid", + ); + } + throw cause; + } + } + if (retained?.record.state === "active") { + return Err( + new AgentSessionRecoveryRequired( + `session ownership requires recovery; inspect with xmd session ownership --key-digest ${digest} --json; explicit recovery requires --expect ${retained.recordRevision} --force`, + { keyDigest: digest, recordRevision: retained.recordRevision }, + ), + ); + } + const active: AgentSessionOwnershipRecord = { + keyDigest: digest, + state: "active", + ownerKind: owner.kind, + operationId: owner.operationId, + release: "none", + lastForcedRecovery: retained?.record.lastForcedRecovery ?? null, + }; + yield* publish(digest, active); + let quiesced = false; + yield* ensure(function* () { + if (quiesced) { + yield* publish(digest, { ...active, state: "idle", release: "quiesced" }); + } + }); + return Ok( + yield* body({ + quiesced() { + quiesced = true; + }, + }), + ); + }); + } catch (cause) { + if (cause instanceof AgentSessionOwnershipError && cause.code === "busy") { + return Err(new AgentSessionBusy("another live owner holds this session")); + } + if (cause instanceof AgentSessionOwnershipError && cause.code === "record-invalid") { + return Err(cause); + } + throw cause; } - yield* publish(ownership, record, { ...active, state: "idle" }); }); - - return Ok( - yield* body({ - quiesced() { - quiesced = true; - }, - }), - ); - }); + }, + }, + ownershipAdministration: { + *inspectOwnership(digest) { + if (!digestPattern.test(digest)) { + return Err(error("record-invalid")); + } + try { + return Ok(yield* read(digest)); + } catch (cause) { + return Err( + cause instanceof AgentSessionOwnershipError ? cause : error("io-failure", cause), + ); + } + }, + *forceRecoverOwnership(request) { + if ( + request.force !== true || + !digestPattern.test(request.keyDigest) || + !digestPattern.test(request.expectedRevision) + ) { + return Err(error("record-invalid")); + } + try { + const receipt = yield* exclusive(request.keyDigest, function* () { + const old = yield* read(request.keyDigest); + if (old.recordRevision !== request.expectedRevision) { + throw error("stale-expectation"); + } + if (old.record.state !== "active") { + throw error("not-recovery-required"); + } + const recoveryId = randomUUID(); + const record: AgentSessionOwnershipRecord = { + keyDigest: request.keyDigest, + state: "idle", + ownerKind: old.record.ownerKind, + operationId: old.record.operationId, + release: "forced", + lastForcedRecovery: { + recoveryId, + abandonedOwnerKind: old.record.ownerKind, + abandonedOperationId: old.record.operationId, + expectedRevision: request.expectedRevision, + backendQuiescenceProved: false, + interruptedOutcome: "unknown", + }, + }; + const recordRevision = yield* publish(request.keyDigest, record); + const receipt: AgentSessionForceRecoveryReceipt = { + keyDigest: request.keyDigest, + previousRecordRevision: request.expectedRevision, + recordRevision, + recoveryId, + release: "forced", + backendQuiescenceProved: false, + interruptedOutcome: "unknown", + providerRequestSent: false, + }; + return receipt; + }); + return Ok(receipt); + } catch (cause) { + return Err( + cause instanceof AgentSessionOwnershipError ? cause : error("io-failure", cause), + ); + } + }, }, }; } - -function busy(key: AgentSessionKey): AgentSessionBusy { - return new AgentSessionBusy( - `another XMD owner is using session "${key.sessionKey}" — a native UI or a turn in ` + - `another process holds it. Run this again once that owner exits.`, - ); -} diff --git a/packages/runtime/mod.ts b/packages/runtime/mod.ts index 00c65bcfb..c1b84188c 100644 --- a/packages/runtime/mod.ts +++ b/packages/runtime/mod.ts @@ -156,20 +156,28 @@ export type { HostFilesEvent, HostFilesObserver, HostFilesOptions } from "./host export { AgentSessionBusy, agentSessionKeyDigest, + AgentSessionOwnershipError, AgentSessionRecoveryRequired, parseAgentSessionOwnership, serializeAgentSessionOwnership, } from "./agent-session-coordinator.ts"; export type { AgentSessionCoordinator, + AgentSessionForcedRecovery, + AgentSessionForceRecoveryReceipt, + AgentSessionHost, AgentSessionKey, AgentSessionOwner, AgentSessionOwnerKind, AgentSessionOwnership, - AgentSessionOwnershipRecordV1, + AgentSessionOwnershipAdministration, + AgentSessionOwnershipFailureCode, + AgentSessionOwnershipInspection, + AgentSessionOwnershipRecord, } from "./agent-session-coordinator.ts"; export { createDenoAgentSessionCoordinator, + createDenoAgentSessionHost, hasDenoAgentSessionCoordinator, } from "./deno-agent-session-coordinator.ts"; export { ExecutableObservationError } from "./executable-observer.ts"; diff --git a/packages/runtime/tests/agent-session-coordinator.test.ts b/packages/runtime/tests/agent-session-coordinator.test.ts index 670aac904..2fcd22356 100644 --- a/packages/runtime/tests/agent-session-coordinator.test.ts +++ b/packages/runtime/tests/agent-session-coordinator.test.ts @@ -10,19 +10,22 @@ import { describe, it } from "@executablemd/test-support/bdd"; import { expect } from "@executablemd/test-support/expect"; import { useTempDirectory } from "@executablemd/test-support/temp"; -import { until } from "effection"; +import { until, ensure, scoped, spawn, withResolvers, all } from "effection"; import type { Operation } from "effection"; import { readFile, readdir, stat, writeFile } from "node:fs/promises"; import { join } from "node:path"; +import { createHash } from "node:crypto"; import { fileURLToPath } from "node:url"; import process from "node:process"; import { agentSessionKeyDigest, createDenoAgentSessionCoordinator, hasDenoAgentSessionCoordinator, + createDenoAgentSessionHost, parseAgentSessionOwnership, + AgentSessionOwnershipError, } from "@executablemd/runtime"; -import type { AgentSessionKey } from "@executablemd/runtime"; +import type { AgentSessionKey, AgentSessionOwnershipAdministration } from "@executablemd/runtime"; const KEY: AgentSessionKey = { provider: "acpx", agent: "scripted-cmd", sessionKey: "xmd:v1:main" }; const OTHER: AgentSessionKey = { ...KEY, sessionKey: "xmd:v1:elsewhere" }; @@ -48,8 +51,27 @@ function* retained(root: string, key: AgentSessionKey): Operation { } /** A child that owns `key` and reports when it is really holding it. */ -function* holder(root: string, key: AgentSessionKey): Operation<{ kill: () => void }> { - const child = new Deno.Command(Deno.execPath(), { +function* holder(root: string, key: AgentSessionKey): Operation<{ kill: () => Operation }> { + let child: Deno.ChildProcess | undefined; + let killed = false; + yield* ensure(function* () { + if (child) { + if (!killed) { + try { + child.kill("SIGKILL"); + } catch { + /* Already exited. */ + } + } + yield* until(child.status); + try { + yield* until(child.stderr.cancel()); + } catch { + /* Already consumed. */ + } + } + }); + child = new Deno.Command(Deno.execPath(), { args: ["run", "--allow-all", OWNER, root, key.provider, key.agent, key.sessionKey], stdout: "piped", stderr: "piped", @@ -72,9 +94,12 @@ function* holder(root: string, key: AgentSessionKey): Operation<{ kill: () => vo throw new Error(`the owner never took the session: ${seen}${failure}`); } return { - kill: () => { - child.kill("SIGKILL"); - child.stderr.cancel().catch(() => {}); + *kill() { + if (child && !killed) { + child.kill("SIGKILL"); + killed = true; + yield* until(child.status); + } }, }; } @@ -154,7 +179,7 @@ describe("Tier CO — the agent session coordinator", () => { ); expect(elsewhere.ok && elsewhere.value).toBe("ran"); - child.kill(); + yield* child.kill(); }); it("CO4: a killed owner leaves a tombstone the released lock does not clear", function* () { @@ -165,7 +190,7 @@ describe("Tier CO — the agent session coordinator", () => { const child = yield* holder(root, KEY); expect(yield* retained(root, KEY)).toMatchObject({ state: "active" }); - child.kill(); + yield* child.kill(); let outcome = yield* coordinator(root).coordinate( KEY, @@ -250,9 +275,10 @@ describe("Tier CO — the agent session coordinator", () => { expect(text).not.toContain(canary); expect(Object.keys(JSON.parse(text)).sort()).toEqual([ "keyDigest", + "lastForcedRecovery", "operationId", "ownerKind", - "schema", + "release", "state", ]); expect((yield* until(stat(file))).mode & 0o777).toBe(0o600); @@ -285,6 +311,512 @@ describe("Tier CO — the agent session coordinator", () => { } expect(raised).toBeInstanceOf(Error); - expect((raised as Error).name).toBe("AgentSessionRecoveryRequired"); + expect(raised instanceof Error && raised.name).toBe("AgentSessionRecoveryRequired"); + }); +}); + +import { + createAgentSessionHostFromIO, + createDenoAgentSessionHostIO, +} from "../deno-agent-session-coordinator.ts"; +import type { + AgentSessionHostIO, + AgentSessionHostFile, +} from "../deno-agent-session-coordinator.ts"; + +function host(root: string) { + const value = createDenoAgentSessionHost(root); + if (!value) { + throw new Error("Deno ownership host unavailable"); + } + return value; +} +function* seed(root: string) { + const value = host(root); + yield* value.coordinator.coordinate(KEY, owner("prompt", "abandoned"), function* () { + return undefined; + }); + const inspected = yield* value.ownershipAdministration.inspectOwnership( + agentSessionKeyDigest(KEY), + ); + if (!inspected.ok) { + throw inspected.error; + } + return { value, observed: inspected.value }; +} +function* forced( + admin: AgentSessionOwnershipAdministration, + digest: string, + expectedRevision: string, +) { + return yield* admin.forceRecoverOwnership({ keyDigest: digest, expectedRevision, force: true }); +} +function code(result: { ok: boolean; error?: Error }) { + return result.error instanceof AgentSessionOwnershipError ? result.error.code : undefined; +} + +function realIO() { + const io = createDenoAgentSessionHostIO(); + if (!io) { + throw new Error("Deno IO unavailable"); + } + return io; +} + +describe("Explicit ownership recovery", () => { + it("FR1: inspection is byte exact, read-only, bounded and never provisions or locks", function* () { + const root = yield* useTempDirectory("xmd-force-"); + const io = realIO(); + let locks = 0; + let writes = 0; + const probe: AgentSessionHostIO = { + ...io, + mkdir() { + writes++; + return Promise.reject(new Error("inspection mkdir")); + }, + open(path, options) { + return io.open(path, options).then((file) => ({ + ...file, + tryLock(exclusive) { + locks++; + return file.tryLock(exclusive); + }, + })); + }, + }; + const admin = createAgentSessionHostFromIO(root, probe).ownershipAdministration; + expect(code(yield* admin.inspectOwnership(agentSessionKeyDigest(KEY)))).toBe("not-found"); + expect(yield* until(readdir(root))).toEqual([]); + const seeded = yield* seed(root); + const path = join(root, "ownership", agentSessionKeyDigest(KEY) + ".json"); + const text = (yield* until(readFile(path, "utf8"))) + " \n"; + yield* until(writeFile(path, text)); + const result = yield* admin.inspectOwnership(agentSessionKeyDigest(KEY)); + expect(result.ok && result.value.recordRevision).toBe( + createHash("sha256").update(text).digest("hex"), + ); + expect(yield* until(readFile(path, "utf8"))).toBe(text); + expect(locks).toBe(0); + expect(writes).toBe(0); + for (const bytes of [ + new Uint8Array([255]), + new TextEncoder().encode(" ".repeat(65537)), + new TextEncoder().encode( + JSON.stringify({ ...seeded.observed.record, keyDigest: agentSessionKeyDigest(OTHER) }), + ), + ]) { + yield* until(writeFile(path, bytes)); + expect(code(yield* admin.inspectOwnership(agentSessionKeyDigest(KEY)))).toBe( + "record-invalid", + ); + } }); + + it("R1: oversize UTF-8 publication refuses before staging and preserves readable ownership", function* () { + const root = yield* useTempDirectory("xmd-force-size-"); + const io = realIO(); + let stagingCreates = 0; + const guarded = createAgentSessionHostFromIO(root, { + ...io, + open(path, options) { + if (options.createNew && path.endsWith(".staging")) { + stagingCreates++; + } + return io.open(path, options); + }, + }); + yield* guarded.coordinator.coordinate(KEY, owner("prompt", "a".repeat(40000)), function* () { + return undefined; + }); + const digest = agentSessionKeyDigest(KEY); + const path = join(root, "ownership", digest + ".json"); + const before = yield* until(readFile(path)); + expect(before.byteLength).toBeLessThan(65536); + const observed = yield* guarded.ownershipAdministration.inspectOwnership(digest); + if (!observed.ok) { + throw observed.error; + } + expect(observed.value.record).toMatchObject({ + state: "active", + operationId: "a".repeat(40000), + }); + const beforeStagingCreates = stagingCreates; + const result = yield* forced( + guarded.ownershipAdministration, + digest, + observed.value.recordRevision, + ); + expect(code(result)).toBe("record-invalid"); + expect(stagingCreates).toBe(beforeStagingCreates); + expect(yield* until(readFile(path))).toEqual(before); + expect(yield* guarded.ownershipAdministration.inspectOwnership(digest)).toEqual(observed); + expect(yield* until(readdir(join(root, "ownership")))).toEqual([digest + ".json"]); + + yield* guarded.coordinator.coordinate(OTHER, owner("session", "idle-owner"), function* (owned) { + owned.quiesced(); + return undefined; + }); + const otherDigest = agentSessionKeyDigest(OTHER); + const idle = yield* guarded.ownershipAdministration.inspectOwnership(otherDigest); + const otherPath = join(root, "ownership", otherDigest + ".json"); + const idleBytes = yield* until(readFile(otherPath)); + const beforeOrdinaryCreates = stagingCreates; + let entered = false; + const ordinary = yield* guarded.coordinator.coordinate( + OTHER, + owner("prompt", "é".repeat(40000)), + function* () { + entered = true; + return undefined; + }, + ); + expect(code(ordinary)).toBe("record-invalid"); + expect(entered).toBe(false); + expect(stagingCreates).toBe(beforeOrdinaryCreates); + expect(yield* until(readFile(otherPath))).toEqual(idleBytes); + expect(yield* guarded.ownershipAdministration.inspectOwnership(otherDigest)).toEqual(idle); + expect((yield* until(readdir(join(root, "ownership")))).sort()).toEqual( + [digest + ".json", otherDigest + ".json"].sort(), + ); + }); + + it("FR2: force preserves its marker through ordinary ownership and keeps the lease inode", function* () { + const root = yield* useTempDirectory("xmd-force-"); + const { value, observed } = yield* seed(root); + const digest = observed.keyDigest; + const lease = join(root, "leases", digest + ".lease"); + const inode = (yield* until(stat(lease))).ino; + expect(code(yield* forced(value.ownershipAdministration, digest, "0".repeat(64)))).toBe( + "stale-expectation", + ); + const result = yield* forced(value.ownershipAdministration, digest, observed.recordRevision); + expect(result.ok).toBe(true); + const inspect = yield* value.ownershipAdministration.inspectOwnership(digest); + if (!inspect.ok) { + throw inspect.error; + } + expect(inspect.value.record).toMatchObject({ + state: "idle", + release: "forced", + ownerKind: "prompt", + operationId: "abandoned", + lastForcedRecovery: { + expectedRevision: observed.recordRevision, + backendQuiescenceProved: false, + interruptedOutcome: "unknown", + }, + }); + expect(parseAgentSessionOwnership(inspect.value.record)).toEqual(inspect.value.record); + expect(result.ok && result.value.recordRevision).toBe(inspect.value.recordRevision); + const marker = inspect.value.record.lastForcedRecovery; + const next = yield* value.coordinator.coordinate( + KEY, + owner("session", "successor"), + function* (owned) { + const active = yield* value.ownershipAdministration.inspectOwnership(digest); + expect(active.ok && active.value.record).toMatchObject({ + state: "active", + release: "none", + lastForcedRecovery: marker, + }); + expect( + code(yield* forced(value.ownershipAdministration, digest, observed.recordRevision)), + ).toBe("busy"); + owned.quiesced(); + return true; + }, + ); + expect(next.ok).toBe(true); + const idle = yield* value.ownershipAdministration.inspectOwnership(digest); + expect(idle.ok && idle.value.record).toMatchObject({ + state: "idle", + release: "quiesced", + operationId: "successor", + lastForcedRecovery: marker, + }); + expect( + code( + yield* forced( + value.ownershipAdministration, + digest, + idle.ok ? idle.value.recordRevision : "", + ), + ), + ).toBe("not-recovery-required"); + expect((yield* until(stat(lease))).ino).toBe(inode); + for (const path of [lease, join(root, "ownership", digest + ".json")]) { + expect((yield* until(stat(path))).mode & 0o777).toBe(0o600); + } + expect((yield* until(stat(join(root, "ownership")))).mode & 0o777).toBe(0o700); + expect(yield* until(readdir(join(root, "ownership")))).toEqual([digest + ".json"]); + }); + + it("FR3: a real live holder stays busy; killed ownership still requires explicit matching force", function* () { + const root = yield* useTempDirectory("xmd-force-"); + const child = yield* holder(root, KEY); + const value = host(root); + const digest = agentSessionKeyDigest(KEY); + const observed = yield* value.ownershipAdministration.inspectOwnership(digest); + if (!observed.ok) { + throw observed.error; + } + expect( + code(yield* forced(value.ownershipAdministration, digest, observed.value.recordRevision)), + ).toBe("busy"); + const still = yield* value.ownershipAdministration.inspectOwnership(digest); + expect(still).toEqual(observed); + const other = yield* value.coordinator.coordinate( + OTHER, + owner("prompt", "elsewhere"), + function* (owned) { + owned.quiesced(); + return true; + }, + ); + expect(other.ok).toBe(true); + yield* child.kill(); + const refused = yield* value.coordinator.coordinate( + KEY, + owner("prompt", "no-force"), + function* () { + throw new Error("must not enter"); + }, + ); + expect(!refused.ok && refused.error.name).toBe("AgentSessionRecoveryRequired"); + expect( + (yield* forced(value.ownershipAdministration, digest, observed.value.recordRevision)).ok, + ).toBe(true); + expect( + (yield* value.coordinator.coordinate(KEY, owner("session", "after-force"), function* (owned) { + owned.quiesced(); + return true; + })).ok, + ).toBe(true); + }); + + it("FR4: two real recovery processes elect at most one winner; the old expectation cannot reclaim a successor", function* () { + const root = yield* useTempDirectory("xmd-force-"); + const { value, observed } = yield* seed(root); + const children: Deno.ChildProcess[] = []; + yield* ensure(function* () { + for (const child of children) { + try { + child.kill("SIGKILL"); + } catch {} + yield* until(child.status); + } + }); + for (let i = 0; i < 2; i++) { + const child = new Deno.Command(Deno.execPath(), { + args: [ + "run", + "--allow-all", + OWNER, + root, + KEY.provider, + KEY.agent, + KEY.sessionKey, + observed.recordRevision, + ], + stdin: "piped", + stdout: "piped", + stderr: "piped", + }).spawn(); + children.push(child); + } + const readers = children.map((c) => c.stdout.getReader()); + for (const reader of readers) { + const ready = yield* until(reader.read()); + expect(new TextDecoder().decode(ready.value)).toBe("recovery:ready\n"); + } + for (const child of children) { + const writer = child.stdin.getWriter(); + yield* until(writer.write(new Uint8Array([1]))); + yield* until(writer.close()); + } + const outputs = yield* all( + readers.map((reader) => + (function* () { + let text = ""; + while (true) { + const chunk = yield* until(reader.read()); + if (chunk.done) { + break; + } + text += new TextDecoder().decode(chunk.value); + } + return text; + })(), + ), + ); + expect(outputs.filter((s) => s.includes('"ok":true')).length).toBe(1); + yield* value.coordinator.coordinate(KEY, owner("prompt", "successor"), function* () { + return undefined; + }); + expect( + code( + yield* forced(value.ownershipAdministration, observed.keyDigest, observed.recordRevision), + ), + ).toBe("stale-expectation"); + const current = yield* value.ownershipAdministration.inspectOwnership(observed.keyDigest); + expect(current.ok && current.value.record.operationId).toBe("successor"); + for (const child of children) { + yield* until(child.status); + yield* until(child.stderr.cancel()); + } + }); + + for (const phase of ["open", "tryLock", "write", "sync", "rename", "directory-sync"]) { + it(`FR5: cancellation joins pending ${phase} before releasing occupancy or the lock`, function* () { + const root = yield* useTempDirectory("xmd-force-"); + const { observed } = yield* seed(root); + const io = realIO(); + const entered = withResolvers(); + const released = Promise.withResolvers(); + let gated = false; + let unlocked = false; + let synced = false; + function gate(start: () => Promise): Promise { + gated = true; + entered.resolve(); + return released.promise.then(start); + } + const wrapped: AgentSessionHostIO = { + ...io, + open(path, options) { + const opening = io.open(path, options).then((f) => { + const wrappedFile: AgentSessionHostFile = { + ...f, + tryLock(exclusive) { + return !gated && phase === "tryLock" + ? gate(() => f.tryLock(exclusive)) + : f.tryLock(exclusive); + }, + write(bytes) { + return !gated && phase === "write" ? gate(() => f.write(bytes)) : f.write(bytes); + }, + sync() { + if (path === join(root, "ownership")) { + const call = () => + f.sync().then(() => { + synced = true; + }); + return !gated && phase === "directory-sync" ? gate(call) : call(); + } + return !gated && phase === "sync" ? gate(() => f.sync()) : f.sync(); + }, + unlock() { + unlocked = true; + return f.unlock(); + }, + }; + return wrappedFile; + }); + return !gated && phase === "open" && path.endsWith(".lease") + ? opening.then((f) => gate(() => Promise.resolve(f))) + : opening; + }, + rename(from, to) { + return !gated && phase === "rename" + ? gate(() => io.rename(from, to)) + : io.rename(from, to); + }, + }; + const value = createAgentSessionHostFromIO(root, wrapped); + const task = yield* spawn(() => + forced(value.ownershipAdministration, observed.keyDigest, observed.recordRevision), + ); + yield* entered.operation; + const halting = task.halt(); + const contested = yield* forced( + value.ownershipAdministration, + observed.keyDigest, + observed.recordRevision, + ); + expect(code(contested)).toBe("busy"); + expect(unlocked).toBe(false); + if (["write", "sync", "rename", "directory-sync"].includes(phase)) { + const crossHost = yield* forced( + host(root).ownershipAdministration, + observed.keyDigest, + observed.recordRevision, + ); + expect(code(crossHost)).toBe("busy"); + } + released.resolve(); + yield* halting; + expect(unlocked).toBe(phase !== "open" && phase !== "tryLock" ? true : phase === "tryLock"); + const current = yield* host(root).ownershipAdministration.inspectOwnership( + observed.keyDigest, + ); + if (!current.ok) { + throw current.error; + } + if (phase === "rename" || phase === "directory-sync") { + expect(current.value.record).toMatchObject({ state: "idle", release: "forced" }); + expect(synced).toBe(true); + } else { + expect(current.value.record).toEqual(observed.record); + } + expect(yield* until(readdir(join(root, "ownership")))).toEqual([ + observed.keyDigest + ".json", + ]); + const again = yield* forced( + value.ownershipAdministration, + observed.keyDigest, + observed.recordRevision, + ); + expect(code(again)).not.toBe("busy"); + }); + } + + for (const phase of ["write", "directory-sync"]) { + it(`FR6: ${phase} failure retains first cause and reports publication uncertainty only after replacement`, function* () { + const root = yield* useTempDirectory("xmd-force-"); + const { observed } = yield* seed(root); + const io = realIO(); + let failed = false; + const fault = new Error("fixture-owned IO fault"); + const wrapped: AgentSessionHostIO = { + ...io, + open(path, options) { + return io.open(path, options).then((f) => ({ + ...f, + write(bytes) { + if (!failed && phase === "write") { + failed = true; + return Promise.reject(fault); + } + return f.write(bytes); + }, + sync() { + if (!failed && phase === "directory-sync" && path === join(root, "ownership")) { + failed = true; + return Promise.reject(fault); + } + return f.sync(); + }, + })); + }, + }; + const result = yield* forced( + createAgentSessionHostFromIO(root, wrapped).ownershipAdministration, + observed.keyDigest, + observed.recordRevision, + ); + expect(code(result)).toBe(phase === "write" ? "io-failure" : "publication-unknown"); + expect(!result.ok && result.error.cause).toBe(fault); + const current = yield* host(root).ownershipAdministration.inspectOwnership( + observed.keyDigest, + ); + if (!current.ok) { + throw current.error; + } + expect(current.value.record.state).toBe(phase === "write" ? "active" : "idle"); + expect(yield* until(readdir(join(root, "ownership")))).toEqual([ + observed.keyDigest + ".json", + ]); + }); + } }); diff --git a/packages/runtime/tests/fixtures/agent-session-owner.ts b/packages/runtime/tests/fixtures/agent-session-owner.ts index 9b0bfd6d1..93e2e89e2 100644 --- a/packages/runtime/tests/fixtures/agent-session-owner.ts +++ b/packages/runtime/tests/fixtures/agent-session-owner.ts @@ -9,12 +9,46 @@ * It lives in the repository rather than a temporary directory because it * imports workspace packages, which resolve only inside the project. */ -import { main, suspend } from "effection"; -import { createDenoAgentSessionCoordinator } from "@executablemd/runtime"; +import { main, suspend, until } from "effection"; +import { + createDenoAgentSessionCoordinator, + createDenoAgentSessionHost, + agentSessionKeyDigest, +} from "@executablemd/runtime"; -const [root, provider, agent, sessionKey] = Deno.args; +const [root, provider, agent, sessionKey, expectedRevision] = Deno.args; await main(function* () { + if (expectedRevision) { + const host = createDenoAgentSessionHost(root!); + if (!host) { + throw new Error("fixture host unavailable"); + } + console.log("recovery:ready"); + const reader = Deno.stdin.readable.getReader(); + try { + yield* until(reader.read()); + } finally { + reader.releaseLock(); + } + const result = yield* host.ownershipAdministration.forceRecoverOwnership({ + keyDigest: agentSessionKeyDigest({ + provider: provider!, + agent: agent!, + sessionKey: sessionKey!, + }), + expectedRevision, + force: true, + }); + console.log( + JSON.stringify( + result.ok + ? { ok: true, receipt: result.value } + : { ok: false, code: Reflect.get(result.error, "code") }, + ), + ); + return; + } const coordinator = createDenoAgentSessionCoordinator(root!); if (!coordinator) { console.log("owner:unavailable"); diff --git a/specs/acp-client-spec.md b/specs/acp-client-spec.md index f25048ace..a0adfb5b5 100644 --- a/specs/acp-client-spec.md +++ b/specs/acp-client-spec.md @@ -104,6 +104,14 @@ routes an opaque request and cannot manufacture that result. effect — no availability probe, no runtime, no route read, no identity, no child. An agent that is not advertised keeps ordinary ACP behavior on every host. +- Exact-target session ownership inspection and explicit operator force are + separate CLI capabilities. They never enter provider dependencies. Default + acquisition remains conservative; force compares exact record bytes under the + same lock and records backend quiescence unproved and interrupted outcome + unknown. A live owner stays busy. The one current ownership shape retains + the latest forced-release marker across ordinary ownership. Obsolete shapes + refuse; there is no ownership format selection or compatibility reader. + Recovery sends no provider request and repairs no identity, route or journal. - The host also passes in a **construction-route store** and an **executable observer**, directly and from the same trusted root as the coordinator. Both are required only for an advertised agent whose adapter binds a build. Which diff --git a/specs/native-agent-session-launch-spec.md b/specs/native-agent-session-launch-spec.md index f7d2a0cd2..695f471ba 100644 --- a/specs/native-agent-session-launch-spec.md +++ b/specs/native-agent-session-launch-spec.md @@ -760,8 +760,8 @@ An active record is a recovery tombstone. A crash releases the kernel lock but not the record, and nothing observable afterwards distinguishes a session whose owner died mid-turn from one it left cleanly — so the next owner receives `session-recovery-required` rather than inferring safety from a pid, an elapsed -time, an empty transcript, or the lock being free. V1 exposes no recovery -operation. +time, an empty transcript, or the lock being free. Explicit operator recovery +accepts this uncertainty without representing it as quiescence. The Deno and compiled hosts combine three things a single one of which would not be enough: a process-local occupancy table, one non-blocking advisory lock, and @@ -778,13 +778,13 @@ each hold "the" session: ``` Directories are `0700` and files `0600`; lease sidecars are never unlinked. The -record is exactly `schema`, `keyDigest`, `state`, `ownerKind` and `operationId`, -and a record carrying any other member is refused rather than read partially. +record has the single current shape described below. Missing or extra members +refuse rather than read partially; ownership has no format discriminator. It holds no instructions, credential, cwd, executable path, argv, environment, transcript, or temporary path — the digest is the only name in the namespace. Node and Bun build no coordinator. Neither runtime exposes a cross-process -advisory lock, and V1 emulates none: a pid, heartbeat, or stale-file timeout +advisory lock, and neither emulates one: a pid, heartbeat, or stale-file timeout calls a paused process dead and admits two owners. So every advertised session, prompt, launch and incomplete replay refuses there with `unsupported-capability`, before contacting an agent, while read-only resolution @@ -795,6 +795,73 @@ whichever side names the session. An adapter that binds no build keeps its ordinary behavior without those dependencies. +### Exact-target ownership inspection and recovery + +```sh +xmd session ownership --key-digest D --json +xmd session ownership --provider P --agent-command A --session-key K --json +xmd session recover --key-digest D --expect R --force --json +``` + +`D` and `R` are lowercase SHA-256 digests. The triple form hashes the exact +provider, already resolved configured agent command and logical session key; +it never resolves an alias or contacts an agent. Inspection reads at most 64KiB +of one exact record, decodes strict UTF-8 and validates the complete schema and +key. Its revision hashes the exact bytes, including whitespace and newline. +It creates no directory or lease, probes no lock and makes no liveness assertion. + +Recovery requires the digest form, the inspected byte revision and explicit +`--force`. Under the shared nonblocking lock it rereads the record. Busy, missing, +invalid, idle and stale-expectation records refuse. Only an active record with +matching bytes can become idle/forced. The persistent lease inode is never +removed. A forced release neither cancels a live owner nor sends, resends, +initializes, closes or kills provider work. It changes no route, session identity, +build binding, journal or prior outcome. + +Ownership has one current unversioned shape with exactly `keyDigest`, `state`, +`ownerKind`, `operationId`, `release` and `lastForcedRecovery`. Active requires +release `none`; idle requires `quiesced` or `forced`. The latest-force member is +null before any force. A force requires a non-null object containing only a +fresh UUID `recoveryId`, `abandonedOwnerKind`, `abandonedOperationId`, the matched +`expectedRevision`, `backendQuiescenceProved:false` and +`interruptedOutcome:"unknown"`. A forced idle record retains the abandoned owner +in its outer fields. Ordinary subsequent ownership preserves this object, +writes its own active owner with release `none`, and writes idle/quiesced only +on genuine acknowledgement. Later force replaces the latest summary. + +The prerelease persistence policy amends this shape in place. There is no +ownership schema discriminator, historical-format reader, migration or upgrade. +Missing fields, extra fields and obsolete version-tagged records refuse unchanged. + +All valued options accept separate or `--name=value` spelling. Mixing digest +and triple, missing or empty values, duplicate or unknown options, positional +extras and values on `--force` or `--json` refuse. Inspection accepts no force or +expectation; recovery accepts no triple. There is no configurable root, path, +listing, history, provider selection or plugin option. `session --help`, +`session ownership --help` and `session recover --help` exit0 without capability +access. Node and Bun refuse inspection and recovery as unsupported. + +Success exits0. Refusal, grammar and persistence failures exit1 with safe stderr. +JSON mode emits one closed object and newline: inspection uses +`agent-session-ownership-inspection` with digest, recordRevision and record; +force uses `agent-session-force-recovery` with digest, previousRecordRevision, +recordRevision, recoveryId, release forced, backendQuiescenceProved false, +interruptedOutcome unknown and providerRequestSent false. Errors use +`agent-session-ownership-error`, code and safe message, a validated digest +when available, and applied unknown only for publication uncertainty. Human +output states the same distinction between forced release and proven quiescence. + +Publication exclusively creates a fresh UUID staging file, writes it whole, +syncs and closes it, atomically replaces the record, then syncs the directory +before success. Pending I/O is joined before releasing resources or occupancy. +Before rename, cancellation retains the active record and removes only its own +staging file. A rename already started settles while locked, and a successful +late replacement is directory-synced and retained. Failures after replacement +may leave publication UNKNOWN; neither rollback nor automatic repeat is allowed. +A new inspection observes current state and cannot prove that an earlier command +completed. Another operator action compares the current bytes explicitly. + + ## Construction route Ownership and construction answer different questions, and this contract keeps @@ -1226,8 +1293,9 @@ Bun keep the same advertised names and assemble none of the answers, so every advertised operation refuses before provider work rather than acting while a native UI may be in the conversation — as *Ownership and concurrency* describes. -Only ordinary `xmd run` receives that assembly. Every other command receives -none, and a host profile whose session capabilities differ from ordinary `xmd run` +Ordinary `xmd run` receives the provider view of that assembly. The `session` +command receives only its separate ownership-administration view. Other commands +receive none, and a host profile whose session capabilities differ from ordinary `xmd run` states its capability sets explicitly rather than inheriting the provider package's. The workflow Agent profile selects both sets empty and installs no native foreground launcher, so `Session.Launch` is unsupported there without @@ -1484,7 +1552,8 @@ Implementation review checks these frozen invariants: coordinator reaches the installed provider directly. 14. Every operation that can act on an advertised session takes ownership under one natural key first, contention refuses rather than queues, and an owner - that did not prove it stopped leaves the session owned. + that did not prove it stopped leaves the session owned until an explicit, + matching operator force records release with quiescence unproved. 15. A session is constructed once, by one mechanism, and its create-once route says which. Routes never convert, and the loser of a race adopts the winner rather than replacing it.