diff --git a/.changes/unreleased/controlled-session-late-workload-exit.yaml b/.changes/unreleased/controlled-session-late-workload-exit.yaml
new file mode 100644
index 00000000..d3b62838
--- /dev/null
+++ b/.changes/unreleased/controlled-session-late-workload-exit.yaml
@@ -0,0 +1,2 @@
+kind: Fixed
+body: Preserve the controller channel when workload-exit observation arrives after the controlled-session output-finalization deadline.
diff --git a/.changes/unreleased/controlled-session-local-controller-image.yaml b/.changes/unreleased/controlled-session-local-controller-image.yaml
new file mode 100644
index 00000000..f189b890
--- /dev/null
+++ b/.changes/unreleased/controlled-session-local-controller-image.yaml
@@ -0,0 +1,2 @@
+kind: Fixed
+body: Allow controlled-session controller packaging to use the locked local image identity when a staged generation has no registry digest.
diff --git a/.changes/unreleased/controlled-session-omegaflow-conformance.yaml b/.changes/unreleased/controlled-session-omegaflow-conformance.yaml
new file mode 100644
index 00000000..647489ec
--- /dev/null
+++ b/.changes/unreleased/controlled-session-omegaflow-conformance.yaml
@@ -0,0 +1,2 @@
+kind: Docs
+body: Document and continuously prove the public controlled-session boundary with an OmegaFlow-shaped asciinema, Playwright, and Chromium conformance workflow, including retained artifacts after failed workload-output finalization.
diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml
index f6d5f67e..7ef998f9 100644
--- a/.github/workflows/integration.yml
+++ b/.github/workflows/integration.yml
@@ -117,6 +117,38 @@ jobs:
go test -timeout 15m ./internal/dockerdeploy
-run '^(TestControlledSessionSupervisorDockerIntegration|TestControlledSessionNetworkingDockerIntegration|TestControlledSessionControllerPackageDockerIntegration)$'
+ - name: Install pinned OmegaFlow conformance fixtures
+ if: matrix.name == 'Linux amd64'
+ run: |
+ asciinema_fixture=testdata/controlled-session/asciinema-v3-linux-amd64.json
+ asciinema_url=$(jq -r .url "$asciinema_fixture")
+ asciinema_sha256=$(jq -r .sha256 "$asciinema_fixture")
+ curl --fail --location --silent --show-error "$asciinema_url" --output "$RUNNER_TEMP/asciinema"
+ echo "$asciinema_sha256 $RUNNER_TEMP/asciinema" | sha256sum --check --strict
+ chmod 0755 "$RUNNER_TEMP/asciinema"
+
+ conformance_fixture=testdata/controlled-session/omegaflow-conformance-v1.json
+ playwright_url=$(jq -r '.packages[] | select(.name == "playwright") | .url' "$conformance_fixture")
+ playwright_sha256=$(jq -r '.packages[] | select(.name == "playwright") | .sha256' "$conformance_fixture")
+ playwright_core_url=$(jq -r '.packages[] | select(.name == "playwright-core") | .url' "$conformance_fixture")
+ playwright_core_sha256=$(jq -r '.packages[] | select(.name == "playwright-core") | .sha256' "$conformance_fixture")
+ curl --fail --location --silent --show-error "$playwright_url" --output "$RUNNER_TEMP/playwright.tgz"
+ curl --fail --location --silent --show-error "$playwright_core_url" --output "$RUNNER_TEMP/playwright-core.tgz"
+ echo "$playwright_sha256 $RUNNER_TEMP/playwright.tgz" | sha256sum --check --strict
+ echo "$playwright_core_sha256 $RUNNER_TEMP/playwright-core.tgz" | sha256sum --check --strict
+
+ echo "REPLOY_ASCIINEMA_FIXTURE=$RUNNER_TEMP/asciinema" >> "$GITHUB_ENV"
+ echo "REPLOY_PLAYWRIGHT_FIXTURE=$RUNNER_TEMP/playwright.tgz" >> "$GITHUB_ENV"
+ echo "REPLOY_PLAYWRIGHT_CORE_FIXTURE=$RUNNER_TEMP/playwright-core.tgz" >> "$GITHUB_ENV"
+
+ - name: Run OmegaFlow controlled-session conformance
+ if: matrix.name == 'Linux amd64'
+ env:
+ REPLOY_DOCKER_INTEGRATION: "1"
+ run: >-
+ go test -count=1 -timeout 15m ./internal/dockerdeploy
+ -run '^TestOmegaFlowControlledSessionConformanceDockerIntegration$'
+
- name: Run CLI runtime integration
if: ${{ matrix.persistent_install == false }}
env:
diff --git a/docs/CONTROLLED_SESSION_DESIGN.md b/docs/CONTROLLED_SESSION_DESIGN.md
index 1aee17c4..4b1b3b5e 100644
--- a/docs/CONTROLLED_SESSION_DESIGN.md
+++ b/docs/CONTROLLED_SESSION_DESIGN.md
@@ -1870,6 +1870,30 @@ reports the session failure. The fixture is a conformance test for the generic
Reploy boundary; OmegaFlow continues to own command completion, cwd reporting,
action markers, browser orchestration, and media rendering.
+Implementation status: complete for the initial Linux `amd64` conformance
+profile. A repository-owned OmegaFlow-shaped controller drives only the public
+JSON Lines broker stream and the terminal attachment beneath unmodified,
+checksum-pinned asciinema 3.2.1. The public host command runs it against two
+ordinary current staged deployments. The controller proves shell continuity
+across marked operations and Ctrl-C, a later resize, a terminal-started HTTP
+service reached by checksum-pinned Playwright 1.55.0 and Chromium, and retained
+cast, text, screenshot, and proof artifacts in the declared output directory.
+The browser controller base is pinned by multi-platform image digest; the
+conformance run is initially required in Linux `amd64` CI while the independent
+client packaging and Docker integration suites continue to cover Linux
+`arm64`.
+
+The same fixture injects a host-side PTY-observation timeout by suspending the
+public host process, after it publishes `terminating`, across the absolute
+output-finalization deadline. The controller, recorder, and private channel
+remain live while the workload exits. After the host resumes, the fixture
+proves the actionable failed-finalization event reaches the OmegaFlow-shaped
+controller, which retains and closes the partial cast, records the failure,
+sends `complete`, consumes `terminated`, acknowledges it, and exits cleanly
+while the public host command still reports the session failure. The fixture
+adds no OmegaFlow code, private-protocol dependency, runtime dependency
+installer, or public Reploy surface.
+
### Slice 6: User-Facing Documentation
After the generic runtime and at least one integration profile are proven,
diff --git a/internal/controlledsession/asciinema_fixture_linux_test.go b/internal/controlledsession/asciinema_fixture_linux_test.go
index 1faad431..17570ffb 100644
--- a/internal/controlledsession/asciinema_fixture_linux_test.go
+++ b/internal/controlledsession/asciinema_fixture_linux_test.go
@@ -16,6 +16,7 @@ import (
"os"
"os/exec"
"path/filepath"
+ "runtime"
"strings"
"testing"
"time"
@@ -140,7 +141,8 @@ func TestTerminalAttachmentRunsBeneathPinnedUnmodifiedAsciinemaV1(t *testing.T)
func readAsciinemaFixtureMetadataV1(t *testing.T) asciinemaFixtureMetadataV1 {
t.Helper()
- payload, err := os.ReadFile("../../testdata/controlled-session/asciinema-v3-linux-amd64.json")
+ fixturePath := "../../testdata/controlled-session/asciinema-v3-linux-" + runtime.GOARCH + ".json"
+ payload, err := os.ReadFile(fixturePath)
if err != nil {
t.Fatal(err)
}
diff --git a/internal/dockerdeploy/controlled_session_controller_package.go b/internal/dockerdeploy/controlled_session_controller_package.go
index fcb0e443..b4a5a625 100644
--- a/internal/dockerdeploy/controlled_session_controller_package.go
+++ b/internal/dockerdeploy/controlled_session_controller_package.go
@@ -228,7 +228,11 @@ func inspectControlledSessionControllerSourceV1(ctx context.Context, current Cur
if err != nil {
return InspectedImageCandidate{}, fmt.Errorf("inspect controlled-session controller generation %q: %w", current.Generation.Reference, err)
}
- inspection, err := parseDockerImageInspectionDetails(current.Generation.Reference, current.Lock.Platform, []byte(output))
+ // Reploy generation references are local mutable tags and deliberately do
+ // not have registry RepoDigests. Parse the inspected record through the
+ // immutable config identity already frozen in the build lock; the parser
+ // then also proves that the generation tag still resolves to that ID.
+ inspection, err := parseDockerImageInspectionDetails(string(current.Lock.FinalImage.ConfigDigest), current.Lock.Platform, []byte(output))
if err != nil {
return InspectedImageCandidate{}, fmt.Errorf("inspect controlled-session controller generation %q: %w", current.Generation.Reference, err)
}
diff --git a/internal/dockerdeploy/controlled_session_controller_package_test.go b/internal/dockerdeploy/controlled_session_controller_package_test.go
index 25614382..ee0b11a5 100644
--- a/internal/dockerdeploy/controlled_session_controller_package_test.go
+++ b/internal/dockerdeploy/controlled_session_controller_package_test.go
@@ -2,6 +2,7 @@ package dockerdeploy
import (
"context"
+ "fmt"
"os"
"os/exec"
"path/filepath"
@@ -19,6 +20,47 @@ import (
"github.com/omry/reploy/internal/providerstore"
)
+func TestInspectControlledSessionControllerSourceAcceptsLocalGenerationTagV1(t *testing.T) {
+ platform, err := blueprint.ParsePlatform("linux/amd64")
+ if err != nil {
+ t.Fatal(err)
+ }
+ configID := canonical.Digest("sha256:" + strings.Repeat("4", 64))
+ diffID := canonical.Digest("sha256:" + strings.Repeat("5", 64))
+ rootFS, err := deploy.RootFSSubject([]canonical.Digest{diffID})
+ if err != nil {
+ t.Fatal(err)
+ }
+ current := CurrentBuild{
+ Generation: deploy.EnvironmentGenerationState{Reference: "reploy/env/controller:g-current"},
+ Lock: deploy.BuildLockV1{
+ Platform: platform,
+ FinalImage: providers.RealizedImageV1{
+ Digest: configID, ConfigDigest: configID, RootFSSubject: rootFS,
+ },
+ },
+ }
+ inspection := fmt.Sprintf(
+ `[{"Id":%q,"RepoDigests":[],"Os":"linux","Architecture":"amd64","RootFS":{"Layers":[%q]},"Config":{}}]`,
+ configID,
+ diffID,
+ )
+ var calls [][]string
+ source, err := inspectControlledSessionControllerSourceV1(t.Context(), current, func(_ context.Context, args ...string) (string, error) {
+ calls = append(calls, append([]string{}, args...))
+ return inspection, nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(calls) != 1 || strings.Join(calls[0], " ") != "image inspect "+current.Generation.Reference {
+ t.Fatalf("Docker inspection calls = %#v", calls)
+ }
+ if source.Descriptor.AuthorReference != string(configID) || source.Descriptor.ImmutableReference != string(configID) || source.Image != current.Lock.FinalImage {
+ t.Fatalf("controller source = %#v", source)
+ }
+}
+
func TestControlledSessionControllerPackageDockerfileAddsOnlyControllerTool(t *testing.T) {
source := applicationRuntimeLayerTestRequest(t).Source
extracted := probearchive.ExtractedSessionClient{
diff --git a/internal/dockerdeploy/controlled_session_omegaflow_conformance_integration_test.go b/internal/dockerdeploy/controlled_session_omegaflow_conformance_integration_test.go
new file mode 100644
index 00000000..9b2383ce
--- /dev/null
+++ b/internal/dockerdeploy/controlled_session_omegaflow_conformance_integration_test.go
@@ -0,0 +1,508 @@
+//go:build linux
+
+package dockerdeploy
+
+import (
+ "bytes"
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "runtime"
+ "strings"
+ "syscall"
+ "testing"
+ "time"
+)
+
+type omegaFlowConformanceFixtureV1 struct {
+ Schema string `json:"schema"`
+ PlaywrightVersion string `json:"playwright_version"`
+ ControllerImage string `json:"controller_image"`
+ Packages []omegaFlowConformancePackageV1 `json:"packages"`
+}
+
+type omegaFlowConformancePackageV1 struct {
+ Name string `json:"name"`
+ URL string `json:"url"`
+ SHA256 string `json:"sha256"`
+}
+
+type omegaFlowHostResultV1 struct {
+ Schema string `json:"schema"`
+ OK bool `json:"ok"`
+ Error *string `json:"error"`
+ ResultDelivered *bool `json:"result_delivered"`
+ ResultAcknowledged *bool `json:"result_acknowledged"`
+ SessionResult *struct {
+ WorkloadOutputFinalizationStatus struct {
+ Kind string `json:"kind"`
+ Reason string `json:"reason,omitempty"`
+ } `json:"workload_output_finalization_status"`
+ ControllerFinalizationStatus struct {
+ Kind string `json:"kind"`
+ } `json:"controller_finalization_status"`
+ CleanupStatus struct {
+ Kind string `json:"kind"`
+ } `json:"cleanup_status"`
+ } `json:"session_result"`
+ ControllerOutput *struct {
+ Kind string `json:"kind"`
+ } `json:"controller_output"`
+}
+
+type omegaFlowControllerProofV1 struct {
+ Schema string `json:"schema"`
+ Scenario string `json:"scenario"`
+ Cast string `json:"cast"`
+ CastBytes int64 `json:"cast_bytes"`
+ BrowserScreenshot string `json:"browser_screenshot,omitempty"`
+ TerminalMarkersVerified bool `json:"terminal_markers_verified,omitempty"`
+ RecorderFailed bool `json:"recorder_failed,omitempty"`
+ OutputFinalizationStatus string `json:"output_finalization_status"`
+ OutputFinalizationReason string `json:"output_finalization_reason,omitempty"`
+}
+
+func TestOmegaFlowControlledSessionConformanceDockerIntegration(t *testing.T) {
+ if os.Getenv("REPLOY_DOCKER_INTEGRATION") != "1" {
+ t.Skip("set REPLOY_DOCKER_INTEGRATION=1 to run Docker integration evidence")
+ }
+ if runtime.GOOS != "linux" || runtime.GOARCH != "amd64" && runtime.GOARCH != "arm64" {
+ t.Skipf("OmegaFlow conformance requires a supported Linux host, got %s/%s", runtime.GOOS, runtime.GOARCH)
+ }
+ assets := omegaFlowConformanceAssetsV1(t)
+ ctx, cancel := context.WithTimeout(context.Background(), 14*time.Minute)
+ defer cancel()
+
+ repositoryRoot := repositoryRootForControllerPackageTestV1(t)
+ host := buildOmegaFlowConformanceHostV1(t, ctx, repositoryRoot)
+ controllerImage := buildOmegaFlowConformanceControllerImageV1(t, ctx, repositoryRoot, assets)
+ controllerDir, workloadDir := prepareOmegaFlowConformanceDeploymentsV1(t, ctx, host, controllerImage)
+
+ t.Run("terminal and browser handoff", func(t *testing.T) {
+ outputDir := filepath.Join(t.TempDir(), "output")
+ result, stderr, exitCode := runOmegaFlowConformanceSessionV1(t, ctx, host, controllerDir, workloadDir, outputDir, "success")
+ if exitCode != 0 || !result.OK || result.Error != nil {
+ t.Fatalf("public success result = %#v, exit=%d, stderr=%s, controller=%s", result, exitCode, stderr, readOmegaFlowControllerErrorV1(outputDir))
+ }
+ assertOmegaFlowHostResultV1(t, result, "drained", outputDir)
+ proof := readOmegaFlowControllerProofV1(t, outputDir, "success")
+ if !proof.TerminalMarkersVerified || proof.BrowserScreenshot == "" || proof.CastBytes == 0 {
+ t.Fatalf("success controller proof = %#v", proof)
+ }
+ png, err := os.ReadFile(filepath.Join(outputDir, proof.BrowserScreenshot))
+ if err != nil || !bytes.HasPrefix(png, []byte("\x89PNG\r\n\x1a\n")) {
+ t.Fatalf("browser screenshot = %d bytes, %v", len(png), err)
+ }
+ for _, artifact := range []string{proof.Cast, "terminal.txt", proof.BrowserScreenshot} {
+ if info, err := os.Stat(filepath.Join(outputDir, artifact)); err != nil || info.Size() == 0 {
+ t.Fatalf("retained success artifact %q = %#v, %v", artifact, info, err)
+ }
+ }
+ })
+
+ t.Run("failed output finalization", func(t *testing.T) {
+ outputDir := filepath.Join(t.TempDir(), "output")
+ result, stderr, exitCode := runOmegaFlowConformanceSessionV1(t, ctx, host, controllerDir, workloadDir, outputDir, "failed-output-finalization")
+ if exitCode != 1 || result.OK || result.Error == nil {
+ t.Fatalf("public failure result = %#v, exit=%d, stderr=%s, controller=%s", result, exitCode, stderr, readOmegaFlowControllerErrorV1(outputDir))
+ }
+ assertOmegaFlowHostResultV1(t, result, "failed", outputDir)
+ if result.SessionResult.WorkloadOutputFinalizationStatus.Reason != "workload PTY output finalization timed out" ||
+ !strings.Contains(stderr, "context deadline exceeded") {
+ t.Fatalf("failure diagnostics = result %#v stderr %q", result.SessionResult.WorkloadOutputFinalizationStatus, stderr)
+ }
+ proof := readOmegaFlowControllerProofV1(t, outputDir, "failed-output-finalization")
+ if !proof.RecorderFailed || proof.CastBytes == 0 || proof.OutputFinalizationStatus != "failed" || proof.OutputFinalizationReason == "" {
+ t.Fatalf("failed-finalization controller proof = %#v", proof)
+ }
+ payload, err := os.ReadFile(filepath.Join(outputDir, proof.Cast))
+ if err != nil || len(payload) == 0 || payload[len(payload)-1] != '\n' {
+ t.Fatalf("retained partial cast = %d bytes, %v", len(payload), err)
+ }
+ })
+}
+
+func readOmegaFlowControllerErrorV1(outputDir string) string {
+ payload, err := os.ReadFile(filepath.Join(outputDir, "controller-error.txt"))
+ if err != nil {
+ return err.Error()
+ }
+ return string(payload)
+}
+
+type omegaFlowConformanceAssets struct {
+ fixture omegaFlowConformanceFixtureV1
+ asciinema string
+ playwright string
+ playwrightCore string
+}
+
+func omegaFlowConformanceAssetsV1(t *testing.T) omegaFlowConformanceAssets {
+ t.Helper()
+ asciinema := os.Getenv("REPLOY_ASCIINEMA_FIXTURE")
+ playwright := os.Getenv("REPLOY_PLAYWRIGHT_FIXTURE")
+ playwrightCore := os.Getenv("REPLOY_PLAYWRIGHT_CORE_FIXTURE")
+ if asciinema == "" || playwright == "" || playwrightCore == "" {
+ t.Skip("set REPLOY_ASCIINEMA_FIXTURE, REPLOY_PLAYWRIGHT_FIXTURE, and REPLOY_PLAYWRIGHT_CORE_FIXTURE to pinned fixture files")
+ }
+ root := repositoryRootForControllerPackageTestV1(t)
+ payload, err := os.ReadFile(filepath.Join(root, "testdata", "controlled-session", "omegaflow-conformance-v1.json"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var fixture omegaFlowConformanceFixtureV1
+ if err := json.Unmarshal(payload, &fixture); err != nil {
+ t.Fatal(err)
+ }
+ if fixture.Schema != "reploy-omegaflow-conformance-fixture-v1" || fixture.PlaywrightVersion == "" || fixture.ControllerImage == "" || len(fixture.Packages) != 2 {
+ t.Fatalf("invalid OmegaFlow conformance fixture = %#v", fixture)
+ }
+ paths := map[string]string{"playwright": playwright, "playwright-core": playwrightCore}
+ for _, item := range fixture.Packages {
+ path, ok := paths[item.Name]
+ if !ok || item.URL == "" {
+ t.Fatalf("unexpected Playwright fixture package = %#v", item)
+ }
+ assertOmegaFlowFixtureDigestV1(t, path, item.SHA256)
+ }
+ asciinemaMetadataPath := filepath.Join(root, "testdata", "controlled-session", "asciinema-v3-linux-"+runtime.GOARCH+".json")
+ asciinemaMetadata, err := os.ReadFile(asciinemaMetadataPath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var recorder struct {
+ Version string `json:"version"`
+ SHA256 string `json:"sha256"`
+ }
+ if err := json.Unmarshal(asciinemaMetadata, &recorder); err != nil {
+ t.Fatal(err)
+ }
+ assertOmegaFlowFixtureDigestV1(t, asciinema, recorder.SHA256)
+ version, err := exec.Command(asciinema, "--version").CombinedOutput()
+ if err != nil || strings.TrimSpace(string(version)) != "asciinema "+recorder.Version {
+ t.Fatalf("asciinema fixture version = %q, %v", version, err)
+ }
+ return omegaFlowConformanceAssets{fixture: fixture, asciinema: asciinema, playwright: playwright, playwrightCore: playwrightCore}
+}
+
+func assertOmegaFlowFixtureDigestV1(t *testing.T, path string, want string) {
+ t.Helper()
+ payload, err := os.ReadFile(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ digest := sha256.Sum256(payload)
+ if got := hex.EncodeToString(digest[:]); got != want {
+ t.Fatalf("fixture %s SHA-256 = %s, want %s", path, got, want)
+ }
+}
+
+func buildOmegaFlowConformanceHostV1(t *testing.T, ctx context.Context, root string) string {
+ t.Helper()
+ outdir := t.TempDir()
+ command := exec.CommandContext(ctx, filepath.Join(root, "tools", "build_reploy"), "--target", "linux-"+runtime.GOARCH, "--outdir", outdir)
+ command.Dir = root
+ command.Env = append(os.Environ(), "GOCACHE="+filepath.Join(t.TempDir(), "go-cache"))
+ if output, err := command.CombinedOutput(); err != nil {
+ t.Fatalf("build packaged conformance host: %v\n%s", err, output)
+ }
+ return filepath.Join(outdir, "linux-"+runtime.GOARCH, "reploy")
+}
+
+func buildOmegaFlowConformanceControllerImageV1(t *testing.T, ctx context.Context, root string, assets omegaFlowConformanceAssets) string {
+ t.Helper()
+ if output, err := exec.CommandContext(ctx, "docker", "pull", assets.fixture.ControllerImage).CombinedOutput(); err != nil {
+ t.Fatalf("pull pinned Playwright controller image: %v\n%s", err, output)
+ }
+ workspace := t.TempDir()
+ helper := filepath.Join(workspace, "omegaflow-conformance-controller")
+ build := exec.CommandContext(ctx, "go", "build", "-buildvcs=false", "-o", helper, "./internal/dockerdeploy/testdata/omegaflow_conformance_controller")
+ build.Dir = root
+ build.Env = append(os.Environ(), "CGO_ENABLED=0", "GOOS=linux", "GOARCH="+runtime.GOARCH, "GOCACHE="+filepath.Join(workspace, "go-cache"))
+ if output, err := build.CombinedOutput(); err != nil {
+ t.Fatalf("build OmegaFlow-shaped controller: %v\n%s", err, output)
+ }
+ for source, name := range map[string]string{
+ assets.asciinema: "asciinema", assets.playwright: "playwright.tgz", assets.playwrightCore: "playwright-core.tgz",
+ } {
+ payload, err := os.ReadFile(source)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(workspace, name), payload, 0o600); err != nil {
+ t.Fatal(err)
+ }
+ }
+ browserProof, err := os.ReadFile(filepath.Join(root, "internal", "dockerdeploy", "testdata", "omegaflow_conformance_controller", "browser-proof.js"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(workspace, "browser-proof.js"), browserProof, 0o600); err != nil {
+ t.Fatal(err)
+ }
+ dockerfile := "FROM " + assets.fixture.ControllerImage + "\n" +
+ "USER 0:0\n" +
+ "COPY --chmod=0555 omegaflow-conformance-controller /usr/local/bin/omegaflow-conformance-controller\n" +
+ "COPY --chmod=0555 asciinema /usr/local/bin/asciinema\n" +
+ "COPY playwright.tgz playwright-core.tgz /tmp/\n" +
+ "RUN mkdir -p /opt/omegaflow/node_modules/playwright /opt/omegaflow/node_modules/playwright-core && " +
+ "tar -xzf /tmp/playwright.tgz --strip-components=1 -C /opt/omegaflow/node_modules/playwright && " +
+ "tar -xzf /tmp/playwright-core.tgz --strip-components=1 -C /opt/omegaflow/node_modules/playwright-core && " +
+ "rm /tmp/playwright.tgz /tmp/playwright-core.tgz\n" +
+ "COPY --chmod=0444 browser-proof.js /opt/omegaflow/browser-proof.js\n"
+ if err := os.WriteFile(filepath.Join(workspace, "Dockerfile"), []byte(dockerfile), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ image := uniqueDockerIntegrationName("reploy-omegaflow-controller")
+ command := exec.CommandContext(ctx, "docker", "build", "--pull=false", "--tag", image, workspace)
+ if output, err := command.CombinedOutput(); err != nil {
+ t.Fatalf("build OmegaFlow controller fixture image: %v\n%s", err, output)
+ }
+ imageID := strings.TrimSpace(runDockerIntegration(t, ctx, "image", "inspect", "--format", "{{.Id}}", image))
+ t.Cleanup(func() {
+ if output, err := exec.CommandContext(context.Background(), "docker", "image", "rm", image).CombinedOutput(); err != nil {
+ t.Errorf("remove OmegaFlow controller fixture image: %v\n%s", err, output)
+ }
+ })
+ return imageID
+}
+
+func prepareOmegaFlowConformanceDeploymentsV1(t *testing.T, ctx context.Context, host string, controllerImage string) (string, string) {
+ t.Helper()
+ root := shortControlledSessionChannelTestDirectoryV1(t)
+ controllerBlueprint := filepath.Join(root, "controller.blueprint.yaml")
+ workloadBlueprint := filepath.Join(root, "workload.blueprint.yaml")
+ platform := "linux/" + runtime.GOARCH
+ controller := fmt.Sprintf(`blueprint:
+ schema: 1
+ version: 0.1.0
+ compatibility:
+ platforms: [%s]
+environment:
+ id: omegaflow-conformance-controller
+ base:
+ image: %s
+ applications:
+ controller:
+ packages:
+ os:
+ - package: bash
+ exports:
+ shell:
+ executable: /usr/bin/bash
+ - util-linux
+ executables:
+ shell:
+ source: os
+ binary: shell
+ commands:
+ conformance:
+ executable: controller.shell
+ trigger: [conformance]
+ native_command: true
+ argv: [-c, 'exec /usr/local/bin/omegaflow-conformance-controller "$@"', conformance]
+docker: {}
+`, platform, controllerImage)
+ workload := fmt.Sprintf(`blueprint:
+ schema: 1
+ version: 0.1.0
+ compatibility:
+ platforms: [%s]
+environment:
+ id: omegaflow-conformance-workload
+ base:
+ image: python:3.13-slim
+ applications:
+ shell:
+ packages:
+ os:
+ - package: bash
+ exports:
+ shell:
+ executable: /usr/bin/bash
+ executables:
+ shell:
+ source: os
+ binary: shell
+ commands:
+ shell:
+ executable: shell.shell
+ workload:
+ command: shell
+ endpoints:
+ web:
+ scheme: http
+ port: 8080
+docker:
+ workload:
+ endpoints:
+ web:
+ extends: environment.workload.endpoints.web
+ bind: {address: 0.0.0.0}
+ publish: {address: 127.0.0.1, staging: 18080, deployed: 18081}
+`, platform)
+ if err := os.WriteFile(controllerBlueprint, []byte(controller), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(workloadBlueprint, []byte(workload), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ controllerDir := filepath.Join(root, "controller")
+ workloadDir := filepath.Join(root, "workload")
+ for _, deployment := range []struct {
+ dir string
+ blueprint string
+ }{
+ {dir: controllerDir, blueprint: controllerBlueprint},
+ {dir: workloadDir, blueprint: workloadBlueprint},
+ } {
+ runOmegaFlowHostCommandV1(t, ctx, host, "stage", "--dir", deployment.dir, "--platform", platform, "file:"+deployment.blueprint)
+ runOmegaFlowHostCommandV1(t, ctx, host, "build", "--dir", deployment.dir)
+ }
+ return controllerDir, workloadDir
+}
+
+func runOmegaFlowHostCommandV1(t *testing.T, ctx context.Context, host string, args ...string) string {
+ t.Helper()
+ command := exec.CommandContext(ctx, host, args...)
+ command.Env = append(os.Environ(), "NO_UPDATE_NOTIFIER=1")
+ output, err := command.CombinedOutput()
+ if err != nil {
+ t.Fatalf("run %s %s: %v\n%s", host, strings.Join(args, " "), err, output)
+ }
+ return string(output)
+}
+
+func runOmegaFlowConformanceSessionV1(
+ t *testing.T,
+ ctx context.Context,
+ host string,
+ controllerDir string,
+ workloadDir string,
+ outputDir string,
+ scenario string,
+) (omegaFlowHostResultV1, string, int) {
+ t.Helper()
+ command := exec.CommandContext(ctx, host,
+ "controlled-session", "run",
+ "--controller-dir", controllerDir,
+ "--workload-dir", workloadDir,
+ "--endpoint", "web",
+ "--columns", "80", "--rows", "24",
+ "--output-dir", outputDir,
+ "--controller-finalization-timeout", "1m",
+ "--result-acknowledgement-timeout", "15s",
+ "--cleanup-timeout", "30s",
+ "--", "conformance", scenario,
+ )
+ command.Env = append(os.Environ(), "NO_UPDATE_NOTIFIER=1")
+ var stdout, stderr bytes.Buffer
+ command.Stdout, command.Stderr = &stdout, &stderr
+ var err error
+ if scenario == "failed-output-finalization" {
+ before := omegaFlowControllerContainersV1(t, ctx)
+ if err = command.Start(); err == nil {
+ resumed := false
+ defer func() {
+ if !resumed && command.Process != nil {
+ _ = command.Process.Signal(syscall.SIGCONT)
+ }
+ }()
+ waitForOmegaFlowControllerLogV1(t, ctx, before, "OMEGAFLOW-CONFORMANCE-TERMINATING", 30*time.Second)
+ if signalErr := command.Process.Signal(syscall.SIGSTOP); signalErr != nil {
+ t.Fatalf("suspend public controlled-session host: %v", signalErr)
+ }
+ time.Sleep(31 * time.Second)
+ if signalErr := command.Process.Signal(syscall.SIGCONT); signalErr != nil {
+ t.Fatalf("resume public controlled-session host: %v", signalErr)
+ }
+ resumed = true
+ err = command.Wait()
+ }
+ } else {
+ err = command.Run()
+ }
+ exitCode := 0
+ if err != nil {
+ var exitErr *exec.ExitError
+ if !errors.As(err, &exitErr) {
+ t.Fatalf("run public controlled-session command: %v\n%s", err, stderr.String())
+ }
+ exitCode = exitErr.ExitCode()
+ }
+ var result omegaFlowHostResultV1
+ if decodeErr := json.Unmarshal(stdout.Bytes(), &result); decodeErr != nil {
+ t.Fatalf("decode public controlled-session result: %v\nstdout=%s\nstderr=%s", decodeErr, stdout.String(), stderr.String())
+ }
+ if result.Schema != "reploy-controlled-session-run-result-v1" {
+ t.Fatalf("public result schema = %q", result.Schema)
+ }
+ return result, stderr.String(), exitCode
+}
+
+func omegaFlowControllerContainersV1(t *testing.T, ctx context.Context) map[string]bool {
+ t.Helper()
+ command := exec.CommandContext(ctx, "docker", "ps", "--filter", "label=io.reploy.session.role=controller", "--format", "{{.ID}}")
+ payload, err := command.Output()
+ if err != nil {
+ t.Fatalf("list controlled-session controller containers: %v", err)
+ }
+ result := map[string]bool{}
+ for _, id := range strings.Fields(string(payload)) {
+ result[id] = true
+ }
+ return result
+}
+
+func waitForOmegaFlowControllerLogV1(t *testing.T, ctx context.Context, before map[string]bool, marker string, timeout time.Duration) {
+ t.Helper()
+ deadline := time.Now().Add(timeout)
+ for time.Now().Before(deadline) {
+ for id := range omegaFlowControllerContainersV1(t, ctx) {
+ if before[id] {
+ continue
+ }
+ command := exec.CommandContext(ctx, "docker", "logs", id)
+ payload, err := command.CombinedOutput()
+ if err == nil && bytes.Contains(payload, []byte(marker)) {
+ return
+ }
+ }
+ time.Sleep(100 * time.Millisecond)
+ }
+ t.Fatalf("timed out waiting for controlled-session controller log marker %q", marker)
+}
+
+func assertOmegaFlowHostResultV1(t *testing.T, result omegaFlowHostResultV1, outputStatus string, outputDir string) {
+ t.Helper()
+ if result.SessionResult == nil || result.ResultDelivered == nil || !*result.ResultDelivered ||
+ result.ResultAcknowledged == nil || !*result.ResultAcknowledged || result.ControllerOutput == nil ||
+ result.ControllerOutput.Kind != "directory-retained" ||
+ result.SessionResult.WorkloadOutputFinalizationStatus.Kind != outputStatus ||
+ result.SessionResult.ControllerFinalizationStatus.Kind != "completed" ||
+ result.SessionResult.CleanupStatus.Kind != "succeeded" {
+ t.Fatalf("public controlled-session result invariants = %#v, controller=%s", result, readOmegaFlowControllerErrorV1(outputDir))
+ }
+}
+
+func readOmegaFlowControllerProofV1(t *testing.T, outputDir string, scenario string) omegaFlowControllerProofV1 {
+ t.Helper()
+ payload, err := os.ReadFile(filepath.Join(outputDir, scenario+"-proof.json"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var proof omegaFlowControllerProofV1
+ if err := json.Unmarshal(payload, &proof); err != nil {
+ t.Fatal(err)
+ }
+ if proof.Schema != "reploy-omegaflow-conformance-proof-v1" || proof.Scenario != scenario {
+ t.Fatalf("controller proof identity = %#v", proof)
+ }
+ return proof
+}
diff --git a/internal/dockerdeploy/controlled_session_supervisor.go b/internal/dockerdeploy/controlled_session_supervisor.go
index 130e651c..12d99db3 100644
--- a/internal/dockerdeploy/controlled_session_supervisor.go
+++ b/internal/dockerdeploy/controlled_session_supervisor.go
@@ -1055,6 +1055,16 @@ func (supervisor *controlledSessionSupervisorV1) observeWorkloadResult(result co
return
}
if supervisor.transportHealthy {
+ // The output-finalization deadline also bounds lifecycle events that
+ // precede the one authoritative finalization outcome. If workload exit
+ // observation arrives after that deadline, omitting this advisory event
+ // must not be misclassified as loss of an otherwise healthy controller
+ // transport. The terminal result still carries the recorded status.
+ deadline := supervisor.terminationDeadline()
+ if !supervisor.backend.now().Before(deadline) {
+ supervisor.recordLateWorkloadExitPublicationV1()
+ return
+ }
if err := supervisor.sendPreFinalizationLifecycleEvent(controlledsession.EventV1{
Kind: controlledsession.EventWorkloadExitV1,
WorkloadExit: &controlledsession.WorkloadExitV1{Status: result.status},
@@ -1064,6 +1074,10 @@ func (supervisor *controlledSessionSupervisorV1) observeWorkloadResult(result co
}
}
+func (supervisor *controlledSessionSupervisorV1) recordLateWorkloadExitPublicationV1() {
+ supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, fmt.Errorf("controlled-session workload exit was observed after its publication deadline"))
+}
+
func (supervisor *controlledSessionSupervisorV1) observeControllerLoss(reason string, detail error) {
snapshot := supervisor.machine.Snapshot()
if snapshot.State == controlledsession.StateTerminatedV1 || snapshot.ControllerFinalizationStatus.Kind == controlledsession.ControllerFinalizationLostV1 {
diff --git a/internal/dockerdeploy/controlled_session_supervisor_test.go b/internal/dockerdeploy/controlled_session_supervisor_test.go
index 3915163a..0f5b8627 100644
--- a/internal/dockerdeploy/controlled_session_supervisor_test.go
+++ b/internal/dockerdeploy/controlled_session_supervisor_test.go
@@ -179,6 +179,42 @@ func TestControlledSessionSupervisorIgnoresPTYRequestsRacingTerminatingEvent(t *
}
}
+func TestControlledSessionSupervisorPreservesTransportForLateWorkloadExitV1(t *testing.T) {
+ plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"})
+ machine, err := controlledsession.NewMachineV1(plan.Authorization)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := machine.Observe(controlledsession.ObservationV1{Kind: controlledsession.ObservationActivatedV1}); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := machine.Observe(controlledsession.ObservationV1{
+ Kind: controlledsession.ObservationHostCancelV1, Reason: "test termination",
+ }); err != nil {
+ t.Fatal(err)
+ }
+ now := time.Now()
+ code := 0
+ supervisor := &controlledSessionSupervisorV1{
+ machine: machine, transportHealthy: true,
+ terminationAt: now.Add(-controlledSessionOutputFinalizationTimeoutV1),
+ backend: controlledSessionSupervisorBackendV1{now: func() time.Time { return now }},
+ }
+ supervisor.observeWorkloadResult(controlledSessionProcessResultV1{
+ status: controlledsession.ProcessStatusV1{Kind: controlledsession.ProcessStatusExitedV1, Code: &code},
+ })
+ if !supervisor.transportHealthy {
+ t.Fatal("late workload-exit observation poisoned the controller transport")
+ }
+ snapshot := machine.Snapshot()
+ if snapshot.WorkloadStatus.Kind != controlledsession.ProcessStatusExitedV1 || snapshot.WorkloadStatus.Code == nil || *snapshot.WorkloadStatus.Code != code {
+ t.Fatalf("late workload status = %#v", snapshot.WorkloadStatus)
+ }
+ if supervisor.diagnosticErr == nil || !strings.Contains(supervisor.diagnosticErr.Error(), "after its publication deadline") {
+ t.Fatalf("late workload-exit diagnostic = %v", supervisor.diagnosticErr)
+ }
+}
+
func TestRunControlledSessionV1PreparesAttachesAndCleansLeaseNetwork(t *testing.T) {
plan := controlledSessionNetworkPlanFixtureV1(t)
requests := make(chan controlledsession.RequestV1, 8)
diff --git a/internal/dockerdeploy/testdata/omegaflow_conformance_controller/browser-proof.js b/internal/dockerdeploy/testdata/omegaflow_conformance_controller/browser-proof.js
new file mode 100644
index 00000000..0c317022
--- /dev/null
+++ b/internal/dockerdeploy/testdata/omegaflow_conformance_controller/browser-proof.js
@@ -0,0 +1,41 @@
+const { chromium } = require("playwright");
+
+async function main() {
+ const [url, screenshot] = process.argv.slice(2);
+ if (!url || !screenshot) {
+ throw new Error("usage: browser-proof URL SCREENSHOT");
+ }
+ const browser = await chromium.launch({
+ headless: true,
+ args: ["--disable-dev-shm-usage", "--no-sandbox"],
+ });
+ try {
+ const page = await browser.newPage({ viewport: { width: 800, height: 600 } });
+ let lastError;
+ for (let attempt = 0; attempt < 40; attempt += 1) {
+ try {
+ await page.goto(url, { waitUntil: "networkidle", timeout: 1000 });
+ lastError = undefined;
+ break;
+ } catch (error) {
+ lastError = error;
+ await new Promise((resolve) => setTimeout(resolve, 250));
+ }
+ }
+ if (lastError) {
+ throw lastError;
+ }
+ const text = await page.locator("#proof").textContent();
+ if (text !== "browser-proof") {
+ throw new Error(`unexpected endpoint content: ${JSON.stringify(text)}`);
+ }
+ await page.screenshot({ path: screenshot });
+ } finally {
+ await browser.close();
+ }
+}
+
+main().catch((error) => {
+ console.error(error.stack || String(error));
+ process.exit(1);
+});
diff --git a/internal/dockerdeploy/testdata/omegaflow_conformance_controller/main.go b/internal/dockerdeploy/testdata/omegaflow_conformance_controller/main.go
new file mode 100644
index 00000000..d95ebd85
--- /dev/null
+++ b/internal/dockerdeploy/testdata/omegaflow_conformance_controller/main.go
@@ -0,0 +1,770 @@
+package main
+
+import (
+ "bufio"
+ "bytes"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "strings"
+ "sync"
+ "syscall"
+ "time"
+)
+
+const controllerStreamSchema = "reploy-controlled-session-client-v1"
+
+type endpoint struct {
+ ID string `json:"id"`
+ Scheme string `json:"scheme"`
+ Host string `json:"host"`
+ Port int `json:"port"`
+}
+
+type lifecycleResult struct {
+ Cause string `json:"cause"`
+ WorkloadStatus struct {
+ Kind string `json:"kind"`
+ Code *int `json:"code,omitempty"`
+ Reason string `json:"reason,omitempty"`
+ } `json:"workload_status"`
+ WorkloadOutputFinalizationStatus struct {
+ Kind string `json:"kind"`
+ Reason string `json:"reason,omitempty"`
+ } `json:"workload_output_finalization_status"`
+ RuntimeObservationStatus struct {
+ Kind string `json:"kind"`
+ Reason string `json:"reason,omitempty"`
+ } `json:"runtime_observation_status"`
+ ControllerFinalizationStatus struct {
+ Kind string `json:"kind"`
+ Reason string `json:"reason,omitempty"`
+ } `json:"controller_finalization_status"`
+ CleanupStatus struct {
+ Kind string `json:"kind"`
+ Message string `json:"message,omitempty"`
+ } `json:"cleanup_status"`
+ RecoveryAction string `json:"recovery_action"`
+}
+
+type event struct {
+ Schema string `json:"schema"`
+ Type string `json:"type"`
+ TerminalSocket string `json:"terminal_socket,omitempty"`
+ Operations []string `json:"operations,omitempty"`
+ Endpoints []endpoint `json:"endpoints,omitempty"`
+ Columns uint32 `json:"columns,omitempty"`
+ Rows uint32 `json:"rows,omitempty"`
+ OutputFinalizationTimeoutMilliseconds uint32 `json:"output_finalization_timeout_milliseconds,omitempty"`
+ Status json.RawMessage `json:"status,omitempty"`
+ Cause string `json:"cause,omitempty"`
+ Reason string `json:"reason,omitempty"`
+ Code string `json:"code,omitempty"`
+ Message string `json:"message,omitempty"`
+ Result *lifecycleResult `json:"result,omitempty"`
+}
+
+type broker struct {
+ command *exec.Cmd
+ input io.WriteCloser
+ events *bufio.Scanner
+ stderr bytes.Buffer
+ waited bool
+}
+
+type recorder struct {
+ command *exec.Cmd
+ input io.WriteCloser
+ stderr bytes.Buffer
+ cast string
+ output synchronizedBuffer
+}
+
+type synchronizedBuffer struct {
+ mu sync.Mutex
+ payload []byte
+}
+
+func (buffer *synchronizedBuffer) Write(payload []byte) (int, error) {
+ buffer.mu.Lock()
+ defer buffer.mu.Unlock()
+ buffer.payload = append(buffer.payload, payload...)
+ return len(payload), nil
+}
+
+func (buffer *synchronizedBuffer) contains(marker []byte) bool {
+ buffer.mu.Lock()
+ defer buffer.mu.Unlock()
+ return bytes.Contains(buffer.payload, marker)
+}
+
+type proof struct {
+ Schema string `json:"schema"`
+ Scenario string `json:"scenario"`
+ Cast string `json:"cast"`
+ CastBytes int64 `json:"cast_bytes"`
+ BrowserScreenshot string `json:"browser_screenshot,omitempty"`
+ TerminalMarkersVerified bool `json:"terminal_markers_verified,omitempty"`
+ RecorderFailed bool `json:"recorder_failed,omitempty"`
+ OutputFinalizationStatus string `json:"output_finalization_status"`
+ OutputFinalizationReason string `json:"output_finalization_reason,omitempty"`
+}
+
+func main() {
+ if err := run(os.Args[1:]); err != nil {
+ fmt.Fprintf(os.Stderr, "omegaflow conformance controller: %v\n", err)
+ if outputDir := os.Getenv("REPLOY_OUTPUT_DIR"); outputDir != "" {
+ _ = os.WriteFile(filepath.Join(outputDir, "controller-error.txt"), []byte(err.Error()+"\n"), 0o644)
+ }
+ os.Exit(1)
+ }
+}
+
+func run(args []string) (resultErr error) {
+ if len(args) != 1 || args[0] != "success" && args[0] != "failed-output-finalization" {
+ return fmt.Errorf("usage: omegaflow-conformance-controller {success | failed-output-finalization}")
+ }
+ outputDir := os.Getenv("REPLOY_OUTPUT_DIR")
+ if outputDir == "" {
+ return fmt.Errorf("REPLOY_OUTPUT_DIR is required")
+ }
+ client, err := startBroker()
+ if err != nil {
+ return err
+ }
+ defer func() {
+ if client.command.Process != nil && client.command.ProcessState == nil {
+ _ = client.command.Process.Kill()
+ }
+ if !client.waited {
+ resultErr = errors.Join(resultErr, client.wait())
+ }
+ }()
+
+ ready, err := client.read("broker-ready", 10*time.Second)
+ if err != nil {
+ return err
+ }
+ cast := filepath.Join(outputDir, args[0]+".cast")
+ recording, err := startRecorder(ready.TerminalSocket, cast)
+ if err != nil {
+ return err
+ }
+ defer func() {
+ if recording.command.Process != nil && recording.command.ProcessState == nil {
+ _ = syscall.Kill(-recording.command.Process.Pid, syscall.SIGKILL)
+ }
+ }()
+
+ opened, err := client.read("opened", 10*time.Second)
+ if err != nil {
+ return err
+ }
+ if _, err := client.read("ready", 30*time.Second); err != nil {
+ return err
+ }
+ if args[0] == "success" {
+ return runSuccess(outputDir, opened, client, recording)
+ }
+ if err := runFailedOutputFinalization(outputDir, client, recording); err != nil {
+ return err
+ }
+ return nil
+}
+
+func runSuccess(outputDir string, opened event, client *broker, recording *recorder) error {
+ web, err := selectEndpoint(opened.Endpoints, "web")
+ if err != nil {
+ return err
+ }
+ if err := recording.writeString("printf 'OPERATION-ONE\\n'; cd /tmp; trap 'printf \\\"\\\\036INTERRUPT-DONE\\\\n\\\"' INT; printf '\\036SLEEP-ACTIVE\\n'; sleep 30; trap - INT\n"); err != nil {
+ return err
+ }
+ if err := recording.waitFor(append([]byte{0x1e}, []byte("SLEEP-ACTIVE")...), 10*time.Second); err != nil {
+ return err
+ }
+ if _, err := recording.input.Write([]byte{0x03}); err != nil {
+ return fmt.Errorf("send terminal Ctrl-C: %w", err)
+ }
+ if err := recording.waitFor(append([]byte{0x1e}, []byte("INTERRUPT-DONE")...), 10*time.Second); err != nil {
+ return err
+ }
+ if err := client.write(map[string]any{"schema": controllerStreamSchema, "type": "resize", "columns": 100, "rows": 31}); err != nil {
+ return err
+ }
+ command := "printf 'OPERATION-TWO\\n'; printf 'CWD='; pwd; printf 'SIZE='; stty size; " +
+ "mkdir -p /mnt/reploy-home/site; printf '%s\\n' 'OmegaFlow handoffbrowser-proof
' > /mnt/reploy-home/site/index.html; " +
+ "python3 -m http.server 8080 --bind 0.0.0.0 --directory /mnt/reploy-home/site >/mnt/reploy-home/server.log 2>&1 & " +
+ "for attempt in 1 2 3 4 5 6 7 8 9 10; do python3 -c 'import urllib.request; urllib.request.urlopen(\"http://127.0.0.1:8080/\", timeout=1).read()' >/dev/null 2>&1 && break; sleep 0.25; done; " +
+ "python3 -c 'import urllib.request; urllib.request.urlopen(\"http://127.0.0.1:8080/\", timeout=1).read()' >/dev/null && printf '\\036SERVICE-STARTED\\n'\n"
+ if err := recording.writeString(command); err != nil {
+ return err
+ }
+ if err := recording.waitFor(append([]byte{0x1e}, []byte("SERVICE-STARTED")...), 10*time.Second); err != nil {
+ return err
+ }
+ screenshot := filepath.Join(outputDir, "browser.png")
+ url := fmt.Sprintf("%s://%s:%d/", web.Scheme, web.Host, web.Port)
+ if err := runBrowserProof(url, screenshot); err != nil {
+ return err
+ }
+ if err := recording.writeString("printf 'BROWSER-DONE\\n'; exit 0\n"); err != nil {
+ return err
+ }
+ if err := recording.input.Close(); err != nil {
+ return fmt.Errorf("close recorder input: %w", err)
+ }
+ finalization, err := client.readUntilOutputFinalization(30 * time.Second)
+ if err != nil {
+ return err
+ }
+ if finalization.status != "drained" {
+ return fmt.Errorf("workload output finalization = %q: %s", finalization.status, finalization.reason)
+ }
+ if err := recording.command.Wait(); err != nil {
+ return fmt.Errorf("asciinema success recording: %w: %s", err, recording.stderr.String())
+ }
+ raw := filepath.Join(outputDir, "terminal.txt")
+ convert := exec.Command("asciinema", "convert", "--quiet", "--output-format", "raw", recording.cast, raw)
+ if output, err := convert.CombinedOutput(); err != nil {
+ return fmt.Errorf("convert success cast: %w: %s", err, output)
+ }
+ payload, err := os.ReadFile(raw)
+ if err != nil {
+ return fmt.Errorf("read converted cast: %w", err)
+ }
+ for _, marker := range []string{"OPERATION-ONE", "OPERATION-TWO", "CWD=/tmp", "SIZE=31 100", "SERVICE-STARTED", "BROWSER-DONE", "^C"} {
+ if !bytes.Contains(payload, []byte(marker)) {
+ return fmt.Errorf("converted cast is missing %q", marker)
+ }
+ }
+ info, err := os.Stat(recording.cast)
+ if err != nil {
+ return err
+ }
+ if err := writeProof(outputDir, proof{
+ Schema: "reploy-omegaflow-conformance-proof-v1", Scenario: "success",
+ Cast: filepath.Base(recording.cast), CastBytes: info.Size(), BrowserScreenshot: filepath.Base(screenshot),
+ TerminalMarkersVerified: true, OutputFinalizationStatus: finalization.status,
+ }); err != nil {
+ return err
+ }
+ return finishBroker(client, "drained")
+}
+
+func runFailedOutputFinalization(outputDir string, client *broker, recording *recorder) error {
+ command := "cat > /mnt/reploy-home/fail.py <<'PY'\n" +
+ "import os\nimport signal\nimport time\n\n" +
+ "def terminate(*_):\n time.sleep(5)\n raise SystemExit(0)\n\n" +
+ "signal.signal(signal.SIGTERM, terminate)\n" +
+ "os.write(1, b'\\x1eFAILURE-READY\\n')\n" +
+ "signal.pause()\nPY\n" +
+ "exec python3 /mnt/reploy-home/fail.py\n"
+ if err := recording.writeString(command); err != nil {
+ return err
+ }
+ if err := recording.waitFor(append([]byte{0x1e}, []byte("FAILURE-READY")...), 10*time.Second); err != nil {
+ return err
+ }
+ if err := client.write(map[string]any{"schema": controllerStreamSchema, "type": "terminate"}); err != nil {
+ return err
+ }
+ if err := client.readUntil("terminating", 10*time.Second); err != nil {
+ return err
+ }
+ // The host-side test waits for this container-log marker before suspending
+ // the public host process across its absolute output-finalization deadline.
+ // The controller and its private channel remain live throughout the injected
+ // observation fault.
+ fmt.Fprintln(os.Stderr, "OMEGAFLOW-CONFORMANCE-TERMINATING")
+ finalization, err := client.readUntilOutputFinalization(45 * time.Second)
+ if err != nil {
+ return err
+ }
+ if finalization.status != "failed" || finalization.reason == "" {
+ return fmt.Errorf("workload output finalization = %q: %s", finalization.status, finalization.reason)
+ }
+ if err := recording.input.Close(); err != nil {
+ return fmt.Errorf("close failed recorder input: %w", err)
+ }
+ recorderErr := recording.command.Wait()
+ if recorderErr == nil {
+ return fmt.Errorf("asciinema unexpectedly reported successful attachment finalization")
+ }
+ payload, err := os.ReadFile(recording.cast)
+ if err != nil {
+ return fmt.Errorf("read partial cast: %w", err)
+ }
+ if len(payload) == 0 || payload[len(payload)-1] != '\n' || !bytes.Contains(payload, []byte("FAILURE-READY")) {
+ return fmt.Errorf("partial cast was not retained as a closed asciinema stream")
+ }
+ if err := writeProof(outputDir, proof{
+ Schema: "reploy-omegaflow-conformance-proof-v1", Scenario: "failed-output-finalization",
+ Cast: filepath.Base(recording.cast), CastBytes: int64(len(payload)), RecorderFailed: true,
+ OutputFinalizationStatus: finalization.status, OutputFinalizationReason: finalization.reason,
+ }); err != nil {
+ return err
+ }
+ return finishBroker(client, "failed")
+}
+
+func (client *broker) readUntil(want string, timeout time.Duration) error {
+ deadline := time.Now().Add(timeout)
+ for time.Now().Before(deadline) {
+ event, err := client.readAny(time.Until(deadline))
+ if err != nil {
+ return err
+ }
+ if event.Type == want {
+ return nil
+ }
+ if event.Type != "diagnostic" && event.Type != "workload-exit" {
+ return fmt.Errorf("controller event = %q while waiting for %q", event.Type, want)
+ }
+ }
+ return fmt.Errorf("timed out waiting for controller event %q", want)
+}
+
+type outputFinalization struct {
+ status string
+ reason string
+}
+
+func (client *broker) readUntilOutputFinalization(timeout time.Duration) (outputFinalization, error) {
+ deadline := time.Now().Add(timeout)
+ for time.Now().Before(deadline) {
+ event, err := client.readAny(time.Until(deadline))
+ if err != nil {
+ return outputFinalization{}, err
+ }
+ switch event.Type {
+ case "diagnostic", "workload-exit", "terminating":
+ continue
+ case "workload-outputs-finalized":
+ var status string
+ if err := json.Unmarshal(event.Status, &status); err != nil {
+ return outputFinalization{}, fmt.Errorf("decode workload output status: %w", err)
+ }
+ return outputFinalization{status: status, reason: event.Reason}, nil
+ case "client-error":
+ return outputFinalization{}, fmt.Errorf("controlled-session client error %q: %s", event.Code, event.Message)
+ default:
+ return outputFinalization{}, fmt.Errorf("unexpected controller event before output finalization: %s", event.Type)
+ }
+ }
+ return outputFinalization{}, fmt.Errorf("timed out waiting for workload output finalization")
+}
+
+func finishBroker(client *broker, wantOutputStatus string) error {
+ if err := client.write(map[string]any{"schema": controllerStreamSchema, "type": "complete"}); err != nil {
+ return err
+ }
+ terminated, err := client.read("terminated", 30*time.Second)
+ if err != nil {
+ return err
+ }
+ if terminated.Result == nil || terminated.Result.WorkloadOutputFinalizationStatus.Kind != wantOutputStatus || terminated.Result.ControllerFinalizationStatus.Kind != "completed" {
+ return fmt.Errorf("authoritative terminated result is inconsistent: %#v", terminated.Result)
+ }
+ if err := client.write(map[string]any{"schema": controllerStreamSchema, "type": "acknowledge-terminated"}); err != nil {
+ return err
+ }
+ if err := client.input.Close(); err != nil {
+ return fmt.Errorf("close broker input: %w", err)
+ }
+ if err := client.wait(); err != nil {
+ return fmt.Errorf("session broker: %w: %s", err, client.stderr.String())
+ }
+ return nil
+}
+
+func (client *broker) wait() error {
+ client.waited = true
+ return client.command.Wait()
+}
+
+func startBroker() (*broker, error) {
+ command := exec.Command("reploy-session-client", "client")
+ input, err := command.StdinPipe()
+ if err != nil {
+ return nil, err
+ }
+ output, err := command.StdoutPipe()
+ if err != nil {
+ return nil, err
+ }
+ client := &broker{command: command, input: input}
+ client.events = bufio.NewScanner(output)
+ client.events.Buffer(make([]byte, 4096), 1<<20)
+ command.Stderr = &client.stderr
+ if err := command.Start(); err != nil {
+ return nil, fmt.Errorf("start session broker: %w", err)
+ }
+ return client, nil
+}
+
+func (client *broker) read(want string, timeout time.Duration) (event, error) {
+ event, err := client.readAny(timeout)
+ if err != nil {
+ return event, err
+ }
+ if event.Type != want {
+ return event, fmt.Errorf("controller event = %q, want %q", event.Type, want)
+ }
+ return event, nil
+}
+
+func (client *broker) readAny(timeout time.Duration) (event, error) {
+ type result struct {
+ event event
+ err error
+ }
+ done := make(chan result, 1)
+ go func() {
+ if !client.events.Scan() {
+ err := client.events.Err()
+ if err == nil {
+ err = io.EOF
+ }
+ done <- result{err: err}
+ return
+ }
+ value, err := decodeEvent(client.events.Bytes())
+ done <- result{event: value, err: err}
+ }()
+ select {
+ case value := <-done:
+ if value.err != nil {
+ return event{}, fmt.Errorf("read controller event: %w: %s", value.err, client.stderr.String())
+ }
+ return value.event, nil
+ case <-time.After(timeout):
+ return event{}, fmt.Errorf("timed out waiting for controller event")
+ }
+}
+
+func decodeEvent(payload []byte) (event, error) {
+ if err := rejectDuplicateJSONFields(payload); err != nil {
+ return event{}, err
+ }
+ var envelope struct {
+ Schema string `json:"schema"`
+ Type string `json:"type"`
+ }
+ if err := json.Unmarshal(payload, &envelope); err != nil {
+ return event{}, err
+ }
+ if envelope.Schema != controllerStreamSchema {
+ return event{}, fmt.Errorf("controller event schema = %q", envelope.Schema)
+ }
+ value := event{Schema: envelope.Schema, Type: envelope.Type}
+ switch envelope.Type {
+ case "broker-ready":
+ var wire struct {
+ Schema string `json:"schema"`
+ Type string `json:"type"`
+ TerminalSocket string `json:"terminal_socket"`
+ }
+ if err := decodeStrictEvent(payload, &wire); err != nil {
+ return event{}, err
+ }
+ value.TerminalSocket = wire.TerminalSocket
+ case "opened":
+ var wire struct {
+ Schema string `json:"schema"`
+ Type string `json:"type"`
+ Operations []string `json:"operations"`
+ Endpoints []endpoint `json:"endpoints"`
+ Columns uint32 `json:"columns"`
+ Rows uint32 `json:"rows"`
+ OutputFinalizationTimeoutMilliseconds uint32 `json:"output_finalization_timeout_milliseconds"`
+ }
+ if err := decodeStrictEvent(payload, &wire); err != nil {
+ return event{}, err
+ }
+ value.Operations, value.Endpoints = wire.Operations, wire.Endpoints
+ value.Columns, value.Rows = wire.Columns, wire.Rows
+ value.OutputFinalizationTimeoutMilliseconds = wire.OutputFinalizationTimeoutMilliseconds
+ case "ready":
+ if err := decodeStrictEvent(payload, &struct {
+ Schema string `json:"schema"`
+ Type string `json:"type"`
+ }{}); err != nil {
+ return event{}, err
+ }
+ case "workload-exit":
+ var wire struct {
+ Schema string `json:"schema"`
+ Type string `json:"type"`
+ Status json.RawMessage `json:"status"`
+ }
+ if err := decodeStrictEvent(payload, &wire); err != nil {
+ return event{}, err
+ }
+ var status struct {
+ Kind string `json:"kind"`
+ Code *int `json:"code,omitempty"`
+ Reason string `json:"reason,omitempty"`
+ }
+ if err := decodeStrictEvent(wire.Status, &status); err != nil {
+ return event{}, fmt.Errorf("decode workload-exit status: %w", err)
+ }
+ value.Status = wire.Status
+ case "terminating":
+ var wire struct {
+ Schema string `json:"schema"`
+ Type string `json:"type"`
+ Cause string `json:"cause"`
+ }
+ if err := decodeStrictEvent(payload, &wire); err != nil {
+ return event{}, err
+ }
+ value.Cause = wire.Cause
+ case "diagnostic", "client-error":
+ var wire struct {
+ Schema string `json:"schema"`
+ Type string `json:"type"`
+ Code string `json:"code"`
+ Message string `json:"message"`
+ }
+ if err := decodeStrictEvent(payload, &wire); err != nil {
+ return event{}, err
+ }
+ if !validDiagnosticCode(wire.Code) {
+ return event{}, fmt.Errorf("controller event code %q is invalid", wire.Code)
+ }
+ value.Code, value.Message = wire.Code, wire.Message
+ case "workload-outputs-finalized":
+ var wire struct {
+ Schema string `json:"schema"`
+ Type string `json:"type"`
+ Status json.RawMessage `json:"status"`
+ Reason string `json:"reason,omitempty"`
+ }
+ if err := decodeStrictEvent(payload, &wire); err != nil {
+ return event{}, err
+ }
+ value.Status, value.Reason = wire.Status, wire.Reason
+ case "terminated":
+ var wire struct {
+ Schema string `json:"schema"`
+ Type string `json:"type"`
+ Result *lifecycleResult `json:"result"`
+ }
+ if err := decodeStrictEvent(payload, &wire); err != nil {
+ return event{}, err
+ }
+ value.Result = wire.Result
+ default:
+ return event{}, fmt.Errorf("controller event type %q is unsupported", envelope.Type)
+ }
+ return value, nil
+}
+
+func decodeStrictEvent(payload []byte, value any) error {
+ decoder := json.NewDecoder(bytes.NewReader(payload))
+ decoder.DisallowUnknownFields()
+ if err := decoder.Decode(value); err != nil {
+ return err
+ }
+ var trailing any
+ if err := decoder.Decode(&trailing); err != io.EOF {
+ return fmt.Errorf("controller event must contain exactly one JSON object")
+ }
+ return nil
+}
+
+func rejectDuplicateJSONFields(payload []byte) error {
+ decoder := json.NewDecoder(bytes.NewReader(payload))
+ if err := scanJSONValue(decoder); err != nil {
+ return err
+ }
+ if token, err := decoder.Token(); err != io.EOF {
+ if err == nil {
+ return fmt.Errorf("JSON contains trailing token %v", token)
+ }
+ return err
+ }
+ return nil
+}
+
+func scanJSONValue(decoder *json.Decoder) error {
+ token, err := decoder.Token()
+ if err != nil {
+ return err
+ }
+ delimiter, compound := token.(json.Delim)
+ if !compound {
+ return nil
+ }
+ switch delimiter {
+ case '{':
+ seen := map[string]bool{}
+ for decoder.More() {
+ keyToken, err := decoder.Token()
+ if err != nil {
+ return err
+ }
+ key, ok := keyToken.(string)
+ if !ok {
+ return fmt.Errorf("JSON object key is not a string")
+ }
+ if !validJSONFieldName(key) {
+ return fmt.Errorf("JSON object field %q is not lowercase ASCII snake_case", key)
+ }
+ if seen[key] {
+ return fmt.Errorf("JSON object repeats field %q", key)
+ }
+ seen[key] = true
+ if err := scanJSONValue(decoder); err != nil {
+ return err
+ }
+ }
+ closing, err := decoder.Token()
+ if err != nil {
+ return err
+ }
+ if closing != json.Delim('}') {
+ return fmt.Errorf("JSON object is not closed")
+ }
+ case '[':
+ for decoder.More() {
+ if err := scanJSONValue(decoder); err != nil {
+ return err
+ }
+ }
+ closing, err := decoder.Token()
+ if err != nil {
+ return err
+ }
+ if closing != json.Delim(']') {
+ return fmt.Errorf("JSON array is not closed")
+ }
+ default:
+ return fmt.Errorf("unexpected JSON delimiter %q", delimiter)
+ }
+ return nil
+}
+
+func validJSONFieldName(value string) bool {
+ if value == "" {
+ return false
+ }
+ for index := 0; index < len(value); index++ {
+ character := value[index]
+ if character >= 'a' && character <= 'z' {
+ continue
+ }
+ if index != 0 && (character == '_' || character >= '0' && character <= '9') {
+ continue
+ }
+ return false
+ }
+ return true
+}
+
+func validDiagnosticCode(value string) bool {
+ if value == "" || value[0] < 'a' || value[0] > 'z' {
+ return false
+ }
+ for index := 1; index < len(value); index++ {
+ character := value[index]
+ if character >= 'a' && character <= 'z' || character >= '0' && character <= '9' || character == '_' {
+ continue
+ }
+ return false
+ }
+ return true
+}
+
+func (client *broker) write(value any) error {
+ payload, err := json.Marshal(value)
+ if err != nil {
+ return err
+ }
+ if _, err := client.input.Write(append(payload, '\n')); err != nil {
+ return fmt.Errorf("write controller request: %w", err)
+ }
+ return nil
+}
+
+func startRecorder(socket string, cast string) (*recorder, error) {
+ if socket == "" {
+ return nil, fmt.Errorf("broker did not report a terminal socket")
+ }
+ attachment := "reploy-session-client attach --socket " + shellQuote(socket)
+ asciinema := "asciinema record --quiet --window-size 80x24 --return --command " + shellQuote(attachment) + " " + shellQuote(cast)
+ command := exec.Command("script", "--quiet", "--return", "--echo", "never", "--command", asciinema, "/dev/null")
+ command.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+ output, err := command.StdoutPipe()
+ if err != nil {
+ return nil, err
+ }
+ input, err := command.StdinPipe()
+ if err != nil {
+ return nil, err
+ }
+ result := &recorder{command: command, input: input, cast: cast}
+ command.Stderr = &result.stderr
+ if err := command.Start(); err != nil {
+ return nil, fmt.Errorf("start asciinema: %w", err)
+ }
+ go func() { _, _ = io.Copy(&result.output, output) }()
+ return result, nil
+}
+
+func (recording *recorder) writeString(value string) error {
+ if _, err := io.WriteString(recording.input, value); err != nil {
+ return fmt.Errorf("write recorder input: %w", err)
+ }
+ return nil
+}
+
+func (recording *recorder) waitFor(marker []byte, timeout time.Duration) error {
+ deadline := time.Now().Add(timeout)
+ for time.Now().Before(deadline) {
+ if recording.output.contains(marker) {
+ return nil
+ }
+ time.Sleep(25 * time.Millisecond)
+ }
+ return fmt.Errorf("timed out waiting for terminal action marker %q", marker)
+}
+
+func selectEndpoint(endpoints []endpoint, id string) (endpoint, error) {
+ for _, candidate := range endpoints {
+ if candidate.ID == id && candidate.Scheme != "" && candidate.Host != "" && candidate.Port != 0 {
+ return candidate, nil
+ }
+ }
+ return endpoint{}, fmt.Errorf("opened event did not grant endpoint %q", id)
+}
+
+func runBrowserProof(url string, screenshot string) error {
+ command := exec.Command("node", "/opt/omegaflow/browser-proof.js", url, screenshot)
+ command.Env = append(os.Environ(), "NODE_PATH=/opt/omegaflow/node_modules")
+ if output, err := command.CombinedOutput(); err != nil {
+ return fmt.Errorf("Playwright Chromium proof: %w: %s", err, output)
+ }
+ return nil
+}
+
+func writeProof(outputDir string, value proof) error {
+ payload, err := json.MarshalIndent(value, "", " ")
+ if err != nil {
+ return err
+ }
+ payload = append(payload, '\n')
+ if err := os.WriteFile(filepath.Join(outputDir, value.Scenario+"-proof.json"), payload, 0o644); err != nil {
+ return fmt.Errorf("write conformance proof: %w", err)
+ }
+ return nil
+}
+
+func shellQuote(value string) string {
+ return "'" + strings.ReplaceAll(value, "'", "'\\''") + "'"
+}
diff --git a/internal/dockerdeploy/testdata/omegaflow_conformance_controller/main_test.go b/internal/dockerdeploy/testdata/omegaflow_conformance_controller/main_test.go
new file mode 100644
index 00000000..5e7ab7e4
--- /dev/null
+++ b/internal/dockerdeploy/testdata/omegaflow_conformance_controller/main_test.go
@@ -0,0 +1,92 @@
+package main
+
+import (
+ "bufio"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+)
+
+func TestDecodeEventEnforcesPublicReaderContract(t *testing.T) {
+ for _, test := range []struct {
+ name string
+ payload string
+ wantErr string
+ }{
+ {
+ name: "unknown field",
+ payload: `{"schema":"reploy-controlled-session-client-v1","type":"ready","future":true}`,
+ wantErr: "unknown field",
+ },
+ {
+ name: "multiple objects",
+ payload: `{"schema":"reploy-controlled-session-client-v1","type":"ready"}{}`,
+ wantErr: "trailing token",
+ },
+ {
+ name: "unknown nested field",
+ payload: `{"schema":"reploy-controlled-session-client-v1","type":"workload-exit","status":{"kind":"exited","code":0,"future":true}}`,
+ wantErr: "unknown field",
+ },
+ {
+ name: "duplicate field",
+ payload: `{"schema":"reploy-controlled-session-client-v1","type":"ready","type":"ready"}`,
+ wantErr: "repeats field",
+ },
+ {
+ name: "unknown event type",
+ payload: `{"schema":"reploy-controlled-session-client-v1","type":"future-event"}`,
+ wantErr: "unsupported",
+ },
+ {
+ name: "field from another event type",
+ payload: `{"schema":"reploy-controlled-session-client-v1","type":"ready","code":"wrong_shape"}`,
+ wantErr: "unknown field",
+ },
+ {
+ name: "case variant field",
+ payload: `{"schema":"reploy-controlled-session-client-v1","Type":"ready"}`,
+ wantErr: "lowercase ASCII snake_case",
+ },
+ {
+ name: "malformed diagnostic code",
+ payload: `{"schema":"reploy-controlled-session-client-v1","type":"diagnostic","code":"Not Valid","message":"bad code"}`,
+ wantErr: "code \"Not Valid\" is invalid",
+ },
+ } {
+ t.Run(test.name, func(t *testing.T) {
+ if _, err := decodeEvent([]byte(test.payload)); err == nil || !strings.Contains(err.Error(), test.wantErr) {
+ t.Fatalf("decodeEvent() error = %v, want %q", err, test.wantErr)
+ }
+ })
+ }
+
+ value, err := decodeEvent([]byte(`{"schema":"reploy-controlled-session-client-v1","type":"client-error","code":"future_code","message":"future diagnostic"}`))
+ if err != nil || value.Type != "client-error" || value.Code != "future_code" {
+ t.Fatalf("open diagnostic code event = %#v, %v", value, err)
+ }
+}
+
+func TestDecodeEventAcceptsPublicGoldenShapes(t *testing.T) {
+ path := filepath.Join("..", "..", "..", "..", "testdata", "controlled-session", "client-v1-events.jsonl")
+ file, err := os.Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer file.Close()
+ scanner := bufio.NewScanner(file)
+ line := 0
+ for scanner.Scan() {
+ line++
+ if _, err := decodeEvent(scanner.Bytes()); err != nil {
+ t.Fatalf("decode public event line %d: %v", line, err)
+ }
+ }
+ if err := scanner.Err(); err != nil {
+ t.Fatal(err)
+ }
+ if line == 0 {
+ t.Fatal("public event fixture is empty")
+ }
+}
diff --git a/testdata/controlled-session/asciinema-v3-linux-arm64.json b/testdata/controlled-session/asciinema-v3-linux-arm64.json
new file mode 100644
index 00000000..ebab7cd5
--- /dev/null
+++ b/testdata/controlled-session/asciinema-v3-linux-arm64.json
@@ -0,0 +1,7 @@
+{
+ "schema": "reploy-test-fixture-v1",
+ "version": "3.2.1",
+ "asset": "asciinema-aarch64-unknown-linux-gnu",
+ "url": "https://github.com/asciinema/asciinema/releases/download/v3.2.1/asciinema-aarch64-unknown-linux-gnu",
+ "sha256": "b516a6d896844c0ffbc96e0a55afe4cbcc79216abde0fc64fdda4e39bee421ea"
+}
diff --git a/testdata/controlled-session/omegaflow-conformance-v1.json b/testdata/controlled-session/omegaflow-conformance-v1.json
new file mode 100644
index 00000000..47c9f07e
--- /dev/null
+++ b/testdata/controlled-session/omegaflow-conformance-v1.json
@@ -0,0 +1,17 @@
+{
+ "schema": "reploy-omegaflow-conformance-fixture-v1",
+ "playwright_version": "1.55.0",
+ "controller_image": "mcr.microsoft.com/playwright:v1.55.0-noble@sha256:b27e719ecbfef153e13fd24e8341736733bf2658b229677eb21ff57ff5d7fb29",
+ "packages": [
+ {
+ "name": "playwright",
+ "url": "https://registry.npmjs.org/playwright/-/playwright-1.55.0.tgz",
+ "sha256": "f9b9876bde4f0508941775648c673a2a94d34aefe38312a1fa3c9440d686d073"
+ },
+ {
+ "name": "playwright-core",
+ "url": "https://registry.npmjs.org/playwright-core/-/playwright-core-1.55.0.tgz",
+ "sha256": "6b63fdf41725afb924880728d39cf560d770f785a861d18c15611e99dd3172bd"
+ }
+ ]
+}