Add independently owned reconnectable execution units (WI-03)

This commit is contained in:
Timothy J. Aveni
2026-09-21 12:44:36 -07:00
parent 884b12f759
commit 89414107f9
5 changed files with 395 additions and 2 deletions
+1 -2
View File
@@ -6,7 +6,6 @@ import type {
DependencyPort, DependencyPort,
ExecutionContract, ExecutionContract,
} from "../capability-model/types.js"; } from "../capability-model/types.js";
import { hexDigest } from "../execution-world/plan.js";
export class PortableBindingError extends Error { export class PortableBindingError extends Error {
constructor( constructor(
@@ -302,5 +301,5 @@ function generateBindings(
methods: all.map((m) => ({ ...m, input: portableValueSchema(m.input), output: portableValueSchema(m.output) })), methods: all.map((m) => ({ ...m, input: portableValueSchema(m.input), output: portableValueSchema(m.output) })),
}); });
} }
return `// Generated ${remote ? "remote" : "portable"} bindings. Regenerate from the checked capability schema; do not edit.\nimport { define${mode}Registry, type ${mode}Export, type PreparedChange, type ManagedReference, type ExactReference, type ${remote ? "ServerMutationOptions" : "MutationKey"} } from ${q(options.runtimeModule ?? "@quixos/camino-replica-core")};\nexport const bindingDigest = ${q(hexDigest(schema))};\nexport const packageRevision = ${q(revision)};\nexport const target = ${q(options.target)};\nexport type Contexts = { ${contexts.join(";\n")} };\nexport type ${mode}Implementation = { ${implementations.join(";\n")} };\nexport const declarations = ${JSON.stringify(descriptors, null, 2)} as const satisfies readonly ${mode}Export[];\nexport function createRegistry(handlers: ${mode}Implementation) { const installed = { ...handlers }; return define${mode}Registry(declarations, { ${adapters.join(",\n")} }); }\n`; return `// Generated ${remote ? "remote" : "portable"} bindings. Regenerate from the checked capability schema; do not edit.\nimport { define${mode}Registry, type ${mode}Export, type PreparedChange, type ManagedReference, type ExactReference, type ${remote ? "ServerMutationOptions" : "MutationKey"} } from ${q(options.runtimeModule ?? "@quixos/camino-replica-core")};\nexport const target = ${q(options.target)};\nexport type Contexts = { ${contexts.join(";\n")} };\nexport type ${mode}Implementation = { ${implementations.join(";\n")} };\nexport const declarations = ${JSON.stringify(descriptors, null, 2)} as const satisfies readonly ${mode}Export[];\nexport function createRegistry(handlers: ${mode}Implementation) { const installed = { ...handlers }; return define${mode}Registry(declarations, { ${adapters.join(",\n")} }); }\n`;
} }
+255
View File
@@ -0,0 +1,255 @@
import type {
BoundDependency,
DependencyBinding,
PackageExport,
PersistentAttachment,
} from "../capability-model/types.js";
import { type ExecutionInstallation, type ExecutionGroupIdentity } from "./installation.js";
import { type ExecutionPlan, hexDigest, ExecutionPlanError } from "./plan.js";
export interface ExecutionEnvironment {
/** Explicit host placement. Merely including an optional module does not select it. */
readonly placement: "service" | "module";
/** Exact host executable/isolation inputs, computed by the trusted installer. */
readonly hostDigest: string;
readonly policyDigest: string;
/** Identifier of the secret/config generation; never secret material. */
readonly credentialGeneration: string;
}
export interface InstalledExecutionGroup extends ExecutionGroupIdentity {
readonly packageRevision: string;
readonly consumedPackages: readonly string[];
}
/** Conservative outbound closure. Runtime graph values, reverse callers, source
* provenance and the workspace epoch are deliberately absent from this identity. */
export function compileExecutionGroups(
plan: ExecutionPlan,
installation: ExecutionInstallation,
environments: Readonly<Record<string, ExecutionEnvironment>>,
): InstalledExecutionGroup[] {
if (installation.plan !== plan.digest || hexDigest(plan.definition) !== plan.digest)
throw new ExecutionPlanError("GROUP_PLAN_MISMATCH", plan.digest, "Expected one checked installation");
const workspace = plan.definition.workspace;
const packages = new Map(workspace.packageImports.map((p) => [p.revisionId as string, p]));
const operations = new Map(plan.definition.operations.map((op) => [op.key, op]));
const attachments = new Map<string, PersistentAttachment>(
[...workspace.sharedAttachments, ...workspace.conformances.flatMap((c) => c.privateAttachments)].map((a) => [
a.id,
a,
]),
);
const logical = (revision: string) => {
const pkg = packages.get(revision);
if (!pkg) throw new ExecutionPlanError("GROUP_PACKAGE_MISSING", revision, "Uninstalled consumed package");
return pkg.packageId;
};
const environment = (revision: string) => {
const value = environments[revision];
if (
!value ||
!["service", "module"].includes(value.placement) ||
!/^[0-9a-f]{64}$/.test(value.hostDigest) ||
!/^[0-9a-f]{64}$/.test(value.policyDigest) ||
!value.credentialGeneration ||
value.credentialGeneration.length > 4096
)
throw new ExecutionPlanError(
"GROUP_ENVIRONMENT_MISSING",
revision,
"Expected checked host/policy digests and a credential generation",
);
return value;
};
const implementations = new Map(
installation.packages.map((pkg) => {
if (!pkg.service && !pkg.module) return [pkg.revision, undefined] as const;
const implementation = pkg[environment(pkg.revision).placement];
if (!implementation)
throw new ExecutionPlanError(
"GROUP_IMPLEMENTATION_MISSING",
pkg.revision,
"Host selected an unavailable implementation",
);
return [pkg.revision, implementation] as const;
}),
);
const attachmentContract = (value: PersistentAttachment) => {
const { displayName: _name, ...rest } = value;
return value.kind === "state"
? rest
: { ...rest, endpoints: value.endpoints.map(({ displayName: _name, ...endpoint }) => endpoint) };
};
const bindingContract = (value: DependencyBinding) =>
value.kind === "query" ? { ...value, packageRevisionId: logical(value.packageRevisionId) } : value;
return installation.packages
.filter((p) => !!implementations.get(p.revision))
.map((root) => {
const consumed = new Map<string, unknown>(),
visited = new Set<string>(),
visitedQueries = new Set<string>(),
consumedPackages = new Set<string>();
const remember = (key: string, value: unknown) => consumed.set(key, value);
const storage = (id: string) => {
const value = attachments.get(id);
if (!value) throw new ExecutionPlanError("GROUP_STORAGE_MISSING", id, "Uninstalled consumed storage");
remember(`storage:${id}`, attachmentContract(value));
};
// Deletion may visit related objects and their owned state. Conservatively
// include the connected schema component, including possible future edges.
const lifecycle = (atom: string) => {
const visited = new Set<string>(),
pending = [atom];
const atoms = (
constraint: Extract<PersistentAttachment, { kind: "edge" }>["endpoints"][number]["constraint"],
) =>
constraint.kind === "atom"
? [constraint.atomId as string]
: workspace.conformances
.filter((c) => c.interfaceRevisionId === constraint.interfaceRevisionId)
.map((c) => c.atomId as string);
while (pending.length) {
const current = pending.pop()!;
if (visited.has(current)) continue;
visited.add(current);
for (const owner of workspace.conformances.filter((c) => c.atomId === current))
for (const attachment of owner.privateAttachments) storage(attachment.id);
for (const attachment of attachments.values())
if (attachment.kind === "edge" && attachment.endpoints.some((e) => atoms(e.constraint).includes(current))) {
storage(attachment.id);
pending.push(...attachment.endpoints.flatMap((e) => atoms(e.constraint)));
}
}
};
const targets = (binding: DependencyBinding, atom: string): string[] => {
if (binding.kind === "interface" && binding.receiver === "input")
return workspace.conformances
.filter((c) => c.interfaceRevisionId === binding.interfaceRevisionId)
.map((c) => c.atomId);
if (!("via" in binding) || !binding.via) return [atom];
storage(binding.via.edgeTypeId);
const edge = attachments.get(binding.via.edgeTypeId);
if (!edge || edge.kind !== "edge")
throw new ExecutionPlanError("GROUP_EDGE_MISSING", atom, "Missing traversal edge");
const endpoint = edge.endpoints.find((e) => e.projectionId !== binding.via!.projectionId)!;
const constraint = endpoint.constraint;
return constraint.kind === "atom"
? [constraint.atomId]
: workspace.conformances
.filter((c) => c.interfaceRevisionId === constraint.interfaceRevisionId)
.map((c) => c.atomId);
};
const query = (revision: string, id: string) => {
const key = JSON.stringify([revision, id]);
if (visitedQueries.has(key)) return;
visitedQueries.add(key);
const value = plan.definition.queries.find((q) => q.id === key);
if (!value) throw new ExecutionPlanError("GROUP_QUERY_MISSING", key, "Missing consumed query");
remember(`query:${JSON.stringify([logical(revision), id])}`, value.query.checked.definitionDigest);
for (const operation of value.operations) visit(operation);
};
const dependencies = (entry: PackageExport, bindings: readonly BoundDependency[], atom: string) => {
for (const bound of bindings) {
const target = bound.binding,
port = entry.dependencyPorts.find((p) => p.id === bound.portId)!;
if ("via" in target && target.via) storage(target.via.edgeTypeId);
if (target.kind === "state") storage(target.slotId);
else if (target.kind === "edge") storage(target.edgeTypeId);
else if (target.kind === "query") query(target.packageRevisionId, target.queryId);
else if (target.kind === "interface" && port.requirement.kind === "interface") {
const iface = workspace.interfaceImports.find((i) => i.revisionId === target.interfaceRevisionId)!;
for (const targetAtom of targets(target, atom))
for (const op of port.requirement.operations ??
iface.members.flatMap((m) => m.operations.map((o) => o.id)))
visit(JSON.stringify([targetAtom, target.interfaceRevisionId, op]));
} else if (target.kind === "creation") {
for (const seed of target.owners) {
const owner = workspace.conformances.find(
(c) => c.atomId === target.atomId && c.interfaceRevisionId === seed.interfaceRevisionId,
)!;
remember(`birth:${owner.id}`, {
id: owner.id,
atom: owner.atomId,
interface: owner.interfaceRevisionId,
initialization: owner.initialization ?? [],
});
for (const attachment of owner.privateAttachments) storage(attachment.id);
}
} else
throw new ExecutionPlanError(
"GROUP_DEPENDENCY_UNSUPPORTED",
bound.portId,
"Cannot establish execution reuse for this dependency",
);
}
};
const visit = (key: string) => {
if (visited.has(key)) return;
visited.add(key);
const operation = operations.get(key);
if (!operation) throw new ExecutionPlanError("GROUP_OPERATION_MISSING", key, "Missing consumed operation");
const binding = operation.binding;
if (binding.kind === "package") {
const pkg = packages.get(binding.packageRevisionId)!;
const implementation = implementations.get(pkg.revisionId);
if (!implementation)
throw new ExecutionPlanError(
"GROUP_IMPLEMENTATION_MISSING",
pkg.revisionId,
"Consumed package has no selected server implementation",
);
consumedPackages.add(pkg.packageId);
remember(`implementation:${pkg.packageId}`, {
digest: implementation.contentDigest,
environment: environment(pkg.revisionId),
});
const entry = pkg.exports.find((e) => e.id === binding.exportId)!;
const bindings = binding.dependencies.map((d) => ({ portId: d.portId, binding: bindingContract(d.binding) }));
const acceptance = (binding.acceptanceDependencies ?? []).map((d) => ({
portId: d.portId,
binding: bindingContract(d.binding),
}));
const contract = {
input: operation.input,
output: operation.output,
execution: operation.execution,
scope: operation.scope,
binding: {
package: pkg.packageId,
export: binding.exportId,
dependencies: bindings,
acceptanceDependencies: acceptance,
},
};
// Deduplicate aliases of the same bound export, rather than keying its owner by reverse callers.
remember(`binding:${hexDigest(contract)}`, contract);
dependencies(entry, binding.dependencies, operation.atomId);
if (entry.acceptanceExportId)
dependencies(
pkg.exports.find((e) => e.id === entry.acceptanceExportId)!,
binding.acceptanceDependencies ?? [],
operation.atomId,
);
} else {
remember(`operation:${key}`, operation);
if (binding.kind === "state") storage(binding.slotId);
if (binding.kind === "edge") storage(binding.edgeTypeId);
if (binding.kind === "lifecycle") lifecycle(operation.atomId);
}
};
for (const operation of plan.definition.operations)
if (operation.binding.kind === "package" && operation.binding.packageRevisionId === root.revision)
visit(operation.key);
const own = implementations.get(root.revision)!;
const exports = packages.get(root.revision)!.exports.map(({ displayName: _name, dependencyPorts, ...entry }) => ({
...entry,
dependencyPorts: dependencyPorts.map(({ displayName: _name, ...port }) => port),
}));
const entries = [...consumed].sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0));
return {
logicalId: logical(root.revision),
packageRevision: root.revision,
executionDigest: `sha256:${hexDigest({ implementation: own.contentDigest, exports, environment: environment(root.revision), versions: plan.definition.versions, consumed: entries })}`,
consumedPackages: [...consumedPackages].sort(),
};
});
}
+1
View File
@@ -4,6 +4,7 @@ export * from "./ports.js";
export * from "./files.js"; export * from "./files.js";
export * from "./installation.js"; export * from "./installation.js";
export * from "./client-contracts.js"; export * from "./client-contracts.js";
export * from "./execution-groups.js";
export * from "./receipts.js"; export * from "./receipts.js";
+122
View File
@@ -0,0 +1,122 @@
import assert from "node:assert/strict";
import { test } from "node:test";
import { loadReaderWorkspace } from "./reader-fixture.js";
import { buildArtifactFixture } from "./artifact-fixture.js";
import { capabilityId as id, type WorkspaceRevision } from "../src/capability-model/types.js";
import {
compileExecutionPlan,
compileExecutionGroups,
checkExecutionWorld,
checkedWorldDigest,
type ExecutionInstallation,
type ExecutionEnvironment,
} from "../src/execution-world/index.js";
const environment: ExecutionEnvironment = {
placement: "service",
hostDigest: "a".repeat(64),
policyDigest: "b".repeat(64),
credentialGeneration: "config:1",
};
const prepare = async (workspace: WorkspaceRevision) => {
const plan = compileExecutionPlan(workspace);
const fixtures = plan.definition.packages.map((pkg) => buildArtifactFixture(plan, pkg.revision));
const world = await checkExecutionWorld(
plan,
fixtures.map((f) => f.manifest),
fixtures[0]!.read,
);
const installation: ExecutionInstallation = {
codec: "camino-execution-installation-v1",
plan: plan.digest,
world: checkedWorldDigest(world),
routes: {},
components: [],
packages: fixtures.map((f) => ({
revision: f.manifest.packageRevisionId,
root: "/fixture",
service: f.manifest.implementations[0]!,
module: f.manifest.implementations[1]!,
})),
};
const environments = Object.fromEntries(plan.definition.packages.map((pkg) => [pkg.revision, { ...environment }]));
return { plan, installation, environments, groups: () => compileExecutionGroups(plan, installation, environments) };
};
test("execution reuse ignores world/source/display changes and unrelated packages", async () => {
const workspace = await loadReaderWorkspace(),
original = await prepare(workspace);
const before = original.groups()[0]!.executionDigest;
workspace.sourceRootCommit = "f".repeat(40);
workspace.packageImports[0]!.displayName = "Renamed";
workspace.packageImports[0]!.source.commit = "e".repeat(40);
workspace.packageImports.push({
packageId: id.package("unrelated"),
revisionId: id.packageRevision("unrelated@1"),
displayName: "Other",
source: workspace.packageImports[0]!.source,
exports: [],
});
const after = await prepare(workspace);
assert.notEqual(after.plan.digest, original.plan.digest);
assert.equal(
after.groups().find((g) => g.logicalId === workspace.packageImports[0]!.packageId)!.executionDigest,
before,
);
original.environments[workspace.packageImports[0]!.revisionId]!.credentialGeneration = "config:2";
assert.notEqual(original.groups()[0]!.executionDigest, before);
original.environments[workspace.packageImports[0]!.revisionId]!.placement = "module";
assert.notEqual(original.groups()[0]!.executionDigest, before);
});
test("outbound provider changes invalidate consumers while caller changes preserve providers", async () => {
const workspace = await loadReaderWorkspace((name, text) =>
name === "DocumentData.qx" ? text.replace("queryable value content", "queryable portable value content") : text,
);
const revision = id.packageRevision("provider@1"),
exportId = id.packageExport("provider:content"),
portId = id.dependencyPort("provider:state");
workspace.packageImports.push({
packageId: id.package("provider"),
revisionId: revision,
displayName: "Provider",
source: workspace.packageImports[0]!.source,
exports: [
{
kind: "operation",
id: exportId,
displayName: "content",
mode: "call",
inputType: { kind: "builtin", name: "unit" },
outputType: { kind: "scalar", name: "string" },
execution: { kind: "snapshot", targets: ["browser", "server"] },
receiverRequirement: { kind: "exact-atom", atomId: id.atom("atom:reader:document") },
dependencyPorts: [
{
id: portId,
displayName: "state",
requirement: { kind: "state", valueType: { kind: "scalar", name: "string" }, primitives: ["read"] },
},
],
},
],
});
const owner = workspace.conformances.find((c) => c.interfaceRevisionId === "interface:reader:document-data@1")!;
owner.operationBindings.find((b) => b.operationId === "document:content:get")!.binding = {
kind: "package",
packageRevisionId: revision,
exportId,
dependencies: [{ portId, binding: { kind: "state", slotId: id.slot("slot:document:content") } }],
};
const prepared = await prepare(workspace);
const values = () => new Map(prepared.groups().map((g) => [g.logicalId, g.executionDigest]));
const before = values();
prepared.installation.packages.find((p) => p.revision === "package:reader-runtime@1")!.service!.contentDigest =
"c".repeat(64);
const callerChanged = values();
assert.equal(callerChanged.get("provider"), before.get("provider"));
assert.notEqual(callerChanged.get("package:reader-runtime"), before.get("package:reader-runtime"));
prepared.installation.packages.find((p) => p.revision === revision)!.service!.contentDigest = "d".repeat(64);
const providerChanged = values();
assert.notEqual(providerChanged.get("provider"), callerChanged.get("provider"));
assert.notEqual(providerChanged.get("package:reader-runtime"), callerChanged.get("package:reader-runtime"));
});
+16
View File
@@ -4,6 +4,22 @@ import { generatePortableTypeScriptBindings } from "../src/bindings/portable.js"
import { compileExecutionPlan } from "../src/execution-world/plan.js"; import { compileExecutionPlan } from "../src/execution-world/plan.js";
import { loadReaderWorkspace } from "./reader-fixture.js"; import { loadReaderWorkspace } from "./reader-fixture.js";
test("service bindings depend on their consumed contract, not workspace or source provenance", async () => {
const schema = compileExecutionPlan(await loadReaderWorkspace()).definition.bindingSchema;
const pkg = schema.packages[0]!;
const before = generatePortableTypeScriptBindings(schema, pkg.revisionId, { target: "server" });
const other = structuredClone(pkg);
other.revisionId = "unrelated@1" as typeof other.revisionId;
other.exports = [];
schema.packages.push(other);
pkg.source.commit = "b".repeat(40);
pkg.revisionId = "same-service-new-source@2" as typeof pkg.revisionId;
assert.equal(generatePortableTypeScriptBindings(schema, pkg.revisionId, { target: "server" }), before);
assert.doesNotMatch(before, /export const (bindingDigest|packageRevision)/);
pkg.exports[0]!.inputType = { kind: "scalar", name: "bool" };
assert.notEqual(generatePortableTypeScriptBindings(schema, pkg.revisionId, { target: "server" }), before);
});
test("reader portable bindings have selected facets, creation inputs and stable keyed calls", async () => { test("reader portable bindings have selected facets, creation inputs and stable keyed calls", async () => {
const schema = compileExecutionPlan(await loadReaderWorkspace()).definition.bindingSchema; const schema = compileExecutionPlan(await loadReaderWorkspace()).definition.bindingSchema;
const source = generatePortableTypeScriptBindings(schema, "package:reader-runtime@1", { target: "browser" }); const source = generatePortableTypeScriptBindings(schema, "package:reader-runtime@1", { target: "browser" });