From 21bd048e91021f4197a76f313d720cdd16edfc65 Mon Sep 17 00:00:00 2001
From: antoinetoussaint-byte
 <312536060+antoinetoussaint-byte@users.noreply.github.com>
Date: Sun, 25 Oct 2026 16:19:32 -0400
Subject: [PATCH 1/3] fix(connect): the generated main serves the Connect face;
 bounded relay; unknown JSON fields

connect-endpoint advertised a port nothing listened on: Sync wrote
src/rpc/connect_gen.py and the runtime and deploy opened CODEFLY_CONNECT_PORT
and 8081, but the scaffold's src/main.py started only the gRPC server.

- The scaffold's main starts the face once its gRPC server listens, through
  connect_gen.start, whenever src/rpc/connect_gen.py is there, and stops it
  before the gRPC server. Sync writes that module when connect-endpoint is on
  and now removes the one it wrote when it is off, so the module is the
  setting's one projection into the process. connect-endpoint is set after
  create, which is why the switch is the sync-written module and not a
  create-time template guard.
- connect_gen.start serves nothing while a face started in this process is
  open, so a service that starts its own with serve (the model gateway's
  hosted lifespan) never gets a second listener.
- The gRPC relay closes a connection that carries no byte either way for
  idle_timeout, as the docstring promised.
- JSON requests ignore unknown fields, as connect-go's codec does.

TestGeneratedMainServesConnect boots the scaffold's own main under uvicorn:
a Connect call answers with connect-endpoint on, and nothing listens on the
port with it off. It fails with "connection refused" on the old main.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 builder.go                              |  36 ++++-
 connect_test.go                         | 170 ++++++++++++++++++++++++
 main.go                                 |   7 +-
 templates/connect/connect_gen.py        |  64 +++++++--
 templates/factory/code/src/main.py.tmpl |  17 ++-
 testdata/connect/test_connect_gen.py    |  84 ++++++++++++
 6 files changed, 359 insertions(+), 19 deletions(-)

diff --git a/builder.go b/builder.go
index 71a490c..be141ff 100644
--- a/builder.go
+++ b/builder.go
@@ -1,6 +1,7 @@
 package main
 
 import (
+	"bytes"
 	"context"
 	"embed"
 	"errors"
@@ -150,10 +151,8 @@ func (s *Builder) Sync(ctx context.Context, _ *builderv0.SyncRequest) (*builderv
 			return s.Base.Builder.SyncError(err)
 		}
 	}
-	if s.FastAPI.Settings.ConnectEndpoint {
-		if err := s.writeConnectFace(); err != nil {
-			return s.Base.Builder.SyncError(err)
-		}
+	if err := s.syncConnectFace(); err != nil {
+		return s.Base.Builder.SyncError(err)
 	}
 	return s.Base.Builder.SyncResponse()
 }
@@ -170,6 +169,35 @@ const connectFacePath = "code/src/rpc/connect_gen.py"
 //go:embed templates/connect/connect_gen.py
 var connectFace []byte
 
+// connectFaceMarker is the line every Connect face the agent wrote carries, and
+// what lets Sync tell its own file from one the service wrote at that path.
+const connectFaceMarker = "This code is generated by the python-fastapi agent (``connect-endpoint: true``)"
+
+// syncConnectFace writes the Connect face when the service sets
+// connect-endpoint and removes the one the agent wrote when it does not. The
+// generated src/main.py starts the face when the module is there, so a face
+// left behind would listen on a port the service no longer declares.
+func (s *Builder) syncConnectFace() error {
+	if s.FastAPI.Settings.ConnectEndpoint {
+		return s.writeConnectFace()
+	}
+	target := s.Local(connectFacePath)
+	content, err := os.ReadFile(target)
+	if errors.Is(err, os.ErrNotExist) {
+		return nil
+	}
+	if err != nil {
+		return s.Wool.Wrapf(err, "cannot read %s", connectFacePath)
+	}
+	if !bytes.Contains(content, []byte(connectFaceMarker)) {
+		return nil
+	}
+	if err := os.Remove(target); err != nil {
+		return s.Wool.Wrapf(err, "cannot remove the Connect face")
+	}
+	return nil
+}
+
 // writeConnectFace writes the Connect face, replacing whatever is there: the
 // file is the agent's, and a hand edit is drift the next sync removes.
 func (s *Builder) writeConnectFace() error {
diff --git a/connect_test.go b/connect_test.go
index 452c957..a032b57 100644
--- a/connect_test.go
+++ b/connect_test.go
@@ -2,10 +2,18 @@ package main
 
 import (
 	"context"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
 	"os"
 	"os/exec"
 	"path/filepath"
+	"strings"
+	"sync"
+	"syscall"
 	"testing"
+	"time"
 
 	agenttesting "github.com/codefly-dev/core/agents/testing"
 	agentv0 "github.com/codefly-dev/core/generated/go/codefly/services/agent/v0"
@@ -130,3 +138,165 @@ func TestBuilderLoadReadsTheAgentsSettings(t *testing.T) {
 	require.True(t, loaded.FastAPI.Settings.ConnectEndpoint)
 	require.Equal(t, filepath.Join(created.Location, "code"), loaded.Service.SourceLocation)
 }
+
+// TestSyncConnectFaceFollowsTheSetting proves Sync removes the face it wrote
+// once connect-endpoint is off (the scaffold's main starts the face whenever
+// the module is there) and leaves a file at that path the agent did not write.
+func TestSyncConnectFaceFollowsTheSetting(t *testing.T) {
+	builder := NewBuilder(NewService())
+	builder.Location = t.TempDir()
+	builder.FastAPI.Settings.GRPCServer.Enabled = true
+	target := filepath.Join(builder.Location, connectFacePath)
+
+	builder.FastAPI.Settings.ConnectEndpoint = true
+	require.NoError(t, builder.syncConnectFace())
+	require.FileExists(t, target)
+
+	builder.FastAPI.Settings.ConnectEndpoint = false
+	require.NoError(t, builder.syncConnectFace())
+	require.NoFileExists(t, target)
+	require.NoError(t, builder.syncConnectFace(), "nothing to remove is not an error")
+
+	require.NoError(t, os.WriteFile(target, []byte("# the service's own\n"), 0o644))
+	require.NoError(t, builder.syncConnectFace())
+	require.FileExists(t, target)
+}
+
+// TestGeneratedMainServesConnect boots a scaffolded gRPC service's own
+// src/main.py under uvicorn, as the runtime does. With connect-endpoint on,
+// Sync's face answers a Connect call on CODEFLY_CONNECT_PORT and stops with
+// the app; with it off, nothing listens there even with the variable set.
+func TestGeneratedMainServesConnect(t *testing.T) {
+	if testing.Short() {
+		t.Skip("boots a scaffolded service with uv")
+	}
+	if _, err := exec.LookPath("uv"); err != nil {
+		t.Skip("uv is not installed: the scaffolded service cannot run")
+	}
+	builder, root := createServiceForTest(t, map[string]*agentv0.Answer{
+		HotReload:      confirmAnswer(false),
+		PublicEndpoint: confirmAnswer(false),
+		GRPCServer:     confirmAnswer(true),
+	})
+	code := filepath.Join(root, "code")
+	// The stubs Sync generates with Buf, generated the way the scaffold's own
+	// tests/rpc/conftest.py does when run directly: same plugins, no network
+	// plugin registry.
+	generated := filepath.Join(code, "src", "rpc", "_generated")
+	require.NoError(t, os.MkdirAll(generated, 0o755))
+	protoc := exec.Command("uv", "run", "python", "-m", "grpc_tools.protoc",
+		"-I../proto", "--python_out=src/rpc/_generated", "--grpc_python_out=src/rpc/_generated", "../proto/api.proto")
+	protoc.Dir = code
+	out, err := protoc.CombinedOutput()
+	require.NoError(t, err, "%s", out)
+
+	builder.FastAPI.Settings.ConnectEndpoint = true
+	require.NoError(t, builder.syncConnectFace())
+	app := bootScaffold(t, code)
+	request, err := http.NewRequest(http.MethodPost,
+		fmt.Sprintf("http://127.0.0.1:%d/api.EchoService/Echo", app.connect),
+		strings.NewReader(`{"message":"over connect","addedLater":1}`))
+	require.NoError(t, err)
+	request.Header.Set("Content-Type", "application/json")
+	response, err := http.DefaultClient.Do(request)
+	require.NoError(t, err)
+	body, err := io.ReadAll(response.Body)
+	_ = response.Body.Close()
+	require.NoError(t, err)
+	require.Equal(t, http.StatusOK, response.StatusCode, "%s", body)
+	require.JSONEq(t, `{"message":"over connect"}`, string(body))
+	app.stop(t)
+	require.False(t, listens(app.connect), "the face outlived the app")
+
+	builder.FastAPI.Settings.ConnectEndpoint = false
+	require.NoError(t, builder.syncConnectFace())
+	app = bootScaffold(t, code)
+	require.True(t, listens(app.grpc), "the gRPC server is up")
+	require.False(t, listens(app.connect), "a service without connect-endpoint opened the Connect port")
+	app.stop(t)
+}
+
+type scaffoldApp struct {
+	cmd                 *exec.Cmd
+	exited              chan error
+	output              *lockedOutput
+	rest, grpc, connect uint16
+}
+
+// bootScaffold runs the scaffold's app under uvicorn and returns once it
+// answers on its REST port: uvicorn listens only after every startup handler,
+// gRPC and Connect included, has run.
+func bootScaffold(t *testing.T, code string) *scaffoldApp {
+	t.Helper()
+	app := &scaffoldApp{output: &lockedOutput{}, exited: make(chan error, 1),
+		rest: freePort(t), grpc: freePort(t), connect: freePort(t)}
+	app.cmd = exec.Command("uv", "run", "uvicorn", "src.main:app", "--host", "127.0.0.1", "--port", fmt.Sprint(app.rest))
+	app.cmd.Dir = code
+	app.cmd.Env = append(os.Environ(),
+		fmt.Sprintf("CODEFLY_GRPC_PORT=%d", app.grpc),
+		fmt.Sprintf("CODEFLY_CONNECT_PORT=%d", app.connect))
+	app.cmd.Stdout, app.cmd.Stderr = app.output, app.output
+	require.NoError(t, app.cmd.Start())
+	go func() { app.exited <- app.cmd.Wait() }()
+	t.Cleanup(func() { _ = app.cmd.Process.Kill() })
+	deadline := time.Now().Add(3 * time.Minute) // a cold uv sync
+	for !answers(app.rest) {
+		select {
+		case err := <-app.exited:
+			t.Fatalf("the app exited before serving: %v\n%s", err, app.output)
+		case <-time.After(200 * time.Millisecond):
+		}
+		require.True(t, time.Now().Before(deadline), "the app never served:\n%s", app.output)
+	}
+	return app
+}
+
+// stop asks uvicorn to shut down, which runs the app's shutdown handlers.
+func (a *scaffoldApp) stop(t *testing.T) {
+	t.Helper()
+	require.NoError(t, a.cmd.Process.Signal(syscall.SIGTERM))
+	select {
+	case <-a.exited:
+	case <-time.After(30 * time.Second):
+		t.Fatalf("the app did not stop:\n%s", a.output)
+	}
+	require.NotContains(t, a.output.String(), "Traceback", a.output.String())
+}
+
+// answers reports whether the scaffold's /version route answers.
+func answers(port uint16) bool {
+	client := http.Client{Timeout: 500 * time.Millisecond}
+	resp, err := client.Get(fmt.Sprintf("http://127.0.0.1:%d/version", port))
+	if err != nil {
+		return false
+	}
+	_ = resp.Body.Close()
+	return resp.StatusCode == http.StatusOK
+}
+
+// lockedOutput is the app's combined output, read while the app writes it.
+type lockedOutput struct {
+	mu  sync.Mutex
+	buf strings.Builder
+}
+
+func (o *lockedOutput) Write(p []byte) (int, error) {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	return o.buf.Write(p)
+}
+
+func (o *lockedOutput) String() string {
+	o.mu.Lock()
+	defer o.mu.Unlock()
+	return o.buf.String()
+}
+
+func listens(port uint16) bool {
+	conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", port), time.Second)
+	if err != nil {
+		return false
+	}
+	_ = conn.Close()
+	return true
+}
diff --git a/main.go b/main.go
index d0640b6..45964ee 100644
--- a/main.go
+++ b/main.go
@@ -86,9 +86,10 @@ type Settings struct {
 	// speaks Connect and gRPC-Web (HTTP/1.1) and gRPC (HTTP/2) on one port:
 	// the face a browser reaches through an L7 hop. Every call is transcoded
 	// onto the grpc.aio server, so it needs grpc-server and adds no API of its
-	// own. The agent writes the face (code/src/rpc/connect_gen.py) at sync and
-	// hands the process its port as CODEFLY_CONNECT_PORT; the app's lifespan
-	// starts it beside the gRPC server. Off by default.
+	// own. The agent writes the face (code/src/rpc/connect_gen.py) at sync,
+	// removes it when the setting is off, and hands the process its port as
+	// CODEFLY_CONNECT_PORT; the scaffold's src/main.py starts it once the gRPC
+	// server listens whenever the module is there. Off by default.
 	ConnectEndpoint bool `yaml:"connect-endpoint"`
 
 	// RuntimeImage overrides the default codefly-built runtime image.
diff --git a/templates/connect/connect_gen.py b/templates/connect/connect_gen.py
index 73f8d19..c7f9e99 100644
--- a/templates/connect/connect_gen.py
+++ b/templates/connect/connect_gen.py
@@ -18,12 +18,19 @@
 end to end, so they are served on the gRPC protocol only.
 
 Request bodies are bounded by ``max_message_bytes`` (gRPC's 4 MiB unless the service
-raises it) before they are read, and every read is bounded in time.
-
-The service starts it from its lifespan, beside its gRPC server, on the port the agent
-hands the process as ``CODEFLY_CONNECT_PORT`` (``DEFAULT_PORT`` where the mapping is
-fixed). It needs ``grpcio``, ``protobuf`` and ``h11`` (uvicorn's own HTTP/1.1 parser) in
-the service's dependencies.
+raises it) before they are read, and every read is bounded in time: an HTTP/1.1 read by
+``read_timeout`` (``idle_timeout`` between requests), and a relayed gRPC connection is
+closed once it has carried no byte either way for ``idle_timeout``. A gRPC stream that
+must stay silent longer keeps its connection with HTTP/2 keepalive pings.
+A JSON request field the service's proto does not know is ignored, as connect-go's codec
+ignores it, so a client built from a newer proto calls a Python service and a Go one alike.
+
+The agent's generated ``src/main.py`` calls ``start`` once its gRPC server listens, on
+the port the agent hands the process as ``CODEFLY_CONNECT_PORT`` (``DEFAULT_PORT`` where
+the mapping is fixed), and stops the face before that server. A service that starts the
+face from its own code with ``serve`` keeps that one: ``start`` serves nothing while a
+face started in this process is open. It needs ``grpcio``, ``protobuf`` and ``h11``
+(uvicorn's own HTTP/1.1 parser) in the service's dependencies.
 """
 
 from __future__ import annotations
@@ -32,6 +39,7 @@
 import base64
 import json
 import logging
+import os
 import re
 import struct
 import zlib
@@ -42,7 +50,7 @@
 import h11
 from google.protobuf import any_pb2, descriptor_pool, json_format, message_factory
 
-__all__ = ["ConnectFace", "serve"]
+__all__ = ["ConnectFace", "serve", "start"]
 
 PORT_ENV = "CODEFLY_CONNECT_PORT"
 DEFAULT_PORT = 8081
@@ -96,6 +104,9 @@
 }
 
 _log = logging.getLogger("codefly.connect")
+# The faces open in this process, so ``start`` never opens a second listener beside one
+# the service's own code opened.
+_faces: set[ConnectFace] = set()
 
 
 class CallError(Exception):
@@ -132,6 +143,7 @@ def __init__(self, server, channel, port: int):
         self._connections: set[asyncio.Task] = set()
 
     async def stop(self, grace: float | None = 5.0) -> None:
+        _faces.discard(self)
         self._server.close()
         for task in list(self._connections):
             task.cancel()
@@ -182,9 +194,22 @@ async def accepted(reader, writer):
     server = await asyncio.start_server(accepted, host, port)
     bound = server.sockets[0].getsockname()[1]
     face = ConnectFace(server, channel, bound)
+    _faces.add(face)
     return face
 
 
+async def start(grpc_port: int, **options) -> ConnectFace | None:
+    """Start the face the agent's generated ``src/main.py`` starts: on
+    ``CODEFLY_CONNECT_PORT`` (``DEFAULT_PORT`` when unset) over the gRPC server on
+    ``grpc_port``, with ``serve``'s ``options``. Return None, starting nothing, while a
+    face started in this process is open: the service started its own."""
+    if _faces:
+        _log.info("connect face: already served on port %d", next(iter(_faces)).port)
+        return None
+    port = int(os.environ.get(PORT_ENV) or DEFAULT_PORT)
+    return await serve(port, grpc_port, **options)
+
+
 class _Connection:
     def __init__(self, channel, grpc_address, limit, read_timeout, idle_timeout):
         self.channel, self.grpc_address, self.limit = channel, grpc_address, limit
@@ -204,13 +229,18 @@ async def serve(self, reader, writer):
             await self._http1(reader, writer, head)
 
     async def _relay(self, reader, writer, head):
-        """gRPC over HTTP/2: the bytes go to the gRPC server as they came."""
+        """gRPC over HTTP/2: the bytes go to the gRPC server as they came, until the
+        connection has carried nothing either way for ``idle_timeout``."""
         upstream_reader, upstream_writer = await asyncio.open_connection(*self.grpc_address)
         upstream_writer.write(head)
+        loop = asyncio.get_running_loop()
+        last = loop.time()
 
         async def pump(source, sink):
+            nonlocal last
             try:
                 while chunk := await source.read(65536):
+                    last = loop.time()
                     sink.write(chunk)
                     await sink.drain()
                 if sink.can_write_eof():
@@ -218,9 +248,20 @@ async def pump(source, sink):
             except (ConnectionError, OSError):
                 pass
 
+        async def idle():
+            # One bound for both directions: a server stream whose client sends nothing
+            # is not idle while the server answers.
+            while (remaining := last + self.idle_timeout - loop.time()) > 0:
+                await asyncio.sleep(remaining)
+
+        pumps = asyncio.gather(pump(reader, upstream_writer), pump(upstream_reader, writer))
+        watchdog = asyncio.ensure_future(idle())
         try:
-            await asyncio.gather(pump(reader, upstream_writer), pump(upstream_reader, writer))
+            await asyncio.wait((pumps, watchdog), return_when=asyncio.FIRST_COMPLETED)
         finally:
+            pumps.cancel()
+            watchdog.cancel()
+            await asyncio.gather(pumps, watchdog, return_exceptions=True)
             upstream_writer.close()
 
     async def _read(self, reader, timeout):
@@ -326,7 +367,10 @@ def request(self, cls, data: bytes) -> bytes:
         if self.name == "proto":
             return data
         try:
-            return json_format.Parse(data or b"{}", cls()).SerializeToString()
+            # Unknown fields are dropped, as connect-go's JSON codec does: a client built
+            # from a newer contract still reaches a service on the older one.
+            message = json_format.Parse(data or b"{}", cls(), ignore_unknown_fields=True)
+            return message.SerializeToString()
         except (json_format.ParseError, UnicodeDecodeError, ValueError) as exc:
             raise CallError(grpc.StatusCode.INVALID_ARGUMENT, f"unmarshal JSON: {exc}") from exc
 
diff --git a/templates/factory/code/src/main.py.tmpl b/templates/factory/code/src/main.py.tmpl
index 8019333..3ee11b1 100644
--- a/templates/factory/code/src/main.py.tmpl
+++ b/templates/factory/code/src/main.py.tmpl
@@ -1,4 +1,5 @@
-import os
+{{ if .GRPCEnabled }}import importlib.util
+{{ end }}import os
 
 from codefly_sdk import codefly
 from fastapi import FastAPI
@@ -59,11 +60,16 @@ for plugin in plugins:
 # The port is provided by codefly at runtime; it falls back to the standard
 # gRPC port in container/Kubernetes where the mapping is fixed.
 _grpc_server = None
+# The Connect face (Connect, gRPC-Web and gRPC on one listener over the gRPC
+# server). codefly writes src/rpc/connect_gen.py at sync when the service sets
+# connect-endpoint and removes it when it does not, so the module being there is
+# the switch; it listens on CODEFLY_CONNECT_PORT.
+_connect_face = None
 
 
 @app.on_event("startup")
 async def _start_grpc():
-    global _grpc_server
+    global _grpc_server, _connect_face
     # Imported inside the handler, not at module load: the generated stubs
     # are produced by `codefly sync` and importing this module must not depend
     # on them (openapi.py imports src.main to emit the schema before sync).
@@ -71,10 +77,17 @@ async def _start_grpc():
 
     port = int(os.environ.get("CODEFLY_GRPC_PORT", "9090"))
     _grpc_server = await _grpc_serve(port)
+    if importlib.util.find_spec("src.rpc.connect_gen") is not None:
+        from src.rpc import connect_gen
+
+        # None when this process already serves a face the service started itself.
+        _connect_face = await connect_gen.start(port)
 
 
 @app.on_event("shutdown")
 async def _stop_grpc():
+    if _connect_face is not None:
+        await _connect_face.stop()
     if _grpc_server is not None:
         await _grpc_server.stop(grace=5)
 {{- end }}
diff --git a/testdata/connect/test_connect_gen.py b/testdata/connect/test_connect_gen.py
index 3459d33..c9268d9 100644
--- a/testdata/connect/test_connect_gen.py
+++ b/testdata/connect/test_connect_gen.py
@@ -284,3 +284,87 @@ async def test_one_connection_serves_several_calls(face):
         assert chunk, seen
         seen += chunk
     writer.close()
+
+
+async def test_connect_json_ignores_unknown_fields(face):
+    # connect-go's JSON codec discards unknown fields: a client built from a newer
+    # contract reaches a Python service exactly as it reaches a Go one.
+    async with httpx.AsyncClient() as client:
+        r = await client.post(
+            _url(face, "Say"),
+            json={"text": "hi", "count": 1, "addedLater": {"x": 1}},
+            headers={"content-type": "application/json", **STAMP},
+        )
+    assert r.status_code == 200, r.text
+    assert r.json() == {"text": "hi|maya", "count": 1}
+
+
+async def test_a_silent_grpc_connection_is_closed_after_the_idle_bound():
+    # An upstream that accepts and never answers: only the face's own bound ends it.
+    upstream = await asyncio.start_server(lambda r, w: None, "127.0.0.1", 0)
+    listener = await connect_gen.serve(
+        0,
+        upstream.sockets[0].getsockname()[1],
+        host="127.0.0.1",
+        read_timeout=0.5,
+        idle_timeout=0.5,
+    )
+    try:
+        reader, writer = await asyncio.open_connection("127.0.0.1", listener.port)
+        writer.write(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n")
+        await writer.drain()
+        assert await asyncio.wait_for(reader.read(), 3) == b""
+        writer.close()
+    finally:
+        await listener.stop()
+        upstream.close()
+
+
+async def test_a_grpc_connection_carrying_traffic_outlives_the_idle_bound():
+    # Bytes in either direction keep the relay open: a server stream whose client sends
+    # nothing is not idle.
+    async def drip(reader, writer):
+        for _ in range(6):
+            writer.write(b".")
+            await writer.drain()
+            await asyncio.sleep(0.2)
+        writer.close()
+
+    upstream = await asyncio.start_server(drip, "127.0.0.1", 0)
+    listener = await connect_gen.serve(
+        0, upstream.sockets[0].getsockname()[1], host="127.0.0.1", idle_timeout=0.5
+    )
+    try:
+        reader, writer = await asyncio.open_connection("127.0.0.1", listener.port)
+        writer.write(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n")
+        await writer.drain()
+        assert await asyncio.wait_for(reader.read(), 3) == b"......"
+        writer.close()
+    finally:
+        await listener.stop()
+        upstream.close()
+
+
+async def test_start_serves_the_agents_port_once(monkeypatch):
+    server = grpc.aio.server()
+    server.add_generic_rpc_handlers((_handler(),))
+    grpc_port = server.add_insecure_port("127.0.0.1:0")
+    await server.start()
+    monkeypatch.setenv(connect_gen.PORT_ENV, "0")
+    try:
+        # The service's own code started a face: the generated start opens no second one.
+        own = await connect_gen.serve(0, grpc_port, host="127.0.0.1")
+        assert await connect_gen.start(grpc_port, host="127.0.0.1") is None
+        await own.stop()
+        started = await connect_gen.start(grpc_port, host="127.0.0.1")
+        assert started is not None
+        async with httpx.AsyncClient() as client:
+            r = await client.post(
+                f"http://127.0.0.1:{started.port}/echo.v1.Echo/Say",
+                json={"text": "s"},
+                headers={"content-type": "application/json", "x-caller": "zoe"},
+            )
+        assert r.json() == {"text": "s|zoe"}
+        await started.stop()
+    finally:
+        await server.stop(None)

From 6c879d1914d3c29ee4c92e69ffa588fbba0f4910 Mon Sep 17 00:00:00 2001
From: antoinetoussaint-byte
 <312536060+antoinetoussaint-byte@users.noreply.github.com>
Date: Sun, 25 Oct 2026 16:41:21 -0400
Subject: [PATCH 2/3] fix(connect): the relay times out only a connection with
 no open call; the variable is the switch

Re-review r20 of #109:

- B1: the relay's idle bound cut calls in flight (a unary handler that
  computes longer than idle_timeout sends nothing, and the face answered
  UNAVAILABLE, retryable, after a model call may have spent tokens). The relay
  now reads the HTTP/2 frame headers both ways: the client's HEADERS opens a
  call, the server's END_STREAM or either side's RST_STREAM ends it, and the
  idle clock ends a connection only while no call is open, as Go's
  http2.Server IdleTimeout does. The docstring no longer claims keepalive
  pings that nothing configures.
  test_a_grpc_call_in_flight_outlives_the_idle_bound (a 2 s call under a 0.5 s
  bound) fails on the previous relay; test_a_grpc_connection_with_no_open_call_is_closed.
- m1: CODEFLY_CONNECT_PORT is the one switch. The deployment renders it when
  connect-endpoint is on (the runtime already set it only then), start()
  serves only when it is set and logs why it does not, and the scaffold's main
  imports the face only when it is set, so a face left on disk binds nothing.
  On but never synced, the main refuses to start and says to sync.
- m2: the process-wide face registry and the double-start guard are gone: a
  service that starts its own face has its own main.
- m3: with connect-endpoint on, Start requires the Connect face to answer once
  the HTTP endpoint does, and otherwise fails naming the setting and
  connect_gen.start, so a main scaffolded before the generated start is a
  refused Start, not a STARTED behind a refused port.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 builder.go                                    |  12 +-
 connect_test.go                               |  63 ++++++++---
 main.go                                       |   6 +-
 runtime.go                                    |  54 ++++++++-
 serve_readiness_test.go                       |  38 +++++++
 templates/connect/connect_gen.py              | 106 +++++++++++++-----
 .../kustomize/base/deployment.yaml.tmpl       |   6 +
 templates/factory/code/src/main.py.tmpl       |  23 ++--
 testdata/connect/test_connect_gen.py          |  73 +++++++++---
 9 files changed, 302 insertions(+), 79 deletions(-)

diff --git a/builder.go b/builder.go
index be141ff..1f70e77 100644
--- a/builder.go
+++ b/builder.go
@@ -174,9 +174,9 @@ var connectFace []byte
 const connectFaceMarker = "This code is generated by the python-fastapi agent (``connect-endpoint: true``)"
 
 // syncConnectFace writes the Connect face when the service sets
-// connect-endpoint and removes the one the agent wrote when it does not. The
-// generated src/main.py starts the face when the module is there, so a face
-// left behind would listen on a port the service no longer declares.
+// connect-endpoint and removes the one the agent wrote when it does not: the
+// file is the setting's projection, and a face left behind is drift (the
+// process serves it only with CODEFLY_CONNECT_PORT, which follows the setting).
 func (s *Builder) syncConnectFace() error {
 	if s.FastAPI.Settings.ConnectEndpoint {
 		return s.writeConnectFace()
@@ -501,9 +501,9 @@ type Parameters struct {
 	// binds (the app defaults to it when CODEFLY_GRPC_PORT is unset).
 	GRPCEnabled bool
 	GRPCPort    int
-	// ConnectEnabled adds the Connect face's containerPort and Service port.
-	// ConnectPort is the fixed in-cluster port it binds (the app defaults to it
-	// when CODEFLY_CONNECT_PORT is unset).
+	// ConnectEnabled adds the Connect face's containerPort, Service port and
+	// CODEFLY_CONNECT_PORT, the variable the app serves the face on.
+	// ConnectPort is the fixed in-cluster port it binds.
 	ConnectEnabled bool
 	ConnectPort    int
 	// AutomountServiceAccountToken projects the ServiceAccount token into the
diff --git a/connect_test.go b/connect_test.go
index a032b57..3eb8e0c 100644
--- a/connect_test.go
+++ b/connect_test.go
@@ -78,11 +78,15 @@ func TestDeploymentRendersConnectPort(t *testing.T) {
 	deployment, err := os.ReadFile(filepath.Join(enabled, "base", "deployment.yaml"))
 	require.NoError(t, err)
 	require.Contains(t, string(deployment), "containerPort: 8081")
+	require.Contains(t, string(deployment), "- name: CODEFLY_CONNECT_PORT\n              value: \"8081\"")
 
 	disabled := agenttesting.AssertKustomizeTemplates(t, deploymentFS, deploymentTestParameters(t, Parameters{GRPCEnabled: true, GRPCPort: 9090}))
 	svc, err = os.ReadFile(filepath.Join(disabled, "base", "service.yaml"))
 	require.NoError(t, err)
 	require.NotContains(t, string(svc), "connect-port")
+	deployment, err = os.ReadFile(filepath.Join(disabled, "base", "deployment.yaml"))
+	require.NoError(t, err)
+	require.NotContains(t, string(deployment), "CODEFLY_CONNECT_PORT")
 }
 
 // TestConnectFaceServesEveryProtocol runs the face's own suite
@@ -163,9 +167,11 @@ func TestSyncConnectFaceFollowsTheSetting(t *testing.T) {
 }
 
 // TestGeneratedMainServesConnect boots a scaffolded gRPC service's own
-// src/main.py under uvicorn, as the runtime does. With connect-endpoint on,
-// Sync's face answers a Connect call on CODEFLY_CONNECT_PORT and stops with
-// the app; with it off, nothing listens there even with the variable set.
+// src/main.py under uvicorn, as the runtime does. With connect-endpoint on
+// (CODEFLY_CONNECT_PORT set, the face synced), the face answers a Connect call
+// and stops with the app. With it off (the variable unset), a face left on
+// disk binds nothing, not even its 8081 default. On but never synced, the app
+// refuses to start and says to sync.
 func TestGeneratedMainServesConnect(t *testing.T) {
 	if testing.Short() {
 		t.Skip("boots a scaffolded service with uv")
@@ -192,7 +198,7 @@ func TestGeneratedMainServesConnect(t *testing.T) {
 
 	builder.FastAPI.Settings.ConnectEndpoint = true
 	require.NoError(t, builder.syncConnectFace())
-	app := bootScaffold(t, code)
+	app := bootScaffold(t, code, true)
 	request, err := http.NewRequest(http.MethodPost,
 		fmt.Sprintf("http://127.0.0.1:%d/api.EchoService/Echo", app.connect),
 		strings.NewReader(`{"message":"over connect","addedLater":1}`))
@@ -208,12 +214,24 @@ func TestGeneratedMainServesConnect(t *testing.T) {
 	app.stop(t)
 	require.False(t, listens(app.connect), "the face outlived the app")
 
-	builder.FastAPI.Settings.ConnectEndpoint = false
-	require.NoError(t, builder.syncConnectFace())
-	app = bootScaffold(t, code)
+	// Off, with the face still on disk (no sync since): the variable is the switch.
+	defaultFree := !listens(8081)
+	app = bootScaffold(t, code, false)
 	require.True(t, listens(app.grpc), "the gRPC server is up")
-	require.False(t, listens(app.connect), "a service without connect-endpoint opened the Connect port")
+	if defaultFree {
+		require.False(t, listens(8081), "a service without connect-endpoint bound the face's default port")
+	}
 	app.stop(t)
+
+	// On, never synced: the app says what is missing instead of serving no face.
+	require.NoError(t, os.Remove(filepath.Join(root, connectFacePath)))
+	refused := startScaffold(t, code, true)
+	select {
+	case <-refused.exited:
+	case <-time.After(3 * time.Minute):
+		t.Fatalf("the app served without its Connect face:\n%s", refused.output)
+	}
+	require.Contains(t, refused.output.String(), "src/rpc/connect_gen.py is missing: run `codefly sync service`")
 }
 
 type scaffoldApp struct {
@@ -223,22 +241,37 @@ type scaffoldApp struct {
 	rest, grpc, connect uint16
 }
 
-// bootScaffold runs the scaffold's app under uvicorn and returns once it
-// answers on its REST port: uvicorn listens only after every startup handler,
-// gRPC and Connect included, has run.
-func bootScaffold(t *testing.T, code string) *scaffoldApp {
+// startScaffold launches the scaffold's app under uvicorn with the variables
+// the runtime hands it: CODEFLY_CONNECT_PORT only when connect is on.
+func startScaffold(t *testing.T, code string, connect bool) *scaffoldApp {
 	t.Helper()
 	app := &scaffoldApp{output: &lockedOutput{}, exited: make(chan error, 1),
 		rest: freePort(t), grpc: freePort(t), connect: freePort(t)}
 	app.cmd = exec.Command("uv", "run", "uvicorn", "src.main:app", "--host", "127.0.0.1", "--port", fmt.Sprint(app.rest))
 	app.cmd.Dir = code
-	app.cmd.Env = append(os.Environ(),
-		fmt.Sprintf("CODEFLY_GRPC_PORT=%d", app.grpc),
-		fmt.Sprintf("CODEFLY_CONNECT_PORT=%d", app.connect))
+	environment := []string{fmt.Sprintf("CODEFLY_GRPC_PORT=%d", app.grpc)}
+	for _, variable := range os.Environ() {
+		if !strings.HasPrefix(variable, "CODEFLY_CONNECT_PORT=") {
+			environment = append(environment, variable)
+		}
+	}
+	if connect {
+		environment = append(environment, fmt.Sprintf("CODEFLY_CONNECT_PORT=%d", app.connect))
+	}
+	app.cmd.Env = environment
 	app.cmd.Stdout, app.cmd.Stderr = app.output, app.output
 	require.NoError(t, app.cmd.Start())
 	go func() { app.exited <- app.cmd.Wait() }()
 	t.Cleanup(func() { _ = app.cmd.Process.Kill() })
+	return app
+}
+
+// bootScaffold starts the scaffold's app and returns once it answers on its
+// REST port: uvicorn listens only after every startup handler, gRPC and
+// Connect included, has run.
+func bootScaffold(t *testing.T, code string, connect bool) *scaffoldApp {
+	t.Helper()
+	app := startScaffold(t, code, connect)
 	deadline := time.Now().Add(3 * time.Minute) // a cold uv sync
 	for !answers(app.rest) {
 		select {
diff --git a/main.go b/main.go
index 45964ee..0464638 100644
--- a/main.go
+++ b/main.go
@@ -88,8 +88,10 @@ type Settings struct {
 	// onto the grpc.aio server, so it needs grpc-server and adds no API of its
 	// own. The agent writes the face (code/src/rpc/connect_gen.py) at sync,
 	// removes it when the setting is off, and hands the process its port as
-	// CODEFLY_CONNECT_PORT; the scaffold's src/main.py starts it once the gRPC
-	// server listens whenever the module is there. Off by default.
+	// CODEFLY_CONNECT_PORT, locally and in the deployment, exactly when it is
+	// on: the scaffold's src/main.py starts the face on that variable once the
+	// gRPC server listens. Start refuses a run where nothing then listens on the
+	// connect port. Off by default.
 	ConnectEndpoint bool `yaml:"connect-endpoint"`
 
 	// RuntimeImage overrides the default codefly-built runtime image.
diff --git a/runtime.go b/runtime.go
index 6709780..4922682 100644
--- a/runtime.go
+++ b/runtime.go
@@ -153,6 +153,10 @@ type Runtime struct {
 	// connectPort is the mapped port for the Connect face, passed to the
 	// Python process as CODEFLY_CONNECT_PORT. Zero when Connect is disabled.
 	connectPort uint16
+	// connectAddress is the address Init resolved for the Connect face, which
+	// Start requires to answer once the HTTP endpoint does. Empty when Connect
+	// is disabled.
+	connectAddress string
 
 	cacheLocation string
 
@@ -506,6 +510,7 @@ func (s *Runtime) Init(ctx context.Context, req *runtimev0.InitRequest) (*runtim
 			return s.Base.Runtime.InitError(connectErr)
 		}
 		s.connectPort = uint16(connectNet.Port)
+		s.connectAddress = connectNet.Address
 		s.Infof("connect will run on %s", connectNet.Address)
 	}
 
@@ -741,11 +746,7 @@ func (s *Runtime) awaitServing(ctx context.Context, handle *runnerHandle) error
 	if s.serveAddress == "" {
 		return nil
 	}
-	target := s.serveAddress
-	if !strings.Contains(target, "://") {
-		target = "http://" + target
-	}
-	target = strings.TrimRight(target, "/") + "/"
+	target := httpTarget(s.serveAddress)
 	timeout := s.serveTimeout
 	if timeout <= 0 {
 		timeout = defaultServeTimeout
@@ -762,7 +763,7 @@ func (s *Runtime) awaitServing(ctx context.Context, handle *runnerHandle) error
 		err := answers(ctx, target)
 		if err == nil {
 			s.Infof("fastapi app serving on %s after %s", s.serveAddress, time.Since(started).Round(100*time.Millisecond))
-			return nil
+			return s.awaitConnect(ctx, answers)
 		}
 		last = err
 		if !s.runnerAlive(ctx, handle) {
@@ -778,6 +779,47 @@ func (s *Runtime) awaitServing(ctx context.Context, handle *runnerHandle) error
 	}
 }
 
+// connectGrace is how long the Connect face may lag the HTTP endpoint. The
+// scaffold's main starts it in the startup uvicorn finishes before it listens,
+// so it normally answers at once; the grace covers an app that starts it from
+// its own lifespan a moment later.
+const connectGrace = 5 * time.Second
+
+// awaitConnect requires the Connect face to answer once the HTTP endpoint
+// does. connect-endpoint advertises the port to every consumer, so a process
+// that never serves it (a main.py scaffolded before the generated main started
+// the face) is refused by name rather than left STARTED with a port that
+// refuses connections. The face answers any HTTP request, a GET with 405.
+func (s *Runtime) awaitConnect(ctx context.Context, answers func(context.Context, string) error) error {
+	if s.connectAddress == "" {
+		return nil
+	}
+	target := httpTarget(s.connectAddress)
+	deadline := time.Now().Add(connectGrace)
+	for {
+		err := answers(ctx, target)
+		if err == nil {
+			return nil
+		}
+		if time.Now().After(deadline) || ctx.Err() != nil {
+			return fmt.Errorf("%s is on but nothing serves the Connect face at %s: the app must start it once its gRPC server listens, as the scaffold's src/main.py does (connect_gen.start(grpc_port) from src/rpc/connect_gen.py): %v", ConnectEndpointSetting, s.connectAddress, err)
+		}
+		select {
+		case <-ctx.Done():
+		case <-time.After(servePollInterval):
+		}
+	}
+}
+
+// httpTarget is address as the URL of its root, whether or not it carries a
+// scheme.
+func httpTarget(address string) string {
+	if !strings.Contains(address, "://") {
+		address = "http://" + address
+	}
+	return strings.TrimRight(address, "/") + "/"
+}
+
 // serveClient is what httpAnswers probes with: one bounded attempt, and a
 // redirect counts as an answer since the endpoint is serving.
 var serveClient = &http.Client{
diff --git a/serve_readiness_test.go b/serve_readiness_test.go
index adc3af5..98ca41e 100644
--- a/serve_readiness_test.go
+++ b/serve_readiness_test.go
@@ -125,3 +125,41 @@ func TestStartFailsWhenTheEndpointNeverAnswers(t *testing.T) {
 	require.Equal(t, runtimev0.StartStatus_STARTED, resp.GetStatus().GetState(), resp.GetStatus().GetMessage())
 	require.Len(t, env.started(), 2)
 }
+
+// TestStartRefusesAConnectFaceNobodyServes: with connect-endpoint on, an app
+// whose HTTP endpoint answers but whose main never starts the Connect face (a
+// src/main.py scaffolded before the generated main started it) is a failed
+// Start naming the setting and the fix, not a STARTED behind a refused port.
+func TestStartRefusesAConnectFaceNobodyServes(t *testing.T) {
+	server := httptest.NewServer(http.NotFoundHandler())
+	t.Cleanup(server.Close)
+	runtime, env := startTestRuntime(t)
+	runtime.serveAddress = server.URL
+	runtime.connectAddress = closedAddress(t)
+
+	resp, err := runtime.Start(context.Background(), &runtimev0.StartRequest{})
+	require.NoError(t, err)
+	require.Equal(t, runtimev0.StartStatus_ERROR, resp.GetStatus().GetState())
+	require.Contains(t, resp.GetStatus().GetMessage(), "connect-endpoint is on but nothing serves the Connect face")
+	require.Contains(t, resp.GetStatus().GetMessage(), "connect_gen.start")
+	require.True(t, env.started()[0].isStopped(), "the app that does not serve its face must be stopped")
+	require.Nil(t, currentRunner(runtime))
+}
+
+// TestStartWaitsForTheConnectFace: a face that answers, whatever the status
+// (it answers a GET with 405), is a STARTED service.
+func TestStartWaitsForTheConnectFace(t *testing.T) {
+	server := httptest.NewServer(http.NotFoundHandler())
+	t.Cleanup(server.Close)
+	face := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.WriteHeader(http.StatusMethodNotAllowed)
+	}))
+	t.Cleanup(face.Close)
+	runtime, _ := startTestRuntime(t)
+	runtime.serveAddress = server.URL
+	runtime.connectAddress = strings.TrimPrefix(face.URL, "http://")
+
+	resp, err := runtime.Start(context.Background(), &runtimev0.StartRequest{})
+	require.NoError(t, err)
+	require.Equal(t, runtimev0.StartStatus_STARTED, resp.GetStatus().GetState(), resp.GetStatus().GetMessage())
+}
diff --git a/templates/connect/connect_gen.py b/templates/connect/connect_gen.py
index c7f9e99..e2d3271 100644
--- a/templates/connect/connect_gen.py
+++ b/templates/connect/connect_gen.py
@@ -20,17 +20,18 @@
 Request bodies are bounded by ``max_message_bytes`` (gRPC's 4 MiB unless the service
 raises it) before they are read, and every read is bounded in time: an HTTP/1.1 read by
 ``read_timeout`` (``idle_timeout`` between requests), and a relayed gRPC connection is
-closed once it has carried no byte either way for ``idle_timeout``. A gRPC stream that
-must stay silent longer keeps its connection with HTTP/2 keepalive pings.
+closed once it has had no open call for ``idle_timeout``, as Go's HTTP/2 server bounds an
+idle connection. A call in flight is never cut by the face, however long its handler
+takes to answer: its deadline and the gRPC server's own policy bound it.
 A JSON request field the service's proto does not know is ignored, as connect-go's codec
 ignores it, so a client built from a newer proto calls a Python service and a Go one alike.
 
-The agent's generated ``src/main.py`` calls ``start`` once its gRPC server listens, on
-the port the agent hands the process as ``CODEFLY_CONNECT_PORT`` (``DEFAULT_PORT`` where
-the mapping is fixed), and stops the face before that server. A service that starts the
-face from its own code with ``serve`` keeps that one: ``start`` serves nothing while a
-face started in this process is open. It needs ``grpcio``, ``protobuf`` and ``h11``
-(uvicorn's own HTTP/1.1 parser) in the service's dependencies.
+The agent hands the process ``CODEFLY_CONNECT_PORT`` exactly when the service sets
+``connect-endpoint``, locally and in its deployment. The agent's generated
+``src/main.py`` then calls ``start`` once its gRPC server listens and stops the face
+before that server; a service with its own main calls ``serve`` itself. It needs
+``grpcio``, ``protobuf`` and ``h11`` (uvicorn's own HTTP/1.1 parser) in the service's
+dependencies.
 """
 
 from __future__ import annotations
@@ -59,6 +60,10 @@
 DEFAULT_IDLE_TIMEOUT_S = 120.0
 
 _PREFACE = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"
+# An HTTP/2 frame header: 24-bit length, type, flags, stream id (RFC 9113 section 4.1).
+_FRAME = struct.Struct(">BHBBI")
+_DATA, _HEADERS, _RST_STREAM = 0x0, 0x1, 0x3
+_END_STREAM_FLAG = 0x1
 _ENVELOPE = struct.Struct(">BI")
 _COMPRESSED, _END_STREAM, _TRAILERS = 0x01, 0x02, 0x80
 _STATUS_DETAILS = "grpc-status-details-bin"
@@ -104,9 +109,6 @@
 }
 
 _log = logging.getLogger("codefly.connect")
-# The faces open in this process, so ``start`` never opens a second listener beside one
-# the service's own code opened.
-_faces: set[ConnectFace] = set()
 
 
 class CallError(Exception):
@@ -143,7 +145,6 @@ def __init__(self, server, channel, port: int):
         self._connections: set[asyncio.Task] = set()
 
     async def stop(self, grace: float | None = 5.0) -> None:
-        _faces.discard(self)
         self._server.close()
         for task in list(self._connections):
             task.cancel()
@@ -194,20 +195,48 @@ async def accepted(reader, writer):
     server = await asyncio.start_server(accepted, host, port)
     bound = server.sockets[0].getsockname()[1]
     face = ConnectFace(server, channel, bound)
-    _faces.add(face)
     return face
 
 
 async def start(grpc_port: int, **options) -> ConnectFace | None:
     """Start the face the agent's generated ``src/main.py`` starts: on
-    ``CODEFLY_CONNECT_PORT`` (``DEFAULT_PORT`` when unset) over the gRPC server on
-    ``grpc_port``, with ``serve``'s ``options``. Return None, starting nothing, while a
-    face started in this process is open: the service started its own."""
-    if _faces:
-        _log.info("connect face: already served on port %d", next(iter(_faces)).port)
+    ``CODEFLY_CONNECT_PORT`` over the gRPC server on ``grpc_port``, with ``serve``'s
+    ``options``. Without that variable the service does not set ``connect-endpoint``:
+    return None, starting nothing, and say why."""
+    port = os.environ.get(PORT_ENV)
+    if not port:
+        _log.warning(
+            "connect face: %s is unset, so connect-endpoint is off: not serving", PORT_ENV
+        )
         return None
-    port = int(os.environ.get(PORT_ENV) or DEFAULT_PORT)
-    return await serve(port, grpc_port, **options)
+    return await serve(int(port), grpc_port, **options)
+
+
+class _Frames:
+    """The HTTP/2 frame headers in one direction of a relayed connection, after the
+    preface: ``on_frame(type, flags, stream)`` for each, payloads skipped."""
+
+    def __init__(self, on_frame):
+        self._on_frame = on_frame
+        self._header = bytearray()
+        self._skip = 0
+
+    def feed(self, data: bytes) -> None:
+        at = 0
+        while at < len(data):
+            if self._skip:
+                step = min(self._skip, len(data) - at)
+                self._skip -= step
+                at += step
+                continue
+            step = min(_FRAME.size - len(self._header), len(data) - at)
+            self._header += data[at : at + step]
+            at += step
+            if len(self._header) == _FRAME.size:
+                high, low, kind, flags, stream = _FRAME.unpack(self._header)
+                self._header.clear()
+                self._skip = high << 16 | low
+                self._on_frame(kind, flags, stream & 0x7FFFFFFF)
 
 
 class _Connection:
@@ -230,17 +259,31 @@ async def serve(self, reader, writer):
 
     async def _relay(self, reader, writer, head):
         """gRPC over HTTP/2: the bytes go to the gRPC server as they came, until the
-        connection has carried nothing either way for ``idle_timeout``."""
+        connection has had no open call for ``idle_timeout``. The frame headers say which
+        calls are open: the client's HEADERS opens one, and the server's END_STREAM or
+        either side's RST_STREAM ends it."""
         upstream_reader, upstream_writer = await asyncio.open_connection(*self.grpc_address)
         upstream_writer.write(head)
         loop = asyncio.get_running_loop()
         last = loop.time()
+        calls: set[int] = set()
+
+        def from_client(kind, flags, stream):
+            if kind == _HEADERS and stream:
+                calls.add(stream)
+            elif kind == _RST_STREAM:
+                calls.discard(stream)
+
+        def from_server(kind, flags, stream):
+            if kind == _RST_STREAM or (kind in (_DATA, _HEADERS) and flags & _END_STREAM_FLAG):
+                calls.discard(stream)
 
-        async def pump(source, sink):
+        async def pump(source, sink, frames):
             nonlocal last
             try:
                 while chunk := await source.read(65536):
                     last = loop.time()
+                    frames.feed(chunk)
                     sink.write(chunk)
                     await sink.drain()
                 if sink.can_write_eof():
@@ -249,12 +292,19 @@ async def pump(source, sink):
                 pass
 
         async def idle():
-            # One bound for both directions: a server stream whose client sends nothing
-            # is not idle while the server answers.
-            while (remaining := last + self.idle_timeout - loop.time()) > 0:
-                await asyncio.sleep(remaining)
-
-        pumps = asyncio.gather(pump(reader, upstream_writer), pump(upstream_reader, writer))
+            # The clock runs from the last byte either way, and only ends the connection
+            # while no call is open: a handler computing an answer sends nothing.
+            while True:
+                remaining = last + self.idle_timeout - loop.time()
+                if remaining <= 0 and not calls:
+                    return
+                await asyncio.sleep(remaining if remaining > 0 else self.idle_timeout)
+
+        client, server = _Frames(from_client), _Frames(from_server)
+
+        pumps = asyncio.gather(
+            pump(reader, upstream_writer, client), pump(upstream_reader, writer, server)
+        )
         watchdog = asyncio.ensure_future(idle())
         try:
             await asyncio.wait((pumps, watchdog), return_when=asyncio.FIRST_COMPLETED)
diff --git a/templates/deployment/kustomize/base/deployment.yaml.tmpl b/templates/deployment/kustomize/base/deployment.yaml.tmpl
index faa9d0f..57fca03 100644
--- a/templates/deployment/kustomize/base/deployment.yaml.tmpl
+++ b/templates/deployment/kustomize/base/deployment.yaml.tmpl
@@ -74,6 +74,12 @@ spec:
             # writable scratch mount so libraries that use ~/.cache still work.
             - name: HOME
               value: /tmp
+{{- if .Deployment.Parameters.ConnectEnabled }}
+            # connect-endpoint is on: the app serves the Connect face on this
+            # port, and only when the variable is set.
+            - name: CODEFLY_CONNECT_PORT
+              value: "{{ .Deployment.Parameters.ConnectPort }}"
+{{- end }}
 {{- if .Restricted }}
 {{- range $environmentVariable, $reference := .SecretReferences }}
             - name: {{ $environmentVariable }}
diff --git a/templates/factory/code/src/main.py.tmpl b/templates/factory/code/src/main.py.tmpl
index 3ee11b1..ce9f825 100644
--- a/templates/factory/code/src/main.py.tmpl
+++ b/templates/factory/code/src/main.py.tmpl
@@ -1,5 +1,4 @@
-{{ if .GRPCEnabled }}import importlib.util
-{{ end }}import os
+import os
 
 from codefly_sdk import codefly
 from fastapi import FastAPI
@@ -61,9 +60,9 @@ for plugin in plugins:
 # gRPC port in container/Kubernetes where the mapping is fixed.
 _grpc_server = None
 # The Connect face (Connect, gRPC-Web and gRPC on one listener over the gRPC
-# server). codefly writes src/rpc/connect_gen.py at sync when the service sets
-# connect-endpoint and removes it when it does not, so the module being there is
-# the switch; it listens on CODEFLY_CONNECT_PORT.
+# server): codefly hands the process CODEFLY_CONNECT_PORT exactly when the
+# service sets connect-endpoint, and `codefly sync` writes the face itself,
+# src/rpc/connect_gen.py.
 _connect_face = None
 
 
@@ -77,10 +76,16 @@ async def _start_grpc():
 
     port = int(os.environ.get("CODEFLY_GRPC_PORT", "9090"))
     _grpc_server = await _grpc_serve(port)
-    if importlib.util.find_spec("src.rpc.connect_gen") is not None:
-        from src.rpc import connect_gen
-
-        # None when this process already serves a face the service started itself.
+    if os.environ.get("CODEFLY_CONNECT_PORT"):
+        try:
+            import src.rpc.connect_gen as connect_gen
+        except ModuleNotFoundError as missing:
+            if missing.name != "src.rpc.connect_gen":
+                raise
+            raise RuntimeError(
+                "connect-endpoint is on but src/rpc/connect_gen.py is missing: "
+                "run `codefly sync service`"
+            ) from missing
         _connect_face = await connect_gen.start(port)
 
 
diff --git a/testdata/connect/test_connect_gen.py b/testdata/connect/test_connect_gen.py
index c9268d9..1f4de1c 100644
--- a/testdata/connect/test_connect_gen.py
+++ b/testdata/connect/test_connect_gen.py
@@ -345,26 +345,73 @@ async def drip(reader, writer):
         upstream.close()
 
 
-async def test_start_serves_the_agents_port_once(monkeypatch):
+async def test_start_serves_only_on_the_agents_port(monkeypatch):
     server = grpc.aio.server()
     server.add_generic_rpc_handlers((_handler(),))
     grpc_port = server.add_insecure_port("127.0.0.1:0")
     await server.start()
-    monkeypatch.setenv(connect_gen.PORT_ENV, "0")
     try:
-        # The service's own code started a face: the generated start opens no second one.
-        own = await connect_gen.serve(0, grpc_port, host="127.0.0.1")
+        # No CODEFLY_CONNECT_PORT: the service does not set connect-endpoint, whatever
+        # file is on disk, so nothing binds.
+        monkeypatch.delenv(connect_gen.PORT_ENV, raising=False)
         assert await connect_gen.start(grpc_port, host="127.0.0.1") is None
-        await own.stop()
+        monkeypatch.setenv(connect_gen.PORT_ENV, "0")
         started = await connect_gen.start(grpc_port, host="127.0.0.1")
         assert started is not None
-        async with httpx.AsyncClient() as client:
-            r = await client.post(
-                f"http://127.0.0.1:{started.port}/echo.v1.Echo/Say",
-                json={"text": "s"},
-                headers={"content-type": "application/json", "x-caller": "zoe"},
-            )
-        assert r.json() == {"text": "s|zoe"}
-        await started.stop()
+        try:
+            async with httpx.AsyncClient() as client:
+                r = await client.post(
+                    f"http://127.0.0.1:{started.port}/echo.v1.Echo/Say",
+                    json={"text": "s"},
+                    headers={"content-type": "application/json", "x-caller": "zoe"},
+                )
+            assert r.json() == {"text": "s|zoe"}
+        finally:
+            await started.stop()
     finally:
         await server.stop(None)
+
+
+@pytest.fixture
+async def short_idle():
+    """A face whose idle bound (0.5 s) is shorter than the slow call (2 s)."""
+    server = grpc.aio.server(interceptors=(Proof(),))
+    server.add_generic_rpc_handlers((_handler(),))
+    grpc_port = server.add_insecure_port("127.0.0.1:0")
+    await server.start()
+    listener = await connect_gen.serve(
+        0, grpc_port, host="127.0.0.1", read_timeout=0.5, idle_timeout=0.5
+    )
+    try:
+        yield listener
+    finally:
+        await listener.stop()
+        await server.stop(None)
+
+
+async def test_a_grpc_call_in_flight_outlives_the_idle_bound(short_idle):
+    # The handler sends nothing for 2 s: an open call is never idle.
+    async with grpc.aio.insecure_channel(f"127.0.0.1:{short_idle.port}") as channel:
+        say_call = channel.unary_unary(
+            "/echo.v1.Echo/Say",
+            request_serializer=Msg.SerializeToString,
+            response_deserializer=Msg.FromString,
+        )
+        answer = await say_call(Msg(text="slow"), metadata=tuple(STAMP.items()), timeout=10)
+    assert answer.text == "slow|maya"
+
+
+async def test_a_grpc_connection_with_no_open_call_is_closed(short_idle):
+    async with grpc.aio.insecure_channel(f"127.0.0.1:{short_idle.port}") as channel:
+        say_call = channel.unary_unary(
+            "/echo.v1.Echo/Say",
+            request_serializer=Msg.SerializeToString,
+            response_deserializer=Msg.FromString,
+        )
+        assert (await say_call(Msg(text="q"), metadata=tuple(STAMP.items()))).text == "q|maya"
+        # The call ended: the relayed connection is idle, and closed past the bound.
+        for _ in range(30):
+            if not short_idle._connections:
+                break
+            await asyncio.sleep(0.1)
+        assert not short_idle._connections

From bca58c0690e74be2d9c35c62c0c038c72847c1c1 Mon Sep 17 00:00:00 2001
From: antoinetoussaint-byte
 <312536060+antoinetoussaint-byte@users.noreply.github.com>
Date: Sun, 25 Oct 2026 17:01:10 -0400
Subject: [PATCH 3/3] fix(connect): bound what the relay tracks per connection;
 end it when either side closes

Re-review r22 of #109:

- N1: the relay recorded every client HEADERS stream as an open call, and
  grpcio ignores an even (server-initiated) stream without a RST, so a client
  could grow the set without bound and hold the connection forever with one
  10-byte frame. Now only a stream above every earlier one opens a call (a
  lower id, a trailer or a stream the server refuses as closed, opens none),
  an even stream closes the connection, and so do more open calls than the
  server's SETTINGS_MAX_CONCURRENT_STREAMS (DEFAULT_MAX_OPEN_CALLS, 1000, when
  it advertises none, as grpcio does). A server GOAWAY drops the calls above
  its last stream, which it will never answer.
- N2: when the gRPC server closed its side, the relay kept the client's open
  while a call was recorded. Either side closing now ends the relay.
- m-a: runtime.go no longer says the face falls back to 8081.

Tests (each fails on 6c879d1): test_a_client_opening_an_even_stream_is_closed,
test_open_calls_on_one_connection_are_bounded,
test_a_stream_id_below_the_highest_opens_no_call,
test_the_relay_ends_when_the_grpc_server_hangs_up.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 runtime.go                           |   3 +-
 templates/connect/connect_gen.py     | 116 +++++++++++++++++++++------
 testdata/connect/test_connect_gen.py | 116 +++++++++++++++++++++++++++
 3 files changed, 208 insertions(+), 27 deletions(-)

diff --git a/runtime.go b/runtime.go
index 4922682..ddc7078 100644
--- a/runtime.go
+++ b/runtime.go
@@ -672,7 +672,8 @@ func (s *Runtime) Start(ctx context.Context, req *runtimev0.StartRequest) (*runt
 	}
 	if s.connectPort != 0 {
 		// The same for the Connect face (src/rpc/connect_gen.py), which the
-		// lifespan starts over the gRPC server; its fixed fallback is 8081.
+		// app starts over the gRPC server on this variable alone: unset, it
+		// serves no face.
 		proc.WithEnvironmentVariables(ctx, resources.Env("CODEFLY_CONNECT_PORT", s.connectPort))
 	}
 
diff --git a/templates/connect/connect_gen.py b/templates/connect/connect_gen.py
index e2d3271..1b80f3b 100644
--- a/templates/connect/connect_gen.py
+++ b/templates/connect/connect_gen.py
@@ -54,6 +54,7 @@
 __all__ = ["ConnectFace", "serve", "start"]
 
 PORT_ENV = "CODEFLY_CONNECT_PORT"
+# The port a deployment hands the face (CODEFLY_CONNECT_PORT); ``start`` reads only the variable.
 DEFAULT_PORT = 8081
 DEFAULT_MAX_MESSAGE_BYTES = 4 * 1024 * 1024
 DEFAULT_READ_TIMEOUT_S = 30.0
@@ -62,8 +63,15 @@
 _PREFACE = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"
 # An HTTP/2 frame header: 24-bit length, type, flags, stream id (RFC 9113 section 4.1).
 _FRAME = struct.Struct(">BHBBI")
-_DATA, _HEADERS, _RST_STREAM = 0x0, 0x1, 0x3
-_END_STREAM_FLAG = 0x1
+_DATA, _HEADERS, _RST_STREAM, _SETTINGS, _GOAWAY = 0x0, 0x1, 0x3, 0x4, 0x7
+_END_STREAM_FLAG, _ACK_FLAG = 0x1, 0x1
+_MAX_CONCURRENT_STREAMS = 0x3
+# Calls one relayed connection may hold open when the gRPC server advertises no
+# SETTINGS_MAX_CONCURRENT_STREAMS (grpcio advertises none); past it the face closes the
+# connection, so what it tracks per connection is bounded.
+DEFAULT_MAX_OPEN_CALLS = 1000
+# The most of a SETTINGS or GOAWAY payload the relay reads; the rest is skipped.
+_CONTROL_PAYLOAD = 256
 _ENVELOPE = struct.Struct(">BI")
 _COMPRESSED, _END_STREAM, _TRAILERS = 0x01, 0x02, 0x80
 _STATUS_DETAILS = "grpc-status-details-bin"
@@ -212,18 +220,34 @@ async def start(grpc_port: int, **options) -> ConnectFace | None:
     return await serve(int(port), grpc_port, **options)
 
 
+class _ProtocolError(Exception):
+    """A client frame no HTTP/2 client may send: the relay closes the connection."""
+
+
 class _Frames:
-    """The HTTP/2 frame headers in one direction of a relayed connection, after the
-    preface: ``on_frame(type, flags, stream)`` for each, payloads skipped."""
+    """The HTTP/2 frames in one direction of a relayed connection, after the preface:
+    ``on_frame(type, flags, stream, payload)`` for each. The payload is read for SETTINGS
+    and GOAWAY only (at most ``_CONTROL_PAYLOAD`` bytes of it) and skipped otherwise."""
 
     def __init__(self, on_frame):
         self._on_frame = on_frame
         self._header = bytearray()
-        self._skip = 0
+        self._frame = None  # (type, flags, stream) while its payload is read
+        self._payload = bytearray()
+        self._keep = 0  # payload bytes still to read
+        self._skip = 0  # payload bytes still to skip
 
     def feed(self, data: bytes) -> None:
         at = 0
         while at < len(data):
+            if self._keep:
+                step = min(self._keep, len(data) - at)
+                self._payload += data[at : at + step]
+                self._keep -= step
+                at += step
+                if not self._keep:
+                    self._emit()
+                continue
             if self._skip:
                 step = min(self._skip, len(data) - at)
                 self._skip -= step
@@ -235,8 +259,20 @@ def feed(self, data: bytes) -> None:
             if len(self._header) == _FRAME.size:
                 high, low, kind, flags, stream = _FRAME.unpack(self._header)
                 self._header.clear()
-                self._skip = high << 16 | low
-                self._on_frame(kind, flags, stream & 0x7FFFFFFF)
+                length = high << 16 | low
+                self._frame = (kind, flags, stream & 0x7FFFFFFF)
+                if kind in (_SETTINGS, _GOAWAY):
+                    self._keep = min(length, _CONTROL_PAYLOAD)
+                    self._skip = length - self._keep
+                else:
+                    self._skip = length
+                if not self._keep:
+                    self._emit()
+
+    def _emit(self):
+        payload = bytes(self._payload)
+        self._payload.clear()
+        self._on_frame(*self._frame, payload)
 
 
 class _Connection:
@@ -258,25 +294,52 @@ async def serve(self, reader, writer):
             await self._http1(reader, writer, head)
 
     async def _relay(self, reader, writer, head):
-        """gRPC over HTTP/2: the bytes go to the gRPC server as they came, until the
-        connection has had no open call for ``idle_timeout``. The frame headers say which
-        calls are open: the client's HEADERS opens one, and the server's END_STREAM or
-        either side's RST_STREAM ends it."""
+        """gRPC over HTTP/2: the bytes go to the gRPC server as they came, until either
+        side closes, or the connection has had no open call for ``idle_timeout``.
+
+        The frame headers say which calls are open: a client HEADERS on a new stream (odd,
+        above every stream before it) opens one, and the server's END_STREAM, either
+        side's RST_STREAM, or the server's GOAWAY above the stream ends it. A client that
+        opens an even stream, or more calls than the server allows at once
+        (SETTINGS_MAX_CONCURRENT_STREAMS, ``DEFAULT_MAX_OPEN_CALLS`` when it advertises
+        none), is closed: no HTTP/2 client does either."""
         upstream_reader, upstream_writer = await asyncio.open_connection(*self.grpc_address)
         upstream_writer.write(head)
         loop = asyncio.get_running_loop()
         last = loop.time()
         calls: set[int] = set()
-
-        def from_client(kind, flags, stream):
-            if kind == _HEADERS and stream:
+        highest = 0
+        max_calls = DEFAULT_MAX_OPEN_CALLS
+
+        def from_client(kind, flags, stream, payload):
+            nonlocal highest
+            if kind == _HEADERS:
+                if stream % 2 == 0:
+                    raise _ProtocolError(f"HEADERS on stream {stream}, which a server opens")
+                if stream <= highest:
+                    # Trailers on a call, or a stream the server refuses as closed:
+                    # never a new call.
+                    return
+                highest = stream
+                if len(calls) >= max_calls:
+                    raise _ProtocolError(f"more than {max_calls} calls open at once")
                 calls.add(stream)
             elif kind == _RST_STREAM:
                 calls.discard(stream)
 
-        def from_server(kind, flags, stream):
+        def from_server(kind, flags, stream, payload):
+            nonlocal max_calls
             if kind == _RST_STREAM or (kind in (_DATA, _HEADERS) and flags & _END_STREAM_FLAG):
                 calls.discard(stream)
+            elif kind == _SETTINGS and not flags & _ACK_FLAG:
+                for at in range(0, len(payload) - 5, 6):
+                    key, value = struct.unpack_from(">HI", payload, at)
+                    if key == _MAX_CONCURRENT_STREAMS:
+                        max_calls = value
+            elif kind == _GOAWAY and len(payload) >= 4:
+                # Streams above the last one the server processes are never answered.
+                last_stream = struct.unpack_from(">I", payload)[0] & 0x7FFFFFFF
+                calls.difference_update({call for call in calls if call > last_stream})
 
         async def pump(source, sink, frames):
             nonlocal last
@@ -286,10 +349,10 @@ async def pump(source, sink, frames):
                     frames.feed(chunk)
                     sink.write(chunk)
                     await sink.drain()
-                if sink.can_write_eof():
-                    sink.write_eof()
             except (ConnectionError, OSError):
                 pass
+            except _ProtocolError as refusal:
+                _log.info("connect face: closing a gRPC connection: %s", refusal)
 
         async def idle():
             # The clock runs from the last byte either way, and only ends the connection
@@ -300,18 +363,19 @@ async def idle():
                     return
                 await asyncio.sleep(remaining if remaining > 0 else self.idle_timeout)
 
-        client, server = _Frames(from_client), _Frames(from_server)
-
-        pumps = asyncio.gather(
-            pump(reader, upstream_writer, client), pump(upstream_reader, writer, server)
+        tasks = (
+            asyncio.ensure_future(pump(reader, upstream_writer, _Frames(from_client))),
+            asyncio.ensure_future(pump(upstream_reader, writer, _Frames(from_server))),
+            asyncio.ensure_future(idle()),
         )
-        watchdog = asyncio.ensure_future(idle())
         try:
-            await asyncio.wait((pumps, watchdog), return_when=asyncio.FIRST_COMPLETED)
+            # Either side closing ends the relay: an HTTP/2 connection is not half-closed,
+            # and a call still recorded open cannot be answered once its server has gone.
+            await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
         finally:
-            pumps.cancel()
-            watchdog.cancel()
-            await asyncio.gather(pumps, watchdog, return_exceptions=True)
+            for task in tasks:
+                task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
             upstream_writer.close()
 
     async def _read(self, reader, timeout):
diff --git a/testdata/connect/test_connect_gen.py b/testdata/connect/test_connect_gen.py
index 1f4de1c..ad54917 100644
--- a/testdata/connect/test_connect_gen.py
+++ b/testdata/connect/test_connect_gen.py
@@ -415,3 +415,119 @@ async def test_a_grpc_connection_with_no_open_call_is_closed(short_idle):
                 break
             await asyncio.sleep(0.1)
         assert not short_idle._connections
+
+
+PREFACE = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"
+
+
+def h2_frame(kind, flags, stream, payload=b""):
+    size = len(payload)
+    return struct.pack(">BHBBI", size >> 16, size & 0xFFFF, kind, flags, stream) + payload
+
+
+SETTINGS = h2_frame(0x4, 0, 0)
+
+
+def opens(stream):
+    """A HEADERS frame opening ``stream`` (END_HEADERS; one indexed field)."""
+    return h2_frame(0x1, 0x4, stream, b"\x82")
+
+
+@pytest.fixture
+async def silent_upstream():
+    """A gRPC address that accepts, reads everything and never answers, so only the face's
+    own rules end a connection relayed to it, and ``face(**options)`` serving over it.
+    ``state["close_after"]`` makes it hang up after that many seconds instead."""
+    held, faces = [], []
+    state = {"close_after": None}
+
+    async def accept(reader, writer):
+        held.append(writer)
+        if state["close_after"] is not None:
+            await asyncio.sleep(state["close_after"])
+            writer.close()
+            return
+        while await reader.read(65536):
+            pass
+
+    upstream = await asyncio.start_server(accept, "127.0.0.1", 0)
+    port = upstream.sockets[0].getsockname()[1]
+
+    async def face(**options):
+        listener = await connect_gen.serve(0, port, host="127.0.0.1", **options)
+        faces.append(listener)
+        return listener
+
+    try:
+        yield face, state
+    finally:
+        for listener in faces:
+            await listener.stop()
+        for writer in held:
+            writer.close()
+        upstream.close()
+
+
+async def closed_within(listener, payload, seconds):
+    reader, writer = await asyncio.open_connection("127.0.0.1", listener.port)
+    writer.write(payload)
+    await writer.drain()
+    try:
+        async with asyncio.timeout(seconds):
+            while await reader.read(65536):
+                pass
+        return True
+    except TimeoutError:
+        return False
+    finally:
+        writer.close()
+
+
+async def test_a_client_opening_an_even_stream_is_closed(silent_upstream):
+    # A client never opens an even stream; grpcio ignores one silently, so a face that
+    # recorded it as a call would hold the connection, and its memory, forever.
+    face, _ = silent_upstream
+    payload = PREFACE + SETTINGS + b"".join(opens(2 * n) for n in range(1, 1000))
+    assert await closed_within(await face(), payload, 2)
+
+
+async def test_open_calls_on_one_connection_are_bounded(silent_upstream):
+    face, _ = silent_upstream
+    listener = await face()
+    calls = connect_gen.DEFAULT_MAX_OPEN_CALLS
+    within = PREFACE + SETTINGS + b"".join(opens(2 * n + 1) for n in range(calls))
+    assert not await closed_within(listener, within, 1)
+    beyond = PREFACE + SETTINGS + b"".join(opens(2 * n + 1) for n in range(calls + 1))
+    assert await closed_within(listener, beyond, 2)
+
+
+async def test_a_stream_id_below_the_highest_opens_no_call(silent_upstream):
+    # 1 and 99 are calls; 3 after 99 is not a new one. Once 1 and 99 are reset no call is
+    # open, and the connection is closed past the idle bound.
+    face, _ = silent_upstream
+    reset = lambda stream: h2_frame(0x3, 0, stream, b"\x00\x00\x00\x08")  # noqa: E731
+    payload = PREFACE + SETTINGS + opens(1) + opens(99) + opens(3) + reset(1) + reset(99)
+    assert await closed_within(await face(idle_timeout=0.5), payload, 2)
+
+
+async def test_the_relay_ends_when_the_grpc_server_hangs_up(silent_upstream):
+    # A call is recorded open and the client is silent: the server closing its side
+    # still ends the connection, rather than leaving the client's open.
+    face, state = silent_upstream
+    state["close_after"] = 0.3
+    listener = await face()
+    reader, writer = await asyncio.open_connection("127.0.0.1", listener.port)
+    writer.write(PREFACE + SETTINGS + opens(1))
+    await writer.drain()
+    try:
+        await asyncio.sleep(0.1)
+        assert listener._connections, "the face took the connection"
+        await asyncio.sleep(0.3)  # the server hangs up
+        for _ in range(20):
+            if not listener._connections:
+                break
+            await asyncio.sleep(0.1)
+        assert not listener._connections, "the face still holds the client's connection"
+        assert await asyncio.wait_for(reader.read(), 1) == b""
+    finally:
+        writer.close()