ci: say why the suite is green — and fix the OOM that saying it uncovered

.github/workflows/conformance.yml

@@ -4,18 +4,27 @@ name: conformance
 # it. The harness needs a real docker daemon; it refuses to run rather than simulate a
 # container, because a harness that fakes the container proves nothing.
 #
-# WHY THIS IS GREEN TODAY. Twenty tests are known to fail against node.py as it stands
-# (see CONFORMANCE-BASELINE.md). They carry the marker `expected_red_until_fixed` and run
-# as STRICT expected failures, so today's run is green with them reported as `xfailed`.
-# The moment somebody fixes one of those the test PASSES, strict xfail turns the run RED,
-# and the failure says which test to promote out of the expected-red group. That is the
-# signal this workflow exists for: it is not a gate on the node being correct, it is a
-# gate on the record of what is wrong staying accurate.
+# WHY THIS IS GREEN TODAY. Because every test passes. There is no
+# `expected_red_until_fixed` test left: the twenty that used to fail against node.py are
+# fixed, and they now sit in `conforms_today`, where a regression turns the run red at once
+# instead of being absorbed as an expected failure. A failing test in this workflow is
+# therefore simply a failure — something to fix in node.py, not something to write down.
 #
-# Six of the twenty are contract violations, three are this repository's own compatibility
-# policy, and eleven are reference-quality expectations the contract permits a real node to
-# skip. The per-basis counts are printed below, because that distinction is what stops a
-# documentation slice teaching a preference as a rule.
+# The expected-red machinery stays, armed and unused, as the tripwire for the next known
+# gap. A test carrying that marker runs as a STRICT expected failure, so CI is green while
+# the gap is on the books, with the gap visible in the report as `xfailed`; the moment
+# somebody fixes it the test PASSES, strict xfail turns the run RED, and the failure says
+# which test to promote into `conforms_today`. What that prevents is a defect described one
+# way in the documents and behaving another way in the image, with nothing that ever
+# notices. The counts printed below are all zero today, and that zero is the claim being
+# made, not a query that came back empty by accident.
+#
+# Only `basis_contract` failures are conformance violations; the other two bases are this
+# repository's own compatibility policy and reference-quality expectations the contract
+# permits a real node to skip. They are counted separately because that distinction is what
+# stops a documentation slice teaching a preference as a rule, and CONFORMANCE-BASELINE.md
+# records what it was worth while the twenty were real: six were contract violations, three
+# were policy, and eleven were reference quality.
 #
 # NOT CHECKED HERE: that each basis_contract citation quotes its source verbatim. That
 # needs a checkout of the orchestrator, which this repository does not have — run it with
@@ -67,15 +76,19 @@ jobs:
             python -m pytest --collect-only -q -m "$group" | tail -1
           done
 
-      - name: What the red tests actually claim
+      - name: What an expected-red test would claim
         if: always()
         # The distinction that keeps this harness honest: only `basis_contract` failures
         # are conformance violations. The others are worth fixing and must never be
         # written down as requirements — so they are counted separately, in the log,
-        # every run.
+        # every run. All three counts are zero today, and printing the zeros is the point:
+        # it is the standing record that nothing is being excused. The count is extracted
+        # as a number rather than left as pytest's own summary line, because "no tests
+        # collected" reads like a query that broke instead of an answer of none.
         run: |
           for basis in basis_contract basis_our_policy basis_reference_quality; do
-            echo -n "expected-red on $basis: "
-            python -m pytest --collect-only -q -m "expected_red_until_fixed and $basis" | tail -1
+            n=$(python -m pytest --collect-only -q -m "expected_red_until_fixed and $basis" | grep -c '::' || true)
+            echo "expected-red on $basis: $n"
           done
-          python -m pytest --collect-only -q --print-labels 2>/dev/null | grep -E '^expected_red' || true
+          python -m pytest --collect-only -q --print-labels 2>/dev/null | grep -E '^expected_red' \
+            || echo 'expected-red label table: empty — every test declares another group'

CLAUDE.md

@@ -91,11 +91,16 @@ covered by a test, and none of them is enforced by the platform.
    `/lspo/creds/creds.json`. Never bake a credentials path into the image.
 2. **Verify every input** against both the pinned `sha256` and the pinned `size`, and
    refuse one with no pin. The platform verifies what you *wrote*, never what you *read*.
-3. **Stream, in both directions.** A single object may legally be 1 GiB against a 2 GiB
-   container. Holding one in memory twice is an out-of-memory kill, and an OOM kill leaves
-   no chance to write a marker at all. This is why there is **no HTTP library dependency**:
-   `requests` builds a multipart body in memory, so using it would contradict this rule in
-   the file that exists to demonstrate it.
+3. **Stream, in both directions — and release the page cache while you do.** A single
+   object may legally be 1 GiB against a 2 GiB container. Holding one in memory twice is
+   an out-of-memory kill, and an OOM kill leaves no chance to write a marker at all. This
+   is why there is **no HTTP library dependency**: `requests` builds a multipart body in
+   memory, so using it would contradict this rule in the file that exists to demonstrate
+   it. Streaming is necessary and not sufficient: the container's memory limit counts the
+   page cache your own reads and writes create, so a perfectly streaming step is still
+   killed for moving a large object through a temporary file. `posix_fadvise(…,
+   POSIX_FADV_DONTNEED)` every few megabytes is what closes that, and the difference is
+   measured — 128 MiB through a 64 MiB container dies without it.
 4. **Re-read the credentials.** The agent replaces the file underneath a running container,
    atomically and with no signal. A node that reads it once cannot upload anything —
    including its own marker — after about fifteen minutes.

docs/PROTOCOL.md

@@ -595,6 +595,21 @@ chunks, hash while you stream, and upload from a file handle rather than from a
 object. Do this even for a node that "only handles small files": the input size is chosen
 by whoever wires the pipeline, not by you.
 
+**RECOMMENDATION, and streaming alone does not buy it. This one is measured.** The
+container's memory limit counts the **page cache** that its own reads and writes create,
+not only what your process holds. So a node that streams perfectly — never more than one
+block in memory — is still OOM-killed for moving a large object through a temporary file:
+your program's footprint stays flat while the kernel's cache for that file grows to the
+size of the object. Ask for those pages back as you go. On Linux that is
+`posix_fadvise(fd, 0, 0, POSIX_FADV_DONTNEED)` every few megabytes, after an `fsync` on
+the write side, because dirty pages cannot be dropped; reads need no sync. It is advice
+to the kernel rather than a guarantee, and it costs nothing when it is ignored.
+
+Measured against this repository's own image: a 128 MiB input through a 64 MiB container
+is OOM-killed without that call and survives with it, and with it the same transfer also
+survives a 32 MiB container. The step's own code was identical in both runs — the only
+difference is who was holding the bytes.
+
 ---
 
 ## 4. Outputs

node.py

@@ -141,6 +141,9 @@
 UPLOAD_TIMEOUT_S = 10.0
 #: Every streaming copy moves this much at a time.
 CHUNK_BYTES = 1024 * 1024
+#: How much a streaming copy may leave in the kernel's page cache before asking for it
+#: back. See :func:`_release_page_cache` — this is not a performance knob.
+CACHE_DROP_BYTES = 8 * 1024 * 1024
 #: Re-read the envelope this long before it says it expires.
 CREDS_MARGIN_S = 5.0
 
@@ -379,6 +382,29 @@ def _is_int(value: object) -> bool:
     return isinstance(value, int) and not isinstance(value, bool)
 
 
+def _release_page_cache(handle, *, sync: bool = False) -> None:
+    """Ask the kernel to drop the pages this file has put in the cache.
+
+    **Streaming is not enough on its own.** The container's memory limit counts the page
+    cache created by its own reads and writes, so a step that never holds more than one
+    block in memory can still be OOM-killed for moving a large object through a temporary
+    file: the program's own footprint stays flat while the kernel's cache for that file
+    grows to the size of the object. This was measured — a 128 MiB input through a 64 MiB
+    container is killed without this call and survives with it.
+
+    Dirty pages cannot be dropped, which is why a write has to be flushed and synced
+    first. Reads need no sync. Both are best-effort: ``posix_fadvise`` is advice, and it
+    does not exist everywhere, so a platform without it simply keeps its cache.
+    """
+    try:
+        if sync:
+            handle.flush()
+            os.fsync(handle.fileno())
+        os.posix_fadvise(handle.fileno(), 0, 0, os.POSIX_FADV_DONTNEED)
+    except (AttributeError, OSError, ValueError):
+        pass
+
+
 # ------------------------------------------------------------------------- http
 
 
@@ -472,6 +498,7 @@ def __init__(self, head: bytes, source_path: str, file_size: int, tail: bytes) -
         self._tail = memoryview(tail)
         self._source_path = source_path
         self._remaining = file_size
+        self._uncached = 0
         self._handle = None
         self._stage = 0  # 0 head, 1 file, 2 tail, 3 done
         self._head_at = 0
@@ -499,6 +526,12 @@ def read(self, size: int = -1) -> bytes:
                 piece = self._handle.read(min(size, self._remaining))
                 if piece:
                     self._remaining -= len(piece)
+                    self._uncached += len(piece)
+                    if self._uncached >= CACHE_DROP_BYTES:
+                        # Reading the file back fills the page cache just as writing it
+                        # did, and that cache counts against the container's memory limit.
+                        _release_page_cache(self._handle)
+                        self._uncached = 0
                     return piece
                 self._handle.close()
                 self._handle = None
@@ -705,6 +738,7 @@ def fetched_and_verified(creds: Credentials, index: int, scratch: str):
             size = 0
             try:
                 with open(path, 'wb') as target:
+                    uncached = 0
                     for chunk in _stream(source, name):
                         _check_stopped()
                         digest.update(chunk)
@@ -714,6 +748,11 @@ def fetched_and_verified(creds: Credentials, index: int, scratch: str):
                                 f'input {name!r} is larger than the {expected_size} bytes the job '
                                 f'pinned for it — this is not the object this run was built from')
                         target.write(chunk)
+                        uncached += len(chunk)
+                        if uncached >= CACHE_DROP_BYTES:
+                            _release_page_cache(target, sync=True)
+                            uncached = 0
+                    _release_page_cache(target, sync=True)
                 break
             except _Expired:
                 if attempt == 2:
@@ -851,12 +890,18 @@ def process(creds: Credentials, manifest: dict, scratch: str) -> dict:
 
 def _count_lines(path: str) -> int:
     lines = 0
+    uncached = 0
     with open(path, 'rb') as handle:
         while True:
             chunk = handle.read(CHUNK_BYTES)
             if not chunk:
+                _release_page_cache(handle)
                 return lines
             lines += chunk.count(b'\n')
+            uncached += len(chunk)
+            if uncached >= CACHE_DROP_BYTES:
+                _release_page_cache(handle)
+                uncached = 0
 
 
 def _write_result(creds: Credentials, manifest: dict, metrics: dict, scratch: str) -> None:
@@ -1050,6 +1095,7 @@ def _sha256_file(path: str) -> str:
     with open(path, 'rb') as handle:
         for chunk in iter(lambda: handle.read(CHUNK_BYTES), b''):
             digest.update(chunk)
+        _release_page_cache(handle)
     return digest.hexdigest()