be9f14ce34
A small helpdesk system: an office Pixel running a patched GrapheneOS Dialer answers technician calls, records both call legs as separate channels, and a Ruby backend transcribes them through Whisper and files an AI summary against the caller. Squashed to a single commit for sharing. No credentials are included; secrets live outside the repo in /etc/helpdesk/env on the server or a gitignored .claude/env.local locally. See .claude/env.local.example for the shape. Start at README.md, then docs/architecture.md.
570 lines
34 KiB
Ruby
570 lines
34 KiB
Ruby
# 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/<uuid>" (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=<secs>
|
|
# (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/<uuid> — download the uploaded call audio. The device uploads raw AAC-in-MP4 (the phone
|
|
# records .m4a), stored as <uuid>.audio; serve it back as an attachment with a human filename built
|
|
# from the call metadata so the operator gets "call_<when>_<number>.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__
|