Escalating Unresolvable Conflicts to Operators #

A conflict landed in your review queue eleven hours ago, it is blocking a customer’s order from settling, and nobody has looked at it — because “pending in a database” is not the same as “someone was told.” A queue holds work; it does not summon a human. This page builds the policy layer that does: an escalation engine that scans review tasks, decides which ones deserve a human’s attention right now based on age against an SLA and the confidence your resolver reported, and drives them up a chain — tier 1, then tier 2, then the on-call pager — with de-duplicated, signed notifications. It sits directly above Manual Review Sync Queues and turns the tasks produced by Building a Conflict Review Queue in Python into paged, tracked incidents that actually get resolved.

Immediate Triage & Prerequisites #

Before wiring notifications, find out whether your queue already has aged tasks that nobody escalated — those are the incidents your customers are feeling right now. List pending tasks by age:

curl -s "http://localhost:5984/conflict_reviews/_all_docs?include_docs=true" | \
  python3 -c "import sys,json,time;now=time.time();\
[print(r['doc']['_id'], r['doc'].get('state'), r['doc'].get('created_at')) \
 for r in json.load(sys.stdin)['rows'] if r['doc'].get('state')=='pending']"

If the oldest created_at is hours in the past, you have an escalation gap, not a merge gap. Prerequisites: Python 3.8+ with requests, the review database from Building a Conflict Review Queue in Python, and a webhook receiver you control. Decide two numbers up front: a confidence threshold below which an auto-resolver’s guess is never trusted (say 0.8), and an SLA budget per tier (say 15 minutes at tier 1, 30 at tier 2). Both belong in config, not code, because you will tune them against real mean-time-to-resolution. This page governs escalation policy; the resolvers whose confidence it reads are the fallthrough links in Designing Fallback Resolution Chains in Python.

Step-by-Step Implementation #

Each step ends with a check you can run before moving on.

1. Attach confidence and an SLA deadline at enqueue. When a task is created, record the resolver’s confidence (or 0.0 if no resolver ran) and a sla_deadline timestamp. Verify a fresh task carries both fields; a task without them cannot be scheduled.

2. Scan for escalation candidates. On a fixed interval, query the review database for tasks in state pending or escalated and evaluate two rules per task: confidence below threshold, or now > sla_deadline. Verify the scan selects only actionable states — never acknowledged or resolved.

3. Compute the target tier from elapsed budgets. Map how many SLA budgets the task has burned to a tier index: zero breaches stay at tier 1, one breach climbs to tier 2, two or more reach on-call. Assert the tier only ever increases, so a task cannot silently de-escalate.

4. Notify through the de-duplication gate. Before sending, compare the task’s last_alert_tier to the target tier. Emit a signed notification only when the tier changed; otherwise suppress it. Verify that running the scan twice with no clock change produces exactly one alert.

5. Record acknowledgement and resolution. An operator acknowledging a task sets state acknowledged and an owner, which removes it from the escalation scan. Completing the merge (via resolve_task) sets resolved. Verify an acknowledged task never re-alerts even as time passes.

Complete Working Example #

The engine below scans the review database, applies the age and confidence rules, climbs the tier ladder, and emits HMAC-signed notifications with per-tier de-duplication. Notification transport is abstracted behind send_notification() so email, webhook, and pager share one signing path.

import hashlib
import hmac
import json
import sys
import time

import requests

REVIEW = "http://localhost:5984/conflict_reviews"
SIGNING_KEY = b"rotate-me-via-secret-manager"   # HMAC key for signed alerts
CONFIDENCE_THRESHOLD = 0.8                        # below this, always escalate
SLA_SECONDS = [15 * 60, 30 * 60]                  # tier 1 budget, tier 2 budget
TIERS = ["tier1-rota", "tier2-lead", "oncall"]    # ladder, low to high


def _now() -> float:
    return time.time()


def target_tier(task, now: float) -> int:
    """How far up the ladder this task belongs, from elapsed SLA budgets."""
    age = now - _parse(task["created_at"])
    burned, elapsed = 0, 0.0
    for budget in SLA_SECONDS:
        elapsed += budget
        if age > elapsed:
            burned += 1
    # Low resolver confidence starts one rung up even before any SLA breach.
    if task.get("confidence", 0.0) < CONFIDENCE_THRESHOLD:
        burned = max(burned, 1)
    return min(burned, len(TIERS) - 1)


def _parse(ts: str) -> float:
    return time.mktime(time.strptime(ts, "%Y-%m-%dT%H:%M:%SZ"))


def sign(payload: dict) -> str:
    """Detached HMAC-SHA256 over the canonical JSON, so receivers can verify."""
    body = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
    return hmac.new(SIGNING_KEY, body, hashlib.sha256).hexdigest()


def send_notification(tier: str, task: dict):
    """Transport stub: route by tier. Returns True if actually dispatched."""
    payload = {"task_id": task["_id"], "doc_id": task["doc_id"],
               "tier": tier, "diffs": task.get("diffs", {}),
               "sent_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}
    signature = sign(payload)
    # In production: email for tier1, POST signed webhook for tier2, pager API
    # for oncall. Here we just prove the signature travels with the payload.
    print(f"NOTIFY {tier}: {task['_id']} sig={signature[:12]}…")
    return True


def scan_and_escalate(session: requests.Session):
    now = _now()
    rows = session.get(f"{REVIEW}/_all_docs", params={"include_docs": "true"},
                       timeout=30).json()["rows"]
    escalated = 0
    for row in rows:
        task = row["doc"]
        if task.get("type") != "conflict_review":
            continue
        if task.get("state") not in ("pending", "escalated"):
            continue  # acknowledged/resolved tasks are out of scope
        tier_idx = target_tier(task, now)
        tier = TIERS[tier_idx]
        # De-duplication: only alert when the rung actually changes.
        if task.get("last_alert_tier") == tier:
            continue
        if send_notification(tier, task):
            task["state"] = "escalated"
            task["current_tier"] = tier
            task["last_alert_tier"] = tier
            task.setdefault("escalation_log", []).append(
                {"tier": tier, "at": time.strftime(
                    "%Y-%m-%dT%H:%M:%SZ", time.gmtime())})
            session.put(f"{REVIEW}/{task['_id']}", json=task, timeout=30)
            escalated += 1
    return escalated


def acknowledge(session: requests.Session, task_id: str, owner: str):
    """Operator claims a task: halts escalation without resolving it yet."""
    task = session.get(f"{REVIEW}/{task_id}", timeout=30).json()
    if task["state"] == "resolved":
        return
    task["state"] = "acknowledged"
    task["owner"] = owner
    task["acknowledged_at"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    session.put(f"{REVIEW}/{task_id}", json=task, timeout=30)
    print(f"{task_id} acknowledged by {owner}; escalation stops")


if __name__ == "__main__":
    s = requests.Session()
    moved = scan_and_escalate(s)
    print(f"escalated {moved} task(s)")
    # Second pass with no clock change must escalate zero: dedup holds.
    assert scan_and_escalate(s) == 0, "de-duplication failed"
    print("dedup verified: repeat scan produced no new alerts")
    sys.exit(0)

Run it against a review database with an aged, low-confidence task and it fires exactly one notification, records the tier on the task, and the assertion proves a second immediate pass stays silent — the de-duplication gate is doing its job.

Gotchas & Edge Cases #

• Alert fatigue kills escalation systems. If every scan re-notifies, operators mute the channel and real incidents die in a muted inbox. The last_alert_tier field is the whole defense: emit once per rung, and only re-alert when the task climbs. Never notify on a task already acknowledged.
• Escalation loops are a design bug, not bad luck. A task that oscillates between tiers because a clock or budget comparison is off will page on-call forever. Make the tier monotonic — it may only increase — and terminate the ladder at a top rung that pages once, not repeatedly.
• On-call rotation and timezones are not optional. “Notify on-call” must resolve through your rotation schedule at send time, in the operator’s timezone, or you will page someone asleep and off-shift. Compute sla_deadline in UTC and let the rotation layer localize; never hardcode a person.
• An escalated conflict is not a resolved one. Paging closes the notification loop, not the conflict loop. The task stays open until a merge commits via the queue’s resolve_task. Track both: time-to-acknowledge and time-to-resolve are different numbers.
• Measure MTTR or you are flying blind. Stamp created_at, acknowledged_at, and the resolution time, and derive mean-time-to-resolution per tier. Rising MTTR at a tier means that rung is understaffed — feed the number into Conflict-Resolution Metrics & SLOs for CouchDB Sync.

Verification & Observability #

Confirm the policy behaves: aged tasks escalate, acknowledged ones fall silent, and each rung alerts once. Inspect the escalation log a task accumulates:

curl -s "http://localhost:5984/conflict_reviews/task:abc123" | \
  python3 -c "import sys,json;t=json.load(sys.stdin);print('state:',t['state'],'tier:',t.get('current_tier'));print('log:',t.get('escalation_log'))"

For fleet-wide health, break tasks down by state and current tier so you can see where work is piling up:

curl -s "http://localhost:5984/conflict_reviews/_all_docs?include_docs=true" | \
  python3 -c "import sys,json,collections;\
c=collections.Counter((r['doc'].get('state'),r['doc'].get('current_tier')) for r in json.load(sys.stdin)['rows'] if r['doc'].get('type')=='conflict_review');\
print(dict(c))"

Healthy output shows few tasks stuck at oncall and a steady flow into acknowledged then resolved. A pileup of tasks parked at tier 2 means tier 1 never acted — the SLA is too loose or the rota is understaffed. Emit the count of tasks per tier and the age of the oldest un-acknowledged task as gauges, and alert when the on-call tier is non-empty for longer than one rotation.

FAQ #

Related #

• Building a Conflict Review Queue in Python
• Designing Fallback Resolution Chains in Python
• Conflict-Resolution Metrics & SLOs for CouchDB Sync

Part of: Manual Review Sync Queues