lweberru · Oct 17, 2026 · Oct 17, 2026
@@ -419,9 +419,15 @@ und verwendet weder Kunden-Credentials noch Kunden-State:
 LZC_NATIVE_RUNNER_TEST=true LZC_RUNNER_PACKAGE_DIR=../.local/runner-local-20261003-mirror-init npm exec -- vitest run tests/platform-worker.test.ts -t 'real native providers'
 ```
 
-Unter **OpenTofu-Ausgabe** zeigt jeder lokale Auftrag die normale CLI-Ausgabe.
-Laufende Jobs werden alle zwei Sekunden aktualisiert; **Ausgabe folgen** laesst
-sich zum Pruefen aelterer Zeilen anhalten. Erfolgreiche Plattform-Plans werden
+Unter **OpenTofu-Ausgabe** zeigen Plattformauftraege auch in Cloud Foundry
+Live-Fortschritt ueber einen authentifizierten NDJSON-Kanal. Der Worker liest
+vollstaendige Zeilen alle 500 ms; nur geaenderter, redigierter Text wird
+verschluesselt gesichert und weitergegeben. Der Plan liefert native
+OpenTofu-UI-Ereignisse, bevor der fertige Diff verfuegbar ist; rohe Ereignisobjekte
+und deren Output-Daten werden nicht an den Browser gegeben. **Ausgabe folgen** laesst sich zum Pruefen aelterer
+Zeilen anhalten. Einklappen und Navigation trennen die Verbindung; erneutes
+Oeffnen setzt beim aktuellen Mitschnitt fort. JSON-Snapshots bleiben kompatibel.
+Erfolgreiche Plattform-Plans werden
 read-only mittels `tofu show -no-color` aus dem unveraenderten gespeicherten
 Artefakt dargestellt, einschliesslich Ressourcen und Attributaenderungen.
 OpenTofu maskiert sensitive Werte wie in der CLI. Die separate Ausgabe-Route
@@ -430,15 +436,19 @@ rohes Plan-Binary oder `show -json` werden nicht an den Browser geliefert.
 Jeder Plan bleibt an sein unveraenderliches Runner-Paket gebunden. Ein Paketwechsel
 erfordert einen neuen Plan und dessen ausdrueckliche Freigabe.
 Laufende Apply-Ausgaben werden ebenfalls aktualisiert. Vor dem terminalen Ergebnis
-sichert der Worker den redigierten CLI-Text einmalig und verschluesselt beim Broker;
+sichert der Worker den letzten redigierten CLI-Text verschluesselt beim Broker;
 er bleibt nach Cleanup und Reload unter derselben privaten Zugriffskontrolle lesbar.
 **Apply-Nachweise** zeigen Plan-ID, SHA-256, Zielorganisation, State-Backend des
 Auftrags und Abschlusszeit. Die bestehende exakte Organisationsbestaetigung und
 gegebenenfalls destruktive Freigabe bleiben erforderlich; kein automatischer Retry.
 Ausgaben sind auf 2 MiB begrenzt; eine Kuerzung wird angezeigt. Die Sicherung ist
 best-effort: bei hartem Prozessabbruch oder fehlgeschlagenem Upload ist ein
-vollstaendiger historischer Mitschnitt nicht garantiert. CF-Live-Ausgabetransport
-bleibt offen; bereits gesicherter Text benoetigt keinen laufenden nativen Runner.
+vollstaendiger historischer Mitschnitt nicht garantiert. Bereits gesicherter
+Text benoetigt keinen laufenden nativen Runner. Waehrend des Streams werden
+Sitzung, Tenant, Rolle und Eigentuemerschaft erneut geprueft. Verbindungen werden
+spaetestens nach 35 Sekunden beendet und bei aktivem Auftrag neu geoeffnet;
+Backpressure, Byte-Limit und hoechstens acht Streams pro Eigentuemer/Tenant und
+API-Instanz begrenzen die Ressourcen. Details: [Plan-Ausfuehrung](../docs/plan-execution.md).
 
 Für neue Plattformen ist **Accelerator-Standard · Management-S3** der Default.
 Der ausgewählte Service Account führt den freigegebenen Bootstrap-Apply aus;

@@ -18,6 +18,7 @@ import { join } from "node:path";
 import { promisify } from "node:util";
 import {
   applicationRunnerSourceSchema,
+  openTofuProgress,
   platformFirewallImage,
   platformRunnerSourceSchema,
 } from "@lzc/contracts";
@@ -291,7 +292,14 @@ export class LocalPlanRunner implements PlanRunner {
         (value) => JSON.parse(value) as Record<string, unknown>,
         () => ({}),
       );
-      for (const phase of ["init", "validate", "plan", "apply", "migration"]) {
+      for (const phase of [
+        "init",
+        "validate",
+        "plan-events",
+        "plan",
+        "apply",
+        "migration",
+      ]) {
         const file = await open(join(work, `${phase}.log`), "r").catch(
           () => null,
         );
@@ -303,6 +311,15 @@ export class LocalPlanRunner implements PlanRunner {
           );
           const { bytesRead } = await file.read(buffer, 0, buffer.length, 0);
           let chunk = buffer.subarray(0, bytesRead).toString("utf8");
+          chunk = chunk.slice(0, chunk.lastIndexOf("\n") + 1);
+          if (phase === "plan-events") {
+            const progress = openTofuProgress(chunk);
+            chunk = progress.text;
+            if (progress.invalid)
+              console.error(
+                JSON.stringify({ event: "runner_progress_invalid" }),
+              );
+          }
           const values: unknown[] = [credentials];
           while (values.length) {
             const value = values.pop();
@@ -320,11 +337,15 @@ export class LocalPlanRunner implements PlanRunner {
             /-----BEGIN [^-]*PRIVATE KEY-----[\s\S]*?(?:-----END [^-]*PRIVATE KEY-----|$)/g,
             "(sensitive value)",
           );
+          chunk = `\n$ tofu ${phase === "plan-events" ? "plan (progress)" : phase}\n${chunk}`;
           if (Buffer.byteLength(chunk) > limit) {
-            chunk = Buffer.from(chunk).subarray(0, limit).toString("utf8");
+            chunk = new TextDecoder().decode(
+              Buffer.from(chunk).subarray(0, limit),
+              { stream: true },
+            );
             truncated = true;
           }
-          text += `\n$ tofu ${phase}\n${chunk}`;
+          text += chunk;
           truncated ||= bytesRead < (await file.stat()).size;
         } finally {
           await file.close();

@@ -0,0 +1,72 @@
+import { once } from "node:events";
+import { PassThrough } from "node:stream";
+import { setTimeout as delay } from "node:timers/promises";
+import type { PlanOutput } from "@lzc/contracts";
+import type { FastifyReply } from "fastify";
+import { CredentialError } from "../credentials/profiles.js";
+
+export function sendPlanOutput(
+  reply: FastifyReply,
+  initial: PlanOutput,
+  read: () => Promise<PlanOutput>,
+  wait: (signal: AbortSignal) => Promise<unknown> = (signal) =>
+    delay(1000, undefined, { signal }),
+) {
+  const controller = new AbortController();
+  const output = new PassThrough({ highWaterMark: 64 * 1024 });
+  const close = () => controller.abort();
+  reply.raw.once("close", close);
+  reply
+    .type("application/x-ndjson")
+    .header("Cache-Control", "private, no-store")
+    .header("X-Accel-Buffering", "no")
+    .send(output);
+  const write = async (event: unknown) => {
+    if (controller.signal.aborted) return;
+    if (!output.write(`${JSON.stringify(event)}\n`))
+      await once(output, "drain", { signal: controller.signal });
+  };
+  void (async () => {
+    const deadline = Date.now() + 30000;
+    const lifetime = setTimeout(() => controller.abort(), 35000);
+    try {
+      let current = initial;
+      let previous = "";
+      while (!controller.signal.aborted) {
+        const serialized = JSON.stringify(current);
+        if (serialized !== previous) {
+          await write({ type: "output", data: current });
+          previous = serialized;
+        } else await write({ type: "heartbeat" });
+        if (current.kind !== "live" || Date.now() >= deadline) break;
+        await wait(controller.signal);
+        if (controller.signal.aborted) break;
+        current = await read();
+      }
+      await write({ type: "complete" });
+    } catch (error) {
+      if (!controller.signal.aborted) {
+        reply.log.warn(
+          { event: "plan_output_stream_failed" },
+          "Plan output stream failed",
+        );
+        output.write(
+          `${JSON.stringify({
+            type: "error",
+            error:
+              error instanceof CredentialError
+                ? error.code
+                : "plan_output_unavailable",
+          })}\n`,
+        );
+      }
+    } finally {
+      clearTimeout(lifetime);
+      reply.raw.removeListener("close", close);
+      const aborted = controller.signal.aborted;
+      controller.abort();
+      if (aborted) output.destroy();
+      else output.end();
+    }
+  })();
+}
@@ -4,10 +4,12 @@ import { z } from "zod";
 import {
   type AuthServices,
   authenticatedSession,
+  cookie,
   validMutation,
 } from "../auth/routes.js";
 import { CredentialError } from "../credentials/profiles.js";
 import { RepositoryError } from "../github/repositories.js";
+import { sendPlanOutput } from "./output-stream.js";
 import type { Plans } from "./service.js";
 export type PlanRoutesServices = Pick<
   Plans,
@@ -21,6 +23,7 @@ export type PlanRoutesServices = Pick<
       | "state"
       | "outputs"
       | "output"
+      | "waitForOutput"
       | "recordOutput"
       | "migration"
       | "recovery"
@@ -34,6 +37,7 @@ export function registerPlans(
   auth: AuthServices,
   plans: PlanRoutesServices,
 ) {
+  const outputStreams = new Map<string, number>();
   app.register(async (routes) => {
     routes.setErrorHandler((error, _request, reply) => {
       if ((error as { statusCode?: number }).statusCode === 413)
@@ -78,7 +82,44 @@ export function registerPlans(
           if (!session) return reply.code(401).send();
           const { id } = z.object({ id: z.uuid() }).parse(request.params);
           reply.header("Cache-Control", "private, no-store");
-          return plans.output!(session, id);
+          const initial = await plans.output!(session, id);
+          if (!request.headers.accept?.includes("application/x-ndjson"))
+            return initial;
+          const owner = `${session.tenantId}:${session.userId}`;
+          const count = outputStreams.get(owner) ?? 0;
+          if (count >= 8)
+            return reply
+              .code(429)
+              .header("Retry-After", "2")
+              .send({ error: "plan_output_stream_limit" });
+          outputStreams.set(owner, count + 1);
+          reply.raw.once("close", () => {
+            const remaining = (outputStreams.get(owner) ?? 1) - 1;
+            if (remaining) outputStreams.set(owner, remaining);
+            else outputStreams.delete(owner);
+          });
+          sendPlanOutput(
+            reply,
+            initial,
+            async () => {
+              const token = cookie(request, "__Host-lzc-session");
+              const current = token
+                ? await auth.store.resolveSession(token)
+                : null;
+              if (
+                !current ||
+                current.id !== session.id ||
+                current.tenantId !== session.tenantId ||
+                current.userId !== session.userId
+              )
+                throw new CredentialError(401, "authentication_required");
+              return plans.output!(current, id);
+            },
+            plans.waitForOutput
+              ? (signal) => plans.waitForOutput!(id, signal)
+              : undefined,
+          );
+          return reply;
         });
       for (const action of ["checkpoint", "reconcile"] as const) {
         if (!plans[action]) continue;

@@ -1,4 +1,5 @@
 import { createHash, randomBytes, randomUUID } from "node:crypto";
+import { EventEmitter, once } from "node:events";
 import { isIP } from "node:net";
 import {
   planResultSchema,
@@ -50,6 +51,18 @@ const lockHash = providerLockHash;
 const invalid = (code: string, status = 409) =>
   new CredentialError(status, code);
 export class Plans {
+  private readonly outputUpdates = new EventEmitter();
+
+  async waitForOutput(id: string, signal: AbortSignal) {
+    try {
+      await once(this.outputUpdates, id, {
+        signal: AbortSignal.any([signal, AbortSignal.timeout(1000)]),
+      });
+    } catch (error) {
+      if (!(error instanceof Error && error.name === "AbortError")) throw error;
+    }
+  }
+
   constructor(
     private readonly pool: pg.Pool,
     private readonly profiles: Pick<
@@ -488,6 +501,7 @@ export class Plans {
       );
       if (!updated.rowCount) throw invalid("invalid_runner_transition");
     });
+    this.outputUpdates.emit(id);
   }
 
   private async preparation(session: Session, id: string) {

@@ -9,6 +9,7 @@ import { platformContractSchema } from "@lzc/domain";
 import { useEffect, useRef, useState } from "react";
 import { currentLanguage, t } from "../i18n";
 import type { DeploymentStep } from "../navigation";
+import { readPlanOutput } from "../plan-output";
 import type { Session } from "./Account";
 
 type Run = {
@@ -197,11 +198,15 @@ function PlanOutput({ run }: { run: Run }) {
     let timer: ReturnType<typeof setTimeout> | undefined;
     const deadline = Date.now() + 20 * 60 * 1000;
     async function load() {
+      let reconnectAfter = 2000;
       setLoading(true);
       try {
         const response = await fetch(`/api/v1/plans/${run.id}/output`, {
           credentials: "same-origin",
           cache: "no-store",
+          headers: {
+            Accept: running ? "application/x-ndjson" : "application/json",
+          },
           signal: controller.signal,
         });
         if (!response.ok)
@@ -212,24 +217,23 @@ function PlanOutput({ run }: { run: Run }) {
                 ? "Keine Berechtigung für diese Ausgabe."
                 : "Die OpenTofu-Ausgabe ist momentan nicht verfügbar.",
           );
-        const result = await response.json();
-        if (
-          typeof result.text !== "string" ||
-          result.text.length > 3 * 1024 * 1024 ||
-          typeof result.truncated !== "boolean"
-        )
-          throw new Error(
-            "Die OpenTofu-Ausgabe konnte nicht bestätigt werden.",
-          );
-        if (controller.signal.aborted) return;
+        const kind = await readPlanOutput(response, (result) => {
+          if (controller.signal.aborted) return;
+          if (
+            result.text ||
+            result.kind === "saved-plan" ||
+            result.kind === "execution"
+          )
+            setText(result.text);
+          setTruncated(result.truncated);
+          setLoading(false);
+          setError("");
+        });
         if (
-          result.text ||
-          result.kind === "saved-plan" ||
-          result.kind === "execution"
+          kind === "live" &&
+          response.headers.get("content-type")?.includes("application/x-ndjson")
         )
-          setText(result.text);
-        setTruncated(result.truncated);
-        setError("");
+          reconnectAfter = 100;
       } catch (cause) {
         if (!controller.signal.aborted)
           setError(
@@ -239,7 +243,7 @@ function PlanOutput({ run }: { run: Run }) {
         if (!controller.signal.aborted) {
           setLoading(false);
           if (running && Date.now() < deadline)
-            timer = setTimeout(() => void load(), 2000);
+            timer = setTimeout(() => void load(), reconnectAfter);
         }
       }
     }