Build workspace agent, capability graph, and versioned cutovers

dist/index.js

@@ -135,19 +135,47 @@ export const createRuntimeContext = (camino, orch, request) => {
                 ports.set(dependency.portId, edge);
                 break;
             }
-            case "receiverInterfaceRevisionId": {
+            case "interfaceRevisionId": {
                 const interfaceRevisionId = dependency.binding.value;
+                const dependencyObjectId = dependency.objectId || request.objectId;
+                const invoke = async (operationId, input) => {
+                    const response = await orch.invokeCapability({
+                        capability: create(CapabilityRefSchema, { interfaceRevisionId, operationId }),
+                        objectId: dependencyObjectId,
+                        input: Object.fromEntries(Object.entries(input).map(([key, value]) => [key, jsToProtoValue(value)])),
+                    });
+                    if (!response.ok)
+                        throw new Error(response.error || `Capability ${operationId} failed`);
+                    for (const dependency of response.dependencies) {
+                        if (dependency.kind === "state") {
+                            await recordDependency({
+                                kind: "state",
+                                objectId: dependency.objectId,
+                                attachmentId: dependency.attachmentId,
+                            });
+                        }
+                        else if (dependency.kind === "edge") {
+                            await recordDependency({
+                                kind: "edge",
+                                objectId: dependency.objectId,
+                                attachmentId: dependency.attachmentId,
+                                projectionId: dependency.projectionId,
+                            });
+                        }
+                    }
+                    return response.result;
+                };
                 const capability = {
+                    objectId: dependencyObjectId,
                     interfaceRevisionId,
                     async invoke(operationId, input = {}) {
-                        const response = await orch.invokeCapability({
-                            capability: create(CapabilityRefSchema, { interfaceRevisionId, operationId }),
-                            objectId: request.objectId,
-                            input: Object.fromEntries(Object.entries(input).map(([key, value]) => [key, jsToProtoValue(value)])),
-                        });
-                        if (!response.ok)
-                            throw new Error(response.error || `Capability ${operationId} failed`);
-                        return protoValueToJs(response.result);
+                        return protoValueToJs(await invoke(operationId, input));
+                    },
+                    async live(operationId, input = {}) {
+                        const value = await invoke(operationId, input);
+                        if (!value)
+                            throw new Error(`Capability ${operationId} returned no value`);
+                        return liveValue(value);
                     },
                 };
                 ports.set(dependency.portId, capability);
@@ -236,7 +264,11 @@ export const createPackageRuntimeRoutes = (config) => {
                 throw new ConnectError(`Unknown export ${exportId ?? ""}`, Code.NotFound);
             try {
                 const result = await evaluate(handler, createRuntimeContext(camino, orch, request));
-                return create(InvokeResponseSchema, { ok: true, result: result.value });
+                return create(InvokeResponseSchema, {
+                    ok: true,
+                    result: result.value,
+                    dependencies: protoDependencies(result.dependencies),
+                });
             }
             catch (error) {
                 return create(InvokeResponseSchema, {
@@ -255,6 +287,7 @@ export const createPackageRuntimeRoutes = (config) => {
             const runtimeContext = createRuntimeContext(camino, orch, request);
             const subscriptions = new Map();
             const establishing = new Map();
+            let subscriptionEpoch = 0;
             const ensureSubscription = async (dependency) => {
                 const key = dependencyKey(dependency);
                 if (subscriptions.has(key))
@@ -280,6 +313,7 @@ export const createPackageRuntimeRoutes = (config) => {
                         };
                         subscription.next = waitNext();
                         subscriptions.set(key, subscription);
+                        subscriptionEpoch += 1;
                     }
                     catch (error) {
                         controller.abort();
@@ -300,7 +334,21 @@ export const createPackageRuntimeRoutes = (config) => {
                 }
             };
             context.signal.addEventListener("abort", abortAll, { once: true });
-            let current = await evaluate(handler, runtimeContext, ensureSubscription);
+            const evaluateWithStableSubscriptions = async () => {
+                // A direct state/edge port records its dependency before reading it,
+                // but a nested interface invocation can only report its transitive
+                // dependencies after that invocation returns. Once a new dependency
+                // stream is established, evaluate again so every read contributing to
+                // the emitted value happened after its stream became live.
+                for (let pass = 0; pass < 32; pass += 1) {
+                    const before = subscriptionEpoch;
+                    const result = await evaluate(handler, runtimeContext, ensureSubscription);
+                    if (subscriptionEpoch === before)
+                        return result;
+                }
+                throw new Error("Derived dependency discovery did not stabilize after 32 passes");
+            };
+            let current = await evaluateWithStableSubscriptions();
             yield create(WatchEventSchema, {
                 watchId,
                 value: current.value,
@@ -333,7 +381,7 @@ export const createPackageRuntimeRoutes = (config) => {
                     if (outcome.done)
                         throw new Error(`Dependency stream ${outcome.key} ended unexpectedly`);
                     subscription.next = subscription.waitNext();
-                    const updated = await evaluate(handler, runtimeContext, ensureSubscription);
+                    const updated = await evaluateWithStableSubscriptions();
                     const active = new Set(updated.dependencies.map(dependencyKey));
                     for (const [key, entry] of subscriptions) {
                         if (!active.has(key)) {