_OBSERVE_RUN_CACHE = {}
_OBSERVE_SCAN_TRUNCATED = False
_PROCESS_SCAN_TRUNCATED = False
_OBSERVE_SCAN_ERROR = None
_PROCESS_SCAN_ERROR = None
_MEMORY_RECLAIM_CONFIG_ERROR = None
_OBSERVE_ROOT_DIAGNOSTICS = {"configured": [], "readable": [], "missing": []}

def configured_observe_roots():
    roots = config_list(MEMORY_CONFIG, "observeStateRoots", config_list(MEMORY_CONFIG, "webObserveRoots", []))
    return [os.path.abspath(item) for item in roots if isinstance(item, str) and item.startswith("/")]

def configured_observe_runs(root):
    global _OBSERVE_SCAN_TRUNCATED, _OBSERVE_SCAN_ERROR
    resolved_root = os.path.abspath(root)
    if resolved_root in _OBSERVE_RUN_CACHE:
        return list(_OBSERVE_RUN_CACHE[resolved_root])
    max_depth = config_int(MEMORY_CONFIG, "observeRunDepth", 1, minimum=1, maximum=12)
    scan_limit = config_int(MEMORY_CONFIG, "observeScanLimit", 1000, minimum=1, maximum=100000)
    markers = set(["manifest.json", "heartbeat.json", "pid", "observer.pid", "runner.pid"])
    runs = []
    scanned = 0
    if not os.path.isdir(resolved_root) or os.path.islink(resolved_root):
        return runs
    def record_walk_error(error):
        global _OBSERVE_SCAN_ERROR
        _OBSERVE_SCAN_ERROR = "observe-state-walk-failed:%s" % type(error).__name__

    for current, directories, files in os.walk(resolved_root, topdown=True, followlinks=False, onerror=record_walk_error):
        relative = os.path.relpath(current, resolved_root)
        depth = 0 if relative == "." else relative.count(os.sep) + 1
        directories[:] = sorted(name for name in directories if not os.path.islink(os.path.join(current, name)))
        if depth >= max_depth:
            directories[:] = []
        if depth == 0:
            continue
        scanned += 1
        if scanned > scan_limit:
            _OBSERVE_SCAN_TRUNCATED = True
            break
        if markers.intersection(files):
            runs.append(os.path.abspath(current))
            directories[:] = []
    if not _OBSERVE_SCAN_TRUNCATED:
        _OBSERVE_RUN_CACHE[resolved_root] = list(runs)
    return runs

def is_observe_run_path(path):
    resolved = os.path.abspath(path)
    for root in configured_observe_roots():
        prefix = root.rstrip("/") + "/"
        if resolved.startswith(prefix) and resolved in set(configured_observe_runs(root)):
            return True
    return False

def is_direct_observe_run_path(path):
    # Kept as a compatibility name for generic artifact retention. Validation is
    # now marker/depth based because the configured roots contain dated trees.
    return is_observe_run_path(path)

def path_has_open_fd(path):
    resolved = os.path.realpath(path)
    prefix = resolved.rstrip("/") + "/"
    proc_root = "/proc"
    try:
        pids = [name for name in os.listdir(proc_root) if name.isdigit()]
    except OSError:
        return True
    for pid in pids:
        base = os.path.join(proc_root, pid)
        for name in ["cwd", "root"]:
            try:
                target = os.path.realpath(os.readlink(os.path.join(base, name)))
            except OSError:
                continue
            if target == resolved or target.startswith(prefix):
                return True
        fd_dir = os.path.join(base, "fd")
        try:
            fds = os.listdir(fd_dir)
        except OSError:
            continue
        for fd in fds:
            try:
                target = os.path.realpath(os.readlink(os.path.join(fd_dir, fd)))
            except OSError:
                continue
            if target == resolved or target.startswith(prefix):
                return True
    return False

def assert_web_observe_candidate(path):
    resolved = os.path.abspath(path)
    if not is_observe_run_path(resolved):
        raise RuntimeError("refusing to remove web-observe path outside configured marker/depth run roots: %s" % path)
    if os.path.islink(resolved) or not os.path.isdir(resolved):
        raise RuntimeError("refusing to remove non-directory or symlink web-observe path: %s" % path)
    stale_hours = config_float(MEMORY_CONFIG, "staleRunMaxAgeHours", 6.0, minimum=0.0)
    record = observe_run_record(resolved, stale_hours)
    if record.get("pidAlive"):
        raise RuntimeError("refusing to remove active web-observe run with live pid: %s" % path)
    if not record.get("staleSignal"):
        raise RuntimeError("refusing to remove web-observe run without stale signal: %s" % path)
    if path_has_open_fd(resolved):
        raise RuntimeError("refusing to remove web-observe run with open fd/cwd reference: %s" % path)
    return record

def read_proc_process_table():
    global _PROCESS_SCAN_TRUNCATED, _PROCESS_SCAN_ERROR
    scan_limit = config_int(MEMORY_CONFIG, "processScanLimit", 16384, minimum=1, maximum=1000000)
    try:
        clock_ticks = int(os.sysconf("SC_CLK_TCK"))
        uptime = float(open("/proc/uptime", "r", encoding="utf-8").read().split()[0])
        page_size = int(os.sysconf("SC_PAGE_SIZE"))
        pid_names = sorted((name for name in os.listdir("/proc") if name.isdigit()), key=int)
        if len(pid_names) > scan_limit:
            _PROCESS_SCAN_TRUNCATED = True
            return {}
    except Exception as exc:
        _PROCESS_SCAN_ERROR = "proc-process-table-read-failed:%s" % type(exc).__name__
        return {}
    rows = {}
    stat_read_failures = 0
    for pid_name in pid_names:
        base = os.path.join("/proc", pid_name)
        try:
            with open(os.path.join(base, "stat"), "r", encoding="utf-8") as handle:
                stat_text = handle.read().strip()
            close = stat_text.rfind(")")
            if close < 0:
                continue
            fields = stat_text[close + 2:].split()
            if len(fields) < 22:
                continue
            pid = int(pid_name)
            ppid = int(fields[1])
            sid = int(fields[3])
            start_ticks = int(fields[19])
            try:
                with open(os.path.join(base, "cmdline"), "rb") as handle:
                    command_line = handle.read(65536).replace(b"\0", b" ").decode("utf-8", errors="replace").strip()
            except OSError:
                command_line = ""
            try:
                with open(os.path.join(base, "statm"), "r", encoding="utf-8") as handle:
                    rss_bytes = safe_int(handle.read().split()[1]) * page_size
            except Exception:
                rss_bytes = 0
            rows[pid] = {
                "pid": pid,
                "ppid": ppid,
                "sid": sid,
                "startTicks": start_ticks,
                "ageSeconds": max(0, int(uptime - (float(start_ticks) / float(clock_ticks)))),
                "rssBytes": rss_bytes,
                "commandPreview": redact_command_preview(command_line),
                "commandLine": command_line,
            }
        except (OSError, ValueError, IndexError):
            stat_read_failures += 1
            continue
    if not rows:
        _PROCESS_SCAN_ERROR = "proc-process-table-empty"
    elif stat_read_failures > max(32, len(pid_names) // 2):
        _PROCESS_SCAN_ERROR = "proc-process-stat-read-failure-threshold:%s-of-%s" % (stat_read_failures, len(pid_names))
    return rows

def validate_observe_run_markers(path):
    for name in ["heartbeat.json", "manifest.json"]:
        marker = os.path.join(path, name)
        if not os.path.lexists(marker):
            continue
        if os.path.islink(marker) or not os.path.isfile(marker):
            raise OSError("observe marker is not a regular file: %s" % marker)
        with open(marker, "r", encoding="utf-8") as handle:
            value = json.load(handle)
        if not isinstance(value, dict):
            raise OSError("observe JSON marker is not an object: %s" % marker)
    for name in ["pid", "observer.pid", "browser.pid", "runner.pid"]:
        marker = os.path.join(path, name)
        if not os.path.lexists(marker):
            continue
        if os.path.islink(marker) or not os.path.isfile(marker):
            raise OSError("observe pid marker is not a regular file: %s" % marker)
        with open(marker, "r", encoding="utf-8") as handle:
            raw = handle.read().strip()
        if not re.match(r"^[1-9]\d*$", raw):
            raise OSError("observe pid marker is invalid: %s" % marker)

def observer_run_records():
    global _OBSERVE_SCAN_ERROR, _OBSERVE_ROOT_DIAGNOSTICS
    stale_hours = config_float(MEMORY_CONFIG, "staleRunMaxAgeHours", 6.0, minimum=0.0)
    records = []
    roots = configured_observe_roots()
    accessible_roots = []
    missing_roots = []
    for root in roots:
        try:
            root_stat = os.stat(root)
            if os.path.isdir(root) and not os.path.islink(root):
                accessible_roots.append(root)
            elif root_stat:
                _OBSERVE_SCAN_ERROR = "observe-state-root-not-directory"
        except FileNotFoundError:
            missing_roots.append(root)
            continue
        except OSError as exc:
            _OBSERVE_SCAN_ERROR = "observe-state-root-read-failed:%s" % type(exc).__name__
    minimum_readable = config_int(MEMORY_CONFIG, "minimumReadableObserveRoots", 1, minimum=1, maximum=100)
    _OBSERVE_ROOT_DIAGNOSTICS = {"configured": list(roots), "readable": list(accessible_roots), "missing": missing_roots}
    if len(accessible_roots) < minimum_readable:
        _OBSERVE_SCAN_ERROR = _OBSERVE_SCAN_ERROR or "minimum-readable-observe-roots-not-met:%s-of-%s" % (len(accessible_roots), minimum_readable)
    for root in accessible_roots:
        for path in configured_observe_runs(root):
            try:
                validate_observe_run_markers(path)
                records.append(observe_run_record(path, stale_hours))
            except (OSError, ValueError, json.JSONDecodeError) as exc:
                _OBSERVE_SCAN_ERROR = "observe-run-marker-read-failed:%s" % type(exc).__name__
                continue
    return records

def process_descendants(processes, root_pid):
    selected = set([int(root_pid)])
    changed = True
    while changed:
        changed = False
        for pid, row in processes.items():
            if pid not in selected and safe_int(row.get("ppid")) in selected:
                selected.add(pid)
                changed = True
    return selected

def session_closed_tree(processes, sid):
    members = {pid: row for pid, row in processes.items() if safe_int(row.get("sid")) == int(sid)}
    if not members:
        return None
    roots = [pid for pid, row in members.items() if safe_int(row.get("ppid")) not in members]
    if len(roots) != 1:
        return None
    root_pid = roots[0]
    if process_descendants(members, root_pid) != set(members.keys()):
        return None
    return {"rootPid": root_pid, "members": members}

def command_matches(row, patterns):
    haystack = str(row.get("commandLine") or "").lower()
    return any(str(pattern).lower() in haystack for pattern in patterns if str(pattern))

def memory_process_candidate_id(kind, root):
    return "%s:%s:%s:%s" % (kind, safe_int(root.get("sid")), safe_int(root.get("pid")), safe_int(root.get("startTicks")))

def collect_memory_reclaim_candidates():
    global _OBSERVE_SCAN_TRUNCATED, _PROCESS_SCAN_TRUNCATED, _OBSERVE_SCAN_ERROR, _PROCESS_SCAN_ERROR, _MEMORY_RECLAIM_CONFIG_ERROR, _OBSERVE_ROOT_DIAGNOSTICS
    _OBSERVE_SCAN_TRUNCATED = False
    _PROCESS_SCAN_TRUNCATED = False
    _OBSERVE_SCAN_ERROR = None
    _PROCESS_SCAN_ERROR = None
    _MEMORY_RECLAIM_CONFIG_ERROR = None
    _OBSERVE_ROOT_DIAGNOSTICS = {"configured": [], "readable": [], "missing": []}
    _OBSERVE_RUN_CACHE.clear()
    required_keys = [
        "observeStateRoots", "minimumReadableObserveRoots", "observeRunDepth", "observeScanLimit", "staleRunMaxAgeHours",
        "processScanLimit", "processStaleMinAgeSeconds", "maxMemoryReclaimCandidates",
        "terminationGraceSeconds", "terminationPollMilliseconds", "observerRootPatterns",
        "browserRootPatterns", "browserProcessPatterns",
    ]
    if any(key not in MEMORY_CONFIG for key in required_keys):
        _MEMORY_RECLAIM_CONFIG_ERROR = "required-yaml-memory-pressure-policy-missing"
        return []
    list_keys = ["observeStateRoots", "observerRootPatterns", "browserRootPatterns", "browserProcessPatterns"]
    integer_keys = [
        "observeRunDepth", "observeScanLimit", "processScanLimit", "processStaleMinAgeSeconds",
        "maxMemoryReclaimCandidates", "terminationGraceSeconds", "terminationPollMilliseconds",
    ]
    if any(not isinstance(MEMORY_CONFIG.get(key), list) or not MEMORY_CONFIG.get(key) for key in list_keys):
        _MEMORY_RECLAIM_CONFIG_ERROR = "required-yaml-memory-pressure-list-invalid"
        return []
    if any(any(not isinstance(item, str) or not item.strip() or "\0" in item for item in MEMORY_CONFIG.get(key)) for key in list_keys):
        _MEMORY_RECLAIM_CONFIG_ERROR = "required-yaml-memory-pressure-list-entry-invalid"
        return []
    if any(not os.path.isabs(item) for item in MEMORY_CONFIG.get("observeStateRoots")):
        _MEMORY_RECLAIM_CONFIG_ERROR = "required-yaml-observe-root-not-absolute"
        return []
    integer_ranges = {
        "observeRunDepth": (1, 12),
        "minimumReadableObserveRoots": (1, 100),
        "observeScanLimit": (1, 100000),
        "processScanLimit": (1, 1000000),
        "processStaleMinAgeSeconds": (60, 604800),
        "maxMemoryReclaimCandidates": (1, 100),
        "terminationGraceSeconds": (1, 30),
        "terminationPollMilliseconds": (50, 1000),
    }
    if any(isinstance(MEMORY_CONFIG.get(key), bool) or not isinstance(MEMORY_CONFIG.get(key), int)
           or MEMORY_CONFIG.get(key) < minimum or MEMORY_CONFIG.get(key) > maximum
           for key, (minimum, maximum) in integer_ranges.items()):
        _MEMORY_RECLAIM_CONFIG_ERROR = "required-yaml-memory-pressure-integer-invalid"
        return []
    if MEMORY_CONFIG.get("minimumReadableObserveRoots") > len(MEMORY_CONFIG.get("observeStateRoots")):
        _MEMORY_RECLAIM_CONFIG_ERROR = "minimum-readable-observe-roots-exceeds-configured-roots"
        return []
    stale_hours = MEMORY_CONFIG.get("staleRunMaxAgeHours")
    if isinstance(stale_hours, bool) or not isinstance(stale_hours, (int, float)) or stale_hours <= 0 or stale_hours > 8760:
        _MEMORY_RECLAIM_CONFIG_ERROR = "required-yaml-stale-age-invalid"
        return []
    browser_patterns = config_list(MEMORY_CONFIG, "browserProcessPatterns", [])
    root_patterns = config_list(MEMORY_CONFIG, "browserRootPatterns", [])
    observer_patterns = config_list(MEMORY_CONFIG, "observerRootPatterns", [])
    stale_seconds = config_int(MEMORY_CONFIG, "processStaleMinAgeSeconds", 3600, minimum=60, maximum=604800)
    if not browser_patterns or not root_patterns or not observer_patterns:
        return []
    processes = read_proc_process_table()
    records = observer_run_records()
    if not processes and _PROCESS_SCAN_ERROR is None:
        _PROCESS_SCAN_ERROR = "proc-process-table-empty"
    if _PROCESS_SCAN_TRUNCATED or _OBSERVE_SCAN_TRUNCATED or _PROCESS_SCAN_ERROR or _OBSERVE_SCAN_ERROR:
        return []
    fresh_observer_pids = set()
    stale_observer_pids = set()
    for record in records:
        pid = safe_int(record.get("pid"))
        if pid <= 0 or pid not in processes:
            continue
        if record.get("staleLiveSignal"):
            stale_observer_pids.add(pid)
        else:
            fresh_observer_pids.add(pid)
    protected_pids = set()
    for pid in fresh_observer_pids:
        protected_pids.update(process_descendants(processes, pid))
    candidates = []
    for sid in sorted(set(safe_int(row.get("sid")) for row in processes.values() if safe_int(row.get("sid")) > 0)):
        closed = session_closed_tree(processes, sid)
        if closed is None:
            continue
        members = closed["members"]
        member_pids = set(members.keys())
        if member_pids.intersection(protected_pids):
            continue
        root = members[closed["rootPid"]]
        if safe_int(root.get("ageSeconds")) < stale_seconds:
            continue
        has_browser = any(command_matches(row, browser_patterns) for row in members.values())
        intersecting_stale = member_pids.intersection(stale_observer_pids)
        kind = None
        reason = None
        if intersecting_stale and command_matches(root, observer_patterns):
            kind = "stale-web-observer-process-tree"
            reason = "stale-observer-heartbeat"
        elif not member_pids.intersection(set(safe_int(item.get("pid")) for item in records if item.get("pidAlive"))):
            if command_matches(root, root_patterns) and has_browser:
                kind = "browser-orphan-process-tree"
                reason = "browser-session-without-observer-record"
        if kind is None:
            continue
        identities = [
            {
                "pid": pid,
                "ppid": safe_int(row.get("ppid")),
                "sid": safe_int(row.get("sid")),
                "startTicks": safe_int(row.get("startTicks")),
                "rssBytes": safe_int(row.get("rssBytes")),
                "commandPreview": row.get("commandPreview"),
            }
            for pid, row in sorted(members.items())
        ]
        expected_memory = sum(safe_int(item.get("rssBytes")) for item in identities)
        candidates.append({
            "id": memory_process_candidate_id(kind, root),
            "kind": kind,
            "risk": "medium",
            "description": "Terminate a YAML-allowlisted stale WebProbe process session after identity and active-observer revalidation",
            "reason": reason,
            "rootPid": safe_int(root.get("pid")),
            "sid": sid,
            "rootStartTicks": safe_int(root.get("startTicks")),
            "processCount": len(identities),
            "processIdentities": identities,
            "expectedMemoryRssBytes": expected_memory,
            "expectedMemoryRss": fmt_bytes(expected_memory),
            "estimatedDiskBytes": 0,
            "estimatedReclaimBytes": 0,
            "activeObserverIntersection": False,
            "treeClosure": "unique-root-ppid-sid-closed",
            "action": {"op": "term-wait-kill-exact-process-tree", "allowlist": "yaml-web-probe-memory-pressure"},
        })
    max_candidates = config_int(MEMORY_CONFIG, "maxMemoryReclaimCandidates", 20, minimum=1, maximum=100)
    return sorted(candidates, key=lambda item: safe_int(item.get("expectedMemoryRssBytes")), reverse=True)[:max_candidates]

def memory_reclaim_scan_diagnostics():
    blocked = bool(_MEMORY_RECLAIM_CONFIG_ERROR or _OBSERVE_SCAN_TRUNCATED or _PROCESS_SCAN_TRUNCATED or _OBSERVE_SCAN_ERROR or _PROCESS_SCAN_ERROR)
    return {
        "ok": not blocked,
        "blocked": blocked,
        "configError": _MEMORY_RECLAIM_CONFIG_ERROR,
        "observeScanTruncated": bool(_OBSERVE_SCAN_TRUNCATED),
        "processScanTruncated": bool(_PROCESS_SCAN_TRUNCATED),
        "observeScanError": _OBSERVE_SCAN_ERROR,
        "processScanError": _PROCESS_SCAN_ERROR,
        "observeRoots": {
            "configuredCount": len(_OBSERVE_ROOT_DIAGNOSTICS.get("configured") or []),
            "readableCount": len(_OBSERVE_ROOT_DIAGNOSTICS.get("readable") or []),
            "missingCount": len(_OBSERVE_ROOT_DIAGNOSTICS.get("missing") or []),
            "minimumReadable": safe_int(MEMORY_CONFIG.get("minimumReadableObserveRoots")),
            "configured": _OBSERVE_ROOT_DIAGNOSTICS.get("configured") or [],
            "readable": _OBSERVE_ROOT_DIAGNOSTICS.get("readable") or [],
            "missing": _OBSERVE_ROOT_DIAGNOSTICS.get("missing") or [],
        },
        "policySource": "config/unidesk-cli.yaml#gc.remote.targets.%s.memoryPressure" % PROVIDER_ID,
    }

def matching_live_identity(identity, processes):
    pid = safe_int(identity.get("pid"))
    row = processes.get(pid)
    return row if row \
        and safe_int(row.get("startTicks")) == safe_int(identity.get("startTicks")) \
        and safe_int(row.get("ppid")) == safe_int(identity.get("ppid")) \
        and safe_int(row.get("sid")) == safe_int(identity.get("sid")) else None

def matching_surviving_identity(identity, processes):
    pid = safe_int(identity.get("pid"))
    row = processes.get(pid)
    return row if row \
        and safe_int(row.get("startTicks")) == safe_int(identity.get("startTicks")) else None

def require_complete_process_scan(processes, phase):
    if not processes or _PROCESS_SCAN_ERROR or _PROCESS_SCAN_TRUNCATED:
        raise RuntimeError("memory process identity scan blocked during %s: %s" % (
            phase,
            _PROCESS_SCAN_ERROR or "process-scan-truncated",
        ))
    return processes

def extend_with_allowlisted_session_members(identities, processes, sid, browser_patterns):
    known = {(safe_int(item.get("pid")), safe_int(item.get("startTicks"))) for item in identities}
    added = []
    for pid, row in processes.items():
        identity_key = (safe_int(pid), safe_int(row.get("startTicks")))
        if identity_key in known or safe_int(row.get("sid")) != safe_int(sid) or not command_matches(row, browser_patterns):
            continue
        item = {
            "pid": safe_int(pid),
            "ppid": safe_int(row.get("ppid")),
            "sid": safe_int(row.get("sid")),
            "startTicks": safe_int(row.get("startTicks")),
            "rssBytes": safe_int(row.get("rssBytes")),
            "commandPreview": row.get("commandPreview"),
            "discoveredAfterTerm": True,
        }
        identities.append(item)
        added.append(item)
        known.add(identity_key)
    return added

def execute_memory_process_candidate(candidate):
    current = next((item for item in collect_memory_reclaim_candidates() if item.get("id") == candidate.get("id")), None)
    if current is None:
        raise RuntimeError("memory candidate no longer has the same closed identity or is now protected")
    before = collect_memory_snapshot()
    identities = [dict(item) for item in (current.get("processIdentities") or [])]
    candidate_sid = safe_int(current.get("sid"))
    browser_patterns = config_list(MEMORY_CONFIG, "browserProcessPatterns", [])
    late_members = []
    processes = require_complete_process_scan(read_proc_process_table(), "pre-term-revalidation")
    live = [item for item in identities if matching_live_identity(item, processes)]
    for item in sorted(live, key=lambda value: safe_int(value.get("pid")), reverse=True):
        try:
            os.kill(safe_int(item.get("pid")), signal.SIGTERM)
        except ProcessLookupError:
            pass
    grace_seconds = config_int(MEMORY_CONFIG, "terminationGraceSeconds", 5, minimum=1, maximum=30)
    poll_ms = config_int(MEMORY_CONFIG, "terminationPollMilliseconds", 200, minimum=50, maximum=1000)
    deadline = time.time() + grace_seconds
    residual = []
    while time.time() < deadline:
        processes = require_complete_process_scan(read_proc_process_table(), "term-grace-poll")
        late_members.extend(extend_with_allowlisted_session_members(identities, processes, candidate_sid, browser_patterns))
        residual = [item for item in identities if matching_surviving_identity(item, processes)]
        if not residual:
            break
        time.sleep(float(poll_ms) / 1000.0)
    processes = require_complete_process_scan(read_proc_process_table(), "pre-kill-revalidation")
    late_members.extend(extend_with_allowlisted_session_members(identities, processes, candidate_sid, browser_patterns))
    residual = [item for item in identities if matching_surviving_identity(item, processes)]
    for item in sorted(residual, key=lambda value: safe_int(value.get("pid")), reverse=True):
        try:
            os.kill(safe_int(item.get("pid")), signal.SIGKILL)
        except ProcessLookupError:
            pass
    final_deadline = time.time() + min(2.0, float(grace_seconds))
    remaining = residual
    while time.time() < final_deadline:
        processes = require_complete_process_scan(read_proc_process_table(), "kill-grace-poll")
        late_members.extend(extend_with_allowlisted_session_members(identities, processes, candidate_sid, browser_patterns))
        remaining = [item for item in identities if matching_surviving_identity(item, processes)]
        if not remaining:
            break
        time.sleep(float(poll_ms) / 1000.0)
    after = collect_memory_snapshot()
    before_available = safe_int((before.get("memory") or {}).get("availableBytes"), None)
    after_available = safe_int((after.get("memory") or {}).get("availableBytes"), None)
    return {
        "status": "failed" if remaining else "succeeded",
        "reclaimedBytes": 0,
        "reclaimedMemoryBytes": max(0, after_available - before_available) if before_available is not None and after_available is not None else None,
        "expectedMemoryRssBytes": current.get("expectedMemoryRssBytes"),
        "termSignalCount": len(live),
        "killSignalCount": len(residual),
        "remainingIdentityCount": len(remaining),
        "lateAllowlistedIdentityCount": len(late_members),
        "remainingIdentities": [
            {
                "pid": safe_int(item.get("pid")),
                "startTicks": safe_int(item.get("startTicks")),
                "sid": safe_int(item.get("sid")),
                "originalPpid": safe_int(item.get("ppid")),
                "observedPpid": safe_int((matching_surviving_identity(item, processes) or {}).get("ppid")),
            }
            for item in remaining
        ],
        "memoryBefore": before,
        "memoryAfter": after,
        "identityRevalidated": True,
        "activeObserverIntersection": False,
    }