incidents: severity-scaled quiet timer, reopen-on-link, thin calls don't fill the cap

Hand-labelling the 09-22 10:00-12:00 ET replay window (server-26#170,
answer key replay_groundtruth_0922.json) found ~25 real incidents, of which
only ~5 had an audible clear — most jobs clear by MDT, so the quiet timer is
the close for most incidents and a flat 90 minutes left a lockout or a plate
check "active" on the portal an hour after it ended.

- summarizer: timer close after 30 min quiet for routine/minor, 60 moderate,
  90 major/unknown. A timer close is provisional: reopenable=True.
- correlator: reopenable incidents inside incident_reopen_window_minutes
  (90, since last substantive call) stay candidates; linking a call to one
  reopens it (status active, reopened_count++). The sweep expires the flag
  so the reopenable pool stays bounded. Real clears (units_cleared,
  llm_closure) are never reopenable.
- cap: incident_max_calls counts substantive calls only
  (substantive_call_count). The bridge MVA hit 40 in 32 min with ~40% thin
  replies, split in half, and the second half took another job's title.

c2-core: 467 pass.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Logan Cusano
2026-09-26 19:14:35 -04:00
co-authored by Claude Opus 5.5
parent 731b54bed9
commit 969d175a67
5 changed files with 100 additions and 8 deletions
@@ -66,3 +66,30 @@ def test_only_numbered_units_hold_an_incident_open():
assert not ic._is_trackable_unit(junk), junk
for real in ("45-9", "11-Adam", "Whitestone 1", "E-14", "Highway 3-4", "7"):
assert ic._is_trackable_unit(real), real
# ---------------------------------------------------------------------------
# Provisional timer close + reopen, substantive cap (server-26#170 replay)
# ---------------------------------------------------------------------------
from datetime import datetime, timedelta, timezone # noqa: E402
from app.config import settings # noqa: E402
from app.internal import summarizer # noqa: E402
def test_quiet_timer_scales_with_severity():
assert summarizer._auto_resolve_minutes({"severity": "routine"}) == settings.incident_auto_resolve_minutes_routine
assert summarizer._auto_resolve_minutes({"severity": "minor"}) == settings.incident_auto_resolve_minutes_routine
assert summarizer._auto_resolve_minutes({"severity": "moderate"}) == settings.incident_auto_resolve_minutes_moderate
assert summarizer._auto_resolve_minutes({"severity": "major"}) == settings.incident_auto_resolve_minutes
assert summarizer._auto_resolve_minutes({}) == settings.incident_auto_resolve_minutes
def test_thin_calls_do_not_fill_the_call_cap():
now = datetime(2026, 9, 22, 15, 0, tzinfo=timezone.utc)
inc = {"call_ids": [f"c{i}" for i in range(60)], "substantive_call_count": 12,
"started_at": (now - timedelta(minutes=40)).isoformat(),
"updated_at": now.isoformat()}
assert ic._incident_at_capacity(inc, now) is None
legacy = {k: v for k, v in inc.items() if k != "substantive_call_count"}
assert ic._incident_at_capacity(legacy, now).startswith("call_cap")
+5 -3
View File
@@ -269,12 +269,14 @@ async def test_run_writes_only_to_its_sandbox_and_pins_the_clock(store):
# The two Car 12 calls are one job; the Car 40 call five hours later is another.
groups = sorted(sorted(i["call_ids"]) for i in sb_incidents.values())
assert groups == [["call-1", "call-2"], ["call-3"]]
# Each aged out on the replayed clock the way it would have live —
# incident_auto_resolve_minutes after its last activity, not "now".
# Each aged out on the replayed clock the way it would have live — its
# severity's quiet timer after its last activity, not "now".
assert run["metrics"]["resolved_via"] == {"idle_timeout": 2}
first = next(i for i in sb_incidents.values() if "call-1" in i["call_ids"])
idle = datetime.fromisoformat(first["resolved_at"]) - datetime.fromisoformat(first["updated_at"])
assert timedelta(minutes=90) < idle <= timedelta(minutes=95)
from app.internal.summarizer import _auto_resolve_minutes
limit = timedelta(minutes=_auto_resolve_minutes(first))
assert limit < idle <= limit + timedelta(minutes=5)
assert run["metrics"]["calls"] == 3
assert set(store.data[f"{root}/scenes"]) == {"call-1", "call-2", "call-3"}