Share committed-response-loss fixture for command retries

.github/workflows/ci.yml

@@ -74,6 +74,7 @@ jobs:
           tests.test_temporal_runtime
           tests.test_cancellation_workflow_policy
           tests.test_sdk_metrics_contract
+          tests.test_command_response_loss
           tests.test_ci_trust
           tests.test_pip_effective_trust
           tests.test_cli_effective_trust

.github/workflows/sonar.yml

@@ -53,7 +53,7 @@ jobs:
       - name: Produce source coverage
         run: |
           python -m pip install --only-binary :all: coverage==7.10.7 jsonschema==4.23.0 PyYAML==6.0.2
-          coverage run --source=scripts,conformance/cancellation/temporal,.github/actions/bootstrap-patched-pip -m unittest tests.test_qualification_policy tests.test_protected_policy_context tests.test_protected_policy_check tests.test_protected_policy_inputs tests.test_protected_policy_issue tests.test_native_policy_check tests.test_native_policy_workflow tests.test_regression_corpus tests.test_regression_evidence_schema tests.test_server_codec_counterfactual_schema tests.test_visual_evidence tests.test_temporal_runtime tests.test_cancellation_workflow_policy tests.test_sdk_metrics_contract tests.test_ci_trust tests.test_pip_effective_trust tests.test_cli_effective_trust tests.test_composer_trust_contract tests.test_composer_container_trust tests.test_composer_container_effective_trust tests.test_php_linux_bootstrap tests.test_php_linux_archive tests.test_php_linux_normalize tests.test_php_linux_recipe tests.test_php_linux_tools tests.test_php_linux_alternatives tests.test_setup_php_linux tests.test_php_linux_qualification tests.test_php_linux_qualification_workflow tests.test_php_linux_configuration
+          coverage run --source=scripts,conformance/cancellation/temporal,conformance/command-keys,.github/actions/bootstrap-patched-pip -m unittest tests.test_qualification_policy tests.test_protected_policy_context tests.test_protected_policy_check tests.test_protected_policy_inputs tests.test_protected_policy_issue tests.test_native_policy_check tests.test_native_policy_workflow tests.test_regression_corpus tests.test_regression_evidence_schema tests.test_server_codec_counterfactual_schema tests.test_visual_evidence tests.test_temporal_runtime tests.test_cancellation_workflow_policy tests.test_sdk_metrics_contract tests.test_command_response_loss tests.test_ci_trust tests.test_pip_effective_trust tests.test_cli_effective_trust tests.test_composer_trust_contract tests.test_composer_container_trust tests.test_composer_container_effective_trust tests.test_php_linux_bootstrap tests.test_php_linux_archive tests.test_php_linux_normalize tests.test_php_linux_recipe tests.test_php_linux_tools tests.test_php_linux_alternatives tests.test_setup_php_linux tests.test_php_linux_qualification tests.test_php_linux_qualification_workflow tests.test_php_linux_configuration
           coverage xml -o coverage.xml
       - uses: SonarSource/sonarqube-scan-action@d209202bc7d53ff1cc128f7f907dac145c9d6ae9 # v8.3.0
         env:

conformance/README.md

@@ -141,6 +141,11 @@ actions work. Report its scope separately from the stable protocol experiments.
 
 ## Evidence scope
 
+Command-admission retry experiments under #216 can reuse the
+[committed-response-loss fixture](command-keys/README.md). Its socket control
+proves response loss after upstream success; the installed SDK/CLI adapter and
+producer-owned legacy fixtures supply the remaining admission/retention proof.
+
 A pinned or successfully installed SDK is not evidence that its worker executed
 an experiment. For each claimed language or cross-language direction, check the
 scenario-level result for the actual caller/parent and worker/child runtime,

conformance/command-keys/README.md

@@ -0,0 +1,53 @@
+# Committed command-response loss
+
+This shared transport fixture supports [#216](https://github.com/durable-workflow/.github/issues/216).
+It forwards one selected request to a loopback HTTP runtime, reads the complete
+successful upstream response, then closes the client connection before sending
+an HTTP status line. The next request forwards normally. The fixture does not
+retry a request or decide whether command admission was idempotent.
+
+`CommittedResponseLoss(upstream)` accepts only `http://127.0.0.1:<port>` with a
+canonical explicit port. Credentials, hostnames, paths, query and fragment
+components refuse. It binds an ephemeral loopback port. Start its
+`serve_forever` method in a thread and route the installed client through `url`.
+Call `arm(method, path)` with the exact origin-form admission route before
+the selected request. Query strings and fragments refuse before forwarding or
+recording; put no credentials in route identifiers. The caller owns shutdown, the runtime, synthetic data and cleanup.
+
+Requests and responses each have a fixed 1 MiB wire-body cap and bounded socket
+timeouts. POST requires one canonical Content-Length; request transfer encodings
+refuse. GET is bodyless and may omit Content-Length. Complete bounded chunked
+upstream responses are supported. Invalid, oversized or incomplete upstream
+responses produce a generic 502 and no success record. A matching valid request
+consumes the arm even if its upstream attempt fails; re-arming is explicit.
+At most 1024 metadata records, including reserved in-flight slots, are allowed.
+Capacity exhaustion returns a generic 503 before contacting upstream and does
+not consume the arm. Failed upstream attempts release their reserved slot.
+
+`records` contains method/path, request byte count/hash, complete upstream status,
+response byte count/hash and the dropped flag. It retains no headers, parsed
+JSON, caller keys, results or payloads. SDK adapters compare their own request
+and retained-admission identities independently. Use synthetic inputs and
+retain only the metadata needed by the experiment.
+
+Run the socket controls without dependencies or an external runtime:
+
+```sh
+python3 -I -S -B conformance/command-keys/test_committed_response_loss.py
+```
+
+The controls prove upstream success was complete before the client received
+`RemoteDisconnected`, and that one incoming request caused exactly one upstream
+request. It also checks one-shot path selection, ordinary forwarding, errors
+and numeric loopback confinement. Negative controls cover framing, byte caps,
+incomplete upstream responses and body/header evidence minimization. The earlier
+CLI fixture is preserved in Git history; this correction tightens its reusable
+authority, resource and evidence boundaries.
+
+An SDK/CLI adapter still verifies the actual published tuple, original/generated
+key, durable admission count, concurrent identical retries, changed-intent
+conflicts, closure/continuation, namespace/run fences and retained identities.
+Record those outcomes separately from this transport control. Producer-owned
+legacy-shape and physical-upgrade seeds remain separate; the proxy creates
+neither. The [original published reproduction](https://github.com/durable-workflow/.github/issues/216#issuecomment-6296529480)
+describes the earlier loss scenarios and is not current installed acceptance.

conformance/command-keys/committed_response_loss.py

@@ -0,0 +1,195 @@
+"""Bounded loopback fixture: consume upstream success, then lose one response."""
+import hashlib
+import http.client
+import http.server
+import re
+import socket
+import threading
+
+
+MAX_REQUEST_BYTES = 1_048_576
+MAX_RESPONSE_BYTES = 1_048_576
+MAX_RECORDS = 1024
+SOCKET_TIMEOUT_SECONDS = 40
+
+
+def origin_path(path):
+    return (isinstance(path, str) and len(path) <= 4096 and not path.startswith("//")
+            and re.fullmatch(r"/[^\x00-\x20\x7f#?]*", path) is not None)
+
+
+def content_length(headers, required):
+    values = headers.get_all("Content-Length", [])
+    if not values and not required:
+        return None
+    if len(values) != 1 or re.fullmatch(r"0|[1-9]\d{0,6}", values[0], re.ASCII) is None:
+        raise ValueError("A single canonical bounded Content-Length is required")
+    return int(values[0])
+
+
+class RequestRefused(ValueError):
+    def __init__(self, status, reason):
+        super().__init__(reason)
+        self.status = status
+
+
+class CommittedResponseLoss(http.server.ThreadingHTTPServer):
+    daemon_threads = True
+
+    def __init__(self, upstream):
+        match = re.fullmatch(r"http://127\.0\.0\.1:([1-9]\d{0,4})", upstream, re.ASCII) if isinstance(upstream, str) else None
+        if match is None or not 1 <= int(match.group(1)) <= 65535:
+            raise ValueError("Fixture upstream must be canonical numeric loopback HTTP with an explicit port")
+        self.upstream_port = int(match.group(1))
+        self.records = []
+        self.pending_records = 0
+        self.armed = None
+        self.lock = threading.Lock()
+        super().__init__(("127.0.0.1", 0), _Handler)
+
+    def arm(self, method, path):
+        if method not in {"GET", "POST"} or not origin_path(path):
+            raise ValueError("Arm requires an exact supported method and origin-form path")
+        with self.lock:
+            if self.armed is not None:
+                raise ValueError("a response loss is already armed")
+            self.armed = (method, path)
+
+    def reserve_record(self, method, path):
+        with self.lock:
+            if len(self.records) + self.pending_records >= MAX_RECORDS:
+                raise RequestRefused(503, "Fixture evidence capacity exhausted")
+            self.pending_records += 1
+            selected = self.armed == (method, path)
+            if selected:
+                self.armed = None
+            return selected
+
+    def finish_record(self, record):
+        with self.lock:
+            self.pending_records -= 1
+            if record is not None:
+                self.records.append(record)
+
+    @property
+    def url(self):
+        return f"http://127.0.0.1:{self.server_port}"
+
+
+class _Handler(http.server.BaseHTTPRequestHandler):
+    timeout = SOCKET_TIMEOUT_SECONDS
+
+    def log_message(self, *_args):
+        # Evidence uses closed metadata records rather than HTTP request logs.
+        pass
+
+    def send_error(self, code, message=None, explain=None):
+        # Base parser messages can include an untrusted raw request line.
+        reasons = {400: "Fixture request framing refused", 413: "Fixture byte cap exceeded",
+                   501: "Fixture method unsupported", 502: "Fixture upstream response unverified",
+                   503: "Fixture evidence capacity exhausted"}
+        super().send_error(code, reasons.get(code, "Fixture request refused"),
+                           "The fixture refused the request without retaining its content.")
+
+    def request_body(self):
+        original_path = self.requestline.split()[1]
+        if (self.request_version not in {"HTTP/1.0", "HTTP/1.1"} or original_path != self.path
+                or not origin_path(original_path) or self.headers.get_all("Transfer-Encoding", [])):
+            raise RequestRefused(400, "Fixture requires an origin path and no request transfer encoding")
+        try:
+            length = content_length(self.headers, self.command == "POST")
+        except ValueError as error:
+            raise RequestRefused(400, "Fixture request length is invalid") from error
+        if self.command == "GET" and length not in (None, 0):
+            raise RequestRefused(400, "Fixture GET requests must be bodyless")
+        length = length or 0
+        if length > MAX_REQUEST_BYTES:
+            raise RequestRefused(413, "Fixture request exceeds its fixed byte cap")
+        self.connection.settimeout(SOCKET_TIMEOUT_SECONDS)
+        try:
+            body = self.rfile.read(length)
+        except OSError as error:
+            raise RequestRefused(400, "Fixture request body is incomplete") from error
+        if len(body) != length:
+            raise RequestRefused(400, "Fixture request body is incomplete")
+        return body
+
+    def response_body(self, response):
+        encodings = response.headers.get_all("Transfer-Encoding", [])
+        length = content_length(response.headers, False)
+        if encodings and (encodings != ["chunked"] or length is not None):
+            raise ValueError("Fixture upstream framing is unsupported")
+        if length is not None and length > MAX_RESPONSE_BYTES:
+            raise ValueError("Fixture upstream response exceeds its fixed byte cap")
+        body = response.read(MAX_RESPONSE_BYTES + 1)
+        if len(body) > MAX_RESPONSE_BYTES or (length is not None and len(body) != length):
+            raise ValueError("Fixture upstream body is oversized or incomplete")
+        if response.read(1):
+            raise ValueError("Fixture upstream body exceeds its fixed byte cap")
+        return body
+
+    def upstream_response(self, raw):
+        # The destination address is literal; only the validated port varies.
+        upstream = http.client.HTTPConnection("127.0.0.1", self.server.upstream_port,
+                                              timeout=SOCKET_TIMEOUT_SECONDS)
+        try:
+            upstream.connect()
+            if upstream.sock.getpeername()[0] != "127.0.0.1":
+                raise ValueError("Fixture upstream peer is not loopback")
+            headers = {k: v for k, v in self.headers.items()
+                       if k.lower() not in {"host", "connection", "transfer-encoding", "proxy-authorization"}}
+            headers["Connection"] = "close"
+            upstream.request(self.command, self.path, body=raw or None, headers=headers)
+            response = upstream.getresponse()
+            body = self.response_body(response)
+            return response.status, response.getheaders(), body
+        finally:
+            upstream.close()
+
+    def refuse(self, status):
+        self.close_connection = True
+        try:
+            self.send_error(status)
+        except OSError:
+            # A failed client may already have closed its side of the fixture.
+            pass
+
+    def forward(self):
+        try:
+            raw = self.request_body()
+            selected = self.server.reserve_record(self.command, self.path)
+        except RequestRefused as error:
+            self.refuse(error.status)
+            return
+        try:
+            status, headers, body = self.upstream_response(raw)
+            drop = selected and 200 <= status < 300
+            record = {"method": self.command, "path": self.path,
+                      "request_body_bytes": len(raw), "request_body_sha256": hashlib.sha256(raw).hexdigest(),
+                      "upstream_status": status, "upstream_complete": True,
+                      "upstream_body_bytes": len(body), "upstream_body_sha256": hashlib.sha256(body).hexdigest(),
+                      "dropped": drop}
+        except (http.client.HTTPException, OSError, ValueError):
+            self.server.finish_record(None)
+            self.refuse(502)
+            return
+        self.server.finish_record(record)
+        self.close_connection = True
+        try:
+            if drop:
+                self.connection.shutdown(socket.SHUT_RDWR)
+                self.connection.close()
+                return
+            self.send_response(status)
+            for key, value in headers:
+                if key.lower() not in {"content-length", "transfer-encoding", "connection", "server", "date"}:
+                    self.send_header(key, value)
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+        except OSError:
+            # Unplanned client loss is never recorded as intentional injection.
+            pass
+
+    do_GET = forward
+    do_POST = forward