fix(configurator): reduce hosted Plan stop latency

landing-zone-configurator/app/apps/worker/src/observation.ts

@@ -1,8 +1,72 @@
+import { watch } from "node:fs";
+import { open } from "node:fs/promises";
+import { resolve } from "node:path";
 import {
   applicationPlanJobSchema,
   applicationRunnerBindingSchema,
+  openTofuProgress,
 } from "@lzc/contracts";
 
+// Read only the bounded first version record; observation failure never fails a Plan.
+export function watchPlanVersion(
+  directory: string,
+  active: () => boolean,
+  observed: () => void,
+) {
+  let closed = false;
+  let reading = false;
+  let close = () => {
+    closed = true;
+  };
+  const check = async () => {
+    if (closed || reading || !active()) return;
+    reading = true;
+    try {
+      const file = await open(resolve(directory, "plan-events.log"), "r");
+      try {
+        const bytes = Buffer.alloc(16384);
+        const { bytesRead } = await file.read(bytes, 0, bytes.length, 0);
+        const content = bytes.subarray(0, bytesRead).toString("utf8");
+        const newline = content.indexOf("\n");
+        if (newline < 0) return;
+        const progress = openTofuProgress(content.slice(0, newline + 1));
+        if (
+          !progress.invalid &&
+          /^OpenTofu 1\.12\.6\n/.test(progress.text) &&
+          !closed &&
+          active()
+        ) {
+          close();
+          observed();
+        }
+      } finally {
+        await file.close();
+      }
+    } catch {
+      // A partial/unavailable file is retried by the existing output publication loop.
+    } finally {
+      reading = false;
+    }
+  };
+  try {
+    const watcher = watch(
+      directory,
+      { persistent: false },
+      (_event, filename) => {
+        if (filename === "plan-events.log") void check();
+      },
+    );
+    close = () => {
+      closed = true;
+      watcher.close();
+    };
+    watcher.on("error", close);
+  } catch {
+    /* Keep the existing bounded publication fallback. */
+  }
+  return () => close();
+}
+
 // Operator logs contain identifiers only, never inputs, engine messages or State.
 export function applicationPlanObservation(
   jobId: unknown,

landing-zone-configurator/app/apps/worker/src/runner.ts

@@ -29,7 +29,7 @@ import {
   s3RunnerBackendSchema,
   summarizePlan,
 } from "@lzc/contracts";
-import { applicationPlanObservation } from "./observation.js";
+import { applicationPlanObservation, watchPlanVersion } from "./observation.js";
 
 export const brokerOrigin =
   "https://lzc-dev-configurator.apps.01.cf.eu01.stackit.cloud";
@@ -610,62 +610,77 @@ export async function runWorker(
       outputTimer = setTimeout(pollOutput, 500);
       try {
         activeCommand = phase;
-        await new Promise<void>((done, fail) => {
-          const child = spawn(
-            "/bin/bash",
-            [
-              resolve(root, "run-plan.sh"),
-              phase,
-              applying
-                ? "platform-apply"
-                : applicationBinding?.purpose === "destroy"
-                  ? "application-destroy-plan"
-                  : applicationBinding?.purpose === "drift"
-                    ? "application-drift-plan"
-                    : mode === "initial-plan-only"
-                      ? mode
-                      : "platform-plan",
-            ],
-            {
-              cwd: work as string,
-              env,
-              stdio: ["ignore", "ignore", "ignore"],
-              detached: true,
-            },
-          );
-          let killTimer: ReturnType<typeof setTimeout> | undefined;
-          const stop = (signal: NodeJS.Signals) => {
-            try {
-              if (child.pid) process.kill(-child.pid, signal);
-            } catch {}
-          };
-          const timeout = setTimeout(
-            () => {
-              errorCode = "timed_out";
-              stop("SIGTERM");
-              killTimer = setTimeout(() => stop("SIGKILL"), 10000);
-            },
-            Math.max(1, deadline - Date.now()),
-          );
-          const clear = () => {
-            clearTimeout(timeout);
-            if (killTimer) clearTimeout(killTimer);
-          };
-          child.once("error", () => {
-            clear();
-            fail(new Error("Runner process failed"));
-          });
-          child.once("exit", (code) => {
-            activeCommand = undefined;
-            if (phase === "planning" && planObserved)
-              observePlan("plan-command-finished");
-            if (errorCode === "timed_out") stop("SIGKILL");
-            clear();
-            code === 0 && errorCode !== "timed_out"
-              ? done()
-              : fail(new Error("Runner phase failed"));
+        const stopObservation =
+          phase === "planning" && mode === "application-plan"
+            ? watchPlanVersion(
+                work as string,
+                () => activeCommand === "planning" && !planObserved,
+                () => {
+                  planObserved = true;
+                  observePlan("native-plan-observed");
+                },
+              )
+            : () => {};
+        try {
+          await new Promise<void>((done, fail) => {
+            const child = spawn(
+              "/bin/bash",
+              [
+                resolve(root, "run-plan.sh"),
+                phase,
+                applying
+                  ? "platform-apply"
+                  : applicationBinding?.purpose === "destroy"
+                    ? "application-destroy-plan"
+                    : applicationBinding?.purpose === "drift"
+                      ? "application-drift-plan"
+                      : mode === "initial-plan-only"
+                        ? mode
+                        : "platform-plan",
+              ],
+              {
+                cwd: work as string,
+                env,
+                stdio: ["ignore", "ignore", "ignore"],
+                detached: true,
+              },
+            );
+            let killTimer: ReturnType<typeof setTimeout> | undefined;
+            const stop = (signal: NodeJS.Signals) => {
+              try {
+                if (child.pid) process.kill(-child.pid, signal);
+              } catch {}
+            };
+            const timeout = setTimeout(
+              () => {
+                errorCode = "timed_out";
+                stop("SIGTERM");
+                killTimer = setTimeout(() => stop("SIGKILL"), 10000);
+              },
+              Math.max(1, deadline - Date.now()),
+            );
+            const clear = () => {
+              clearTimeout(timeout);
+              if (killTimer) clearTimeout(killTimer);
+            };
+            child.once("error", () => {
+              clear();
+              fail(new Error("Runner process failed"));
+            });
+            child.once("exit", (code) => {
+              activeCommand = undefined;
+              if (phase === "planning" && planObserved)
+                observePlan("plan-command-finished");
+              if (errorCode === "timed_out") stop("SIGKILL");
+              clear();
+              code === 0 && errorCode !== "timed_out"
+                ? done()
+                : fail(new Error("Runner phase failed"));
+            });
           });
-        });
+        } finally {
+          stopObservation();
+        }
       } finally {
         activeCommand = undefined;
         outputStopped = true;

landing-zone-configurator/app/tests/worker-observation.test.ts

@@ -0,0 +1,72 @@
+import { appendFile, mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { afterEach, expect, it, vi } from "vitest";
+import { watchPlanVersion } from "../apps/worker/src/observation.js";
+
+const roots: string[] = [];
+const version = JSON.stringify({
+  type: "version",
+  ui: "1.2",
+  "@message": "OpenTofu 1.12.6",
+});
+afterEach(async () => {
+  await Promise.all(
+    roots.splice(0).map((root) => rm(root, { recursive: true, force: true })),
+  );
+});
+
+it("observes a completed version record once without forwarding native contents", async () => {
+  const root = await mkdtemp(join(tmpdir(), "lzc-observation-"));
+  roots.push(root);
+  const observed = vi.fn();
+  const stop = watchPlanVersion(root, () => true, observed);
+  try {
+    const file = join(root, "plan-events.log");
+    await writeFile(file, version.slice(0, 20));
+    await new Promise((done) => setTimeout(done, 40));
+    expect(observed).not.toHaveBeenCalled();
+    await appendFile(
+      file,
+      `${version.slice(20)}\n${JSON.stringify({ type: "diagnostic", "@message": "private-engine-value" })}\n`,
+    );
+    await vi.waitFor(() => expect(observed).toHaveBeenCalledExactlyOnceWith());
+    await appendFile(file, `${version}\n`);
+    await new Promise((done) => setTimeout(done, 40));
+    expect(observed).toHaveBeenCalledTimes(1);
+  } finally {
+    stop();
+  }
+});
+
+it.each(["closed", "inactive"])(
+  "does not observe a %s command",
+  async (mode) => {
+    const root = await mkdtemp(join(tmpdir(), "lzc-observation-"));
+    roots.push(root);
+    const observed = vi.fn();
+    const stop = watchPlanVersion(root, () => mode !== "inactive", observed);
+    try {
+      if (mode === "closed") stop();
+      await writeFile(join(root, "plan-events.log"), `${version}\n`);
+      await new Promise((done) => setTimeout(done, 40));
+      expect(observed).not.toHaveBeenCalled();
+    } finally {
+      stop();
+    }
+  },
+);
+
+it("leaves Plan execution available if the optional watcher cannot be created", async () => {
+  const root = await mkdtemp(join(tmpdir(), "lzc-observation-"));
+  roots.push(root);
+  await mkdir(join(root, "existing"));
+  const observed = vi.fn();
+  const stop = watchPlanVersion(
+    join(root, "unavailable"),
+    () => true,
+    observed,
+  );
+  expect(() => stop()).not.toThrow();
+  expect(observed).not.toHaveBeenCalled();
+});

landing-zone-configurator/infra/recovery-lab/README.de.md

@@ -127,6 +127,18 @@ von einmaligen oder warmen Runner-Apps und ignoriert andere Instanzen, ungültig
 IDs, Meldungen vor dem Scharfschalten sowie über drei Sekunden alte Ereignisse.
 Seine Umgebung übernimmt keine sonstigen Cloud-Zugangsdaten.
 
+Die Organisations-, Space- und Template-Grenze wird einmal beim Scharfschalten
+geprüft. Unmittelbar vor dem Stop prüft eine einzelne Abfrage erneut die genaue
+Runner-App-GUID und ihren Space; einmalige Tasks benötigen zusätzlich die
+eindeutig aktive Task. Eine ungeprüfte Grenze kann keinen Stop autorisieren.
+Wiederholte Organisations-, Template- und vollständige App-Listen-Abfragen im
+kurzen Plan-Zeitfenster werden gezielt vermieden. Der Worker beobachtet den
+begrenzten ersten nativen Versionsdatensatz beim Schreiben seiner Datei; die
+normale Ausgabeübertragung bleibt als Rückfall erhalten. Beide Mechanismen
+verzögern den Plan nicht und ändern keine Eingaben. Private Nachweise enthalten
+Zeitstempel für Absicht und Antwort zum Vergleich von tatsächlichem Stop und
+nativem Abschluss.
+
 Bei einem ausführbaren Template mit direkter Bereitstellung startet die normale
 Bestelloberfläche automatisch den ersten Plan. Diesen Lauf beenden lassen und
 sein Ergebnis prüfen. Für den Abbruchtest dieselbe entbehrliche Instanz verwenden:

landing-zone-configurator/infra/recovery-lab/README.md

@@ -120,6 +120,16 @@ CF, follows bounded log streams only from one-shot/warm Runner apps, and ignores
 other instances, malformed IDs, observations before arming and events older than
 three seconds. Its environment carries no ambient cloud credentials.
 
+The organization/space/template boundary is verified once when arming. Immediately
+before stopping, a single lookup rechecks the exact Runner app GUID and its space;
+one-shot tasks additionally require the unique active task. An unverified boundary
+cannot authorize a stop. Repeating organization, template and full app-list reads
+inside the short Plan window is deliberately avoided. The worker observes the
+bounded first native version record as its file is written, with normal output
+publication retained as a fallback. Neither mechanism delays the Plan or changes
+its inputs. Private receipts include intent and response timestamps for comparing
+the actual stop with native completion.
+
 For an executable template with direct deployment policy, the ordinary ordering
 UI automatically starts the first Plan. Let that initial Plan finish and review
 its outcome. Reuse the same disposable instance for the interruption drill:

landing-zone-configurator/infra/recovery-lab/hosted_plan.py

@@ -70,6 +70,7 @@ def __init__(self, cf, cf_home, space, template, instance, confirm=None, receipt
         self.env = {"PATH": "/usr/bin:/bin", "LANG": "C.UTF-8", "CF_HOME": str(home.resolve())}
         self.confirm = confirm
         self.stopped = False
+        self.boundary_verified = False
 
     def call(self, args):
         result = subprocess.run([self.cf, *args], env=self.env, capture_output=True, timeout=15)
@@ -88,13 +89,15 @@ def api(self, path, method="GET"):
         return value
 
     def boundary(self):
+        self.boundary_verified = False
         space = self.api(f"/v3/spaces/{self.space}")
         org = self.api("/v3/organizations/" + guid(space["relationships"]["organization"]["data"]["guid"]))
         if org["name"] != "lzc-dev-runners" or space["name"] != "plans":
             raise RuntimeError("Only the isolated lzc-dev-runners/plans space is supported")
         template = self.api(f"/v3/apps/{self.template}")
         if template["name"] != "lzc-plan-template" or template["relationships"]["space"]["data"]["guid"] != self.space:
             raise RuntimeError("Runner template binding mismatch")
+        self.boundary_verified = True
 
     def apps(self):
         result = self.api(f"/v3/apps?space_guids={self.space}&per_page=100")
@@ -104,13 +107,12 @@ def apps(self):
 
     def stop(self, app_id, event, finished):
         guid(app_id)
-        if not self.confirm or self.stopped or event["jobId"] in finished:
+        if not self.confirm or self.stopped or not self.boundary_verified or event["jobId"] in finished:
             raise RuntimeError("No stop authorization or observation is already finished")
         if observation(json.dumps(event, separators=(",", ":")), self.instance, self.armed_at, time.time()) is None or event["observation"] != "native-plan-observed":
             raise RuntimeError("Fresh matching native Plan observation required")
-        self.boundary()
-        app = self.apps().get(app_id)
-        if not app:
+        app = self.api(f"/v3/apps/{app_id}")
+        if app_id == self.template or app.get("guid") != app_id or app.get("relationships", {}).get("space", {}).get("data", {}).get("guid") != self.space:
             raise RuntimeError("Observed Runner is outside the allowed space")
         # Single-use guard is set before the first mutation, including API failure.
         self.stopped = True
@@ -120,11 +122,11 @@ def fresh():
         def request(path, result):
             fresh()
             with Receipt(self.receipt_path) as receipt:
-                record = {"phase": "stop-intent", "instanceId": self.instance, "jobId": event["jobId"], "observedAt": event["observedAt"], "qualifiesRecovery": False, **result}
+                record = {"phase": "stop-intent", "instanceId": self.instance, "jobId": event["jobId"], "observedAt": event["observedAt"], "intentAt": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"), "qualifiesRecovery": False, **result}
                 receipt.write(record)
                 fresh()
                 self.api(path, "POST")
-                receipt.write({**record, "phase": "stop-requested"})
+                receipt.write({**record, "phase": "stop-requested", "responseAt": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")})
             return result
         if app["name"] == "lzc-plan-" + event["jobId"]:
             tasks = self.api(f"/v3/apps/{app_id}/tasks?per_page=100")