Story 2873: compaction orchestration — trigger + queue + fold persist

Story-2873 (TICKET-MAP T2i-a, RFD-AGT-013 D17, RFD-memory-engine R-3) builds the broker half of automatic memory compaction. When a chat session's ACTIVE memory (turns past the clear watermark, or their estimated tokens) crosses a configured threshold, the clerk worker enqueues a kind='compact' command onto the same agent_invoke SQS rail it consumes. The compactor agent (context_judge, built in pramaan-spec #2874) folds (prior fold_text + active transcript) and emits memory.folded; the worker persists the fold into agents.session_memory and bumps memory_version in one transaction. The transcript (agents.chat_turn) and chat_session.summary are never touched. This ticket is ORCHESTRATION ONLY (pramaan-docs ADR-166 / AGENTS.md R11): no model call, no prompt in the worker — agents think, lambdas move.

What changed

SurfaceChange
src/shared/agent_contracts/invoke.py TurnKind gains "compact": the memory-compaction route. A compact invoke carries the broker-assembled compaction material in input (target scope, prior fold, active turns, token cap).
src/shared/agent_contracts/events.py New registered stream event memory.folded (fold_text: str, fold_token_count: int | None) — the compactor's typed fold effect, parsed by the same canonical boundary as every other brain event.
src/shared/agent_commands.py Command matrix for a compact launch: agent_key='context_judge' (rail routing), mandatory run_id/chat_session_id, exactly one of matter_id/form_id, and NO connection_id/message/history — a broker-initiated maintenance turn, never a user turn. The TARGET memory scope travels in input.
src/clerk_worker/runner.py Trigger: after a converse turn's transcript rows land, _maybe_enqueue_compaction evaluates turn_count - cleared_until_turn_index and the active-bytes/4 token estimate against settings thresholds, reserves a deterministic request_id (compact:{firm}:{target_agent}:{matter|form}:{user}:v{memory_version}) through the same agent_run_key idempotency insert the launch path uses, and enqueues one compact command. FAIL-OPEN: any trigger error logs a warning and never breaks the user's turn. Consumer: process_command routes kind='compact' to _process_compact_command — claim (redrive-safe), assemble the material from the DB, invoke the context_judge runtime, persist memory.folded in ONE transaction (UPDATE-first; a scope with no row INSERTs at version 2), honest failed stamp on any gap (R5).
src/shared/settings.py memory_fold_min_tokens (2000), memory_fold_min_turns (20), memory_cap_tokens (1200) — the RFD design values, tunable via env without a deploy (pramaan-docs ADR-151).
Registry + rail AGENT_CONTEXT_JUDGE = "context-judge" joins the SSM registry roster; context_judge joins the clerk-rail key set. The WS launch edge stays CLOSED to the compactor key — only the worker's trigger mints compact commands.

The no-double-compaction guarantee

The request_id embeds the scope's CURRENT memory_version. While the version is N, at most one compaction run for that scope version exists: SQS redrives and concurrent triggers lose the agent_run_key reserve race. When the fold lands, memory_version bumps to N+1, so the next threshold crossing mints a fresh request_id. Until pramaan-spec #2874 deploys the compactor and publishes its registry entry, resolution fails closed and a compaction run stamps failed honestly — bounded to one failed run per memory_version.

Before and after

PathBeforeAfter
Long active session No automatic fold; the durable memory block grew without bound. Threshold crossing enqueues one compact run on the clerk rail.
Fold persistence No writer for session_memory.fold_text from a model fold. memory.folded → one-transaction fold persist + memory_version bump; transcript untouched.
Compactor absent (#2874 not deployed) — Fail-closed resolution; the compact run stamps failed with a diagnostic (R5), never a silent settle.

Doctrine held

Verification

uv run ruff check <changed files>
# All checks passed! (repo-wide count unchanged vs main: 121 pre-existing)

.venv/bin/python -m pytest tests/unit/clerk_worker/ tests/unit/test_agent_launch_core.py \
  tests/unit/test_memory_version_bump.py tests/unit/test_memory_pack_assembly.py \
  tests/unit/test_agent_commands.py tests/unit/agent_contracts/ tests/unit/test_sqs_span_naming_audit.py -q
# 373 passed

# Red-first: the same tests against pre-change source —
# 14 failed + collection ImportError on the new runner test module.

make openapi
# openapi/app-api.json regenerated: the AgentKey enum gains context_judge,
# the turn Kind enum gains compact (the two shared vocabulary additions);
# no router or DTO shape change.

# DB-backed suites (make test / integration) NOT run locally — unit-only change;
# the agent_key CHECK widening (agent_run_agent_key_valid et al., the 0126/0145
# pattern) rides the pramaan-spec #2874 chain that deploys the compactor.