fix(execution): wire resumeExecution resolver and guard the gap

backend/lib/arbiter-stack.ts

@@ -1189,36 +1189,56 @@ export class ArbiterStack extends cdk.Stack {
         },
       );
 
-      // Least-privilege (decision O4 — resource-scoped, no wildcards). The
-      // watchdog now:
-      //   * Scans + reads-by-key (GetItem/Query) the executions table and
-      //     conditionally UpdateItems it (reconcile writes go through the
-      //     executor's conditional guards; the watchdog itself fails stuck
-      //     executions).
-      //   * Reads the workflows table (GetItem/Query ONLY — never Scan/write)
-      //     to know the DAG graph for reconcile.
-      //   * SendMessage to the worker queue ARN to re-dispatch a stalled node.
-      //   * PutEvents on the shared bus (workflow.failed / workflow.completed)
-      //     and PutMetricData for the best-effort timeout metric (below).
-      // It remains a distinct, higher-trust role than the worker (which is
-      // UpdateItem-only + FGAC) and is triggered ONLY by its EventBridge
-      // schedule — never externally invokable.
+      // Least-privilege (decision O4/a41e12a6, reconciled — finding d037634b,
+      // superseding a41e12a6's original text). SendMessage to the worker
+      // queue ARN re-dispatches a stalled node; PutEvents on the shared bus
+      // emits workflow.failed/workflow.completed; PutMetricData (below) is
+      // the best-effort timeout metric. Distinct, higher-trust role than the
+      // worker (UpdateItem-only + FGAC); triggered ONLY by its EventBridge
+      // schedule, never externally invokable.
+      //
+      // DynamoDB grants verified against every call reachable from the
+      // watchdog's own code path (timeout_watchdog.py + the shared
+      // executor.py primitives it calls: _load_execution, invoke_node,
+      // _reconcile_or_fail_node, handle_node_failure, _finalize_execution):
+      //   * dynamodb:Scan — PRE-EXISTING (predates the durable-execution
+      //     work at c991af7). Load-bearing: the executions table's only
+      //     index is WorkflowIndex (workflowId/startedAt); there is no
+      //     status index, so `_scan_running()`'s filtered Scan for
+      //     status == 'running' is the only viable access pattern for this
+      //     low-frequency sweep (see that function's own docstring).
+      //   * dynamodb:GetItem — added by c991af7 (durable-execution work).
+      //     Load-bearing: `executor._load_execution` reads a single
+      //     execution row by key when reconciling/re-dispatching.
+      //   * dynamodb:UpdateItem — PRE-EXISTING. Load-bearing: `_fail_stuck`,
+      //     `invoke_node`'s conditional pending->running dispatch,
+      //     `_reconcile_or_fail_node`'s stall re-dispatch flip,
+      //     `handle_node_failure`, and `_finalize_execution` all write
+      //     through this table via conditional-guarded UpdateItem calls
+      //     (never a bare write — see executor.py's own ConditionExpression
+      //     guards).
+      //   * dynamodb:Query — REMOVED. Added by c991af7 alongside GetItem but
+      //     never actually called: no function reachable from the watchdog
+      //     (including the shared executor.py primitives above) issues a
+      //     Query against this table. Confirmed by direct grep of the
+      //     watchdog + executor + dag modules for `.query(`.
       workflowTimeoutWatchdogFunction.addToRolePolicy(
         new iam.PolicyStatement({
           effect: iam.Effect.ALLOW,
-          actions: [
-            "dynamodb:Scan",
-            "dynamodb:GetItem",
-            "dynamodb:Query",
-            "dynamodb:UpdateItem",
-          ],
+          actions: ["dynamodb:Scan", "dynamodb:GetItem", "dynamodb:UpdateItem"],
           resources: [props.executionsTable.tableArn],
         }),
       );
+      // Workflows table: GetItem-only (read-only, no write). PRE-EXISTING
+      // GetItem (via `executor._load_workflow`, called to read the DAG graph
+      // for reconcile — see `_load_workflow`'s docstring: "read-only
+      // GetItem"). dynamodb:Query REMOVED for the identical reason as above
+      // — never called; the workflows table is looked up by its partition
+      // key (workflowId) only, never queried.
       workflowTimeoutWatchdogFunction.addToRolePolicy(
         new iam.PolicyStatement({
           effect: iam.Effect.ALLOW,
-          actions: ["dynamodb:GetItem", "dynamodb:Query"],
+          actions: ["dynamodb:GetItem"],
           resources: [props.workflowsTable.tableArn],
         }),
       );

backend/lib/backend-stack.ts

@@ -2944,6 +2944,13 @@ export class BackendStack extends cdk.Stack {
       responseMappingTemplate: appsync.MappingTemplate.lambdaResult(),
     });
 
+    executionLambdaDataSource.createResolver("ResumeExecutionResolver", {
+      typeName: "Mutation",
+      fieldName: "resumeExecution",
+      requestMappingTemplate: appsync.MappingTemplate.lambdaRequest(),
+      responseMappingTemplate: appsync.MappingTemplate.lambdaResult(),
+    });
+
     // publishWorkflowProgress — IAM-only mutation called by fan-out Lambda
     executionLambdaDataSource.createResolver(
       "PublishWorkflowProgressResolver",

backend/test/arbiter-stack-step-runner.test.ts

@@ -511,14 +511,18 @@ describe("ArbiterStack — Step Runner Lambda and EventBridge rules (Task 1.6)",
       });
     });
 
-    test("watchdog DynamoDB grant is resource-scoped: reconcile reads/writes executions, reads workflows (decision O4)", () => {
+    test("watchdog DynamoDB grant is resource-scoped: reconcile reads/writes executions, reads workflows (decision a41e12a6, reconciled — finding d037634b)", () => {
       const actions = actionsForRole("WorkflowTimeoutWatchdog");
-      // Reconcile needs to Scan for running execs, read a single exec/workflow
-      // by key (GetItem/Query), and conditionally UpdateItem (fail/reconcile).
+      // Reconcile needs to Scan for running execs (no status index exists —
+      // see arbiter-stack.ts's grant-site comment), GetItem a single exec by
+      // key, and conditionally UpdateItem (fail/reconcile). dynamodb:Query is
+      // deliberately NOT granted: no function reachable from the watchdog
+      // (timeout_watchdog.py or the shared executor.py primitives it calls)
+      // ever issues a Query against this table.
       expect(actions.has("dynamodb:Scan")).toBe(true);
       expect(actions.has("dynamodb:GetItem")).toBe(true);
-      expect(actions.has("dynamodb:Query")).toBe(true);
       expect(actions.has("dynamodb:UpdateItem")).toBe(true);
+      expect(actions.has("dynamodb:Query")).toBe(false);
       // No table-wide write blast radius — never Put/Delete/BatchWrite.
       expect(actions.has("dynamodb:PutItem")).toBe(false);
       expect(actions.has("dynamodb:DeleteItem")).toBe(false);
@@ -529,10 +533,12 @@ describe("ArbiterStack — Step Runner Lambda and EventBridge rules (Task 1.6)",
       expect(actions.has("cloudwatch:PutMetricData")).toBe(true);
     });
 
-    test("watchdog workflows-table grant is read-only (GetItem/Query, never write/Scan)", () => {
+    test("watchdog workflows-table grant is read-only (GetItem-only, never Query/write/Scan)", () => {
       // Find the statement(s) on the watchdog role that grant a dynamodb read
       // and assert none of them grant a write on the same statement — the
-      // workflows read grant is [GetItem, Query] with no UpdateItem/Put/Delete.
+      // workflows read grant is [GetItem] only (no Query — never called by
+      // `executor._load_workflow`, which reads by key only — and no
+      // UpdateItem/Put/Delete).
       const policies = template.findResources("AWS::IAM::Policy");
       let sawWorkflowsReadOnly = false;
       for (const p of Object.values(policies) as any[]) {
@@ -547,7 +553,7 @@ describe("ArbiterStack — Step Runner Lambda and EventBridge rules (Task 1.6)",
           const acts = Array.isArray(s.Action) ? s.Action : [s.Action];
           const isReadOnly =
             acts.includes("dynamodb:GetItem") &&
-            acts.includes("dynamodb:Query") &&
+            !acts.includes("dynamodb:Query") &&
             !acts.includes("dynamodb:UpdateItem") &&
             !acts.includes("dynamodb:Scan") &&
             !acts.includes("dynamodb:PutItem") &&

backend/test/backend-stack-workflows.test.ts

@@ -192,6 +192,7 @@ describe("BackendStack — Workflow/App/Execution Lambda functions and AppSync w
     const executionMutationFields = [
       "startExecution",
       "cancelExecution",
+      "resumeExecution",
       "publishWorkflowProgress",
     ];
 

backend/test/schema-resolver-parity-guard.test.ts

@@ -0,0 +1,198 @@
+/**
+ * Generic parity guard — every schema.graphql Query/Mutation field must
+ * have a wired AWS::AppSync::Resolver in one of the resolver-owning
+ * stacks' synthesized CloudFormation templates.
+ *
+ * Motivated by finding 24563f6c: `Mutation.resumeExecution` was declared
+ * in schema.graphql with a tested handler in execution-resolver.ts, but no
+ * stack ever called `.createResolver(...)` / `new CfnResolver(...)` for it
+ * — the field returned null in live dev. Neither the type checker, nor the
+ * handler's own unit tests, nor `cdk synth` (which happily synthesizes an
+ * AppSync API with fewer resolvers than schema fields — that is not a CFN
+ * error) can catch a declared-but-unwired field. Only a guard that parses
+ * the schema and cross-checks it against the resolver set closes this
+ * defect CLASS, not just this one instance. (`scripts/split-gates/rails/
+ * rail3-resolver-parity.ts` is a DIFFERENT check — it detects resolver
+ * fields lost/gained relative to a pre-split baseline snapshot during the
+ * backend-stack-split refactor; a field that was NEVER wired in the first
+ * place has no baseline entry to diff against, so rail 3 would not have
+ * caught this.)
+ *
+ * Resolver-owning stacks (confirmed by grep for `createResolver(` /
+ * `new appsyncCfn.CfnResolver(` across backend/lib/*.ts — both produce the
+ * identical `AWS::AppSync::Resolver` CFN resource type, so scanning for
+ * that resource type in the synthesized template covers both wiring
+ * styles uniformly): backend, registry, projects, governance, services,
+ * arbiter. frontend/gateway/telemetry own zero AppSync resolvers.
+ *
+ * Reads already-synthesized templates from cdk.out (produced by
+ * `npx cdk synth citadel-backend-dev citadel-registry-dev
+ * citadel-projects-dev citadel-governance-dev citadel-services-dev
+ * citadel-arbiter-dev`, or a full `npx cdk synth --all`). Skips gracefully
+ * if cdk.out is absent so a bare `npm test` (no prior synth) does not fail
+ * the whole suite — mirrors the existing rail-2 stateful-pin test's skip
+ * convention (test/split-gates-rail2-stateful-pin.test.ts).
+ */
+import * as fs from "fs";
+import * as path from "path";
+import { parse } from "graphql";
+import type { DocumentNode, ObjectTypeDefinitionNode } from "graphql";
+import {
+  loadTemplate,
+  extractResolverKeys,
+} from "../scripts/split-gates/template-utils";
+
+const ENV = process.env.SPLIT_GATES_ENV ?? "dev";
+
+const SDL_PATH = path.resolve(
+  __dirname,
+  "..",
+  "src",
+  "schema",
+  "schema.graphql",
+);
+
+/** Every stack that wires at least one AWS::AppSync::Resolver against the
+ * shared BackendStack GraphQL API. See file header for how this list was
+ * derived. */
+const RESOLVER_STACK_NAMES = [
+  `citadel-backend-${ENV}`,
+  `citadel-registry-${ENV}`,
+  `citadel-projects-${ENV}`,
+  `citadel-governance-${ENV}`,
+  `citadel-services-${ENV}`,
+  `citadel-arbiter-${ENV}`,
+];
+
+/**
+ * Fields deliberately excluded from this guard's enforcement. Every entry
+ * MUST carry a reason. Two categories are valid here:
+ *   (a) genuinely non-Lambda-resolver fields (none exist in this schema
+ *       today — every Query/Mutation field is Lambda-resolver-backed);
+ *   (b) PRE-EXISTING unwired fields discovered BY this guard, tracked as
+ *       separate findings rather than fixed here (scope discipline — see
+ *       reasons below). These are NOT "intentionally unwired" in the
+ *       design sense; they are known, tracked defects of the identical
+ *       class as 24563f6c/resumeExecution, temporarily allowlisted so this
+ *       guard can ship green for the field it was built to catch
+ *       (resumeExecution) without also silently blocking on 7 unrelated,
+ *       already-broken fields it happened to also surface. Removing an
+ *       entry here (because its resolver was wired) should make its test
+ *       start passing; removing it WITHOUT wiring the resolver will make
+ *       the guard fail again, as intended.
+ */
+const DELIBERATELY_UNWIRED: ReadonlyMap<string, string> = new Map([
+  // --- pre-existing, tracked separately (finding 0018a6d7), NOT fixed by
+  // this change (finding 24563f6c/d037634b only) ---
+  [
+    "Mutation.updateAgentStatus",
+    "Pre-existing gap surfaced by this guard, same class as resumeExecution. Handler exists (agent-resolver.ts:125) but no stack wires a resolver for it. Tracked separately: finding 0018a6d7. Not in scope for finding 24563f6c.",
+  ],
+  [
+    "Mutation.updateProjectProgress",
+    "Pre-existing gap surfaced by this guard, same class as resumeExecution. Tracked separately: finding 0018a6d7. Not in scope for finding 24563f6c.",
+  ],
+  [
+    "Mutation.testTool",
+    "Pre-existing gap surfaced by this guard, same class as resumeExecution. Tracked separately: finding 0018a6d7. Not in scope for finding 24563f6c.",
+  ],
+  [
+    "Query.listAvailableDataSources",
+    "Pre-existing gap surfaced by this guard, same class as resumeExecution. Tracked separately: finding 0018a6d7. Not in scope for finding 24563f6c.",
+  ],
+  [
+    "Query.listIntegrationOperations",
+    "Pre-existing gap surfaced by this guard, same class as resumeExecution. Tracked separately: finding 0018a6d7. Not in scope for finding 24563f6c.",
+  ],
+  [
+    "Query.getDashboardMetrics",
+    "Pre-existing gap surfaced by this guard, same class as resumeExecution. Tracked separately: finding 0018a6d7. Not in scope for finding 24563f6c.",
+  ],
+  [
+    "Query.getRecentActivity",
+    "Pre-existing gap surfaced by this guard, same class as resumeExecution. Tracked separately: finding 0018a6d7. Not in scope for finding 24563f6c.",
+  ],
+]);
+
+function cdkOutTemplatePath(stackName: string): string {
+  return path.resolve(__dirname, "..", "cdk.out", `${stackName}.template.json`);
+}
+
+const templatesExist = RESOLVER_STACK_NAMES.every((name) =>
+  fs.existsSync(cdkOutTemplatePath(name)),
+);
+
+describe("schema <-> AppSync resolver parity guard", () => {
+  if (!templatesExist) {
+    const missing = RESOLVER_STACK_NAMES.filter(
+      (name) => !fs.existsSync(cdkOutTemplatePath(name)),
+    );
+    it.skip(`skipped: cdk.out template(s) missing (run cdk synth first). missing=${missing.join(", ")}`, () => {});
+    return;
+  }
+
+  const sdl: DocumentNode = parse(fs.readFileSync(SDL_PATH, "utf8"));
+
+  function schemaFieldNames(typeName: "Query" | "Mutation"): string[] {
+    const def = sdl.definitions.find(
+      (d): d is ObjectTypeDefinitionNode =>
+        d.kind === "ObjectTypeDefinition" && d.name.value === typeName,
+    );
+    if (!def || !def.fields) {
+      throw new Error(
+        `schema.graphql has no ${typeName} type — parser or schema regressed`,
+      );
+    }
+    return def.fields.map((f) => f.name.value);
+  }
+
+  const queryFields = schemaFieldNames("Query");
+  const mutationFields = schemaFieldNames("Mutation");
+
+  it("sanity check: schema declares a non-trivial number of Query and Mutation fields", () => {
+    expect(queryFields.length).toBeGreaterThan(10);
+    expect(mutationFields.length).toBeGreaterThan(10);
+  });
+
+  // Merge resolver keys ("TypeName.fieldName") across every resolver-owning
+  // stack's synthesized template.
+  const wiredKeys = new Set<string>();
+  for (const stackName of RESOLVER_STACK_NAMES) {
+    const template = loadTemplate(cdkOutTemplatePath(stackName));
+    for (const key of extractResolverKeys(template).keys()) {
+      wiredKeys.add(key);
+    }
+  }
+
+  it("sanity check: at least one resolver was found across the scanned stacks (guard is not vacuous)", () => {
+    expect(wiredKeys.size).toBeGreaterThan(50);
+  });
+
+  it.each(queryFields.map((f) => [f] as const))(
+    "Query.%s has a wired AppSync Resolver (or a documented allowlist reason)",
+    (fieldName) => {
+      const key = `Query.${fieldName}`;
+      if (DELIBERATELY_UNWIRED.has(key)) return;
+      expect(wiredKeys.has(key)).toBe(true);
+    },
+  );
+
+  it.each(mutationFields.map((f) => [f] as const))(
+    "Mutation.%s has a wired AppSync Resolver (or a documented allowlist reason)",
+    (fieldName) => {
+      const key = `Mutation.${fieldName}`;
+      if (DELIBERATELY_UNWIRED.has(key)) return;
+      expect(wiredKeys.has(key)).toBe(true);
+    },
+  );
+
+  it("every DELIBERATELY_UNWIRED entry corresponds to a real schema field (no stale allowlist entries)", () => {
+    const allFields = new Set([
+      ...queryFields.map((f) => `Query.${f}`),
+      ...mutationFields.map((f) => `Mutation.${f}`),
+    ]);
+    for (const key of DELIBERATELY_UNWIRED.keys()) {
+      expect(allFields.has(key)).toBe(true);
+    }
+  });
+});