Implement progressive query admission, session boundaries and event-driven Studio observations

src/query/compile.ts

@@ -41,6 +41,7 @@ import { objectRow } from "./relational-schema.js";
 import { workspaceQueryRootId } from "./root.js";
 import { relationalCompiler } from "./relational-compile.js";
 import { fragmentSources, type FragmentResolver } from "./fragments.js";
+import { lowerQueryTree, stripQueryBoundaries } from "./tree.js";
 import {
   QueryCompileError,
   type QueryDeclaration,
@@ -158,7 +159,7 @@ export async function compileQuery(
       if (node.name.value.startsWith("__")) fail("QUERY_UNSUPPORTED_FEATURE", "Introspection is not supported", node);
     },
     Directive(node) {
-      if (!["include", "skip"].includes(node.name.value))
+      if (!["include", "skip", "boundary"].includes(node.name.value))
         fail("QUERY_UNSUPPORTED_FEATURE", `Unsupported directive ${node.name.value}`, node);
     },
   });
@@ -421,6 +422,7 @@ export async function compileQuery(
         for (const selection of set.selections) {
           if (++expandedFields > 10000) fail("QUERY_WORK_LIMIT", "Expanded query exceeds 10000 fields", selection);
           if (selection.kind === "FragmentSpread") {
+            if (selection.directives?.some((directive) => directive.name.value === "boundary")) continue;
             if (fragmentContracts.get(selection.name.value)?.mode === "reference") continue;
             add(
               fragments.get(selection.name.value)!.selectionSet,
@@ -519,7 +521,7 @@ export async function compileQuery(
   };
   const conditions = (directives: readonly DirectiveNode[] = []) =>
     directives
-      .filter((directive) => directive.name.value !== "live")
+      .filter((directive) => ["include", "skip"].includes(directive.name.value))
       .map((directive) => ({
         include: directive.name.value === "include",
         value: argument(directive.arguments!.find((entry) => entry.name.value === "if")!.value),
@@ -540,6 +542,52 @@ export async function compileQuery(
         const contract = fragmentContracts.get(node.name.value);
         if (contract && generated.byName.get(parent.name)?.revisionId !== contract.root)
           fail("FRAGMENT_ROOT_MISMATCH", "Imported fragment requires its exact nominal root interface", node);
+        const directive = node.directives?.find((directive) => directive.name.value === "boundary");
+        if (directive) {
+          const name = directive.arguments?.find((argument) => argument.name.value === "name")?.value;
+          if (name?.kind !== "StringValue" || !/^[A-Za-z_][A-Za-z0-9_]{0,63}$/.test(name.value))
+            return fail(
+              "QUERY_BOUNDARY_NAME",
+              "A boundary needs a literal identifier of at most 64 characters",
+              directive,
+            );
+          const root = generated.byName.get(parent.name)?.revisionId;
+          if (!root || root === workspaceQueryRootId)
+            return fail("QUERY_BOUNDARY_ROOT", "A boundary must select a nominal Graph Object", node);
+          if (contract?.mode === "reference")
+            fail("QUERY_BOUNDARY_REFERENCE", "Use an ordinary checked Fragment spread for a boundary", node);
+          const fragment = fragments.get(node.name.value)!;
+          const token = node.loc?.startToken;
+          return [
+            {
+              name: "$boundary",
+              key: "",
+              arguments: {},
+              conditions: inheritedConditions,
+              boundary: {
+                name: name.value,
+                root,
+                output: required(parent, fragment.selectionSet),
+                ...(token ? { location: { file: node.loc!.source.name, line: token.line, column: token.column } } : {}),
+              },
+              selection: [
+                ...(contract
+                  ? [
+                      {
+                        name: "$fragment",
+                        key: "",
+                        selection: [],
+                        arguments: {},
+                        conditions: [],
+                        fragmentSpread: contract,
+                      },
+                    ]
+                  : []),
+                ...selection(fragment.selectionSet, parent),
+              ],
+            },
+          ];
+        }
         return [
           ...(contract
             ? [
@@ -598,6 +646,13 @@ export async function compileQuery(
       ];
     });
     const contract = generated.byName.get(parent.name);
+    const names = new Set<string>();
+    for (const field of selections)
+      if (field.boundary) {
+        if (names.has(field.boundary.name))
+          fail("QUERY_BOUNDARY_DUPLICATE", `Duplicate boundary ${field.boundary.name}`);
+        names.add(field.boundary.name);
+      }
     if (!contract || contract.revisionId === workspaceQueryRootId) return selections;
     // Fragment expansion may repeat metadata. Combine its children so a later
     // implicit reference cannot overwrite an explicitly selected aggregate.
@@ -632,10 +687,11 @@ export async function compileQuery(
   const normalized = print(document),
     schema = printSchema(generated.schema);
   const selections = selection(operations[0]!.selectionSet, generated.schema.getQueryType()!);
+  const tree = lowerQueryTree(declaration, selections, result);
   const definitionDigest = createHash("sha256")
     .update(
       canonicalJson({
-        semantics: 6,
+        semantics: tree ? 7 : 6,
         declaration,
         normalized,
         fragmentContracts: Object.fromEntries(fragmentContracts),
@@ -667,7 +723,8 @@ export async function compileQuery(
     variables: { kind: "record", fields: variables },
     output: result,
     effects: [...effects.values()],
-    selection: selections,
+    selection: tree ? stripQueryBoundaries(selections) : selections,
+    ...(tree ? { tree } : {}),
     variableDefaults: Object.fromEntries(
       (operations[0]!.variableDefinitions ?? [])
         .filter((entry) => entry.defaultValue)