[TH-9449] fix(harness): answer Dileep's P3 comments on #273, #275, #277 and #285

src/fi/alk/harness/agent_scan.py

@@ -141,6 +141,9 @@ def _check_job(job: HarnessJob) -> HarnessFailure | None:
 
 # The most the source look's list of folders still to visit may hold, by its count (TH-9525).
 _LOOK_HELD = 67_108_864
+# Inside one folder, the clock is read again after this many entries, so a huge folder can't keep
+# the look past its deadline.
+_LOOK_CLOCK_EVERY = 4_096
 
 
 def _path_cost(path: str) -> int:
@@ -158,7 +161,9 @@ def _look(
     `_LOOK_HELD` first) or ``"clock"`` (the deadline passed first), with the folders read and what
     the list held. Each folder is read once, entry by entry: its subfolders are taken on until the
     bound, and the read goes on past the bound for a regular file, which wins wherever it is in the
-    folder. A folder or entry that can't be read is passed over, as os.walk does."""
+    folder. A folder or entry that can't be read is passed over, as os.walk does. The clock is read
+    before each folder after the source's own, and every `_LOOK_CLOCK_EVERY` entries inside any
+    folder; an empty source never reaches either, so it is "none" whatever the time."""
     try:
         if not source.is_dir():
             return "none", 0, 0
@@ -179,7 +184,9 @@ def _look(
         over = False
         try:
             with os.scandir(folder) as entries:
-                for entry in entries:
+                for index, entry in enumerate(entries, 1):
+                    if not index % _LOOK_CLOCK_EVERY and clock() > deadline:
+                        return "clock", folders, held
                     try:
                         if entry.is_dir(follow_symlinks=False):
                             if not over:

src/fi/alk/harness/scan_walk.py

@@ -212,19 +212,23 @@ class _Unread(Exception):
     pass
 
 
-# CPython 3.12's sizes on a 64-bit build, as asked of the allocator: a str (`_text`), a tuple
+# CPython's sizes on a 64-bit build, as asked of the allocator, measured on 3.12, the Python the
+# hosted image builds the kit's venv with (`_SIZES_MEASURED_ON`): a str (`_text`, which uses
+# 3.11's 49-byte header, so on 3.12 and 3.13, at 41, it is 8 bytes high), a tuple
 # (`_TUPLE` and a slot per item), a set's own object, one dict or set entry with its share of
 # the table, a facts object, and the join's share for a file that could be an agent (its sign,
 # its component bookkeeping, its row) beside three path strings `_project_of` and
 # `_subfolder` make. Each is then charged at the block the allocator gives it (`_block`).
+_SIZES_MEASURED_ON = (3, 12)
 _TUPLE = 40
 _SLOT = 8
 _SET = 216
 _ENTRY = 112
 _FACTS = 512
 _JOIN_SHARE = 768
 # What the join holds for every kept file however it's joined: its place in the sorted list of
-# paths and in one search's set of files seen.
+# paths and in one search's set of files seen. Held by
+# `test_a_fan_out_join_is_charged_without_its_base`, which fails with this at 0.
 _JOIN_PER_FILE = 64
 # What the join holds whatever was kept: its result, and up to `_MAX_AGENTS` rows with their
 # names and evidence. Charged once, when the walk starts. A build hazard's own look is
@@ -915,11 +919,20 @@ def _source_files(
     Installed trees and the folders a build skips are never entered. ``side`` False gives only
     files outside test, example and other side folders; True gives only the files in them.
     ``charge`` counts each folder kept to enter, at its full path, which is what `os.walk`
-    holds until it enters it."""
+    holds until it enters it, and each folder's own listing, its subfolder and file names, which
+    `os.walk` holds while the walk is in that folder. With ``side`` True, the second pass, only
+    folders the first pass never entered are counted: side folders, and folders below one."""
     for directory, directories, files in os.walk(start, followlinks=False):
         directories.sort()
         files.sort()
         relative = directory[len(top) :].split(os.sep)[1:]
+        first_pass = side is not True or any(_is_side_part(part) for part in relative)
+        if charge is not None and first_pass:
+            # Summed list by list: one tuple of every name would itself hold a slot a name.
+            charge(
+                sum(_SLOT + _text(name) for name in directories)
+                + sum(_SLOT + _text(name) for name in files)
+            )
         if relative and any(name.endswith(".dist-info") for name in directories):
             # An installed tree, not the customer's code.
             directories[:] = []
@@ -928,9 +941,10 @@ def _source_files(
         if side is False:
             directories[:] = [name for name in directories if not _is_side_part(name)]
         if charge is not None and directories:
-            charge(
-                sum(2 * _SLOT + _text(os.path.join(directory, name)) for name in directories)
-            )
+            new = directories
+            if not first_pass:
+                new = [name for name in directories if _is_side_part(name)]
+            charge(sum(2 * _SLOT + _text(os.path.join(directory, name)) for name in new))
         folder = "/".join(relative)
         for name in files:
             if side is None or _read_with_side_files(_join(folder, name)) is side:
@@ -1473,13 +1487,18 @@ def walk_repo(
         logger.warning("agent walk: joining files into agents ran out of time")
         walk.facts.clear()
         return _empty_result(walk, "clock")
-    except (_ResultTooBig, _CapReached):
+    except _ResultTooBig:
         # Too many candidates reach each other to hold: the same end as the walk's other caps.
-        # The join's lookups all meet folders the walk already charged, so none charges again;
-        # if one ever did, past a held stop, it ends here too rather than in the crash path.
         logger.warning("agent walk: joining files into agents passed its edge budget")
         walk.facts.clear()
         return _empty_result(walk, "limit")
+    except _CapReached:
+        # The join's lookups all meet folders the walk already charged, so none charges again;
+        # if one ever did, past a held stop, it ends here rather than in the crash path. The
+        # rows found so far are dropped with the facts, unlike a stop during the walk.
+        logger.warning("agent walk: joining files into agents passed its memory budget")
+        walk.facts.clear()
+        return _empty_result(walk, "limit")
     logger.info(
         "agent walk: files_read=%d unread=%d reader_faults=%d agents=%d left_out=%d stopped=%s",
         walk.reader.files_read,
@@ -1542,9 +1561,9 @@ def _hazard(
         # folders below it.
         if index and clock() > deadline:
             return ""
-        # Sorted, so that a look stopped by a cap stops at the same folder on every file system.
+        # Folders sorted, so that a look stopped by a cap stops at the same folder on every file
+        # system. Files need no order: only their count and `agent.py`'s are used.
         directories.sort()
-        files.sort()
         directories[:] = [name for name in directories if name not in _BUILD_IGNORED]
         found += files.count("agent.py")
         if found > 1:

tests/harness/test_agent_scan.py

@@ -1536,6 +1536,54 @@ def test_the_clock_is_first_read_after_the_sources_own_folder(tmp_path) -> None:
     assert (word, folders) == ("clock", 1)
 
 
+def test_the_clock_is_read_inside_a_huge_folder(tmp_path) -> None:
+    """One folder of 5,000 empty folders, past the deadline: the look stops partway through that
+    folder's listing, not after reading all of it."""
+    source = tmp_path / "src"
+    folders = _empty_folders(source, [f"f{index:05}" for index in range(5_000)])
+    whole = _cost(str(source)) + sum(_cost(folder) for folder in folders)
+
+    word, read, held = agent_scan._look(source, deadline=0.0, clock=lambda: 1.0)
+
+    assert (word, read) == ("clock", 1)
+    # Read at the 4,096th entry, after 4,095 folders were taken on: where and how often.
+    assert held == (agent_scan._LOOK_CLOCK_EVERY - 1) * _cost(folders[0]) < whole
+
+
+def test_the_clock_is_read_inside_a_huge_folder_of_links(tmp_path, monkeypatch) -> None:
+    """Links are entries too: 5,000 dangling links past the deadline stop at the clock without
+    each being looked at."""
+    source = tmp_path / "src"
+    source.mkdir()
+    for index in range(5_000):
+        os.symlink("nowhere", source / f"l{index:05}")
+    calls = [0]
+    real = os.lstat
+
+    def counted(path, *args, **kwargs):
+        calls[0] += 1
+        return real(path, *args, **kwargs)
+
+    monkeypatch.setattr(agent_scan.os, "lstat", counted)
+
+    word, read, _ = agent_scan._look(source, deadline=0.0, clock=lambda: 1.0)
+
+    assert (word, read) == ("clock", 1)
+    assert calls[0] < agent_scan._LOOK_CLOCK_EVERY
+
+
+def test_the_clock_is_read_inside_a_huge_folder_past_the_bound(tmp_path, monkeypatch) -> None:
+    """Past the bound the read goes on for a regular file; the clock still stops it."""
+    source = tmp_path / "src"
+    folders = _empty_folders(source, [f"f{index:05}" for index in range(5_000)])
+    monkeypatch.setattr(agent_scan, "_LOOK_HELD", 50 * _cost(folders[0]))
+
+    word, read, held = agent_scan._look(source, deadline=0.0, clock=lambda: 1.0)
+
+    assert (word, read) == ("clock", 1)
+    assert held > agent_scan._LOOK_HELD
+
+
 def test_the_look_says_why_it_stopped_in_one_line_of_counts(tmp_path, monkeypatch, caplog) -> None:
     source = tmp_path / "src"
     folders = _empty_folders(source, [f"folder_{index:03}" for index in range(200)])

tests/harness/test_framework_detection.py

@@ -1756,14 +1756,21 @@ def _watch_the_stack(monkeypatch: pytest.MonkeyPatch) -> list[int]:
     def watching(*args: Any, **kwargs: Any) -> Any:
         walk = real_walk(*args, **kwargs)
         for item in walk:
-            stack = walk.gi_frame.f_locals.get("stack", []) if walk.gi_frame else []
+            # A missing `stack` fails loudly rather than reading nothing.
+            stack = walk.gi_frame.f_locals["stack"]
             held.append(sum(_allocated(sys.getsizeof(path)) + 8 for path in stack if isinstance(path, str)))
             yield item
 
     monkeypatch.setattr(fd.os, "walk", watching)
     return held
 
 
+# os.walk keeps its folders still to visit in a `stack` local only from 3.12; before, it recurses.
+_STACK_WALK = pytest.mark.skipif(
+    sys.version_info < (3, 12), reason="os.walk has a `stack` list to watch only from Python 3.12"
+)
+
+
 _STACK_CASES = {
     # top length (None: tmp_path itself), depth, siblings, chain stem, sibling stem
     # Sized so the whole list passes 2 MiB even under a Linux tmp_path of about 50 characters, and
@@ -1777,6 +1784,7 @@ def watching(*args: Any, **kwargs: Any) -> Any:
 }
 
 
+@_STACK_WALK
 @pytest.mark.parametrize("case", list(_STACK_CASES))
 def test_the_walks_own_list_never_passes_the_folder_bound(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch, case: str
@@ -1800,6 +1808,7 @@ def test_the_walks_own_list_never_passes_the_folder_bound(
     assert max(held) <= bound, (max(held), bound)
 
 
+@_STACK_WALK
 def test_the_walks_own_list_grows_past_the_test_bound_without_it(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None:

tests/harness/test_scan_walk.py

@@ -2297,3 +2297,123 @@ def charging(self: object, **kwargs: object) -> object:
     monkeypatch.setattr(scan_walk._Walk, "result", charging)
 
     assert _walk(tmp_path).stopped == "limit"
+
+
+def test_each_folder_is_charged_once_across_the_two_passes(tmp_path: Path) -> None:
+    """The side pass walks the customer's own folders again; only the folders the first pass
+    never entered (side folders, and folders below one) are charged there."""
+    for index in range(40):
+        (tmp_path / f"app{index:02}" / "inner").mkdir(parents=True)
+        (tmp_path / "tests" / f"t{index:02}" / "deep").mkdir(parents=True)
+        (tmp_path / f"app{index:02}" / "examples" / "e").mkdir(parents=True)
+        (tmp_path / f"app{index:02}" / ".hidden" / "x").mkdir(parents=True)
+    (tmp_path / ".github" / "workflows").mkdir(parents=True)
+    charged: dict[str, int] = {"own": 0, "side": 0, "once": 0}
+
+    def into(key: str):
+        def charge(size: int) -> None:
+            charged[key] += size
+
+        return charge
+
+    top = str(tmp_path)
+    for _ in scan_walk._source_files(top, top, side=False, charge=into("own")):
+        pass
+    for _ in scan_walk._source_files(top, top, side=True, charge=into("side")):
+        pass
+    for _ in scan_walk._source_files(top, top, charge=into("once")):
+        pass
+
+    assert charged["own"] > 0 and charged["side"] > 0
+    assert charged["own"] + charged["side"] == charged["once"], charged
+
+
+def test_the_estimates_sizes_were_measured_on_the_hosted_kits_python() -> None:
+    """The sizes are CPython 3.12's. On 3.14 a tuple's header is larger, so the estimate would be
+    low there. This fails when the hosted image's kit venv moves to another
+    Python, so the sizes are measured again before the bound is trusted."""
+    dockerfile = (Path(__file__).resolve().parents[2] / "Dockerfile.hosted").read_text()
+    major, minor = scan_walk._SIZES_MEASURED_ON
+    made = [
+        line
+        for line in dockerfile.splitlines()
+        if "/opt/alk-venv" in line and ("-m venv" in line or "uv venv" in line or "virtualenv" in line)
+    ]
+    assert len(made) == 1, made
+    assert f"python{major}.{minor} -m venv /opt/alk-venv" in made[0], made
+    assert made[0].count("-m venv") == 1, made
+
+
+def test_a_memory_budget_stop_while_joining_says_so(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A held-budget stop during the join ends at `limit`, with its own cause in the log, not the
+    edge budget's."""
+    _files(tmp_path, {"bot.py": CLIENT})
+
+    def reached(self, *, upload: bool):
+        raise scan_walk._CapReached()
+
+    monkeypatch.setattr(scan_walk._Walk, "result", reached)
+    kept: list[int] = []
+    real_empty = scan_walk._empty_result
+
+    def empty(walk, stopped):
+        kept.append(len(walk.facts))
+        return real_empty(walk, stopped)
+
+    monkeypatch.setattr(scan_walk, "_empty_result", empty)
+    with caplog.at_level("WARNING", logger=scan_walk.logger.name):
+        result = scan_walk.walk_repo(tmp_path, deadline=FOREVER, clock=lambda: 0.0)
+
+    assert result.stopped == "limit"
+    assert "passed its memory budget" in caplog.text
+    assert "edge budget" not in caplog.text
+    assert kept == [0]  # the facts are let go before the empty result is built
+
+
+@pytest.mark.parametrize("shape", ["one_entry", "two_entries"])
+def test_a_fan_out_join_is_charged_without_its_base(tmp_path: Path, shape: str) -> None:
+    """110 hubs of leaves that each import one shared module, about 22,000 files: the join walks
+    every leaf. One entry puts the import table just past its growth at 21,846 entries. Two
+    entries are dearer still: the first fills that table, then the second's own search grows its
+    seen set while the table stays full (about 59.5 bytes a file). What the join really holds is
+    covered by the per-file charge alone, without `_JOIN_BASE`, and only with `_JOIN_PER_FILE`
+    at 64: at 56 the two-entry shape fails. Leaves that import nothing share
+    CPython's empty tuple, which would hide the per-file cost."""
+    per_hub = 199 if shape == "one_entry" else 198
+    hubs = {
+        f"h{h:03}.py": "".join(f"import l{h:03}_{n:03}\n" for n in range(per_hub)) for h in range(110)
+    }
+    leaves = {f"l{h:03}_{n:03}.py": "import common\n" for h in range(110) for n in range(per_hub)}
+    entries = {"a.py": "".join(f"import h{h:03}\n" for h in range(110)) + CLIENT}
+    if shape == "two_entries":
+        entries["b.py"] = "".join(f"import h{h:03}\n" for h in range(100)) + CLIENT
+    _files(tmp_path, {**entries, "common.py": "x = 1\n", **hubs, **leaves})
+
+    seen = _measure(tmp_path)
+
+    assert seen["join_rise"] > 1_000_000, seen
+    assert seen["join_rise"] <= seen["join_charged"] - scan_walk._JOIN_BASE, seen
+
+
+@pytest.mark.parametrize("entries", ["folders", "files", "links"])
+def test_a_wide_folder_is_charged_its_listing(tmp_path: Path, entries: str) -> None:
+    """One folder of many entries: what `os.walk` holds for the folder it is in, its names, is
+    charged once when the walk reaches it, before any name is filtered out (links to a folder are
+    listed as folders and then left out), so the charge is never below the walk's peak."""
+    target = tmp_path / "target"
+    target.mkdir()
+    wide = tmp_path / "wide"
+    wide.mkdir()
+    for index in range(80_000 if entries == "files" else 30_000):
+        if entries == "folders":
+            (wide / f"f{index:05}").mkdir()
+        elif entries == "files":
+            (wide / f"f{index:05}.txt").touch()
+        else:
+            (wide / f"f{index:05}").symlink_to(target)
+
+    seen = _measure(tmp_path)
+
+    assert seen["walk_peak"] <= seen["charged"], seen