fix(operator-queue): ingestion depth/rate/size caps at the agent-authored sync seam (#1632)#1682
Open
AndriiPasternak31 wants to merge 10 commits into
Open
fix(operator-queue): ingestion depth/rate/size caps at the agent-authored sync seam (#1632)#1682AndriiPasternak31 wants to merge 10 commits into
AndriiPasternak31 wants to merge 10 commits into
Conversation
…#1632) Requirements-before-impl (Rule #1) for the operator-queue create-path rate/depth/size caps: per-agent DB depth cap (primary, Redis-independent), per-agent + fleet rate limits (fail-open), truncate-with-marker field hygiene, reserved-id guard, leader-locked sync, and the platform exemption. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
- count_pending_for_agent(agent) -> int: DB-measured per-agent pending depth, the Redis-independent primary bound for the ingestion cap (delegated through database.py). - create_item generous belt: reject title>4KiB / question>16KiB / context>64KiB / id>512 (ValueError the sync loop quarantines) — an order of magnitude past the service caps, so platform/clamped items never trip it (#1525 validate-at-boundary-AND-at-sink). - Fix non-dict context AttributeError in execution_id extraction (a str/list context used to raise and hot-loop the sync). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…#1632) Bound a hostile (not just runaway) agent at the one agent-authored create seam (_sync_agent): - DEPTH cap (primary, DB-measured => Redis-independent): stop ingesting once pending+admitted >= OPERATOR_QUEUE_MAX_PENDING_PER_AGENT (25). - RATE cap (burst smoothing, fail-open): per-agent + fleet rate_limiter.check at the create point; denied => hold + break. - Field hygiene (_clamp_ingested_item, total, inside the #1525 try/except): truncate-with-marker title/question; context/options over cap => marker (validated execution_id only); non-dict context => {}; created_at normalized to ingest time; priority validate-only. - Reserved-id guard: reject agent ids with a platform-reserved prefix so an agent can't pre-suppress its own flood alarm / the #1402 poison alert; malformed/oversize ids rejected. - Per-cycle scan bounds: skip an oversized file wholesale; cap scan at 500. - Leader lock (opqueue:leader, mirror monitoring #1464): only the leader worker syncs => no double-charge / double-broadcast / double-scan. - Flood alert: one per episode, un-guessable id, cooldown, emit-failure safe. - validation_service._notify_operator_on_failure => direct DB create (exempt by construction) + fixes the pre-existing bare-list schema bug. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
45 pure/mocked tests covering: depth cap (admit-to-cap-then-hold, no infinite drip, break bounds exists() probes); rate cap (allow-then-deny break, per-agent short-circuits fleet, one real in-process-fallback run); total clamp (truncate-with-marker, context/options markers, non-dict context, non-serializable, boundary lengths, never-raises, no-mutate); reserved/malformed id guard; flood alert (direct create, un-guessable id, cooldown, emit-failure-safe, oversize-file skip); leader lock (election, fail-open, own-lease release, non-leader poll no-op); DB belt + count helper. Registered in tests/registry.json. #1525 _fake_db: stub count_operator_queue_pending_for_agent → 0 so the new depth gate is a no-op (a MagicMock default misfired the >= cap check). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
- architecture.md: Operator Queue subsystem gains the ingestion-caps paragraph (depth/rate/fleet + reserved-id guard + leader lock + exemption + DB belt + flood alert); Background-Services row notes the leader-gated sync; Redis keys (opqueue:leader, operator_queue_create:*); rate_limiter reuse-consumer list updated. - feature-flows/operating-room.md: Sync Service Ingestion-caps section + poll-cycle leader gate + Agent Protocol field limits / held semantics / reserved ids / created_at normalization + queue-flood-* side effect. - feature-flows.md: Recent Updates row + Documented Flows row. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Add the 13 OPERATOR_QUEUE_* tunables to .env.example and wire them
through both docker-compose.yml and docker-compose.prod.yml with
${VAR:-default} so the .env levers actually reach the backend container
(the #1039 packaging discipline — an unplumbed lever is inert in prod).
Defaults match the code getenv defaults exactly (depth 25, rate 60/60s,
fleet 300, field caps). Mirrors the REPORT_RATE_LIMIT (#918) shape.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…ducer (#1632) _leader_ttl is refreshed once at the top of each _poll_cycle, so it must outlast one worst-case cycle (a slow _sync_agent read + write_file, ~10s) plus the inter-cycle sleep. The bare poll_interval*3 (15s at the 5s default) had ~zero headroom — a single slow-writing agent could expire the lease mid-cycle, a sibling would grab it, and leadership would flap, double-emitting the flood alert (its id carries utc_now_iso(), so on_conflict can't dedup it). Floor the TTL at 30s so one slow write can't drop leadership; the *3 rule still governs at larger poll intervals. Also pin the concrete platform producer: validation_service._notify_operator_on_failure was converted to a direct db.create_operator_queue_item (bypasses the agent-file sync caps, reserved `val_` id an agent can't pre-create, best-effort) — the new tests assert the direct create + reserved id + create-failure swallowing. Docs (architecture.md, operating-room.md) and the test registry updated for both. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
AndriiPasternak31
marked this pull request as ready for review
July 17, 2026 17:19
|
Resolve by running |
Resolve conflicts between the #1632 operator-queue ingestion caps and the #1631 (per-agent id/request_id split) + #1644 (extracted retention prune predicate) + #1616 (SSH TTL sweep) work that landed on dev. Code: - db/operator_queue.py: keep both the #1632 DB-sink belt caps AND the #1631 uuid-id / request_id split (insert + re-read, `(agent_name, request_id)` conflict target) + the #1644 `_operator_queue_prune_predicate`. Fixed a latent merge bug where the belt cap referenced the renamed `item_id`. - services/operator_queue_service.py: keep the #1632 caps (depth/rate/field hygiene/reserved-id/leader lock) layered on dev's #1631 API — adopt the tuple-keyed `_create_failures`, `_rejected_reserved`, two-arg `item_exists`/`mark_acknowledged` (uuid return), and the normalized reserved-prefix guard (using the #1632 superset prefix list). Dropped the now-redundant inline reserved-id check; guarded the reserved check against a non-string id so a hostile `"id": 123` can't AttributeError-crash the cycle. Docs: merged the Background Services + operating-room descriptions so both the #1631 id-scoping and the #1632 caps are documented; kept dev's #1616 Cleanup Service row; combined both Recent-Updates blocks. Tests: taught the #1632 stub-engine belt tests the #1631 insert+re-read return contract; neutralized the new depth/rate caps in the #1631 `_wire_sync` helper so those tests exercise id-scoping (caps covered by test_1632). All operator- queue unit tests green across 3 pytest-randomly seeds. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…e + test isolation (#1632) Two root causes surfaced by the CI backend-unit-test regression diff (10 new HEAD-only failures in test_1632, order-dependent across pytest-randomly seeds; not reproducible locally because macOS vs Linux collection order differs under the same seed). 1. PRODUCTION BUG — flood-alert cooldown suppressed the FIRST alert on a freshly-booted process. `_maybe_emit_flood_alert` defaulted "never alerted" to `0.0` and checked `now - last < COOLDOWN`. `time.monotonic()` is seconds from an arbitrary reference and can be < COOLDOWN (300s) right after boot, so `now - 0.0 < 300` was true and the first-ever alert for every agent was dropped for the first 5 minutes of backend uptime. Fixed with a `None` sentinel so the first alert always emits; the cooldown still gates repeats. (This is why the flood tests failed only when they happened to run within the first 300s of the CI runner's uptime — a real defect, not just test flake.) 2. TEST ISOLATION — the TestDbBelt engine stub used string-path `monkeypatch.setattr("db.operator_queue.get_engine", ...)`. Under a full-suite ordering where a prior test left a stand-in in sys.modules that re-exports the real class, the string patch lands on the stand-in while the real method reads the original module dict (module-identity gotcha) → the real engine runs. New `_patch_engine` helper patches the method's __globals__ via monkeypatch.setitem, immune to the swap. Verified against a simulated stand-in. Added a regression test pinning the low-monotonic case. Operator-queue suite green across all 3 CI seeds; ruff + sys.modules lint clean. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fixes #1632
Summary
The operator-queue create path ingested unbounded agent input at the
_sync_agentseam — no per-agent rate limit, no depth bound, no size caps, agent-chosenid/priority/created_at. #1402 makes this queue the approval channel for irreversible actions, so a compromised / prompt-injected agent that floods plausible "approve this" items causes operator fatigue → reflexive approval. Named residual on the pull-mode default-ON gate list (TARGET_ARCHITECTURE.md).This bounds a hostile agent (not just a runaway) with two independent bounds plus field hygiene, all at the one agent-authored seam.
What changed
count_operator_queue_pending_for_agentcomputed once/cycle; admit only whilepending + admitted < OPERATOR_QUEUE_MAX_PENDING_PER_AGENT(25). At the cap → stop ingesting (break, no drip) + hold surplus behind one aggregated alert.rate_limiter.check(60/60s, 300/60s)._clamp_ingested_item, total, inside the bug: operator-queue create_item KeyError on missing created_at causes an infinite sync-retry loop #1525 create try/except): truncate-with-marker title/question; context/options over cap → marker (validatedexecution_idonly); non-dict context →{};created_atnormalized to ingest time; priority validate-only.poison-alert.opqueue:leader, mirror monitoring bug: fleet-health monitoring loop runs once per uvicorn worker — duplicate health checks and doubled circuit-breaker failure feed #1464): single syncing worker under--workers 2— no double-charge / double-broadcast / double-scan. TTL floored at 30s so one slow-writing cycle can't flap leadership (the flood-alert id carriesutc_now_iso(), so a flap would double-emit).queue-flood-*alert per episode (cooldown, emit-failure-safe).validation_service._notify_operator_on_failure→ direct DB create: restores the "platform creates bypass_sync_agent" exemption by construction, and fixes a pre-existing latent bug (it appended to a bare list; the sync loop expects{"requests":[...]}, so the notification was never ingested).create_item(bug: operator-queue create_item KeyError on missing created_at causes an infinite sync-retry loop #1525 two-layer: validate at boundary AND sink) so the exemption invariant is never solely load-bearing.Scope note (please read at review)
complexity-medium, but scope grew to security-complete (depth + fleet + leader-lock + correctness) after a 3-reviewer pass (autoplan strategy+eng, independent security, Codex) unanimously found the literal-AC fix a speed-bump a rate-aware hostile agent defeats. Suggest a complexity re-label.create_item(additive belt) and read-onlyoperator_queue_item_exists— bug: operator-queue create path has no rate limit or size caps — flooding/operator-fatigue surface (blocking pull default-ON) #1632 does NOT re-key the PK to(agent_name, id); that stays bug: operator_queue.id is a global primary key — cross-agent id collision silently swallows requests #1631's. Second-merger rebases (~a few lines).Verification
tests/unit/test_1632_operator_queue_caps.py+test_1525_operator_queue_quarantine.py→ 52 passed (pure/mocked; depth cap, rate short-circuit, real in-process limiter, clamp totality/boundaries, reserved/malformed id, exemption, leader no-op, best-effort producer, DB belt + count helper)./validate-prclean: requirements (§26.7 OPS-001-CAPS) · architecture (OPS-001 caps + leader-lock + Redis keys + rate-limiter reuse) · feature-flow (operating-room.md+ both index rows) · security greps · config packaging (all 13OPERATOR_QUEUE_*env vars wired into both compose files +.env.example).Draft — not for merge yet (SOC-2 separation of duties; I'll flip to ready for an independent reviewer).
🤖 Generated with Claude Code