# frozen_string_literal: true

# Helpdesk domain core — the Store seam and the Service: the monotonic call state machine, people and
# their AI memory, the resolutions ledger, reconciliation, and reporting. Ruby standard library only.
# The schema of record is db/migrate/001_init.sql (PostgreSQL); the in-memory JSON Store here is its
# byte-for-byte parity twin (db/parity_check.rb proves they agree), so the same Service runs on either.
# grep anchors: class Store (the storage seam), class Service (all behaviour), reconcile! (the watchdog).

require "json"
require "securerandom"
require "time"
require "monitor"
require_relative "reporting"
require_relative "ledger"

module Helpdesk
  # --- enums (explicit hash form — reorder-safe; values mirror db/migrate/001_init.sql) ---
  OPERATOR   = { "tmobile" => 0, "vodafone" => 1 }.freeze
  RESOLUTION = { "solved" => 0, "partial" => 1, "unsolved" => 2 }.freeze
  DIRECTION  = { "inbound" => 0, "outbound" => 1 }.freeze

  # Reverse-channel command verbs (operator PC -> device InCallService). Each maps to a Telecom call
  # control: answer/reject/hangup -> Call.answer/reject/disconnect; hold/resume -> Call.hold/unhold;
  # mute/unmute -> setMuted;
  # dtmf -> Call.playDtmfTone; route -> requestCallEndpointChange. The device applies these on
  # the single InCallService and the resulting state flows back via the existing webhooks (the ACK).
  # answer..route are call-scoped (issue_command, keyed by an existing call_id). `dial` is the one
  # DEVICE-scoped verb — it starts a NEW outgoing call, so it's issued via dial() (no call_id) and
  # applied by the phone's CommandChannelClient.placeCall. Listed here so the phone's CommandGate.VERBS
  # (which mirrors this set) accepts it.
  COMMAND_VERBS = %w[answer reject hangup hold resume mute unmute dtmf route dial].freeze

  # Event states + forward-only rank. Terminal side-states carry no rank.
  STATE_RANK = {
    "ringing" => 0, "answered" => 1, "ended" => 2,
    "recording_uploaded" => 3, "transcribed" => 4, "summarised" => 5
  }.freeze
  SIDE_STATES = %w[recording_missing failed no_speech].freeze

  # Reconciliation timings (the watchdog windows reconcile! enforces) — seconds.
  RING_WATCHDOG_S      = 15 * 60
  ANSWERED_WATCHDOG_S  = 6  * 60 * 60
  RECORDING_GRACE_S    = 30 * 60
  QUARANTINE_TTL_S     = 24 * 60 * 60
  DEVICE_PRESENCE_TTL_S = 300 # "connected" if polled/heartbeat within this window (> the ~2 min idle heartbeat)
  MAX_PENDING_COMMANDS  = 20  # per-device reverse-command queue cap — refuse rather than grow unbounded

  # --- tiny in-memory store with JSON-file persistence -----------------------
  class Store
    attr_reader :people, :phone_numbers, :orgs, :campaigns, :dids, :events, :quarantine, :broadcasts, :commands

    def initialize
      @people = {}; @phone_numbers = {}; @orgs = {}; @campaigns = {}
      @dids = {}; @events = {}; @quarantine = {} # recording_uuid => {call_id, at}
      @broadcasts = Hash.new { |h, k| h[k] = [] } # operator_id => [payloads]  (screen-pop pub/sub)
      @commands   = Hash.new { |h, k| h[k] = [] } # device_id => [commands]    (reverse channel, PC->device)
      @seq = Hash.new(0)
    end

    def next_id(kind) = @seq[kind] += 1

    # indexes
    def person_by_e164(e164) = @phone_numbers[e164]&.then { |pn| @people[pn[:person_id]] }
    def did_by_e164(e164)    = @dids[e164]
    def event_by_call_id(cid) = @events.values.find { |e| e[:call_id] == cid }

    # ---- explicit WRITE INTERFACE (shared with PgStore; Service calls these instead of relying on
    # in-place mutation being magically persisted by the whole-store dump). For the JSON store these
    # just (re)index the record hash into the cache — dump/save then serializes it, so the JSON backend
    # keeps working identically. PgStore overrides these to also UPSERT/DELETE the row. Every §3
    # mutation site calls the matching put_*/delete_* after mutating the record. ----
    def put_org(o)            = (@orgs[o[:id]] = o)
    def put_campaign(c)       = (@campaigns[c[:id]] = c)
    def put_person(p)         = (@people[p[:id]] = p)
    def put_event(ev)         = (@events[ev[:id]] = ev)
    def put_phone_number(pn)  = (@phone_numbers[pn[:e164]] = pn)
    def put_did(d)            = (@dids[d[:e164]] = d)
    def put_quarantine(uuid, h) = (@quarantine[uuid] = h)
    def delete_person(id)         = @people.delete(id)
    def delete_phone_number(e164) = @phone_numbers.delete(e164)
    def delete_quarantine(uuid)   = @quarantine.delete(uuid)
    # transaction seam (no-op for the JSON store — the whole-store dump is atomic anyway; PgStore
    # overrides it with a real DB transaction so Service multi-row ops are atomic on both backends).
    def transaction = (yield if block_given?)
    # AI-memory compare-and-set write. Both backends implement it so the CAS+row-lock semantics live in
    # the store, not the Service. JSON store: check the cached rev, write if it matches. Returns the
    # written person hash on success, or nil if stale (caller retries). PgStore does the same inside a
    # SELECT ... FOR UPDATE transaction so it's correct multi-process too.
    def cas_person_ai_context(id, expected_rev, ai_context, built_from)
      p = @people[id] or return nil
      return nil unless (p[:ai_context_rev] || 0) == expected_rev
      p[:ai_context] = ai_context
      p[:ai_context_rev] = expected_rev + 1
      p[:ai_context_built_from] = built_from if built_from
      p
    end

    # ---- JSON-file persistence (durable subset; broadcasts/commands are transient in-flight) ----
    # id-keyed collections are dumped as VALUE ARRAYS and re-indexed on load, so JSON's string keys never
    # corrupt the integer ids. @seq is included so restored ids can never collide with freshly-minted ones.
    # quarantine IS persisted: it holds the on-disk audio_path of a recording uploaded before its
    # /incoming — dropping it on a restart would orphan the .audio file and the call would never
    # transcribe (silent recording-data loss). Keyed by recording_uuid (a string), so it dumps as-is.
    def dump = {
      "seq"           => @seq,
      "orgs"          => @orgs.values,
      "campaigns"     => @campaigns.values,
      "people"        => @people.values,
      "phone_numbers" => @phone_numbers.values,
      "dids"          => @dids.values,
      "events"        => @events.values,
      "quarantine"    => @quarantine
    }

    def restore(h)
      (h["seq"] || {}).each { |k, v| @seq[k.to_sym] = v.to_i }
      @orgs          = index(symbolize(h["orgs"]),          :id)
      @campaigns     = index(symbolize(h["campaigns"]),     :id)
      @people        = index(symbolize(h["people"]),        :id)
      @phone_numbers = index(symbolize(h["phone_numbers"]), :e164)
      @dids          = index(symbolize(h["dids"]),          :e164)
      @events        = index(symbolize(h["events"]),        :id)
      # value hashes come back string-keyed from JSON; re-symbolize (uuid string keys stay as-is)
      @quarantine    = (h["quarantine"] || {}).transform_values { |v| v.transform_keys(&:to_sym) }
      self
    end

    def self.load(path)
      s = new
      s.restore(JSON.parse(File.read(path))) if path && File.exist?(path) && !File.zero?(path)
      s
    rescue => e
      warn "[helpdesk] state load failed (#{e.class}: #{e.message}) — starting fresh"
      new
    end

    # atomic: write a temp file then rename, so a crash mid-write can't corrupt the live snapshot.
    def save(path)
      tmp = "#{path}.tmp.#{Process.pid}"
      File.write(tmp, JSON.pretty_generate(dump))
      File.rename(tmp, path)
    end

    private

    # Symbolize TOP-LEVEL keys only — nested string-keyed structures (action_items [{"text"=>..}],
    # flags ["x"]) must stay string-keyed to match the code that reads them, so we do NOT deep-symbolize.
    def symbolize(list) = (list || []).map { |r| r.transform_keys(&:to_sym) }
    def index(list, key) = list.each_with_object({}) { |r, acc| acc[r[key]] = r }
  end

  # --- E.164 normalization (CZ-aware; the phone's util/CzE164.java mirrors this byte-for-byte) ---
  module Phone
    module_function

    # Minimal CZ-aware normalizer. Returns E.164 or nil for un-normalizable input.
    def e164(raw, region: "CZ")
      return nil if raw.nil?
      s = raw.to_s.strip.gsub(/[\s\-().]/, "")
      return nil if s.empty?
      return s if s =~ /\A\+\d{6,15}\z/          # already E.164
      s = s.sub(/\A00/, "+")                     # 00420… -> +420…
      return s if s =~ /\A\+\d{6,15}\z/
      if region == "CZ" && s =~ /\A\d{9}\z/      # bare 9-digit CZ national -> +420
        return "+420#{s}"
      end
      nil
    end
  end

  # --- the service: lookups, webhooks, state machine, reconciliation ----------
  class Service
    attr_reader :store
    # optional hook: called with the Event when a recording is uploaded. The server wires the
    # Transcriber→Summariser pipeline here; the pure domain stays runner-agnostic (nil = no-op).
    attr_accessor :transcription_runner
    # optional hook: called (no args) after any durable WRITE op completes, under the lock. The server
    # wires this to a dirty-flag so a debounced flusher persists the store. nil = no persistence (tests).
    attr_accessor :after_write

    # write the durable store to disk (called by the server's debounced flusher; nil path = no-op).
    def persist!(path) = path && @lock.synchronize { @store.save(path) }

    def initialize(store = Store.new, clock: -> { Time.now.utc }, transcription_runner: nil)
      @store = store
      @clock = clock
      @transcription_runner = transcription_runner
      @lock = Monitor.new # serializes all public state ops across WEBrick + worker threads (see below)
      @cmd_cond = @lock.new_cond # signalled by issue_command to wake a long-poll drain_commands
      @device_seen = {} # device_id => monotonic last-poll time (phone presence; "waiting for the phone")
    end

    def now = @clock.call

    # locked read for external callers (server.rb); internal code uses @store.event_by_call_id directly
    def event_by_call_id(call_id) = @store.event_by_call_id(call_id)

    # ---- seeding helpers (used by seed.rb and tests) ----
    def add_org(name:, operator:)
      id = @store.next_id(:org); @store.put_org({ id: id, name: name, operator: operator }); id
    end

    def add_person(name:, operator:, position: nil, numbers: [], context: nil, email: nil)
      id = @store.next_id(:person)
      @store.put_person({ id: id, name: name, operator: operator, position: position, context: context, email: email })
      numbers.each { |raw| add_number(person_id: id, raw: raw) }
      id
    end

    def add_number(person_id:, raw:, label: nil)
      e = Phone.e164(raw)
      return nil unless e
      owner = @store.phone_numbers[e]
      return e if owner && owner[:person_id] != person_id # never STEAL a number already owned by someone
      @store.put_phone_number({ person_id: person_id, e164: e, raw_input: raw, label: label })            # else
      e
    end

    def add_campaign(name:, operator:)
      id = @store.next_id(:campaign); @store.put_campaign({ id: id, name: name, operator: operator }); id
    end

    def add_did(e164:, operator:, campaign_id:)
      @store.put_did({ e164: e164, operator: operator, campaign_id: campaign_id }); e164
    end

    # ---- webhook: incoming (pre-answer, fire-and-forget) ----
    def incoming(call_id:, ts:, number:, presentation:, dialed_did: nil, device_id: nil)
      @store.transaction do  # insert/update the event (+ possible quarantine delete) atomically
        ev = @store.event_by_call_id(call_id) || blank_event(@store.next_id(:event), call_id)
        caller_e164 = Phone.e164(number)
        did = @store.did_by_e164(Phone.e164(dialed_did) || dialed_did)
        person = caller_e164 && @store.person_by_e164(caller_e164)

        ev[:started_at]        ||= ts
        ev[:number]            ||= caller_e164
        ev[:presentation]        = presentation
        ev[:dialed_did]          = did&.dig(:e164) || dialed_did
        ev[:did_id]              = did&.dig(:e164)
        ev[:operator_context]    = did&.dig(:operator)
        ev[:campaign_id]         = did&.dig(:campaign_id)
        ev[:person_id]           = person&.dig(:id)
        ev[:device_id]           = device_id
        link_prior_event(ev, person)
        advance!(ev, "ringing")
        attach_quarantined_recording(ev) # a recording that PUT before this /incoming (race) — reattach now
        @store.put_event(ev)             # explicit persist (insert or update)
        broadcast_screen_pop(ev, person)
        { event_id: ev[:id], state: ev[:state] }
      end
    end

    # ---- webhook: outgoing (operator-originated call from the office Pixel; direction=outbound) ----
    # Mirrors incoming() minus the inbound-only fields (presentation, dialed_did — there is no DID on a
    # call we place). The same answered/ended webhooks and the recording pipeline apply unchanged.
    def outgoing(call_id:, ts:, number:, device_id: nil)
      @store.transaction do
        ev = @store.event_by_call_id(call_id) || blank_event(@store.next_id(:event), call_id)
        callee_e164 = Phone.e164(number)
        person = callee_e164 && @store.person_by_e164(callee_e164)
        ev[:direction]          = "outbound"
        ev[:started_at]       ||= ts
        ev[:number]           ||= callee_e164
        ev[:person_id]          = person&.dig(:id)
        ev[:device_id]          = device_id
        link_prior_event(ev, person)
        advance!(ev, "ringing")
        attach_quarantined_recording(ev) # same upload race as inbound — reattach if audio arrived first
        @store.put_event(ev)
        broadcast_screen_pop(ev, person) # the operator sees who they're calling, with history
        { event_id: ev[:id], state: ev[:state] }
      end
    end

    # ---- webhook: answered ----
    def answered(call_id:, ts:)
      ev = ensure_event(call_id)
      ev[:answered_at] ||= ts
      advance!(ev, "answered")
      @store.put_event(ev)
      { event_id: ev[:id], state: ev[:state] }
    end

    # ---- webhook: ended ----
    def ended(call_id:, ts:, duration_s: nil, disconnect_cause: "unknown", recording_uuid: nil)
      ev = ensure_event(call_id)
      ev[:ended_at]          ||= ts
      ev[:disconnect_cause]    = disconnect_cause
      ev[:recording_uuid]      = recording_uuid if recording_uuid
      # duration: prefer device value; else derive
      ev[:duration_s] = duration_s || derive_duration(ev)
      # never-answered inference: missing answered_at is expected for missed/rejected;
      # otherwise the timeline is incomplete (a lost /answered webhook).
      if ev[:answered_at].nil? && !%w[missed rejected].include?(disconnect_cause)
        flag(ev, :timeline_incomplete)
      end
      advance!(ev, "ended")
      @store.put_event(ev)
      { event_id: ev[:id], state: ev[:state] }
    end

    # ---- recording upload complete ----
    def recording_uploaded(recording_uuid:, call_id:, url: nil, audio_path: nil)
      ev = @store.event_by_call_id(call_id)
      unless ev
        # audio with no Event yet -> quarantine WITH the on-disk path, so incoming() can reattach it and
        # the pipeline can still find the file (without audio_path a later reattach couldn't transcribe).
        @store.put_quarantine(recording_uuid, { call_id: call_id, url: url, audio_path: audio_path, at: now.iso8601 })
        return { quarantined: true, call_id: call_id }
      end
      ev[:recording_uuid] = recording_uuid
      ev[:recording_url]  = url
      ev[:audio_path]     = audio_path if audio_path # local file the pipeline transcribes
      unflag(ev, :recording_pending)
      advance!(ev, "recording_uploaded")
      @store.put_event(ev)
      enqueue_transcription(ev)
      { event_id: ev[:id], state: ev[:state] }
    end

    # transcription/summary hooks — server.rb wires @transcription_runner to the real pipeline on a
    # recording upload; the tests leave it nil (the state machine works without audio). grep: transcription-hooks.
    def enqueue_transcription(ev) = @transcription_runner&.call(ev)
    def mark_transcribed(call_id:, transcript:)
      ev = ensure_event(call_id); ev[:transcript] = transcript; advance!(ev, "transcribed"); @store.put_event(ev)
    end
    def mark_summarised(call_id:, summary:, action_items: [], suggested_position: nil,
                        context_digest: nil, caller_facts: [], ledger_deltas: [])
      ev = ensure_event(call_id)
      ev[:ai_summary] = summary; ev[:action_items] = action_items
      ev[:suggested_position] = suggested_position unless suggested_position.to_s.strip.empty?
      ev[:context_digest] = context_digest.to_s unless context_digest.nil?
      ev[:caller_facts]   = Array(caller_facts)
      ev[:ledger_deltas]  = Array(ledger_deltas)
      advance!(ev, "summarised")
      @store.put_event(ev)
    end

    # transcription/pipeline failure → terminal `failed` side-state, so the console shows an actionable
    # error instead of a call stuck "processing" forever. Never clobbers an already-successful outcome.
    def mark_failed(call_id:, reason: nil, retryable: nil)
      ev = ensure_event(call_id)
      return { event_id: ev[:id], state: ev[:state] } if %w[transcribed summarised].include?(ev[:state])
      ev[:failure_reason] = reason
      ev[:failure_retryable] = retryable unless retryable.nil?
      flag(ev, :transcription_failed)
      set_state!(ev, "failed") # `failed` is not in STATE_RANK, so force it (advance! would no-op)
      @store.put_event(ev)
      { event_id: ev[:id], state: ev[:state] }
    end

    # Whisper detected no speech and wrote empty output → distinct terminal `no_speech` side-state.
    # NOT `transcribed`: an empty transcript isn't a successful transcription, and summarising "" is
    # garbage. (transcribe_async must branch on the pipeline `stage`, since an empty string is truthy.)
    def mark_no_speech(call_id:)
      ev = ensure_event(call_id)
      return { event_id: ev[:id], state: ev[:state] } if %w[transcribed summarised].include?(ev[:state])
      ev[:transcript] = ""
      set_state!(ev, "no_speech") # not in STATE_RANK — force it (advance! would no-op)
      @store.put_event(ev)
      { event_id: ev[:id], state: ev[:state] }
    end

    # ---- reconciliation / watchdogs (call periodically; the server ticks this every 60s) ----
    def reconcile!
      t = now
      @store.events.each_value do |ev|
        begin
          case ev[:state]
          when "ringing"
            (close_missed(ev); @store.put_event(ev)) if age(ev[:started_at], t) > RING_WATCHDOG_S
          when "answered"
            if age(ev[:answered_at], t) > ANSWERED_WATCHDOG_S
              ev[:ended_at] ||= t.iso8601; ev[:duration_s] ||= nil
              flag(ev, :timeline_incomplete); set_state!(ev, "ended")
              @store.put_event(ev)
            end
          when "ended"
            # recording promised but not uploaded within grace -> recording_missing. Guard nil ended_at
            # (a /calls/ended with ts:null leaves it nil) so Time.parse can't raise, and wrap each Event
            # in begin/rescue so one malformed Event can't abort the loop and starve every later Event.
            if ev[:ended_at] && age(ev[:ended_at], t) > RECORDING_GRACE_S
              set_state!(ev, "recording_missing"); @store.put_event(ev)
            end
          end
        rescue => e
          warn "[helpdesk] reconcile! skipped event #{ev[:id]} (#{ev[:state]}): #{e.class}: #{e.message}"
        end
      end
      # expire quarantined orphan uploads (explicit delete_quarantine per row)
      @store.quarantine.select { |_uuid, q| age(q[:at], t) > QUARANTINE_TTL_S }.keys.each { |u| @store.delete_quarantine(u) }
      self
    end

    # screen-pop consumer: an in-memory per-operator queue the console drains over GET /screen-pop
    def drain_broadcasts(operator_id)
      if operator_id == "all"
        msgs = @store.broadcasts.values.flatten(1); @store.broadcasts.clear; return msgs
      end
      msgs = @store.broadcasts[operator_id]; @store.broadcasts[operator_id] = []; msgs
    end

    def screen_pop_payload(ev, person)
      {
        type: "screen_pop", call_id: ev[:call_id], event_id: ev[:id],
        person: person && { id: person[:id], name: person[:name],
                            position: person[:position], operator: person[:operator] },
        unknown_caller: person.nil?,
        number: ev[:number],
        direction: ev[:direction] || "inbound",
        recent_events: recent_events_for(person, ev),
        dialed_did: ev[:dialed_did],
        campaign: @store.campaigns.dig(ev[:campaign_id], :name)
      }
    end

    # operator sets notes + resolution (advisory AI never touches resolution)
    def set_notes(call_id:, notes: nil, resolution: nil)
      ev = @store.event_by_call_id(call_id) or return nil
      ev[:notes] = notes unless notes.nil?
      # only touch resolution when a VALID one is supplied — a notes-only update must not clear it
      ev[:resolution] = resolution if RESOLUTION.key?(resolution)
      @store.put_event(ev)
      event_view(ev)
    end

    # ---- reverse channel: operator PC issues a call-control command to the device ----
    # Enqueues for the call's device; the real device long-polls drain_commands, applies via
    # Telecom, and the resulting state returns through the existing webhooks (research §3).
    # Returns nil (unknown call) or an error hash (bad verb/arg); never mutates call state here.
    def issue_command(call_id:, verb:, arg: nil)
      verb = verb.to_s
      return { error: "unknown_verb", verb: verb } unless COMMAND_VERBS.include?(verb)
      return { error: "bad_arg", verb: verb } if verb == "dtmf" && !valid_dtmf?(arg)
      ev = @store.event_by_call_id(call_id) or return nil
      device = ev[:device_id] || "unrouted"
      cmd = { id: SecureRandom.uuid, call_id: call_id, verb: verb, arg: arg,
              issued_at: now.iso8601, status: "pending" }
      @store.commands[device] << cmd
      @cmd_cond.broadcast # wake a long-poll drain for this device -> reverse-command latency ~1 RTT
      { command_id: cmd[:id], call_id: call_id, verb: verb, queued_for: device }
    end

    # Click-to-dial: the console asks the office phone to place an outgoing call. DEVICE-scoped (there is
    # no call yet) — enqueue a `dial` command carrying the number; the phone's CommandChannelClient places
    # it via TelecomManager, and the resulting call flows back through /calls/outgoing → record → display
    # like any other outbound call (no call_id correlation needed — the phone mints it on the outbound edge).
    # NOTE: the operator still talks through the Pixel; the PC can't carry call audio (audio topology).
    def dial(number:, device_id: nil)
      # E.164-only: NO raw-string passthrough. Phone.e164 returns nil for anything it can't normalize,
      # which blocks MMI/USSD codes (*21*<n>#, ##002#), premium/short codes, and injection into the
      # tel: URI on the device — the operator can only place real E.164 phone calls.
      num = Phone.e164(number)
      return { error: "bad_number" } unless num
      device = device_id || most_recent_device
      # Only dial a device that's actually connected (polled within the presence TTL) — otherwise the
      # command would sit queued for a disconnected phone and fire a stale number on reconnect.
      return { error: "no_device" } unless device && device_connected?(device)
      q = @store.commands[device]
      return { error: "device_busy" } if q.size >= MAX_PENDING_COMMANDS # bound the queue; refuse runaway
      cmd = { id: SecureRandom.uuid, call_id: nil, verb: "dial", arg: num, issued_at: now.iso8601, status: "pending" }
      q << cmd
      @cmd_cond.broadcast # wake the device's long-poll so it dials in ~1 RTT
      { command_id: cmd[:id], verb: "dial", number: num, queued_for: device }
    end

    # the office phone most recently seen on the reverse channel / heartbeat — the click-to-dial target
    # when the console doesn't name one (single-device deployment).
    def most_recent_device
      mono = Process.clock_gettime(Process::CLOCK_MONOTONIC)
      @device_seen.reject { |id, t| id.nil? || id == "unrouted" || (mono - t) > DEVICE_PRESENCE_TTL_S }
                  .max_by { |_id, t| t }&.first
    end

    def device_connected?(id)
      t = @device_seen[id] or return false
      (Process.clock_gettime(Process::CLOCK_MONOTONIC) - t) <= DEVICE_PRESENCE_TTL_S
    end

    # Device drains its pending commands. LONG-POLL: block up to wait_s for a command, returning the
    # instant issue_command enqueues one (via @cmd_cond) — this drops reverse-command latency from the
    # client poll interval to ~1 RTT, which matters because mute/hangup are safety controls (up to 2 s of
    # live mic after "mute" is a PII leak). @cmd_cond.wait releases @lock while blocked, so other requests
    # keep flowing. wait_s: 0 (default) = plain drain (dev endpoints + tests); wait_s > 0 = the phone's
    # long-poll, which is how commands reach the device in production (it holds GET /device/commands open).
    def drain_commands(device_id, wait_s: 0)
      id = device_id || "unrouted"
      @device_seen[id] = Process.clock_gettime(Process::CLOCK_MONOTONIC) # this poll = the phone is present
      deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + wait_s.to_f
      loop do
        cmds = @store.commands[id]; @store.commands[id] = []
        return cmds unless cmds.empty?
        remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
        return [] if remaining <= 0
        @cmd_cond.wait(remaining) # released while waiting; woken by issue_command.broadcast or timeout
      end
    end

    # Idle heartbeat: a second presence signal alongside the (now always-on) command long-poll. The
    # phone's AlarmManager pings this every ~2 min; it survives the command service being killed and
    # recreated. Same effect as a /device/commands poll — marks the device seen — but never touches
    # the command queue.
    def device_heartbeat(device_id)
      @device_seen[device_id || "unrouted"] = Process.clock_gettime(Process::CLOCK_MONOTONIC)
      { ok: true }
    end

    # Phone presence for the console's "Waiting for the phone to connect" indicator. A device counts as
    # connected if it polled /device/commands OR sent an idle heartbeat within DEVICE_PRESENCE_TTL_S.
    # Returns overall connected + per-device last-seen.
    def device_status
      mono = Process.clock_gettime(Process::CLOCK_MONOTONIC)
      devices = @device_seen.map do |id, t|
        { device_id: id, last_seen_ago_s: (mono - t).round(1), connected: (mono - t) <= DEVICE_PRESENCE_TTL_S }
      end
      { connected: devices.any? { |d| d[:connected] }, devices: devices }
    end

    # simulated device-side executor (the demo stands in for the InCallService's Telecom calls).
    # answer/reject/hangup drive the real state machine (as the device's webhook would); the live
    # in-call toggles (mute/hold/route/dtmf) set call attributes the console reflects. Idempotent:
    # answered/ended are monotonic, and the toggles are set-to-value.
    def apply_command(cmd)
      cid = cmd[:call_id]
      case cmd[:verb]
      when "answer" then answered(call_id: cid, ts: now.iso8601)
      when "reject" then ended(call_id: cid, ts: now.iso8601, disconnect_cause: "rejected")
      when "hangup" then ended(call_id: cid, ts: now.iso8601, disconnect_cause: "local")
      when "mute"   then set_attr(cid, :muted, true)
      when "unmute" then set_attr(cid, :muted, false)
      when "hold"   then set_attr(cid, :on_hold, true)
      when "resume" then set_attr(cid, :on_hold, false)
      when "dtmf"   then (ev = @store.event_by_call_id(cid)) && (ev[:dtmf_log] << cmd[:arg]) && @store.put_event(ev)
      when "route"  then set_attr(cid, :audio_route, cmd[:arg])
      end
      ev = @store.event_by_call_id(cid)
      { command_id: cmd[:id], verb: cmd[:verb], applied: !ev.nil?, event: ev && event_view(ev) }
    end

    # recent call log, optionally filtered by operator queue
    def recent_events(operator: nil, limit: 20)
      @store.events.values
            .select { |e| operator.nil? || operator == "all" || e[:operator_context] == operator }
            .sort_by { |e| e[:started_at].to_s }.reverse.first(limit)
            .map { |e| event_view(e) }
    end

    # ---- people directory + per-caller context (CRM-lite) --------------------------------------------
    # The operator maintains a persistent free-text context per caller ("what should we remember about
    # this person") and can browse that caller's whole call history. Distinct from per-call notes.
    def people_list
      @store.people.values.map { |p| person_summary(p) }
            .sort_by { |p| [-(p[:call_count] || 0), p[:name].to_s] }
    end

    def person_detail(person_id)
      p = @store.people[person_id.to_i] or return nil
      suggested = p[:position].to_s.strip.empty? ? latest_suggested_position(p[:id]) : nil
      person_summary(p).merge(
        context: p[:context].to_s, suggested_position: suggested,
        ai_context: p[:ai_context] || { "summary" => "", "other_context" => [] },
        ai_context_edited: !!p[:ai_context_edited],
        ai_context_pinned_sections: p[:ai_context_pinned_sections] || {},
        resolutions_ledger: p[:resolutions_ledger] || [],
        recently_solved:   recently_by_resolution(p[:id], "solved"),
        recently_unsolved: recently_by_resolution(p[:id], "unsolved"),
        recently_partial:  recently_by_resolution(p[:id], "partial"),
        undigested_calls:  undigested_count(p[:id]),
        events: person_events(p[:id]).map { |e| event_view(e) })
    end

    def set_person_context(person_id:, context:)
      p = @store.people[person_id.to_i] or return nil
      p[:context] = context.to_s
      p[:notes_rev] = (p[:notes_rev] || 0) + 1
      @store.put_person(p)
      person_detail(p[:id])
    end

    # Apply an Event's ledger_deltas (vectors already attached by transcribe_async) to the caller's
    # resolutions_ledger via the pure Ledger reducer. No network — safe under the store lock. Idempotent.
    def apply_ledger_deltas(person_id:, event_id:)
      p  = @store.people[person_id] or return nil
      ev = @store.events[event_id]  or return nil
      entry_vec = lambda do |entry|
        ref = entry["vec_ref"] or return nil
        src = @store.events[ref["event_id"]] or return nil
        packed = (src[:ledger_deltas] || [])[ref["idx"].to_i]&.dig("question_vec")
        Ledger.unpack_vec(packed)
      end
      r = Ledger.apply(ledger: p[:resolutions_ledger] || [], deltas: ev[:ledger_deltas] || [],
                       event_id: event_id, started_at: ev[:started_at],
                       applied_event_ids: p[:ledger_applied_events] || [],
                       entry_vec: entry_vec, next_id: -> { "led#{@store.next_id(:ledger)}" })
      p[:resolutions_ledger]   = r[:ledger]
      p[:ledger_applied_events] = r[:applied_event_ids]
      @store.put_person(p)
      { person_id: person_id, ledger_size: r[:ledger].size, changed: r[:changed] }
    end

    # active (non-superseded) ledger for the summariser prompt + schema enum; capped, minimal shape.
    def person_active_ledger(person_id)
      p = @store.people[person_id.to_i] or return []
      (p[:resolutions_ledger] || []).reject { |e| e["superseded_by"] }
        .last(50).map { |e| { "id" => e["id"], "question" => e["question"], "decision" => e["decision"] } }
    end

    # compare-and-set write of the rendered dossier (called by ContextMaintainer AFTER its LLM call, under
    # the lock). Rejects if ai_context_rev moved since the maintainer's snapshot (a concurrent edit/rebuild).
    def set_person_ai_context(person_id:, ai_context:, expected_rev:, built_from: nil)
      p = @store.people[person_id] or return nil
      # Delegate the compare-and-set to the store so the RMW is atomic + row-locked (SELECT..FOR UPDATE
      # in PgStore). Returns the updated person on success, nil if the rev moved (concurrent edit/rebuild).
      updated = @store.cas_person_ai_context(person_id, expected_rev, ai_context, built_from)
      return { stale: true, rev: p[:ai_context_rev] || 0 } if updated.nil?
      { ok: true, rev: updated[:ai_context_rev] }
    end

    # operator pen-edit of ONE dossier section (summary=String, other_context=Array). Pins it verbatim and
    # guards it from AI overwrite until released; bumps rev so an in-flight maintainer write goes stale.
    def edit_person_ai_context(person_id:, section:, value:)
      p = @store.people[person_id.to_i] or return nil
      section = section.to_s
      return { error: "bad_section" } unless %w[summary other_context].include?(section)
      p[:ai_context] ||= { "summary" => "", "other_context" => [] }
      p[:ai_context][section] = value
      (p[:ai_context_pinned_sections] ||= {})[section] = value
      p[:ai_context_edited] = true
      p[:ai_context_rev] = (p[:ai_context_rev] || 0) + 1
      @store.put_person(p)
      person_detail(p[:id])
    end

    def release_ai_context_section(person_id:, section:)
      p = @store.people[person_id.to_i] or return nil
      (p[:ai_context_pinned_sections] ||= {}).delete(section.to_s)
      p[:ai_context_edited] = !p[:ai_context_pinned_sections].empty?
      p[:ai_context_rev] = (p[:ai_context_rev] || 0) + 1
      @store.put_person(p)
      person_detail(p[:id])
    end

    # Resolve a flagged contradiction. grep anchor: resolve-contradiction.
    # - no `decision` (or the same as the current one): ACKNOWLEDGE — the current decision stands, the ⚠
    #   clears. - a different `decision`: the operator OVERRIDES. Append an operator-sourced entry with the
    # chosen value and supersede the current one, so the ledger stays append-only (nothing is rewritten) and
    # the caller's active decision becomes the operator's call. Either way the ⚠ clears on the entry + the
    # predecessors it superseded.
    LEDGER_DECISIONS = %w[yes no conditional].freeze
    def resolve_contradiction(person_id:, entry_id:, decision: nil, by: nil)
      p = @store.people[person_id.to_i] or return nil
      ledger = (p[:resolutions_ledger] ||= [])
      e = ledger.find { |x| x["id"] == entry_id.to_s } or return { error: "not_found" }
      decision = decision.to_s
      if LEDGER_DECISIONS.include?(decision) && decision != e["decision"]
        neu = operator_ledger_entry(e["question"], decision, by).merge("acknowledged" => true)
        e["superseded_by"] = neu["id"]
        ledger << neu
      end
      ledger.each { |x| next unless x["id"] == e["id"] || x["superseded_by"] == e["id"]; x["acknowledged"] = true; x.delete("contradiction") }
      @store.put_person(p)  # resolutions_ledger is a jsonb column — persist the whole person row
      person_detail(p[:id])
    end

    # Operator-entered resolution: a yes/no decision the operator records by hand (e.g. from notes or a
    # policy) rather than one extracted from a call. If an active entry already answers the same question,
    # supersede it (append-only) so this becomes the caller's current answer. grep anchor: add-resolution.
    def add_resolution(person_id:, question:, decision:, by: nil)
      p = @store.people[person_id.to_i] or return nil
      q = question.to_s.strip
      return { error: "bad_question" } if q.empty?
      return { error: "bad_decision" } unless LEDGER_DECISIONS.include?(decision.to_s)
      ledger = (p[:resolutions_ledger] ||= [])
      nq = q.downcase.gsub(/\s+/, " ")
      prev = ledger.find { |e| !e["superseded_by"] && e["question"].to_s.downcase.gsub(/\s+/, " ") == nq }
      neu = operator_ledger_entry(q, decision.to_s, by)
      if prev
        prev["superseded_by"] = neu["id"]; prev.delete("contradiction")
        neu["first_at"] = prev["first_at"] if prev["first_at"]
      end
      ledger << neu
      @store.put_person(p)
      person_detail(p[:id])
    end

    # Edit any resolution, any time - not only conflicted or freshly-added ones. Append-only: this
    # supersedes the named active entry with an operator entry carrying the new decision and/or wording,
    # so history is preserved and the newest entry is the caller's current answer. grep anchor:
    # edit-resolution.
    def edit_resolution(person_id:, entry_id:, question:, decision:, by: nil)
      p = @store.people[person_id.to_i] or return nil
      q = question.to_s.strip
      return { error: "bad_question" } if q.empty?
      return { error: "bad_decision" } unless LEDGER_DECISIONS.include?(decision.to_s)
      ledger = (p[:resolutions_ledger] ||= [])
      cur = ledger.find { |e| e["id"] == entry_id && !e["superseded_by"] } or return { error: "not_found" }
      neu = operator_ledger_entry(q, decision.to_s, by)
      neu["first_at"] = cur["first_at"] if cur["first_at"]
      cur["superseded_by"] = neu["id"]; cur.delete("contradiction")
      ledger << neu
      @store.put_person(p)
      person_detail(p[:id])
    end

    # deterministic replay of the reducer over the caller's events in ASCENDING started_at order — the
    # canonical rebuild. Pure: vectors resolved from Events, no network.
    def recompute_ledger(person_id:)
      p = @store.people[person_id.to_i] or return nil
      replay_person_ledger(p)
      { person_id: p[:id], ledger_size: p[:resolutions_ledger].size }
    end

    # operator creates a contact in the console. numbers are free-form strings (E.164-normalized here);
    # position is a free-text caller role/title. Returns the person_detail view, or an error hash.
    def create_person(name:, operator: nil, position: nil, numbers: [], email: nil, context: nil)
      return { error: "name_required" } if name.to_s.strip.empty?
      op = valid_operator(operator)
      @store.transaction do
        id = @store.next_id(:person)
        @store.put_person({ id: id, name: name.to_s.strip, operator: op,
                            position: blank_to_nil(position), context: context, email: blank_to_nil(email) })
        Array(numbers).each { |raw| add_number(person_id: id, raw: raw) }
        replay_person_ledger(@store.people[id]) if backlink_events_for(id) > 0
        person_detail(id)
      end
    end

    # operator edits a contact. Only supplied fields change (nil = leave as-is). `numbers`, when given,
    # REPLACES this person's number set. Never touches call history; context is edited via its own path
    # but may also be set here.
    def update_person(person_id:, name: nil, operator: nil, position: nil, numbers: nil, email: nil, context: nil)
      p = @store.people[person_id.to_i] or return nil
      @store.transaction do
        p[:name]     = name.to_s.strip unless name.nil? || name.to_s.strip.empty?
        p[:operator] = valid_operator(operator) unless operator.nil?
        p[:email]    = blank_to_nil(email) unless email.nil?
        p[:notes_rev] = (p[:notes_rev] || 0) + 1 if !context.nil? && p[:context] != context.to_s
        p[:context]  = context.to_s unless context.nil?
        p[:position] = blank_to_nil(position) unless position.nil?
        @store.put_person(p)
        unless numbers.nil?  # replace only THIS person's numbers (leave others' numbers untouched)
          @store.phone_numbers.select { |_e, pn| pn[:person_id] == p[:id] }.keys.each { |e| @store.delete_phone_number(e) }
          Array(numbers).each { |raw| add_number(person_id: p[:id], raw: raw) }
        end
        person_detail(p[:id])
      end
    end

    # operator deletes a contact. Their past call events are KEPT (they reference person_id, which now
    # dangles → those calls show as unknown/number-only), so call history is never silently lost.
    def delete_person(person_id:)
      p = @store.people[person_id.to_i] or return nil
      @store.transaction do
        @store.phone_numbers.select { |_e, pn| pn[:person_id] == p[:id] }.keys.each { |e| @store.delete_phone_number(e) }
        @store.delete_person(p[:id])  # events keep their (now-dangling) person_id — history is never lost
      end
      { deleted: p[:id] }
    end

    # ---- phone contacts sync (A): the office Pixel pushes its address book here ----
    # Ownership: the PHONE owns identity (name/email/numbers) — matched/UPSERTED by E.164 number — while
    # the CONSOLE owns `context` (never touched here). Upsert-only: a contact removed on the phone is NOT
    # deleted server-side (it still owns call history + context). `source:"phone"` marks synced people.
    def upsert_contacts(contacts)
      upserted = 0
      Array(contacts).each do |c|
        c = c.transform_keys(&:to_s)
        name = c["name"].to_s.strip
        nums = Array(c["numbers"]).map { |x| Phone.e164(x) }.compact.uniq
        next if name.empty? || nums.empty?
        email = Array(c["emails"]).map { |e| e.to_s.strip }.find { |e| !e.empty? }
        @store.transaction do
          existing_id = nums.map { |e| @store.phone_numbers[e]&.dig(:person_id) }.compact.first
          if existing_id
            p = @store.people[existing_id]
            p[:name]  = name            # phone owns identity
            p[:email] = email if email  # phone owns email (fills/refreshes; leaves it if the phone has none)
            p[:source] = "phone"
            @store.put_person(p)
          else
            id = @store.next_id(:person)
            @store.put_person({ id: id, name: name, operator: "unrouted", position: nil,
                                context: nil, email: email, source: "phone" })
            existing_id = id
          end
          nums.each { |e| add_number(person_id: existing_id, raw: e) } # attach all of the contact's numbers
        end
        upserted += 1
      end
      { upserted: upserted }
    end

    # KPI + billing rollup over the Event stream (see lib/helpdesk/reporting.rb)
    def report(**opts) = Report.build(@store.events.values.map { |e| event_view(e) }, **opts)

    # compact, UI-friendly projection of an Event
    def event_view(ev)
      person = ev[:person_id] && @store.people[ev[:person_id]]
      { id: ev[:id], call_id: ev[:call_id], state: ev[:state],
        person_id: ev[:person_id], person_name: person&.dig(:name),
        position: person && person[:position],
        operator: ev[:operator_context], campaign: @store.campaigns.dig(ev[:campaign_id], :name),
        number: ev[:number], dialed_did: ev[:dialed_did], direction: ev[:direction] || "inbound",
        started_at: ev[:started_at], answered_at: ev[:answered_at],
        ended_at: ev[:ended_at], duration_s: ev[:duration_s], notes: ev[:notes],
        transcript: ev[:transcript], ai_summary: ev[:ai_summary], action_items: ev[:action_items],
        suggested_position: ev[:suggested_position],
        recording_url: ev[:recording_url], resolution: ev[:resolution], flags: ev[:flags],
        muted: ev[:muted], on_hold: ev[:on_hold], audio_route: ev[:audio_route],
        failure_reason: ev[:failure_reason], prior_event_id: ev[:prior_event_id] }
    end

    # Atomic snapshot of everything the ContextMaintainer needs (taken under the lock; the LLM call then
    # runs OUTSIDE the lock in the server). Read-op only.
    def context_maintainer_input(person_id, mode: :incremental)
      p = @store.people[person_id.to_i] or return nil
      evs = person_events(p[:id]) # newest-first
      since = p.dig(:ai_context_built_from, "last_event_id")
      events = (mode == :incremental && since) ? evs.select { |e| e[:id].to_i > since.to_i } : evs
      { person_id: p[:id],
        ai_context: p[:ai_context] || { "summary" => "", "other_context" => [] },
        ai_context_rev: p[:ai_context_rev] || 0,
        pinned_sections: p[:ai_context_pinned_sections] || {},
        digests: events.map { |e| e[:context_digest].to_s }.reject(&:empty?),
        caller_facts: events.flat_map { |e| Array(e[:caller_facts]) }.uniq,
        ledger: (p[:resolutions_ledger] || []).reject { |e| e["superseded_by"] }
                  .map { |e| { "question" => e["question"], "decision" => e["decision"], "quote" => e["quote"] } },
        notes: p[:context].to_s,
        transcripts: (mode == :rebuild ? evs.first(5).map { |e| e[:transcript].to_s }.reject(&:empty?) : []),
        last_event_id: evs.first&.dig(:id),
        notes_rev: p[:notes_rev] || 0,
        rebuilt_rev: (p.dig(:ai_context_built_from, "rebuilt_rev") || 0) }
    end

    # --- thread-safety ------------------------------------------------------------------------------
    # WEBrick serves each request on its own thread AND the transcription worker mutates the store from a
    # background thread. Without serialization, concurrent Hash iteration/mutation raises "can't add a new
    # key into hash during iteration" and corrupts the store (both reproduced in the pre-flash review).
    # Wrap every public state op in ONE reentrant Monitor — reentrant so nested public calls
    # (apply_command -> answered/ended) don't self-deadlock. Ample for the single-device prototype; the
    # Rails port replaces this with row locks / DB transactions. Seeding helpers run pre-boot (single-
    # threaded) so they're intentionally left unwrapped.
    SYNCHRONIZED = %i[
      incoming outgoing answered ended recording_uploaded mark_transcribed mark_summarised mark_failed mark_no_speech
      reconcile! drain_broadcasts set_notes issue_command dial drain_commands apply_command
      recent_events report event_by_call_id device_status device_heartbeat
      people_list person_detail set_person_context create_person update_person delete_person upsert_contacts
      apply_ledger_deltas person_active_ledger
      set_person_ai_context edit_person_ai_context release_ai_context_section resolve_contradiction recompute_ledger
      context_maintainer_input
    ].freeze
    # Methods that change DURABLE state (people/orgs/numbers/events) — after these run, notify @after_write
    # so the server can persist. Transient mutations (issue_command/drain_commands over @store.commands)
    # are excluded; apply_command isn't listed because it delegates to answered/ended, which are.
    WRITE_OPS = %i[
      incoming outgoing answered ended recording_uploaded mark_transcribed mark_summarised mark_failed mark_no_speech
      reconcile! set_notes set_person_context create_person update_person delete_person upsert_contacts
      apply_ledger_deltas
      set_person_ai_context edit_person_ai_context release_ai_context_section resolve_contradiction recompute_ledger
    ].freeze
    SYNCHRONIZED.each do |name|
      orig = instance_method(name)
      writes = WRITE_OPS.include?(name)
      define_method(name) do |*args, **kwargs, &blk|
        @lock.synchronize do
          result = orig.bind(self).call(*args, **kwargs, &blk)
          @after_write.call if writes && @after_write
          result
        end
      end
    end

    private

    # A fresh operator-sourced ledger entry (the shape add_resolution / edit_resolution append). Random
    # led-op id so it never collides with next_id on seeded or imported ledgers.
    # Shared shape for every operator-sourced ledger entry. Random id, not next_id(:ledger): it can't
    # collide with existing ids on hand-seeded/imported ledgers where the sequence isn't ahead of the
    # entry ids. `by` is the operator's cert CN (server.rb operator_cn) so the ledger says WHO decided,
    # not just that a human did; omitted when the identity is unknown (local dev with auth off).
    def operator_ledger_entry(question, decision, by = nil)
      e = { "id" => "led-op-#{SecureRandom.hex(4)}", "question" => question, "decision" => decision,
            "first_at" => now.iso8601, "last_at" => now.iso8601, "last_event_id" => nil,
            "quote" => "", "source" => "operator" }
      who = by.to_s.strip
      e["by"] = who unless who.empty?
      e
    end

    # Reattach a recording that was PUT before its /calls/incoming arrived (the quarantine race). Called
    # from incoming() once the Event exists; advances it and kicks off transcription. No-op if none.
    def attach_quarantined_recording(ev)
      uuid, q = @store.quarantine.find { |_u, qq| qq[:call_id] == ev[:call_id] }
      return unless q
      @store.delete_quarantine(uuid)  # caller (incoming/outgoing) persists ev via put_event
      ev[:recording_uuid] = uuid
      ev[:recording_url]  = q[:url]
      ev[:audio_path]     = q[:audio_path] if q[:audio_path]
      advance!(ev, "recording_uploaded")
      enqueue_transcription(ev)
    end

    def blank_event(id, call_id)
      { id: id, call_id: call_id, person_id: nil, campaign_id: nil, did_id: nil,
        recording_uuid: nil, recording_url: nil, started_at: nil, answered_at: nil, ended_at: nil,
        duration_s: nil, direction: "inbound", operator_context: nil, presentation: nil,
        state: nil, notes: nil, transcript: nil, ai_summary: nil, action_items: [],
        resolution: nil, prior_event_id: nil, pomoc_ticket_id: nil, device_id: nil, flags: [],
        muted: false, on_hold: false, audio_route: nil, dtmf_log: [],
        context_digest: nil, caller_facts: [], ledger_deltas: [] }
    end

    def ensure_event(call_id)
      @store.event_by_call_id(call_id) || begin
        ev = blank_event(@store.next_id(:event), call_id)
        advance!(ev, "ringing") # arrived out of order before /incoming
        @store.put_event(ev)    # explicit insert of the out-of-order event
        ev
      end
    end

    # monotonic forward-only transition. `failed` is a hard terminal; `recording_missing`
    # is recoverable — a late recording upload can still advance it.
    def advance!(ev, target)
      return set_state!(ev, target) if ev[:state].nil?
      return ev if ev[:state] == "failed"
      set_state!(ev, target) if STATE_RANK.fetch(target, -1) > effective_rank(ev[:state])
      ev
    end

    # recording_missing ranks as 'ended' so a late upload (higher rank) can revive it.
    def effective_rank(state)
      return STATE_RANK["ended"] if state == "recording_missing"
      STATE_RANK.fetch(state, -1)
    end

    def set_state!(ev, s) = (ev[:state] = s; ev)

    def close_missed(ev)
      ev[:ended_at] ||= now.iso8601; ev[:disconnect_cause] ||= "missed"
      set_state!(ev, "recording_missing")
    end

    def derive_duration(ev)
      return nil unless ev[:answered_at] && ev[:ended_at]
      (Time.parse(ev[:ended_at]) - Time.parse(ev[:answered_at])).round
    end

    def link_prior_event(ev, person)
      return unless person
      prior = @store.events.values
                    .select { |e| e[:person_id] == person[:id] && e[:id] != ev[:id] }
                    .max_by { |e| e[:started_at].to_s }
      ev[:prior_event_id] = prior&.dig(:id)
    end

    def recent_events_for(person, current)
      return [] unless person
      @store.events.values
            .select { |e| e[:person_id] == person[:id] && e[:id] != current[:id] }
            .sort_by { |e| e[:started_at].to_s }.reverse.first(5)
            .map { |e| { id: e[:id], at: e[:started_at], summary_line: e[:ai_summary],
                         resolution: e[:resolution] } }
    end

    def broadcast_screen_pop(ev, person)
      op = ev[:operator_context] || "unrouted"
      @store.broadcasts[op] << screen_pop_payload(ev, person)
    end

    # --- people helpers ---
    def person_numbers(person_id)
      @store.phone_numbers.values.select { |pn| pn[:person_id] == person_id }.map { |pn| pn[:e164] }
    end

    # backlink prior events (by matched number) to a person, but ONLY events that are unowned (nil) or
    # dangling (person_id points to a deleted person). Never steal an event owned by a LIVE person.
    def backlink_events_for(person_id)
      nums = person_numbers(person_id)
      return 0 if nums.empty?
      count = 0
      @store.events.each_value do |e|
        next unless e[:number] && nums.include?(e[:number])
        owner = e[:person_id]
        next if owner && @store.people.key?(owner)
        e[:person_id] = person_id; @store.put_event(e); count += 1
      end
      count
    end

    # pure ledger replay for one person (used by recompute_ledger + create_person backlink). No network.
    def replay_person_ledger(p)
      entry_vec = lambda do |entry|
        ref = entry["vec_ref"] or return nil
        src = @store.events[ref["event_id"]] or return nil
        Ledger.unpack_vec((src[:ledger_deltas] || [])[ref["idx"].to_i]&.dig("question_vec"))
      end
      ledger = []; applied = []
      @store.events.values.select { |e| e[:person_id] == p[:id] }
            .sort_by { |e| e[:started_at].to_s }.each do |ev|
        r = Ledger.apply(ledger: ledger, deltas: ev[:ledger_deltas] || [], event_id: ev[:id],
                         started_at: ev[:started_at], applied_event_ids: applied,
                         entry_vec: entry_vec, next_id: -> { "led#{@store.next_id(:ledger)}" })
        ledger = r[:ledger]; applied = r[:applied_event_ids]
      end
      p[:resolutions_ledger] = ledger; p[:ledger_applied_events] = applied
      @store.put_person(p)
    end

    def person_events(person_id)
      @store.events.values.select { |e| e[:person_id] == person_id }
            .sort_by { |e| e[:started_at].to_s }.reverse
    end
    # recently-* lists derived from the operator resolution enum (NOT the LLM) — can't contradict the
    # resolution badges. person_events is newest-first, so first(limit) = most recent.
    def recently_by_resolution(person_id, resolution, limit: 5)
      person_events(person_id).select { |e| e[:resolution] == resolution }.first(limit).map do |e|
        line = (e[:context_digest].to_s.split("\n").first || e[:ai_summary].to_s.split("\n").first).to_s.strip[0, 160]
        { at: e[:started_at], call_id: e[:call_id], line: line }
      end
    end

    # calls that happened but have no context_digest yet (still processing / needs_review) — provenance hint.
    def undigested_count(person_id)
      person_events(person_id).count do |e|
        e[:started_at] && e[:context_digest].to_s.empty? && !%w[ringing answered no_speech].include?(e[:state])
      end
    end
    # the caller's most recent AI-suggested position (person_events is newest-first), nil if none
    def latest_suggested_position(person_id)
      person_events(person_id).map { |e| e[:suggested_position] }.find { |x| !x.to_s.strip.empty? }
    end
    def person_summary(p)
      evs = person_events(p[:id])
      { id: p[:id], name: p[:name], operator: p[:operator], email: p[:email], position: p[:position],
        numbers: person_numbers(p[:id]),
        call_count: evs.size, last_call_at: evs.first&.dig(:started_at),
        has_context: !p[:context].to_s.strip.empty? }
    end

    # a person's operator is one of the routing queues (or unrouted). "all" is a filter, never stored.
    def valid_operator(op) = %w[tmobile vodafone unrouted].include?(op.to_s) ? op.to_s : "unrouted"
    def blank_to_nil(s) = s.to_s.strip.empty? ? nil : s.to_s.strip

    def flag(ev, f)   = (ev[:flags] |= [f.to_s])
    def unflag(ev, f) = (ev[:flags].delete(f.to_s))
    def age(iso, t)   = iso ? (t - Time.parse(iso)) : Float::INFINITY

    def set_attr(cid, key, val)
      ev = @store.event_by_call_id(cid) or return nil
      ev[key] = val
      @store.put_event(ev)   # persist the live-call attr (fixes the pre-PG WRITE_OPS gap for mute/hold/route)
    end
    def valid_dtmf?(arg) = arg.to_s.match?(/\A[0-9*#]\z/)
  end
end