mesh(safety): mirror estop per-issuer fairness cap on resume cache (closes #267)

[cagataycali]

Summary

_on_safety_resume added entries to _resume_replay_cache unconditionally, while _on_safety_estop bounds each issuer to max(1, _resume_replay_cache_max() // 4) slots. A single wire_zid (or body issuer_id on an attribution-less transport) holding the override code could fill all cache slots and then churn legitimate other-issuer entries out via the eviction oldest-drop branch, suppressing real replay-rejection signals.

Fix

Mirror the estop per-issuer-cap pattern in the resume path using the identical expression (per_issuer_cap = max(1, _resume_replay_cache_max() // 4)), counting slots by the domain-tagged issuer_key. Over-cap resumes are refused (returning before clearing lockout -- the safe direction) and audited as resume_per_issuer_cap_exceeded.

Pin tests

• test_resume_cache_per_issuer_cap_enforced
• test_resume_cache_other_issuer_entries_not_evicted
• test_resume_and_estop_per_issuer_cap_use_same_expression (structural symmetry)

Cap + symmetry tests fail pre-fix, all pass post-fix.

Test output

$ hatch run python -m pytest tests/mesh/test_resume_replay.py -q
============================== 17 passed in 0.95s ==============================

closes #267

[yinsong1986]

Summary

Mirrors the per-issuer fairness cap from _on_safety_estop into _on_safety_resume so a single wire_zid (or body issuer_id on an attribution-less transport) holding the override code cannot fill _resume_replay_cache_max() slots and then churn legitimate other-issuer entries out via the 20%-oldest-drop eviction branch, suppressing real replay-rejection signals. The cap expression (max(1, _resume_replay_cache_max() // 4)) is duplicated verbatim from the estop site and pinned by a structural symmetry test.

The diff is small, surgical, well-justified, scoped to closing #267, and faithful to the canonical estop pattern (eviction first, cap check, refuse with audit-publish best-effort).

What's good

• Mirrors the canonical estop pattern exactly — same cap expression, same eviction-then-cap ordering, same narrow (TypeError, ValueError, OSError) audit-publish guard per AGENTS.md > "Exception Clauses Must Be Narrow".
• Three regression tests, including the structural symmetry pin — test_resume_and_estop_per_issuer_cap_use_same_expression is the right pattern for this kind of two-site invariant per AGENTS.md > Review Learnings (#86) > "Pin every reviewed fix with a regression test".
• Audits the over-cap rejection as resume_per_issuer_cap_exceeded so an operator dashboard can alert on flood attempts.
• Documents the asymmetric safety direction (refused estop still engages lockout; refused resume must NOT clear lockout) directly in the inline comment — makes the divergence intentional and readable.
• No public API or wire-format changes. The new audit event_type is additive in the audit-event taxonomy; no migration burden.

Must fix before merge

(none — PR is ready to merge once any follow-ups are tracked)

Follow-up in v0.4.1

• CHANGELOG entry missing. This is a security hardening fix (closes #267) and the Unreleased section already documents safety-relevant changes (e.g. the camera presign-TTL shrink in #228). Add a short bullet under an ### Added or ### Security heading citing #267 and the resume_per_issuer_cap_exceeded audit event so operators tracking safety-event taxonomies upgrade with intent.
• Test coverage gap: wire_zid-keyed issuer_key path is not exercised. All three new tests rely on _make_envelope which has no source_info, so issuer_key always lands in the ("body", issuer_id) branch (verified by the ("body", "op-flooder") / ("body", "op-A") / ("body", "op-B") assertions). A TLS-attributed flooder filling the cap via ("wire", <zid>) is the more realistic Summit-fleet threat model — add a test that mints a sample with a source_info.source_id.zid set and asserts the cap rejects against the wire-keyed slot. (tests/mesh/test_resume_replay.py:382-460.)
• Test coverage gap: cap-reclaim dynamic. The estop-site comment (core.py:1767-1774) explicitly claims "an attacker who flooded their cap and waited for eviction now has fewer entries" — this dynamic-rate-limit property has no resume-side test. Adding one (set STRANDS_MESH_RESUME_FRESHNESS_WINDOW_S low, sleep past TTL, assert the issuer can reclaim a slot) closes the parallel invariant.
• Symmetry pin is whitespace-fragile. test_resume_and_estop_per_issuer_cap_use_same_expression does an assert cap_expr in resume_src substring match. A future formatter run that wraps the line, adds a # noqa, or splits the expression silently passes/fails based on whitespace rather than semantics. Acceptable as-is for v0.4.0; consider an AST-based check (ast.parse(src) + walk for the Assign node) in v0.4.1.
• Test fixture style: Mesh.__new__(Mesh) + Mesh.__init__(...) is redundant. Mesh defines no __new__ override, so Mesh(robot, peer_id) produces the same object. Pre-existing in _make_mesh (not introduced by this PR), tracked as a fixture-cleanup follow-up.

Verification suggestions

hatch run python -m pytest tests/mesh/test_resume_replay.py -q
hatch run lint
# Spot-check: confirm the cap expression really is byte-identical at both sites
grep -n 'per_issuer_cap = max(1, _resume_replay_cache_max() // 4)' strands_robots/mesh/core.py
# Should print exactly two matches: one in _on_safety_estop, one in _on_safety_resume.

[yinsong1986]

Summary

Mirrors the per-issuer fairness cap that already lives in _on_safety_estop into _on_safety_resume, using the identical expression per_issuer_cap = max(1, _resume_replay_cache_max() // 4) so the two replay-cache fairness defenses cannot drift. Counts owned slots by the domain-tagged issuer_key (matches the resume cache's (issuer_key, proof_nonce) key shape), refuses over-cap resumes (returning before clearing lockout — the safe direction, as called out in the description), and audits the rejection as resume_per_issuer_cap_exceeded. Closes #267.

The fix is symmetric with the estop site: eviction first, then per-issuer count, then conditional add — same ordering, same lock discipline (_resume_replay_lock held throughout), same narrow (TypeError, ValueError, OSError) audit-publish exception tuple per AGENTS.md > "Exception Clauses Must Be Narrow".

What's good

• Identical expression for per_issuer_cap on both sites; structural symmetry test (test_resume_and_estop_per_issuer_cap_use_same_expression) pins it so a future drift fails CI.
• Correct key-shape arithmetic: sum(1 for k in self._resume_replay_cache if k[0] == issuer_key) matches the documented cache shape dict[tuple[tuple[str, str], str], float] (line 341).
• Cap check sits inside self._resume_replay_lock — concurrent floods cannot race the count.
• Variables issuer_id (validated non-empty str at line 2014) and proof_nonce (validated str at line 1953) are already proven safe before the new proof_nonce[:16] slice runs — no NoneType crash on the new path.
• Comment block explicitly explains why the over-cap path returns before the lockout-clear branch ("a refused resume must NOT clear lockout"). Good safety-direction reasoning.
• Two behavioral tests + one structural pin, all asserted to fail pre-fix and pass post-fix.

Must fix before merge

(none — PR is ready to merge once the follow-ups are tracked as v0.4.1 issues)

Follow-up in v0.4.1

• Test gap on the safety-direction invariant (tests/mesh/test_resume_replay.py, around L382-L406). The PR description specifically calls out "a refused resume must NOT clear lockout — returning here is the safe direction," but no test asserts m._estop_lockout.is_set() is True after the over-cap rejection. Easy regression target — add an explicit assertion in test_resume_cache_per_issuer_cap_enforced.
• Symmetry-pin brittleness (tests/mesh/test_resume_replay.py, L454-L460). The structural pin is a literal substring match on source text. Any cosmetic refactor (extracting per_issuer_cap into a _per_issuer_cap() helper, breaking the line at 100 cols, hoisting it to a module constant) silently fails the test even though the invariant still holds. Consider either (a) extracting a shared _per_issuer_cap() helper called from both sites and asserting both sources contain the call, or (b) AST-based comparison.
• Audit publish inside the replay lock (strands_robots/mesh/core.py, L2148-L2162 of the new block). publish_safety_event is called while holding self._resume_replay_lock. Symmetric with the estop site (pre-existing pattern, not a regression here), but if publish_safety_event ever grows synchronous I/O, both sites stall the safety path under load. Worth tracking as a v0.4.x concurrency-hygiene item alongside the estop site.

Verification suggestions

hatch run python -m pytest tests/mesh/test_resume_replay.py -q
# Spot-check the symmetry pin actually catches drift:
# temporarily edit one site's cap expression to `// 5` and confirm
# `test_resume_and_estop_per_issuer_cap_use_same_expression` fails.

[yinsong1986]

[FOLLOW-UP] The PR description correctly justifies why this return lands here ("a refused resume must NOT clear lockout — returning here is the safe direction"), and the test test_resume_cache_per_issuer_cap_enforced exercises the over-cap path, but no test asserts the consequence: m._estop_lockout.is_set() is True after an over-cap resume. That is the very invariant a future refactor could most easily break (e.g. someone moving the lockout-clear above the cap check thinking it's harmless). Suggest adding assert m._estop_lockout.is_set() is True after the loop in test_resume_cache_per_issuer_cap_enforced to pin it. Not blocking — the path is correct as written.

[yinsong1986]

[FOLLOW-UP] The structural pin via literal substring match is brittle: if a future cleanup extracts per_issuer_cap = max(1, _resume_replay_cache_max() // 4) into a _per_issuer_cap() helper called from both sites, or hoists the divisor into a module-level constant, this test fails even though the invariant ("both sites use the same cap expression") still holds. Per AGENTS.md > "Pin regression tests for reviewed fixes" the goal is to catch the regression, not the textual form. Two cleaner options for v0.4.1: (a) extract a shared _per_issuer_cap() helper and assert both inspect.getsource outputs contain the call, or (b) walk the function ASTs and compare the cap subtree. Not blocking for v0.4.0 — current pin is strictly better than no pin.