Implementing Dynamic Priority Scoring for Urgent Municipal Requests

Within Priority Scoring Algorithms, the single highest-leverage task is teaching the intake engine to recognize a genuinely urgent public-records request — an expedited-processing demand, an imminent-threat-to-life inquiry, a journalist on a same-day deadline — and lift it above the routine administrative backlog before a static routing table buries it. This page shows how to build that scorer so it is deterministic, fully auditable, and defensible if the agency’s prioritization is ever challenged in court.

Scenario & Compliance Stakes

A city’s records inbox is processing a normal Tuesday of permit-copy requests when a reporter files for “all communications between the city manager and the developer of parcel 0451-22, requesting expedited processing under imminent public-interest urgency.” Under 5 U.S.C. § 552(a)(6)(E), a properly substantiated expedited-processing request must get a determination within ten calendar days — a far tighter clock than the 20-business-day default in § 552(a)(6)(A)(i). If a static, first-in-first-out queue treats that filing like a routine permit copy, the agency blows the expedited window and hands the requester grounds to litigate.

The opposite failure is just as costly. If the scorer fires on the word “emergency” sitting in an email signature or a legal disclaimer, routine requests flood the urgent lane, real deadlines get starved, and the prioritization looks arbitrary — exactly what a court probes when reviewing whether an agency’s processing was non-arbitrary. A defensible scorer therefore has to do two things at once: extract urgency signals from bounded, sanitized text so it cannot be tricked by boilerplate, and emit an audit trail that shows precisely which weighted signals produced each priority decision. That decision then feeds the rest of Intake & Routing Workflows — queue placement and department assignment all inherit the tier this stage assigns.

Prerequisites

  • Python 3.11+ for the dataclasses slots, structured exception handling, and decimal arithmetic used below.
  • Standard library only for the core scorer — re, html, hashlib, decimal, json, logging, os. No third-party dependency is needed to score; keeping the engine dependency-free shrinks the audit surface.
  • Normalized intake payloads already produced upstream by your Email & Form Parsing Pipelines — multipart MIME, web-form JSON, and fax-to-email all reduced to a single canonical text field plus a sender address.
  • Externalized scoring weights — environment variables or a config service, never hardcoded constants — so a weight change is a reviewable, version-controlled event.
  • Write access to an append-only audit store (a WORM bucket or hash-chained log) so every scoring decision is preserved immutably.
  • Least-privilege boundaries from your Security Boundary Configuration enforced before the scorer ever sees requester PII.

Implementation

The scorer is built in two stateless stages: bounded signal extraction, then deterministic weighted scoring against a compliance threshold. Keeping both stages free of side effects is what makes the result reproducible — feeding the same payload through twice must yield byte-identical output, or the audit trail is worthless.

Start with signal extraction. The two non-negotiable defenses here are a hard length cap (so email signatures and quoted reply chains never reach the matcher) and a pre-compiled, anchored pattern (so a hostile payload cannot trigger catastrophic regex backtracking).

python 
import re
import html
import hashlib
import logging
from dataclasses import dataclass, field

logger = logging.getLogger("municipal.intake.scoring")

@dataclass(frozen=True, slots=True)
class RequestSignal:
    correlation_id: str
    raw_text_hash: str                      # SHA-256, stable across worker restarts
    is_expedited_flagged: bool = False
    contains_statutory_deadline: bool = False
    sender_is_press_or_advocacy: bool = False
    emergency_keywords: list[str] = field(default_factory=list)
    confidence_score: float = 0.0

# Compile once at module load: prevents ReDoS recompilation and is thread-safe.
STATUTORY_PATTERNS = re.compile(
    r"(?:expedited|emergency|imminent\s+threat|life|safety|health|statutory\s+deadline)",
    re.IGNORECASE | re.MULTILINE,
)
PRESS_DOMAINS = {"press", "news", "journal", "media", "ap.org", "reuters.com", "propublica.org"}
MAX_EVALUATION_LENGTH = 1200    # ignore footers/signatures; bounds regex input (anti-ReDoS)

def extract_signals(raw_payload: str, sender_email: str, correlation_id: str) -> RequestSignal:
    """Extract bounded urgency signals from a normalized intake payload.

    Stateless and idempotent so the same payload always yields the same hash and
    signals — a precondition for the audit trail relied on under 5 U.S.C.
    552(a)(6)(E) (expedited-processing determinations must be non-arbitrary).
    """
    try:
        # 1. Truncate BEFORE unescaping so the matcher never sees signature boilerplate.
        cleaned = html.unescape(raw_payload[:MAX_EVALUATION_LENGTH])
    except (TypeError, ValueError) as exc:
        logger.warning('{"event":"decode_failed","correlation_id":"%s","err":"%s"}',
                       correlation_id, exc)
        cleaned = ""

    matches = STATUTORY_PATTERNS.findall(cleaned)               # 2. anchored, pre-compiled
    domain = sender_email.split("@")[-1].lower() if "@" in sender_email else ""
    is_press = any(d in domain for d in PRESS_DOMAINS)

    # 3. Heuristic confidence: weighted term density + sender classification, capped at 1.0.
    confidence = min(1.0, (len(matches) * 0.3) + (0.4 if is_press else 0.0))

    return RequestSignal(
        correlation_id=correlation_id,
        # 4. SHA-256, not built-in hash(): hash() is randomized per process and is NOT audit-safe.
        raw_text_hash=hashlib.sha256(cleaned.encode("utf-8")).hexdigest(),
        is_expedited_flagged=bool(matches),
        contains_statutory_deadline="statutory deadline" in cleaned.lower(),
        sender_is_press_or_advocacy=is_press,
        emergency_keywords=[m.lower() for m in matches],
        confidence_score=round(confidence, 3),
    )

The scoring stage turns those signals into a number and a tier. Weights load from the environment so a policy change is a reviewable config commit, scoring uses Decimal so the result is exact and reproducible (never a float rounding surprise), and every applied weight is recorded in an audit_trail list before the function returns.

python 
import os
import json
from decimal import Decimal, ROUND_HALF_UP

# Externalized weights — a change is a version-controlled, reviewable event, not a code edit.
WEIGHT_EXPEDITED_FLAG = Decimal(os.getenv("WEIGHT_EXPEDITED_FLAG", "40"))
WEIGHT_PRESS         = Decimal(os.getenv("WEIGHT_PRESS", "15"))
WEIGHT_DEADLINE      = Decimal(os.getenv("WEIGHT_DEADLINE", "25"))
WEIGHT_EMERGENCY     = Decimal(os.getenv("WEIGHT_EMERGENCY", "20"))
COMPLIANCE_THRESHOLD = int(os.getenv("COMPLIANCE_THRESHOLD", "65"))

def calculate_priority_score(signal: RequestSignal) -> dict:
    """Return a deterministic score, tier, and full audit trail for one request.

    Every weight and input is logged so a compliance officer can reconstruct
    exactly why a request was placed in the URGENT lane — the record that
    demonstrates non-arbitrary expedited-processing decisions to a reviewing court.
    """
    score = Decimal("0")
    audit_trail: list[str] = []

    if signal.is_expedited_flagged:
        score += WEIGHT_EXPEDITED_FLAG
        audit_trail.append("expedited_flag_applied")
    if signal.sender_is_press_or_advocacy:
        score += WEIGHT_PRESS
        audit_trail.append("press_domain_applied")
    if signal.contains_statutory_deadline:
        score += WEIGHT_DEADLINE
        audit_trail.append("statutory_deadline_applied")
    if signal.emergency_keywords:
        score += WEIGHT_EMERGENCY
        audit_trail.append(f"emergency_keywords:{len(signal.emergency_keywords)}")

    # Cap at 100 so no single request can monopolize the queue and starve others.
    final = min(100, int(score.quantize(Decimal("1"), rounding=ROUND_HALF_UP)))
    tier = "URGENT" if final >= COMPLIANCE_THRESHOLD else "STANDARD"

    result = {
        "event": "priority_scored",
        "correlation_id": signal.correlation_id,
        "raw_text_hash": signal.raw_text_hash,
        "priority_score": final,
        "priority_tier": tier,
        "threshold": COMPLIANCE_THRESHOLD,
        "audit_trail": audit_trail,
        "requires_manual_review": final >= 90,
    }
    logger.info(json.dumps(result))     # append-only JSON line = the litigation-ready record
    return result

Once a request carries a tier, it flows into the rest of the pipeline. URGENT payloads bypass the rate-limited default workers in your Async Queue Management layer and route to dedicated compliance handlers, while Department Routing Logic maps the tier to jurisdictional ownership. Because jurisdictions disagree on what counts as urgent — a California Public Records Act 10-day posture differs from a federal baseline — keep the deadline-derived weights in your State Law Compliance Frameworks config rather than in this module.

Priority scoring flow from normalized payload to queue arbitration A normalized intake payload feeds bounded urgency-signal extraction, which feeds a weighted additive Decimal score loaded from environment weights. The score meets a compliance-threshold decision: at or above the threshold of 65 the request takes the URGENT tier, below it the STANDARD tier. A STANDARD request routes to rate-limited default workers. An URGENT request meets a second decision: at or above 90 it routes to a manual review queue for human sign-off, otherwise straight to dedicated compliance handlers serving the statutory ten-day clock. Normalized payload single text field + sender Extract urgency signals bounded, sanitized text Weighted additive score env weights · exact Decimal Score ≥ threshold (65)? STANDARD tier routine backlog Rate-limited default workers async queue URGENT tier expedited lane Score ≥ 90? Dedicated compliance handlers statutory 10-day clock Manual review queue human sign-off no yes no yes

Expected Output & Verification

Scoring the reporter’s expedited filing — press domain, an explicit expedited demand, and an imminent-threat phrase — emits a single JSON audit line:

json 
{"event": "priority_scored", "correlation_id": "req-0451-22", "raw_text_hash": "a91f...c2", "priority_score": 75, "priority_tier": "URGENT", "threshold": 65, "audit_trail": ["expedited_flag_applied", "press_domain_applied", "emergency_keywords:2"], "requires_manual_review": false}

Verify three invariants before trusting the scorer in production. First, determinism: scoring the same payload twice must return an identical raw_text_hash and priority_score — if it does not, something non-deterministic (a stray hash(), a wall-clock input) has leaked in. Second, threshold correctness: a payload with only a press-domain match (15 points) must stay STANDARD, while expedited + deadline (65) must tip to URGENT. Third, audit completeness: every applied weight appears in audit_trail. A focused test pins all three:

python 
def test_scorer_is_deterministic_and_auditable():
    sig = extract_signals("Requesting expedited processing — imminent threat to safety.",
                          "newsroom@ap.org", "req-0451-22")
    a = calculate_priority_score(sig)
    b = calculate_priority_score(sig)
    assert a == b                                   # determinism: identical re-run
    assert a["priority_tier"] == "URGENT"           # 40 + 15 + 20 = 75 >= 65
    assert "expedited_flag_applied" in a["audit_trail"]   # weight is provable

Common Pitfalls

  • ReDoS and boilerplate false positives on unbounded text. Feeding the whole email body to the matcher lets a 50-line legal disclaimer trigger “emergency” and lets a crafted payload cause catastrophic backtracking. Always truncate to MAX_EVALUATION_LENGTH first and strip > quoted reply chains before matching, so the scorer only ever reads the requester’s actual ask.
  • Using Python’s built-in hash() for the audit hash. hash() is salted per process, so the same text hashes differently after a worker restart and your reconciliation breaks. Use hashlib.sha256 (as above) so the hash is stable and the audit trail survives restarts.
  • Float arithmetic and hardcoded weights. Summing float weights can produce 64.99999 where you expected 65, flipping the tier non-deterministically; and hardcoding weights hides policy changes from review. Use Decimal with explicit ROUND_HALF_UP and load every weight from the environment so each change is reviewable and the math is exact.

Frequently Asked Questions

How do I keep the scorer's prioritization defensible if it is challenged in court?

Make every decision reconstructable. Each scoring call emits one append-only JSON line carrying the correlation_id, the SHA-256 payload hash, the exact weights that fired (audit_trail), the final score, and the threshold in force at the time. Because the weights load from version-controlled config, you can show which policy was active on any date and reproduce the score from the preserved payload hash — that reproducibility is what demonstrates the prioritization was rule-based, not arbitrary, under 5 U.S.C. § 552(a)(6)(E).

Where should the URGENT threshold sit, and how do I tune it without starving the queue?

Start the COMPLIANCE_THRESHOLD so that a single weak signal (a press domain alone) stays STANDARD while a substantiated expedited request clears it — 65 against the default weights does this. Then watch the ratio of URGENT to total volume. If the urgent lane swells past what your compliance handlers can clear inside the statutory window, the threshold is too low and routine work is starving real deadlines; raise it via config hot-reload rather than editing code, and confirm the change in the audit log.

Should the scorer ever auto-route an urgent request without human review?

Auto-route the clear cases, escalate the extreme ones. The requires_manual_review flag fires at a score of 90, sending the most consequential requests to a person before they leave the intake stage, while mid-range URGENT requests flow straight to dedicated compliance handlers. That split keeps throughput high without letting the engine make an unreviewed call on a request severe enough to carry real legal exposure.

← Back to Intake & Routing Workflows