feat(cloud): relay terminal traffic to workers

cloud/.env.example

@@ -135,3 +135,7 @@ AO_CLOUD_WEB_STATE_DIR=
 # Persistent duplex terminal streams (low-latency typing). Off by default;
 # the polled store-and-forward transport is byte-identical when unset.
 # AO_CLOUD_TERMINAL_STREAM=1
+# Experimental local terminal relay. Requires AO_CLOUD_TERMINAL_STREAM=1.
+# It sends output over the active worker tunnel before asynchronously mirroring
+# it to durable replay storage.
+# AO_CLOUD_TERMINAL_RELAY=1

cloud/cmd/ao-cloud/main.go

@@ -397,13 +397,19 @@ func run(logger *slog.Logger) error {
 		SecretCipher:              providerCipher,
 		WebhookMaxBody:            cfg.GitHub.WebhookMaxBody,
 		TerminalStreamEnabled:     cfg.TerminalStreamEnabled,
+		TerminalRelayEnabled:      cfg.TerminalRelayEnabled,
 	}
 	if cfg.Environment == "development" &&
 		os.Getenv("AO_CLOUD_DEVELOPMENT_SKIP_CREDENTIAL_VALIDATION") == "true" {
 		logger.Warn("coding-agent credential validation is disabled for development")
 		apiOptions.CredentialValidator = developmentCredentialValidator{}
 	}
 	api := httpapi.New(apiOptions)
+	if cfg.TerminalRelayEnabled {
+		logger.Info("experimental terminal relay enabled",
+			"terminal_stream_enabled", cfg.TerminalStreamEnabled,
+			"mode", "local_same_replica")
+	}
 	// The work-wait long-poll and terminal streaming both ride a Postgres NOTIFY
 	// listener. Run it wherever workers connect so WaitForWork can be woken on
 	// enqueue; register the terminal channels only when that feature is on.

cloud/cmd/ao-worker/main.go

@@ -665,12 +665,13 @@ func (c *client) FailTransport(
 func (c *client) PublishTerminalOutput(
 	ctx context.Context,
 	terminalID string,
+	id int64,
 	data []byte,
 ) error {
 	return c.do(
 		ctx,
 		"/worker/terminals/"+url.PathEscape(terminalID)+"/output",
-		worker.TerminalOutputRequest{Data: data},
+		worker.TerminalOutputRequest{ID: id, Data: data},
 		nil,
 	)
 }

cloud/compose.terminal-relay-test.yaml

@@ -0,0 +1,31 @@
+# Browser-side control-plane replica for the local terminal-relay split test.
+# It shares the normal local stack's Postgres network, but intentionally has no
+# Docker socket: workers must be connected to the primary control plane on 8081
+# while the browser connects here on 8082.
+services:
+  relay-browser-replica:
+    image: ao-cloud-local-control-plane
+    environment:
+      AO_CLOUD_ENV: development
+      AO_CLOUD_DEVELOPMENT_SKIP_CREDENTIAL_VALIDATION: "true"
+      AO_CLOUD_RELEASE: local-relay-test
+      AO_CLOUD_HTTP_ADDRESS: :8080
+      AO_CLOUD_DATABASE_URL: postgres://ao_cloud_app:ao_cloud_local_app@postgres:5432/ao_cloud?sslmode=disable
+      AO_CLOUD_MIGRATE_ON_STARTUP: "false"
+      AO_CLOUD_LOCAL_AUTH: "true"
+      AO_CLOUD_LOCAL_SESSION_TTL: 720h
+      # A browser-only replica needs no local worker provider. The ECS provider
+      # intentionally starts no reconciler here, so this container cannot touch
+      # Docker or provision a sandbox during the split-routing test.
+      AO_CLOUD_SANDBOX_PROVIDER: ecs
+      AO_CLOUD_TERMINAL_STREAM: "1"
+      AO_CLOUD_TERMINAL_RELAY: "1"
+    ports:
+      - "127.0.0.1:8082:8080"
+    networks:
+      - relay-test
+
+networks:
+  relay-test:
+    name: "${AO_CLOUD_RELAY_TEST_NETWORK:-ao-cloud-local_default}"
+    external: true

cloud/compose.yaml

@@ -47,6 +47,7 @@ services:
       AO_CLOUD_LOCAL_SESSION_TTL: 720h
       AO_CLOUD_SANDBOX_PROVIDER: docker
       AO_CLOUD_TERMINAL_STREAM: "${AO_CLOUD_TERMINAL_STREAM:-}"
+      AO_CLOUD_TERMINAL_RELAY: "${AO_CLOUD_TERMINAL_RELAY:-}"
       AO_CLOUD_PROVIDER_SECRET_KEY: "${AO_CLOUD_PROVIDER_SECRET_KEY:?AO_CLOUD_PROVIDER_SECRET_KEY is required}"
       AO_CLOUD_PUBLIC_URL: http://control-plane:8080
       AO_CLOUD_WORKER_SIGNING_KEY: "${AO_CLOUD_WORKER_SIGNING_KEY:?AO_CLOUD_WORKER_SIGNING_KEY is required}"

cloud/docs/control-plane.md

@@ -96,9 +96,14 @@ requests are also bounded.
 Terminal tickets are random, hashed at rest, short-lived, single-use, and bound
 to a session epoch. Minting or reconnecting with a ticket is not lifecycle
 activity. Actual terminal input renews the short interaction lease; merely
-retaining a hidden WebSocket does not. The WebSocket itself is a stateless
-bridge: input becomes a durable worker request and output is replayed from
-PostgreSQL by sequence.
+retaining a hidden WebSocket does not. The WebSocket itself is normally a
+stateless bridge: input becomes a durable worker request and output is replayed
+from PostgreSQL by sequence. Local relay experiments
+(`AO_CLOUD_TERMINAL_RELAY=1`, with terminal streaming also enabled) retain the
+same ticket and sequence contract but forward a worker frame to an attached
+browser first, then mirror that exact frame to PostgreSQL in order. Durable
+storage remains the reconnect/replay source and the original queue path is the
+fallback whenever no local relay stream is available.
 Workspace shells are supported. Attaching to the coding agent's native TUI is
 deliberately not implemented, so `kind=agent` is rejected instead of being
 silently mapped to a different process. Because arbitrary shell input cannot

cloud/internal/config/config.go

@@ -80,6 +80,11 @@ type Config struct {
 	// replaces the input/output polling loops. Off means the polled
 	// store-and-forward behavior, byte for byte.
 	TerminalStreamEnabled bool
+	// TerminalRelayEnabled forwards terminal output to an attached browser
+	// directly from the worker stream, before the same frame is mirrored to
+	// durable replay storage. It remains opt-in until the hosted entrypoint is
+	// shard-aware across relay replicas.
+	TerminalRelayEnabled bool
 
 	NodeOpsBaseURL       string
 	NodeOpsAPIKey        string
@@ -182,6 +187,7 @@ func Load() (Config, error) {
 		LocalSessionTTL:        durationEnv("AO_CLOUD_LOCAL_SESSION_TTL", 24*time.Hour),
 		AllowAnonymousCheckout: boolEnv("AO_CLOUD_ALLOW_ANONYMOUS_GITHUB_CHECKOUT", false),
 		TerminalStreamEnabled:  boolEnv("AO_CLOUD_TERMINAL_STREAM", false),
+		TerminalRelayEnabled:   boolEnv("AO_CLOUD_TERMINAL_RELAY", false),
 		SandboxProvider: strings.ToLower(
 			envOrDefault("AO_CLOUD_SANDBOX_PROVIDER", defaultSandboxProvider(hosted)),
 		),
@@ -472,6 +478,9 @@ func Load() (Config, error) {
 	if cfg.IdlePauseInterval <= 0 {
 		return Config{}, errors.New("AO_CLOUD_IDLE_PAUSE_INTERVAL must be positive")
 	}
+	if cfg.TerminalRelayEnabled && !cfg.TerminalStreamEnabled {
+		return Config{}, errors.New("AO_CLOUD_TERMINAL_RELAY requires AO_CLOUD_TERMINAL_STREAM")
+	}
 	if cfg.IdlePauseThreshold < time.Minute {
 		return Config{}, errors.New("AO_CLOUD_IDLE_PAUSE_THRESHOLD must be at least 1m")
 	}

cloud/internal/httpapi/server.go

@@ -93,6 +93,7 @@ type Store interface {
 	QueueTerminalResize(context.Context, domain.TerminalSession, uint16, uint16) error
 	CloseTerminal(context.Context, domain.TerminalSession) error
 	AppendTerminalOutput(context.Context, string, string, string, string, int64, []byte) (int64, error)
+	AppendTerminalOutputAt(context.Context, string, string, string, string, int64, int64, []byte) (int64, error)
 	ClaimTerminalInput(context.Context, string, string, string, int64, string, time.Duration) (domain.WorkerRequest, bool, error)
 	MarkTerminalExited(context.Context, string, string, string, string, int64, int) error
 	EnsureWorkerAgentTerminal(context.Context, string, string, string, int64, time.Duration) (domain.TerminalSession, error)
@@ -161,6 +162,7 @@ type Server struct {
 	credentialValidator     credentialValidator
 	webhookMaxBody          int64
 	terminalStreamEnabled   bool
+	terminalRelayEnabled    bool
 	terminalStreams         *terminalStreams
 	workWaiters             *workWaiters
 	// workerBinariesBySHA serves the content-addressed worker/helper binaries
@@ -195,6 +197,7 @@ type Options struct {
 	CredentialValidator       credentialValidator
 	WebhookMaxBody            int64
 	TerminalStreamEnabled     bool
+	TerminalRelayEnabled      bool
 }
 
 func New(options Options) *Server {
@@ -264,6 +267,7 @@ func New(options Options) *Server {
 		credentialValidator:       options.CredentialValidator,
 		webhookMaxBody:            webhookMaxBody,
 		terminalStreamEnabled:     options.TerminalStreamEnabled,
+		terminalRelayEnabled:      options.TerminalRelayEnabled,
 		terminalStreams:           newTerminalStreams(),
 		workWaiters:               newWorkWaiters(),
 	}

cloud/internal/httpapi/terminal_fastpath_test.go

@@ -70,6 +70,41 @@ func TestPushInputFailsAfterStreamRetired(t *testing.T) {
 	}
 }
 
+func TestRelayOutputDeliversLiveFrame(t *testing.T) {
+	registry := newTerminalStreams()
+	output, unsubscribe := registry.subscribeRelayOutput("term")
+	defer unsubscribe()
+
+	if dropped := registry.relayOutput("term", terminalRelayOutput{
+		sequence: 7,
+		data:     []byte("hello"),
+	}); dropped != 0 {
+		t.Fatalf("dropped=%d, want 0", dropped)
+	}
+	select {
+	case got := <-output:
+		if got.sequence != 7 || string(got.data) != "hello" {
+			t.Fatalf("got sequence=%d data=%q", got.sequence, got.data)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("relay output was not delivered")
+	}
+}
+
+func TestRelayOutputReportsSaturatedClient(t *testing.T) {
+	registry := newTerminalStreams()
+	output, unsubscribe := registry.subscribeRelayOutput("term")
+	defer unsubscribe()
+	for range cap(output) {
+		if dropped := registry.relayOutput("term", terminalRelayOutput{sequence: 1, data: []byte("x")}); dropped != 0 {
+			t.Fatalf("unexpected drop while filling buffer: %d", dropped)
+		}
+	}
+	if dropped := registry.relayOutput("term", terminalRelayOutput{sequence: 2, data: []byte("x")}); dropped != 1 {
+		t.Fatalf("dropped=%d, want 1", dropped)
+	}
+}
+
 func TestTerminalRoutingKeyStableAndBounded(t *testing.T) {
 	a := terminalRoutingKey("session-abc")
 	b := terminalRoutingKey("session-abc")