# frozen_string_literal: true # Helpdesk backend HTTP layer — WEBrick + the Ruby standard library. This is the deployed service: # the operator console + operator API and the phone's device API, all in one process. The PostgreSQL # store (lib/helpdesk/pg_store.rb) adds pg + sequel; the JSON-store default needs neither, so # `ruby server.rb` runs with no gems and no database. The store backend is chosen by # HELPDESK_DATABASE_URL (set = PostgreSQL, unset = JSON file). # Device auth: bearer + HMAC + nonce replay (authed?). Operator auth: mTLS terminated at the proxy, # which injects the identity headers (operator_authed?). grep anchors: route, authed?, operator_authed?. # Run: ruby components/backend/server.rb (PORT=4000) require "webrick" require "json" require "openssl" require "securerandom" require "time" require "fileutils" require "tmpdir" require_relative "lib/helpdesk/domain" require_relative "lib/helpdesk/business_hours" require_relative "lib/helpdesk/cert_watch" require_relative "pipeline" require_relative "wiki/embedder" require_relative "context_maintainer" require_relative "seed" module Helpdesk class Server DEVICE_TOKEN = ENV.fetch("HELPDESK_DEVICE_TOKEN", "dev-device-token") DEVICE_SECRET = ENV.fetch("HELPDESK_DEVICE_SECRET", "dev-device-secret") AUTH_ENFORCE = ENV.fetch("HELPDESK_AUTH", "1") == "1" DEV = ENV.fetch("HELPDESK_DEV", "1") == "1" # operator-console demo endpoints # Operator/console auth: operators sign in with their vmin CLIENT CERT (mTLS), terminated at the # reverse proxy which injects the identity + a shared secret. Enforced by default off DEV. See # operator_authed? below. OP_AUTH_ENFORCE = ENV.fetch("HELPDESK_OP_AUTH", DEV ? "0" : "1") == "1" PROXY_SECRET = ENV["HELPDESK_PROXY_SECRET"] # shared secret proving a request came via the mTLS proxy # Which client-cert holders count as operators, by CN, comma-separated. The proxy only forwards certs # its CA verified, but that CA (ca.vmin.cz) belongs to the hosting provider, so "signed by the CA" is # authentication, not authorization - this pins WHICH holders may use the console. It is also the only # revocation we own: no CRL is configured, so dropping a CN here is how a lost laptop's cert is turned # off. Unset = accept any cert the proxy verified, which is the pre-allowlist behaviour and keeps # local dev working. grep anchor: operator-allowlist. OPERATOR_CNS = ENV.fetch("HELPDESK_OPERATOR_CNS", "").split(",").map { |s| s.strip.downcase } .reject(&:empty?).freeze SEED_ON_EMPTY = ENV.fetch("HELPDESK_SEED", "1") == "1" # prod sets HELPDESK_SEED=0 to boot an empty store (no demo seed) TRANSCRIBE = ENV.fetch("HELPDESK_TRANSCRIBE", "1") == "1" # CI e2e sets 0 to skip the transcription pipeline (no Whisper on the runner) MAX_SKEW_S = 300 MAX_BODY_BYTES = 80 * 1024 * 1024 # app-level backstop; nginx enforces the real per-path caps (64k/64m) class BodyTooLarge < StandardError; end CONSOLE = File.join(__dir__, "public", "console.html") DEV_SCENARIOS = [ { number: "+420601234567", did: "+420800111222" }, # Jan Novák — T-Mobile { number: "+420602345678", did: "+420800333444" }, # Petr Svoboda — Vodafone { number: "+420777123456", did: "+420800111222" }, # unknown caller — T-Mobile queue ].freeze def initialize(port: (ENV["PORT"] || 4000).to_i) # Fail-closed: never run a production (non-dev) server with the repo-public dev credentials — a # missed provisioning step would otherwise accept any device signing with the known secret. if AUTH_ENFORCE && !DEV && (DEVICE_TOKEN == "dev-device-token" || DEVICE_SECRET == "dev-device-secret") abort "[helpdesk] refusing to start: device credentials are the public dev defaults on a non-dev " \ "build — set HELPDESK_DEVICE_TOKEN + HELPDESK_DEVICE_SECRET (on the server: /etc/helpdesk/env)." end # Loud posture warning: DEV=1 turns operator auth OFF (console, recordings/transcripts=PII, /dial # all become unauthenticated) and enables the /dev/* call-injection endpoints. Fine for local # testing on loopback; dangerous if this binds a LAN/public interface. The fail-closed check above # only guards *device* creds, so warn explicitly here so a "just run it on the office wifi" deploy # isn't silently wide open. Set HELPDESK_DEV=0 (+ the mTLS proxy) for anything real. if DEV && !OP_AUTH_ENFORCE warn "[helpdesk] WARNING: running in DEV mode — operator endpoints (console, /rec recordings, " \ "/dial) are UNAUTHENTICATED and /dev/* injection is enabled. Do NOT expose this beyond " \ "localhost/trusted-LAN. Set HELPDESK_DEV=0 + HELPDESK_OP_AUTH=1 (mTLS proxy) for production." end # FIX 5b: initialise latch + context maintainer once in the constructor to avoid init races. @latch = {} @latch_mu = Mutex.new @ctx = Helpdesk::ContextMaintainer.new # Durable store. Backend is selected by env (dev/prod parity + rollback): HELPDESK_DATABASE_URL set # -> PostgreSQL (writes are immediate, no snapshot flusher); unset -> the in-memory JSON-file Store # (the debounced flusher below). Rollback = unset HELPDESK_DATABASE_URL; the JSON files are untouched. @state = ENV["HELPDESK_STATE"] || File.join(Dir.home, ".config", "helpdesk", "state.json") if (db_url = ENV["HELPDESK_DATABASE_URL"]) && !db_url.empty? require_relative "lib/helpdesk/pg_store" store = Helpdesk::PgStore.new(db_url) @svc = (store.people.empty? && SEED_ON_EMPTY) ? Seed.call(Service.new(store)) : Service.new(store) warn "[helpdesk] store backend: PostgreSQL (#{db_url.sub(%r{//[^@]*@}, '//***@')})" else FileUtils.mkdir_p(File.dirname(@state)) store = Store.load(@state) seeded = store.people.empty? @svc = (seeded && SEED_ON_EMPTY) ? Seed.call(Service.new(store)) : Service.new(store) @dirty = seeded # persist the initial seed on first boot @svc.after_write = -> { @dirty = true } Thread.new do # debounced persistence — coalesces write bursts into one snapshot every ~2s loop do sleep 2 next unless @dirty @dirty = false begin @svc.persist!(@state); rescue => e; warn "[helpdesk] persist failed: #{e.class}: #{e.message}"; end end end warn "[helpdesk] store backend: JSON file (#{@state})" end # Watchdog scheduler: reconcile! is the resilience layer (ring-timeout -> missed, recording-grace # -> recording_missing, quarantine-TTL expiry). Nothing else calls it, so without this thread a # lost /ended webhook leaves a call "ringing" forever and a failed upload never surfaces the # "recording never arrived" warning. Runs every 60s; one Event blowing up can't starve the loop # (reconcile! wraps each Event), and the whole tick is rescued so the thread never dies. Thread.new do loop do sleep 60 begin @svc.reconcile!; rescue => e; warn "[helpdesk] reconcile failed: #{e.class}: #{e.message}"; end end end @nonces = {} @nonce_lock = Mutex.new # @nonces is touched by every device request thread — guard it (see authed?) # Certificate expiry watch. nil unless something is configured, so local dev probes nothing and # /status keeps its old shape. See lib/helpdesk/cert_watch.rb (grep anchor: cert-watch-sources). cert_cfg = CertWatch.config_from_env @cert_warn_days = cert_cfg[:warn_days] @cert_watch = (CertWatch::Cache.new { CertWatch.check(**cert_cfg) } if CertWatch.configured?(cert_cfg)) # real transcription pipeline: hosted Whisper (WHISPER_URL) or local WhisperX, then the # summariser. Fired on a recording upload; runs off-thread so the upload returns immediately. @rec_dir = ENV["HELPDESK_REC_DIR"] || File.join(Dir.tmpdir, "helpdesk-recordings") FileUtils.mkdir_p(@rec_dir) @pipeline = Pipeline.build @svc.transcription_runner = ->(ev) { transcribe_async(ev) } if TRANSCRIBE # BindAddress: prod sets HELPDESK_BIND=127.0.0.1 so this plain-HTTP backend never listens on a public # interface — nginx (mTLS) is the only public entry. Unset (local dev) = bind all, exactly as before. @http = WEBrick::HTTPServer.new(Port: port, BindAddress: ENV["HELPDESK_BIND"], Logger: WEBrick::Log.new(File::NULL), AccessLog: []) @http.mount_proc("/healthz") { |_r, res| json(res, 200, ok: true, events: @svc.store.events.size) } @http.mount_proc("/api/v1") do |req, res| begin route(req, res) rescue BodyTooLarge json(res, 413, error: "body_too_large") end end # recording download for the console — recording_url is stored as "/rec/" (top-level, not # under /api/v1), so serve it here. Operator-authed like the console API (proxy injects the # headers in prod; permissive in dev). @http.mount_proc("/rec") { |req, res| operator_authed?(req) ? serve_recording(req, res) : unauthorized(res) } @http.mount_proc("/") { |_req, res| serve_console(res) } # operator console (catch-all) trap("INT") { @http.shutdown } trap("TERM") { @http.shutdown } end def start = @http.start private def route(req, res) path = req.path.sub(%r{\A/api/v1}, "") m = req.request_method case [m, path] in ["POST", "/calls/incoming"] then guarded(req, res) { |b| @svc.incoming(**sym(b).slice(:call_id, :ts, :number, :presentation, :dialed_did, :device_id)) } in ["POST", "/calls/outgoing"] then guarded(req, res) { |b| @svc.outgoing(**sym(b).slice(:call_id, :ts, :number, :device_id)) } in ["POST", "/calls/answered"] then guarded(req, res) { |b| @svc.answered(call_id: b["call_id"], ts: b["ts"]) } in ["POST", "/calls/ended"] then guarded(req, res) { |b| @svc.ended(**sym(b).slice(:call_id, :ts, :duration_s, :disconnect_cause, :recording_uuid)) } else case [m, path] when ["POST", "/reconcile"] then operator_authed?(req) ? (@svc.reconcile!; json(res, 200, reconciled: true)) : unauthorized(res) when ["POST", "/dial"] # click-to-dial: console tells the office phone to place an outgoing call if operator_authed?(req) b = (JSON.parse(read(req)) rescue {}) json(res, 200, @svc.dial(number: b["number"], device_id: b["device_id"])) else unauthorized(res) end else route_rest(m, path, req, res) end end end def route_rest(m, path, req, res) case when m == "PUT" && path =~ %r{\A/recordings/([\w-]+)\z} body = read(req) return unauthorized(res) unless authed?(req, body) uuid = $1; call_id = req["X-Call-Id"].to_s # Without a call_id the audio can only be quarantined under nil and orphaned — reject so the # device retries with the header rather than silently losing the only copy of the recording. return json(res, 400, error: "missing_call_id") if call_id.empty? audio_path = File.join(@rec_dir, "#{uuid}.audio"); File.binwrite(audio_path, body) r = @svc.recording_uploaded(recording_uuid: uuid, call_id: call_id, url: "/rec/#{uuid}", audio_path: audio_path) json(res, 200, r) when m == "GET" && path =~ %r{\A/status\z} now = BusinessHours.prague_now open = BusinessHours.open?(now) body = { business_open: open, prague_time: now.strftime("%Y-%m-%d %H:%M %A"), out_of_hours_sms: open ? nil : BusinessHours::OUT_OF_HOURS_SMS } # Certificate expiry rides along on the poll the console already makes every 30s rather than # adding a second one. Reads from a cache that never blocks this request; absent entirely when # nothing is configured, which is the local-dev case. certs = @cert_watch ? @cert_watch.value.dup : [] # Plus the caller's OWN certificate, which we could not otherwise see: there is one per operator, # they live in browsers, and we hold none of them. nginx tells us the expiry of the one just # presented, so each operator is warned about theirs and nobody else's. own = CertWatch.presented_entry(req["X-Operator-Cert-Days"], req["X-Operator-Cert-Expires"], warn_days: @cert_warn_days) certs << own if own body[:certs] = certs.sort_by { |c| c["days_left"] } unless certs.empty? json(res, 200, body) when m == "GET" && path =~ %r{\A/device/status\z} # phone presence for the console's "Waiting for the phone to connect" banner (operator-facing) return unauthorized(res) unless operator_authed?(req) json(res, 200, @svc.device_status) when m == "GET" && path =~ %r{\A/screen-pop\z} return unauthorized(res) unless operator_authed?(req) op = req.query["operator"] || "unrouted" json(res, 200, messages: @svc.drain_broadcasts(op)) when m == "GET" && path == "/events" return unauthorized(res) unless operator_authed?(req) json(res, 200, events: @svc.recent_events(operator: req.query["operator"], limit: (req.query["limit"] || 20).to_i)) when m == "GET" && path == "/report" return unauthorized(res) unless operator_authed?(req) q = req.query json(res, 200, @svc.report(operator: q["operator"], from: q["from"], to: q["to"])) when m == "GET" && path == "/people" return unauthorized(res) unless operator_authed?(req) json(res, 200, people: @svc.people_list) when m == "POST" && path == "/people" return unauthorized(res) unless operator_authed?(req) b = JSON.parse(read(req)) rescue {} v = @svc.create_person(name: b["name"], operator: b["operator"], position: b["position"], numbers: b["numbers"] || [], email: b["email"], context: b["context"]) (v && v[:error]) ? json(res, 422, v) : (refresh_context(v[:id], mode: :rebuild); json(res, 201, v)) when m == "GET" && path =~ %r{\A/people/(\d+)\z} return unauthorized(res) unless operator_authed?(req) p = @svc.person_detail($1) p ? json(res, 200, p.merge(ai_context_pending: context_pending?($1.to_i))) : json(res, 404, error: "not_found") when m == "PUT" && path =~ %r{\A/people/(\d+)\z} return unauthorized(res) unless operator_authed?(req) b = JSON.parse(read(req)) rescue {} v = @svc.update_person(person_id: $1, name: b["name"], operator: b["operator"], position: b["position"], numbers: b["numbers"], email: b["email"], context: b["context"]) v ? json(res, 200, v) : json(res, 404, error: "not_found") when m == "POST" && path =~ %r{\A/people/(\d+)/delete\z} # POST-as-delete: Ruby's WEBrick ProcHandler only routes GET/POST/PUT (no do_DELETE), so a real # DELETE 405s before reaching here. POST keeps it working without a custom servlet. return unauthorized(res) unless operator_authed?(req) v = @svc.delete_person(person_id: $1) v ? json(res, 200, v) : json(res, 404, error: "not_found") when m == "POST" && path =~ %r{\A/people/(\d+)/context\z} return unauthorized(res) unless operator_authed?(req) b = JSON.parse(read(req)) rescue {} v = @svc.set_person_context(person_id: $1, context: b["context"]) v ? json(res, 200, v) : json(res, 404, error: "not_found") when m == "POST" && path =~ %r{\A/people/(\d+)/notes/confirm\z} return unauthorized(res) unless operator_authed?(req) b = JSON.parse(read(req)) rescue {} @svc.set_person_context(person_id: $1, context: b["context"]) if b.key?("context") # accept the flushed notes refresh_context($1.to_i, mode: :rebuild) json(res, 202, ok: true) when m == "PUT" && path =~ %r{\A/people/(\d+)/ai-context\z} return unauthorized(res) unless operator_authed?(req) b = JSON.parse(read(req)) rescue {} v = @svc.edit_person_ai_context(person_id: $1, section: b["section"], value: b["value"]) (v && v[:error]) ? json(res, 422, v) : (v ? json(res, 200, v) : json(res, 404, error: "not_found")) when m == "POST" && path =~ %r{\A/people/(\d+)/ai-context/rebuild\z} return unauthorized(res) unless operator_authed?(req) refresh_context($1.to_i, mode: :rebuild); json(res, 202, ok: true) when m == "POST" && path =~ %r{\A/people/(\d+)/ai-context/(summary|other_context)/release\z} return unauthorized(res) unless operator_authed?(req) v = @svc.release_ai_context_section(person_id: $1, section: $2) v ? json(res, 200, v) : json(res, 404, error: "not_found") when m == "POST" && path =~ %r{\A/people/(\d+)/ledger/([\w-]+)/resolve\z} return unauthorized(res) unless operator_authed?(req) b = JSON.parse(read(req)) rescue {} # optional { "decision": "yes|no|conditional" } to override v = @svc.resolve_contradiction(person_id: $1, entry_id: $2, decision: b["decision"], by: operator_cn(req)) (v && v[:error]) ? json(res, 404, v) : (v ? json(res, 200, v) : json(res, 404, error: "not_found")) when m == "POST" && path =~ %r{\A/people/(\d+)/ledger\z} return unauthorized(res) unless operator_authed?(req) b = JSON.parse(read(req)) rescue {} # { "question": "...", "decision": "yes|no|conditional" } v = @svc.add_resolution(person_id: $1, question: b["question"], decision: b["decision"], by: operator_cn(req)) v.nil? ? json(res, 404, error: "not_found") : (v[:error] ? json(res, 422, v) : json(res, 201, v)) when m == "PUT" && path =~ %r{\A/people/(\d+)/ledger/([\w-]+)\z} return unauthorized(res) unless operator_authed?(req) b = JSON.parse(read(req)) rescue {} # edit any resolution: { "question": "...", "decision": "..." } v = @svc.edit_resolution(person_id: $1, entry_id: $2, question: b["question"], decision: b["decision"], by: operator_cn(req)) v.nil? ? json(res, 404, error: "not_found") : (v[:error] ? json(res, (v[:error] == "not_found" ? 404 : 422), v) : json(res, 200, v)) when m == "POST" && path =~ %r{\A/events/([\w-]+)/notes\z} return unauthorized(res) unless operator_authed?(req) b = JSON.parse(read(req)) rescue {} # FIX 3: detect resolution changes and dispatch an incremental context refresh before = @svc.event_by_call_id($1); old_res = before && before[:resolution] v = @svc.set_notes(call_id: $1, notes: b["notes"], resolution: b["resolution"]) refresh_context(v[:person_id], mode: :incremental) if v && v[:person_id] && v[:resolution] != old_res v ? json(res, 200, v) : json(res, 404, error: "not_found") when m == "POST" && path =~ %r{\A/calls/([\w-]+)/command\z} # operator PC issues a call-control command (reverse channel). Gated on the operator's vmin cert # (mTLS via the proxy) — this drives the live phone, so it must never be open on the LAN. The # device-authed drain (GET /device/commands) is the phone side of this same channel. return unauthorized(res) unless operator_authed?(req) b = JSON.parse(read(req)) rescue {} r = @svc.issue_command(call_id: $1, verb: b["verb"], arg: b["arg"]) r.nil? ? json(res, 404, error: "not_found") : json(res, r[:error] ? 422 : 202, r) when m == "GET" && path =~ %r{\A/device/commands\z} # the device long-polls its pending commands (device-authed, like the webhooks). ?wait= # (bounded 0..30) blocks the request until a command is enqueued -> ~1 RTT command latency. return unauthorized(res) unless authed?(req, read(req)) wait_s = [[req.query["wait"].to_i, 0].max, 30].min json(res, 200, commands: @svc.drain_commands(req.query["device_id"], wait_s: wait_s)) when m == "GET" && path =~ %r{\A/device/heartbeat\z} # idle presence ping (device-authed) — keeps "Phone connected" fresh between calls, when the # command long-poll isn't running. Marks the device seen; never drains the command queue. return unauthorized(res) unless authed?(req, read(req)) json(res, 200, @svc.device_heartbeat(req.query["device_id"])) when m == "POST" && path == "/device/contacts" # the office Pixel pushes its address book (device-authed, like the webhooks). Upsert-only; # the phone owns name/email/numbers, the console owns per-caller context (never touched here). body = read(req) return unauthorized(res) unless authed?(req, body) b = JSON.parse(body) rescue {} json(res, 200, @svc.upsert_contacts(b["contacts"] || [])) when m == "POST" && path =~ %r{\A/dev/(incoming|answer|end|drain)\z} && DEV dev_action($1, req, res) when m == "GET" && path =~ %r{\A/events/([\w-]+)\z} return unauthorized(res) unless operator_authed?(req) ev = @svc.event_by_call_id($1) # locked read (Service delegates to the store under @lock) ev ? json(res, 200, @svc.event_view(ev)) : json(res, 404, error: "not_found") else json(res, 404, error: "no_route", method: m, path: path) end end def serve_console(res) return json(res, 404, error: "console_missing") unless File.exist?(CONSOLE) res.status = 200 res["Content-Type"] = "text/html; charset=utf-8" res.body = File.read(CONSOLE) end # GET /rec/ — download the uploaded call audio. The device uploads raw AAC-in-MP4 (the phone # records .m4a), stored as .audio; serve it back as an attachment with a human filename built # from the call metadata so the operator gets "call__.m4a", not an opaque uuid. def serve_recording(req, res) uuid = req.path.sub(%r{\A/rec/?}, "") return json(res, 400, error: "bad_uuid") unless uuid.match?(/\A[\w-]+\z/) ev = @svc.store.events.values.find { |e| e[:recording_uuid] == uuid } path = ev && ev[:audio_path] return json(res, 404, error: "recording_not_found") unless path && File.exist?(path) stamp = (ev[:started_at] || "unknown").tr(":", "-") fname = "call_#{stamp}_#{ev[:number] || 'unknown'}.m4a".gsub(/[^\w.+-]/, "_") res.status = 200 res["Content-Type"] = "audio/mp4" res["Content-Disposition"] = "attachment; filename=\"#{fname}\"" res.body = File.binread(path) end # dev-only endpoints so the console demos a full call without the device def dev_action(kind, req, res) b = (JSON.parse(read(req)) rescue {}) case kind when "incoming" s = DEV_SCENARIOS.sample; cid = SecureRandom.uuid r = @svc.incoming(call_id: cid, ts: Time.now.utc.iso8601, number: s[:number], presentation: "allowed", dialed_did: s[:did], device_id: "dev-console") json(res, 200, r.merge(call_id: cid)) when "answer" json(res, 200, @svc.answered(call_id: b["call_id"], ts: Time.now.utc.iso8601)) when "end" json(res, 200, @svc.ended(call_id: b["call_id"], ts: Time.now.utc.iso8601, duration_s: b["duration_s"] || 60, disconnect_cause: "local")) when "drain" # simulate the device: drain its queued commands and apply them (closes the PC->device loop) applied = @svc.drain_commands(b["device_id"] || "dev-console").map { |c| @svc.apply_command(c) } json(res, 200, applied: applied) end end # run the transcription→summary pipeline for an uploaded recording, off the request thread. # Transcript is durable; summary is advisory (key-gated). Marks the Event so the console shows it. def transcribe_async(ev) Thread.new do path = ev[:audio_path] unless path && File.exist?(path) @svc.mark_failed(call_id: ev[:call_id], reason: "audio_missing"); next end ledger = ev[:person_id] ? @svc.person_active_ledger(ev[:person_id]) : [] r = @pipeline.process(path, operator: ev[:operator_context], ledger: ledger) case r[:stage] when "summarised" @svc.mark_transcribed(call_id: ev[:call_id], transcript: r[:transcript].to_s) deltas = attach_ledger_vectors(r[:ledger_deltas] || []) @svc.mark_summarised(call_id: ev[:call_id], summary: r[:summary], action_items: r[:action_items] || [], suggested_position: r[:suggested_position], context_digest: r[:context_digest], caller_facts: r[:caller_facts] || [], ledger_deltas: deltas) if ev[:person_id] @svc.apply_ledger_deltas(person_id: ev[:person_id], event_id: ev[:id]) refresh_context(ev[:person_id], mode: :incremental) # defined in Task 8 end when "transcribed" @svc.mark_transcribed(call_id: ev[:call_id], transcript: r[:transcript].to_s) when "no_speech" @svc.mark_no_speech(call_id: ev[:call_id]) else @svc.mark_failed(call_id: ev[:call_id], reason: r[:error] || "transcription_failed", retryable: r.dig(:transcription, :retryable)) end rescue => e warn "[helpdesk] pipeline error for #{ev[:call_id]}: #{e.class}: #{e.message}" @svc.mark_failed(call_id: ev[:call_id], reason: "pipeline_exception") rescue nil end end # Batch-embed the new (entry_id=null) decision questions and attach packed vectors. OUTSIDE the store # lock (this runs on the transcription thread). LOCAL rescue: an embed failure must never bubble to the # outer mark_failed (which would discard the good transcript) — deltas are kept, vecs left nil (the # reducer then flags them for review). def attach_ledger_vectors(deltas) news = deltas.select { |d| d["entry_id"].to_s.empty? } return deltas if news.empty? begin vecs = (@embedder ||= Helpdesk::Embedder.new).embed(news.map { |d| d["question"].to_s }) news.each_with_index { |d, i| d["question_vec"] = Helpdesk::Ledger.pack_vec(vecs[i]) if vecs[i] } rescue => e warn "[helpdesk] ledger embed failed (#{e.class}: #{e.message}) — deltas kept, vectors nil" end deltas end def refresh_context(person_id, mode: :incremental) return if person_id.nil? start = @latch_mu.synchronize do if @latch[person_id] then @latch[person_id] = :rerun; false else @latch[person_id] = :running; true end end return unless start Thread.new do clear = -> { @latch_mu.synchronize { @latch.delete(person_id) } } begin loop do inp = @svc.context_maintainer_input(person_id, mode: mode) (clear.call; break) unless inp # person deleted mid-run — clear the latch on this abnormal exit m = mode m = :rebuild if inp[:ai_context_rev].zero? && inp[:digests].any? # first-run legacy m = :rebuild if m == :incremental && (inp[:ai_context_rev] - inp[:rebuilt_rev]) >= 6 # threshold # FIX 1: if we upgraded to :rebuild but the snapshot was taken with a non-rebuild mode, # re-snapshot with :rebuild so ContextMaintainer gets full history (last-5 transcripts). if m == :rebuild && mode != :rebuild inp = @svc.context_maintainer_input(person_id, mode: :rebuild) (clear.call; break) unless inp # person deleted mid-run end r = @ctx.build(mode: m, old_ai_context: inp[:ai_context], pinned_sections: inp[:pinned_sections], digests: inp[:digests], caller_facts: inp[:caller_facts], ledger: inp[:ledger], notes: inp[:notes], transcripts: inp[:transcripts]) stale = false if r[:ok] built = { "last_event_id" => inp[:last_event_id], "notes_rev" => inp[:notes_rev], "model" => r[:model], "at" => Time.now.utc.iso8601, "rebuilt_rev" => (m == :rebuild ? inp[:ai_context_rev] + 1 : inp[:rebuilt_rev]) } res = @svc.set_person_ai_context(person_id: person_id, ai_context: r[:ai_context], expected_rev: inp[:ai_context_rev], built_from: built) stale = res && res[:stale] end # Decide re-loop vs final exit ATOMICALLY, deleting the latch inside the same sync block on the # final-exit branch. Doing the delete here (not in a blanket ensure) closes the lost-:rerun window # AND avoids clobbering a run that a concurrent trigger legitimately spawns after we release. A # concurrent trigger during our run only ever sets :rerun (never spawns), so it is caught here. again = @latch_mu.synchronize { (@latch[person_id] == :rerun || stale) ? (@latch[person_id] = :running; true) : (@latch.delete(person_id); false) } break unless again end rescue => e warn "[helpdesk] context refresh #{person_id}: #{e.class}: #{e.message}" clear.call # leak fix: clear the latch on the exception path end end end def context_pending?(person_id) @latch_mu.synchronize { !@latch[person_id].nil? } end # device-auth + JSON body → run block → 200 with result (idempotent upserts) def guarded(req, res) body = read(req) return unauthorized(res) unless authed?(req, body) parsed = body.empty? ? {} : JSON.parse(body) json(res, 200, yield(parsed)) rescue BodyTooLarge raise # handled at the /api/v1 mount as a 413 — must not fall into the generic 500 below rescue JSON::ParserError => e json(res, 422, error: "bad_json", detail: e.message) rescue => e # Do NOT echo e.message to the client — it can carry caller numbers / transcript fragments (PII). warn "[helpdesk] server_error on #{req.request_method} #{req.path}: #{e.class}: #{e.message}" json(res, 500, error: "server_error") end def authed?(req, body) return true unless AUTH_ENFORCE return false unless secure_eq(req["Authorization"].to_s, "Bearer #{DEVICE_TOKEN}") # constant-time ts = req["X-Request-Ts"].to_i; nonce = req["X-Request-Nonce"].to_s; sig = req["X-Signature"].to_s return false if nonce.empty? || sig.empty? return false if (Time.now.to_i - ts).abs > MAX_SKEW_S expected = OpenSSL::HMAC.hexdigest("SHA256", DEVICE_SECRET, "#{ts}.#{nonce}.#{body}") return false unless secure_eq(expected, sig) # Replay-check + insert + prune must be ONE atomic critical section. WEBrick runs each request on # its own thread and the device drives 3 concurrent signed streams (webhooks, recording PUT, 2s # command poll); an unguarded @nonces.reject! racing an insert raises "can't add a new key into # hash during iteration" (reproduced) and opens a check-then-act replay window. @nonce_lock.synchronize do return false if @nonces.key?(nonce) # replay @nonces[nonce] = ts @nonces.reject! { |_n, t| Time.now.to_i - t > 600 } # prune expired end true end # Operator/console auth (the vmin-cert mTLS plane). mTLS is terminated at the reverse proxy (nginx/ # caddy), which verifies the operator's vmin client cert against the vmin CA — the same PKI as the # hosted Whisper — and injects X-Operator-Cert (the verified subject) + X-Proxy-Secret. The backend # trusts ONLY the proxy (the shared secret), never a raw client, and is not directly exposed for # these endpoints. WEBrick stays plain HTTP behind the proxy, which is where mTLS is terminated. # The phone never presents a cert — it reaches the DEVICE endpoints with HMAC through its own door. def operator_authed?(req) return true unless OP_AUTH_ENFORCE return false unless PROXY_SECRET && secure_eq(req["X-Proxy-Secret"].to_s, PROXY_SECRET.to_s) Server.operator_allowed?(req["X-Operator-Cert"], OPERATOR_CNS) end # The operator's name for attributing what they record. nil when auth is off (local dev). def operator_cn(req) = Server.cert_cn(req["X-Operator-Cert"]) # Pull the CN out of a client-cert subject DN. nginx writes $ssl_client_s_dn as RFC2253 # ("CN=Jan Novak,O=...,C=CZ") since 1.11.6 and as the older slash form ("/C=CZ/O=.../CN=Jan Novak") # before that, so accept both. nil when the DN carries no CN at all. def self.cert_cn(dn) m = dn.to_s.match(%r{(?:\A|[,/])\s*CN=([^,/]+)}i) or return nil cn = m[1].strip cn.empty? ? nil : cn end # Is this verified cert subject an operator? Pure, so the rule is testable without a server or proxy. # An empty allowlist accepts any DN the proxy's CA verified - the behaviour before HELPDESK_OPERATOR_CNS # existed, which is what local dev and an un-migrated deployment rely on. grep anchor: operator-allowlist. def self.operator_allowed?(dn, allowlist) return false if dn.to_s.empty? return true if allowlist.empty? cn = cert_cn(dn) !cn.nil? && allowlist.include?(cn.downcase) end # Reject oversized bodies: declared length BEFORE reading (normal requests), actual size after # (chunked requests carry no Content-Length, so they're only caught post-buffer — nginx's # client_max_body_size is the hard pre-buffer guarantee on the public doors; this is the backstop). def read(req) # Use the raw Content-Length header, NOT req.content_length: WEBrick's content_length does # Integer(header) and RAISES "can't convert nil into Integer" when the header is absent, which is # every device GET (/device/heartbeat, /device/commands). That 500'd those endpoints, so the phone's # presence heartbeat + reverse-command poll silently failed (console stuck on "Waiting for phone", # Accept-from-PC never applied). req["content-length"].to_i is nil-safe (nil -> 0). raise BodyTooLarge if req["content-length"].to_i > MAX_BODY_BYTES b = req.body.to_s raise BodyTooLarge if b.bytesize > MAX_BODY_BYTES b end def sym(h) = h.transform_keys(&:to_sym) def secure_eq(a, b) = a.bytesize == b.bytesize && OpenSSL.fixed_length_secure_compare(a, b) def unauthorized(res) = json(res, 401, error: "unauthorized") def json(res, status, obj) res.status = status res["Content-Type"] = "application/json" res.body = JSON.generate(obj) end end end Helpdesk::Server.new.start if $PROGRAM_NAME == __FILE__