-- generated by cesdk 0.1.0, do not edit
-- project: discovery 0.2

local __mods, __cache = {}, {}
local function require(name)
  local c = __cache[name]
  if c ~= nil then return c end
  local f = __mods[name]
  if not f then error("cesdk bundle: module '" .. name .. "' not found", 2) end
  local r = f()
  if r == nil then r = true end
  __cache[name] = r
  return r
end

__mods["cesdk.manifest"] = function() return { name = "discovery", version = "0.2", protocols = { "/discovery/1" } } end

__mods["main"] = function()
-- discovery - the CES network registry / crawler. A privileged /s/ extension
-- that walks the network as a CES client: it pages every known server's public
-- peer table (ces.peer_info), validates with ces.ping, keeps a registry of what
-- it finds, and projects a curated active set into its host's peer table. For
-- breadth past the peer table it also publishes a bounded random sample of its
-- registry (/s/discovery.sample) and pulls peers' samples (ces.file_client). No
-- mesh, no persistent connections. Full design: local/PEERING_AGENT_DESIGN.md.
--
-- Identity (CES_MANIFEST: name/version/description) is generated by the cesdk
-- bundler from project.lua; it is not hand-written here.

local registry = require("registry")
local Agent    = require("agent")
local admin    = require("admin")
local panel    = require("panel")
local conf     = require("conf")

-- Operator-deployed config; defaults apply if the file is absent. Production
-- sets a rendezvous in `seeds` and slow intervals; a test deploys a fast file.
local cfg = conf.load("/s/discovery.conf")

-- Our own server pubkey, so the registry recognizes and drops self.
local me = ces.owner_pubkey and ces.owner_pubkey() or nil

-- Build the agent's option set from a config map (the conf.load shape). Used at
-- launch AND on every live on_config push, so both paths share one set of
-- defaults — a knob omitted from the file always reverts to the same default.
local DEFAULT_SEEDS = { "ces.pubcom.org:53830" }  -- production rendezvous default

-- Typed-broadcast dest for the self-announce gossip. 128 leading zero bits mark
-- it as "not a pubkey, a typed broadcast" (CES-wide convention; the gossip core
-- skips sink routing and floods it); the low 128 bits are the discovery announce
-- protocol id. Receivers demux by filtering on_gossip's meta.dest. (16 zero bytes
-- + a 16-byte constant id.)
local ANNOUNCE_DEST = string.rep("\0", 16) .. "CES.DISCOVERY.01"

local function build_opts(c)
  return {
    min_peer_target = conf.num(c, "min_peer_target", 100000000),  -- 1 credit floor
    -- Auto-promote at most this many outbound peers. Deliberately well under the
    -- host's peer-table size so discovery never fills it — it leaves headroom for
    -- the operator's INTENTIONAL peer additions. 20 is a healthy default.
    active_target   = conf.num(c, "active_target", 20),
    crawl_ms        = conf.num(c, "crawl_ms", 3000),
    maint_ms        = conf.num(c, "maint_ms", 10000),
    save_ms         = conf.num(c, "save_ms", 300000),          -- full registry flush (~5 min)
    pull_floor_ms   = conf.num(c, "pull_floor_ms", 60000),     -- 1 min when learning
    pull_ceil_ms    = conf.num(c, "pull_ceil_ms", 1800000),    -- 30 min when converged
    sample_k        = conf.num(c, "sample_k", 16),
    probe_ms        = conf.num(c, "probe_ms", 3000),           -- per-ping reply wait (ces.ping)
    dead_after      = conf.num(c, "dead_after", 10),           -- failed revalidations before give-up
    peer_min_credit = conf.num(c, "peer_min_credit", 100000000),  -- floor balance per peer (1.0)
    -- Self-announce over gossip: `announce` is our own reachable address to
    -- broadcast (empty = off; the operator knows the public address). Pushed once
    -- on boot then every announce_ms; budget is small (pay-to-be-found anti-spam).
    announce        = c.announce or "",
    announce_ms     = conf.num(c, "announce_ms", 86400000),       -- 24h
    announce_budget = conf.num(c, "announce_budget", 1000000),    -- 0.01 credit
    -- A "seeds = ..." line (including empty "seeds =") overrides the default.
    -- Seeds are untrusted rumors, validated by the crawl's ces.ping.
    seeds           = c.seeds ~= nil and conf.list(c, "seeds") or DEFAULT_SEEDS,
  }
end

local opts = build_opts(cfg)
local reg = registry.open(me, opts.dead_after)
for _, addr in ipairs(opts.seeds) do reg:hear(addr, "seed") end

ces.log("discovery: up, registry=" .. reg:count())

local agent = Agent.new(reg, opts)
agent:start()

-- Inbound self-announce from another discovery: a typed broadcast on the announce
-- dest carrying a server address. Fold it into the registry; the crawl pings,
-- validates and promotes it like any rumor (self is dropped by the registry).
function on_gossip(msg, meta)
  if meta and meta.dest == ANNOUNCE_DEST and msg and #msg > 0 then
    reg:hear(msg, "gossip")
  end
end

-- Outbound self-announce: once on boot (immediate, so it is easy to observe),
-- then every announce_ms. Off when `announce` is unset.
local function announce()
  if opts.announce ~= "" and ces.gossip and ces.gossip.send then
    ces.gossip.send(opts.announce, opts.announce_budget, ANNOUNCE_DEST)
  end
end
announce()
ces.every(opts.announce_ms, announce)

admin.attach(reg)   -- observability over the relay (cesh dial); inbound-only, not a mesh

-- One apply path for both the host's on_config push and the panel's config
-- form: rebuild opts (so the form re-renders current values) and retune the
-- running agent.
local function apply_cfg(c)
  opts = build_opts(c)
  agent:reconfigure(opts)
end

-- Extension contract: live registry stats on the dashboard, a dump command, the
-- config defaults the editor seeds, and on_config for LIVE retuning — an edit in
-- the dashboard reconfigures the running agent (re-arms cadences, updates knobs,
-- folds in new seeds) with no restart. On hosts that install the mene library,
-- a declarative panel (registry composition + server table + config form)
-- supersedes the flat status lane in the dashboard; the status map stays for
-- API consumers.
local spec = {
  status = function()
    local s = reg:stats()
    return {
      registry     = tostring(s.total),
      alive        = tostring(s.alive),
      verified     = tostring(s.verified),
      heard        = tostring(s.heard),
      dark         = tostring(s.dark),
      from_samples = tostring(s.sample),
      pull_gap_ms  = tostring(agent.pull_interval or 0),
    }
  end,
  commands = { { id = "dump", label = "Log registry summary" } },
  on_command = function(id)
    if id == "dump" then
      -- Operator-triggered, so INFO is fine (the periodic heartbeat is trace).
      -- Same canonical line as the heartbeat -- one formatter, no drift.
      local line = agent:summary()
      ces.log("discovery: " .. line)
      return line
    end
  end,
  config_defaults =
    "seeds = ces.pubcom.org:53830\n" ..
    "active_target = 20\n" ..       -- max auto-promoted peers (headroom for manual)
    "crawl_ms = 3000\n" ..
    "maint_ms = 10000\n" ..
    "save_ms = 300000\n" ..
    "pull_floor_ms = 60000\n" ..
    "pull_ceil_ms = 1800000\n" ..
    "sample_k = 16\n" ..
    "probe_ms = 3000\n" ..          -- per-ping reply wait (lower to give up on dead hosts faster)
    "dead_after = 10\n" ..          -- failed re-validations before a host is given up
    "peer_min_credit = 100000000\n",   -- floor credits to keep funded at each peer (1.0)
  -- Live reconfigure: the host pushes the edited config map here (no restart).
  -- We rebuild the same opts the launch path uses and hand them to the agent.
  -- The panel's config form applies through the same function.
  on_config = apply_cfg,
}
if mene then
  spec.panel = panel.build(reg, agent, apply_cfg, function() return opts end)
end
ces.extension_admin(spec)

ces.run()

end

__mods["registry"] = function()
-- registry - the discovery agent's library of CES servers it has heard of. The
-- whole thing lives in RAM (far larger than the server's peer table, but capped
-- at MAX_RECORDS): rumors and query responses merge in naturally, deduped by
-- address. Per address it tracks long-range liveness/standing. See
-- local/PEERING_AGENT_DESIGN.md.
--
-- Persistence is a periodic full flush, not an incremental index: every few
-- minutes the agent calls save(), which prunes to the best MAX_RECORDS and
-- rewrites the whole /s/ file (truncate + overwrite). Boot loads it back; a load
-- that fails ANYWHERE wipes the file and starts with zero hosts (re-learned from
-- the seed + gossip) rather than running on a half-parsed library. If file ops
-- are unavailable it runs purely in-memory.

local M = {}
local Registry = {}
Registry.__index = Registry

-- Long-range standing states.
M.HEARD     = "heard"      -- address only, from a rumor; not yet validated
M.ALIVE     = "alive"      -- ces.ping succeeded recently (pubkey known, TOFU)
M.DARK      = "dark"       -- was alive, now unreachable; kept + periodically retried
M.DEAD      = "dead"       -- never validated after many tries / garbage
M.SELF      = "self"       -- our own server: never probe, promote, or gossip it

-- Where the registry persists. A /s/ extension acts under the server owner; if
-- this path/zone is not writable for it, load/save no-op and we run in-memory.
local REG_PATH = "/s/discovery.reg"
local READ_MAX = 1048576   -- ces.file_read caps a single read at 1 MiB

-- Hard cap on how many servers we remember. Enforced at save time (intake is
-- free between flushes); when over, prune() keeps the healthiest. 10k addresses
-- is a few hundred KB of /s/ file and a trivial RAM table, yet far more than any
-- real network needs in flight.
local MAX_RECORDS = 10000

-- Pruning order when over MAX_RECORDS: keep the most useful. State class first
-- (validated + reachable beats an unproven rumor), then reciprocating peers,
-- then most-recently-seen. DEAD garbage is shed first.
local STATE_RANK = { [M.SELF] = 4, [M.ALIVE] = 3, [M.DARK] = 2,
                     [M.HEARD] = 1, [M.DEAD] = 0 }

-- Consecutive failed re-validations before we give a host up (ALIVE/DARK ->
-- DEAD). Far smaller than the peer table's 5000: that re-probes every peer every
-- miner cycle, whereas the crawl re-validates a host only once per full pass, so
-- 10 failures is already many passes. A LIVE ping resets it to 0; merely hearing
-- a host gossiped does NOT (you can't un-fail a host by rumor). DEAD is dropped
-- at the next save(), after which the host is re-learnable from a fresh sighting.
local DEAD_AFTER = 10

-- The published sample lives in the unmetered, server-owned /s/ zone: a /s/
-- program writes it under server authority for free, and /s/ reads are donated
-- (cost 0). Same constant path on every server; the reader picks the server by
-- which one it dials, not by the path.
local SAMPLE_PATH = "/s/discovery.sample"
local SAMPLE_READ_MAX = 65536
M.SAMPLE_PATH = SAMPLE_PATH
M.SAMPLE_READ_MAX = SAMPLE_READ_MAX

local function now_s() return math.floor(ces.now() / 1000000) end

-- host:port shape gate for untrusted intake (rumor / sample lines). Rejects
-- control bytes and malformed tokens; accepts DNS names, IPv4, and bracketed
-- IPv6, each ending in :<port>.
local function valid_addr(a)
  return type(a) == "string" and a:match("^%g+:%d+$") ~= nil
end

local function rand_u32()
  if not ces.random_bytes then return 0 end
  local b = ces.random_bytes(4)
  if not b or #b < 4 then return 0 end
  local x, y, z, w = b:byte(1, 4)
  return ((x * 256 + y) * 256 + z) * 256 + w
end

local function hex(s)
  local o = {}
  for i = 1, #s do o[i] = string.format("%02x", string.byte(s, i)) end
  return table.concat(o)
end
local function unhex(h)
  if not h or h == "" then return "" end
  local o = {}
  for i = 1, #h, 2 do o[#o + 1] = string.char(tonumber(h:sub(i, i + 1), 16)) end
  return table.concat(o)
end

local function blank(addr, source)
  return { addr = addr, pubkey = "", state = M.HEARD, rpc_port = 0,
           last_seen = 0, last_try = 0, recip = false, source = source or "?",
           verified = false, fails = 0 }
end

-- self_pubkey: our own server's 32-byte pubkey, so an address that turns out to
-- be us (learned via a neighbor's gossip, confirmed on probe) can be marked SELF
-- and excluded from probing/promotion/gossip. nil disables the self-check.
function M.open(self_pubkey, dead_after)
  local self = setmetatable({ byaddr = {}, order = {}, cursor = 1, pcursor = 1,
                              me = self_pubkey,
                              dead_after = dead_after or DEAD_AFTER }, Registry)
  self:load()
  return self
end

-- Intake: learn of a server address (seed, rumor, gossip). Untrusted - validated
-- later by ces.ping. Idempotent; just records a fresh HEARD entry.
function Registry:hear(addr, source)
  if not valid_addr(addr) then return end
  if not self.byaddr[addr] then
    self.byaddr[addr] = blank(addr, source)
    self.order[#self.order + 1] = addr
  end
end

-- Direct evidence a server exists: learn the pubkey (TOFU) and mark ALIVE.
-- Creates the entry if unknown (a ces.ping success or a ces.peers observation is
-- itself a first sighting), so observing the local peer table populates the
-- registry without a prior rumor.
-- `verified` (optional) carries the host's cryptographic confirmation of this
-- peer (true/false from ces.peers().verified). It governs gossip, NOT promotion:
-- only verified peers are sampled out. Passed nil by the crawl (a pinged rumor
-- isn't our own verified peer), which leaves it untouched.
function Registry:seen(addr, pubkey, rpc_port, verified)
  if not addr or addr == "" then return end
  local r = self.byaddr[addr]
  if not r then
    r = blank(addr, "peer")
    self.byaddr[addr] = r
    self.order[#self.order + 1] = addr
  end
  if pubkey and #pubkey == 32 then r.pubkey = pubkey end
  if rpc_port and rpc_port > 0 then r.rpc_port = rpc_port end
  if verified ~= nil then r.verified = verified and true or false end
  r.fails = 0   -- a LIVE sighting is the only thing that clears the failure count
  -- An address that resolves to our own pubkey is us: latch it SELF (sticky --
  -- never probed, promoted, or gossiped again) instead of marking it ALIVE.
  if self.me and r.pubkey == self.me then
    r.state = M.SELF
    return
  end
  r.state, r.last_seen = M.ALIVE, now_s()
  r.last_try = r.last_seen
end

-- True once `addr` has been confirmed to be our own server (see seen()).
function Registry:is_self(addr)
  local r = self.byaddr[addr]
  return r ~= nil and r.state == M.SELF
end

-- A failed probe ages the host. First miss: ALIVE -> DARK (kept, retried for
-- recovery). After DEAD_AFTER consecutive misses (only a live ping resets the
-- count): -> DEAD, which the crawl then skips and save() drops. A HEARD rumor we
-- can never validate ages the same way, so junk rumors self-clean too.
function Registry:missed(addr)
  local r = self.byaddr[addr]; if not r then return end
  r.last_try = now_s()
  r.fails = (r.fails or 0) + 1
  if r.fails >= self.dead_after then
    r.state = M.DEAD
  elseif r.state == M.ALIVE then
    r.state = M.DARK
  end
end

function Registry:set_recip(addr, yes)
  local r = self.byaddr[addr]; if r then r.recip = yes and true or false end
end

-- Up to n addresses to (re)probe this tick, round-robin over the whole registry
-- so a 10k library is swept a slice at a time.
function Registry:slice(n)
  local out, total = {}, #self.order
  if total == 0 then return out end
  for _ = 1, (n < total and n or total) do
    if self.cursor > total then self.cursor = 1 end
    out[#out + 1] = self.order[self.cursor]
    self.cursor = self.cursor + 1
  end
  return out
end

-- Next ALIVE entry with a known rpc port, round-robin over the registry on its
-- own cursor, for the sample-pull. nil if none. Bounded to one full pass.
function Registry:next_pullable()
  local total = #self.order
  for _ = 1, total do
    if self.pcursor > total then self.pcursor = 1 end
    local r = self.byaddr[self.order[self.pcursor]]
    self.pcursor = self.pcursor + 1
    if r and r.state == M.ALIVE and r.rpc_port > 0 then return r end
  end
  return nil
end

-- Up to k random ALIVE-and-VERIFIED addresses — the bounded slice we publish for
-- others to pull. We only gossip peers we cryptographically confirmed ourselves
-- (reached + signed server-info), never a merely-reachable one, so an inbound-PoW
-- address lie can never propagate through us. Fewer than k if the proven set is
-- smaller. This also makes us vouch only for peers we economically committed to.
function Registry:sample(k)
  local alive = {}
  for _, addr in ipairs(self.order) do
    local r = self.byaddr[addr]
    if r.state == M.ALIVE and r.verified then alive[#alive + 1] = addr end
  end
  local n = #alive
  if n <= k then return alive end
  local out, picked, tries = {}, {}, 0
  while #out < k and tries < k * 8 do
    tries = tries + 1
    local idx = (rand_u32() % n) + 1
    if not picked[idx] then picked[idx] = true; out[#out + 1] = alive[idx] end
  end
  return out
end

-- Fold a peer's published sample (newline-separated addresses) as HEARD rumors.
-- Returns the count of addresses that were new (drives the pull pacing). `max`
-- caps how many NEW addresses we accept from a single sample: honest nodes
-- publish only sample_k, but the reader can't trust the publisher, so a bound
-- here stops one malicious oversized /s/discovery.sample from flooding us.
function Registry:ingest_sample(data, max)
  local novel = 0
  for line in (data .. "\n"):gmatch("(.-)\n") do
    if max and novel >= max then break end
    local addr = line:match("^%s*(%S+)%s*$")
    if addr and valid_addr(addr) and not self.byaddr[addr] then
      self:hear(addr, "sample")
      novel = novel + 1
    end
  end
  return novel
end

-- Entries worth promoting into the active peer set: validated (pubkey known) and
-- currently alive.
function Registry:promotable()
  local out = {}
  for _, addr in ipairs(self.order) do
    local r = self.byaddr[addr]
    if r.state == M.ALIVE and #r.pubkey == 32 then out[#out + 1] = r end
  end
  return out
end

-- All known addresses, in insertion order. Read-only; for observability.
function Registry:addrs() return self.order end

-- Per-state counts, for status and logging. `verified` cuts across state (it is
-- the gossip gate, so it's the dimension that actually matters now). DEAD is not
-- counted: nothing assigns it yet (no DARK->DEAD aging), so it would only ever
-- report 0 and mislead — re-add it with the aging logic that populates it.
function Registry:stats()
  local s = { total = #self.order, alive = 0, heard = 0, dark = 0,
              me = 0, sample = 0, verified = 0 }
  for _, addr in ipairs(self.order) do
    local r = self.byaddr[addr]
    if r.state == M.ALIVE then s.alive = s.alive + 1
    elseif r.state == M.HEARD then s.heard = s.heard + 1
    elseif r.state == M.DARK then s.dark = s.dark + 1
    elseif r.state == M.SELF then s.me = s.me + 1 end
    if r.source == "sample" then s.sample = s.sample + 1 end
    if r.verified then s.verified = s.verified + 1 end
  end
  return s
end

function Registry:get(addr) return self.byaddr[addr] end
function Registry:count() return #self.order end

-- ---- persistence: prune to cap, then full-flush the whole library ----------
function Registry:serialize()
  local lines = {}
  for _, addr in ipairs(self.order) do
    local r = self.byaddr[addr]
    lines[#lines + 1] = table.concat({
      addr, hex(r.pubkey), r.state, tostring(r.last_seen),
      tostring(r.last_try), r.recip and "1" or "0", r.source,
      tostring(r.rpc_port or 0), r.verified and "1" or "0" }, "\t")
  end
  return table.concat(lines, "\n")
end

-- Drop the registry to the healthiest MAX_RECORDS. No-op under the cap, so the
-- common case (a small library) keeps its natural insertion order, which the
-- crawl/pull cursors page through. Called at save time, so intake never has to
-- fight the cap mid-merge.
function Registry:prune(maxn)
  maxn = maxn or MAX_RECORDS
  if #self.order <= maxn then return end
  local recs = {}
  for _, addr in ipairs(self.order) do recs[#recs + 1] = self.byaddr[addr] end
  table.sort(recs, function(a, b)
    local ra, rb = STATE_RANK[a.state] or 0, STATE_RANK[b.state] or 0
    if ra ~= rb then return ra > rb end
    if a.recip ~= b.recip then return a.recip end          -- reciprocating first
    return (a.last_seen or 0) > (b.last_seen or 0)          -- most recent first
  end)
  local keep, order = {}, {}
  for i = 1, maxn do
    local r = recs[i]; keep[r.addr] = r; order[#order + 1] = r.addr
  end
  self.byaddr, self.order = keep, order
  self.cursor, self.pcursor = 1, 1
end

-- Parse the persisted file into RAM. Throws on a malformed file so load() can
-- catch it and reset; callers never invoke this directly.
function Registry:_load_unsafe()
  if not ces.file_read or not ces.file_stat then return end
  -- READ rejects a range past EOF, so size the read to the file (stat first).
  local st = ces.file_stat(REG_PATH)
  if not st or not st.size or st.size == 0 then return end
  local n = st.size < READ_MAX and st.size or READ_MAX
  local data = ces.file_read(REG_PATH, 0, n)
  if not data or data == "" then return end
  for line in (data .. "\n"):gmatch("(.-)\n") do
    if line ~= "" then
      local f = {}
      for col in (line .. "\t"):gmatch("(.-)\t") do f[#f + 1] = col end
      local addr = f[1]
      if addr and addr ~= "" and not self.byaddr[addr] then
        self.byaddr[addr] = {
          addr = addr, pubkey = unhex(f[2]), state = f[3] or M.HEARD,
          last_seen = tonumber(f[4]) or 0, last_try = tonumber(f[5]) or 0,
          recip = f[6] == "1", source = f[7] or "?",
          rpc_port = tonumber(f[8]) or 0, verified = f[9] == "1", fails = 0 }
        self.order[#self.order + 1] = addr
        if #self.order >= MAX_RECORDS then break end         -- bounded on load too
      end
    end
  end
end

-- Fail-safe load: a corrupt or unreadable file must not leave us running on a
-- half-parsed library. Any error wipes RAM and truncates the file to empty, so
-- we boot knowing zero hosts and re-learn from the seed + gossip.
function Registry:load()
  local ok, err = pcall(function() self:_load_unsafe() end)
  if not ok then
    self.byaddr, self.order = {}, {}
    self.cursor, self.pcursor = 1, 1
    if ces.file_resize then pcall(ces.file_resize, REG_PATH, 0) end
    if ces.log then
      ces.log("discovery: registry load failed, wiped (" .. tostring(err) .. ")")
    end
  end
end

-- Size the /s/ file to exactly #data and overwrite it from offset 0. The order
-- matters: WRITE rejects any range past EOF, so a file that GREW (the registry
-- keeps learning) must be resized UP first or the appended tail is dropped and
-- nothing past the old size ever persists. resize also trims a stale tail when
-- the data shrank. /s/ is unmetered, so resize-grow costs nothing here.
local function write_exact(path, data)
  if ces.file_create then ces.file_create(path, #data, 0, 0) end
  if ces.file_resize then ces.file_resize(path, #data) end
  ces.file_write(path, 0, data)
end

-- Forget the hosts we gave up on. Rebuilds byaddr/order without DEAD entries;
-- the cursors self-correct (slice/next_pullable wrap past the end). Called at
-- save time, so a given-up host lingers at most one flush before it's gone and
-- becomes re-learnable from a fresh sighting.
function Registry:drop_dead()
  local keep, order = {}, {}
  for _, addr in ipairs(self.order) do
    local r = self.byaddr[addr]
    if r.state ~= M.DEAD then keep[addr] = r; order[#order + 1] = addr end
  end
  self.byaddr, self.order = keep, order
end

function Registry:save()
  if not ces.file_write then return end
  self:drop_dead()                                 -- forget given-up hosts
  self:prune(MAX_RECORDS)                          -- bound to the healthiest 10k
  local data = self:serialize()
  if data == "" then return end   -- nothing to persist; file_write rejects empty
  write_exact(REG_PATH, data)
end

-- Republish the readable sample (k random ALIVE addresses) so other agents can
-- pull addresses this one knows.
function Registry:publish(k)
  if not ces.file_write or not self.me then return end
  local addrs = self:sample(k)
  if #addrs == 0 then return end
  write_exact(SAMPLE_PATH, table.concat(addrs, "\n"))
end

return M

end

__mods["agent"] = function()
-- agent - the discovery crawler. A pure CES client: it pages each known server's
-- public peer table one slot at a time (ces.peer_info) and folds in a bounded
-- random sample pulled from peers' published address books (ces.file_client read
-- of /s/discovery.sample). No mesh. Liveness via ces.ping; the active set is
-- projected into the host's peer table via add_peer/remove_peer.
--   crawl (CRAWL_MS): one peer-table slot, or one ping, per tick.
--   maint (MAINT_MS): observe own peers, promote/curate, publish sample, persist.
--   pull  (paced):    read one peer's sample; gap floors on novelty, else grows.

local registry = require("registry")

local M = {}
local Agent = {}
Agent.__index = Agent

local CRAWL_MS = 3000     -- one slot / one ping per tick
local MAINT_MS = 10000    -- promote / curate / publish cadence
local SAVE_MS  = 300000   -- full-flush the registry to /s/ (~5 min); intake is RAM
local LOG_MS   = 60000    -- registry summary to the server log
local PULL_FLOOR_MS = 60000     -- min gap between sample-pulls (still learning)
local PULL_CEIL_MS  = 1800000   -- max gap between sample-pulls (converged)
local SAMPLE_K      = 16        -- addresses per published / pulled sample

local function now_ms() return math.floor(ces.now() / 1000) end

function M.new(reg, opts)
  opts = opts or {}
  return setmetatable({
    reg             = reg,
    min_peer_target = opts.min_peer_target or 100000000,  -- floor: 1 full credit
    active_target   = opts.active_target   or 100,        -- active-set size goal
    crawl_ms        = opts.crawl_ms        or CRAWL_MS,
    maint_ms        = opts.maint_ms        or MAINT_MS,
    save_ms         = opts.save_ms         or SAVE_MS,
    pull_floor_ms   = opts.pull_floor_ms   or PULL_FLOOR_MS,
    pull_ceil_ms    = opts.pull_ceil_ms    or PULL_CEIL_MS,
    sample_k        = opts.sample_k        or SAMPLE_K,
    probe_ms        = opts.probe_ms        or 3000,        -- per-ping reply wait (ces.ping default)
    peer_min_credit = opts.peer_min_credit or 100000000,   -- floor balance to keep at a peer (1.0 credit)
    cur             = nil,    -- address whose peer table we are paging now
    slot            = 0,      -- next peer-table slot to read on cur
    pull_interval   = opts.pull_floor_ms or PULL_FLOOR_MS,  -- current pull gap
    next_pull_at    = 0,      -- ms wall time of next allowed pull (0 = now)
    peer_bal        = {},     -- cached program-account balance per peer addr,
                              -- to skip a remote read on every pull (re-synced
                              -- with a real read only when an op there fails)
  }, Agent)
end

function Agent:start()
  -- Raise the miner's global per-peer reserve goal to our floor if lower; never
  -- lower a higher operator-chosen target.
  if ces.peer_target and ces.set_peer_target then
    local cur = ces.peer_target() or 0
    if cur < self.min_peer_target then ces.set_peer_target(self.min_peer_target) end
  end
  self:observe()   -- seed the registry from our own peer table so the crawl has roots
  self.timers = {}
  self:arm("crawl", self.crawl_ms,      function() self:guard("crawl", function() self:crawl() end) end)
  self:arm("maint", self.maint_ms,      function() self:guard("maint", function() self:maintain() end) end)
  self:arm("save",  self.save_ms,       function() self:guard("save",  function() self.reg:save() end) end)
  self:arm("pull",  self.pull_floor_ms, function() self:guard("pull",  function() self:pull_tick() end) end)
  self:arm("log",   LOG_MS,             function() self:guard("log",   function() self:log_summary() end) end)
end

-- Register a named periodic timer, remembering its handle + interval + closure so
-- reconfigure() can re-arm it at a new cadence (ces.every returns a cancel handle).
function Agent:arm(name, interval_ms, fn)
  self.timers[name] = { handle = ces.every(interval_ms, fn), interval = interval_ms, fn = fn }
end

-- Re-arm a named timer at a new interval (cancel old, register new with the same
-- closure). No-op if unchanged, or if this ces build lacks ces.cancel / returned
-- no handle (then the old cadence stays rather than risk a duplicate timer).
function Agent:rearm(name, interval_ms)
  local t = self.timers[name]
  if not t or t.interval == interval_ms then return end
  if not (ces.cancel and t.handle) then return end
  ces.cancel(t.handle)
  t.handle = ces.every(interval_ms, t.fn)
  t.interval = interval_ms
end

-- Live config update (ces.extension_admin on_config). `opts` has the SAME shape
-- main.lua builds at launch. Read-on-use knobs (active_target, probe_ms, sample_k,
-- peer_min_credit, dead_after, pull bounds) take effect on their next use; the
-- timer cadences are re-armed; new seeds enter as HEARD rumors. So an operator
-- retunes discovery from the dashboard with no restart.
function Agent:reconfigure(opts)
  self.min_peer_target = opts.min_peer_target
  self.active_target   = opts.active_target
  self.sample_k        = opts.sample_k
  self.probe_ms        = opts.probe_ms
  self.peer_min_credit = opts.peer_min_credit
  self.pull_floor_ms   = opts.pull_floor_ms
  self.pull_ceil_ms    = opts.pull_ceil_ms
  self.pull_interval   = opts.pull_floor_ms        -- restart pacing from the floor
  self.reg.dead_after  = opts.dead_after
  for _, addr in ipairs(opts.seeds or {}) do self.reg:hear(addr, "seed") end
  if ces.peer_target and ces.set_peer_target then
    local cur = ces.peer_target() or 0
    if cur < self.min_peer_target then ces.set_peer_target(self.min_peer_target) end
  end
  self:rearm("crawl", opts.crawl_ms)
  self:rearm("maint", opts.maint_ms)
  self:rearm("save",  opts.save_ms)
  self:rearm("pull",  opts.pull_floor_ms)
  -- Lifecycle (operator-triggered, infrequent): confirms the live edit took.
  ces.log("discovery: reconfigured (live)")
end

-- Guard each cadence on its own: a fault in one must not kill its timer or block
-- the others.
function Agent:guard(name, fn)
  local ok, err = pcall(fn)
  if not ok then ces.log("discovery: " .. name .. " error: " .. tostring(err)) end
end

-- Adopt a new target (ping it for liveness + identity) or read the next slot of
-- the current target's peer table.
function Agent:crawl()
  if not ces.peer_info or not ces.ping then return end
  if not self.cur then
    local addr = self:next_target()
    if not addr then return end
    self.cur, self.slot = addr, 0
    local info = ces.ping(addr, self.probe_ms)
    if info and info.pubkey and #info.pubkey == 32 then
      self.reg:seen(addr, info.pubkey, info.rpc_port)   -- ALIVE (or latched SELF)
      if self.reg:is_self(addr) then self.cur = nil end
    else
      self.reg:missed(addr)                     -- ALIVE->DARK; pick another next tick
      self.cur = nil
    end
    return
  end
  local r = ces.peer_info(self.cur, self.slot)
  if not r then                                 -- query failed: drop this host
    self.reg:missed(self.cur); self.cur = nil; return
  end
  if r.found and r.address and r.address ~= "" then
    self.reg:hear(r.address, "crawl")           -- a rumor; validated when we crawl it
  end
  self.slot = self.slot + 1
  if self.slot >= r.count then self.cur = nil end   -- table fully read; next host
end

-- Next address to crawl, round-robin over the registry, skipping ourselves and
-- hosts we've given up on (DEAD). Bounded so a registry of only self/dead/holes
-- can't spin (DEAD is transient — dropped at the next save).
function Agent:next_target()
  for _ = 1, 16 do
    local addr = self.reg:slice(1)[1]
    if not addr then return nil end
    local r = self.reg:get(addr)
    if r and r.state ~= registry.SELF and r.state ~= registry.DEAD then
      return addr
    end
  end
  return nil
end

-- Fold our own peer table into the registry (live crawl roots + promotion
-- candidates) and record which outbound peers reciprocate. We ingest every
-- REACHABLE peer so the agent can promote and mine it — but we carry the host's
-- `verified` flag through, because only verified peers are ever GOSSIPED (see
-- sample()). verified means we reached the address and a signed server-info
-- proved the box holds that key; a merely reachable entry could be an inbound-PoW
-- key->address lie that any live host would satisfy. So: reach it to consider it,
-- prove it to vouch for it.
function Agent:observe()
  if not ces.peers then return end
  for _, p in ipairs(ces.peers() or {}) do
    if p.reachable and p.pubkey and #p.pubkey == 32 then
      self.reg:seen(p.address, p.pubkey, p.rpc_port, p.verified)
    end
    if p.outbound then self.reg:set_recip(p.address, p.inbound) end
  end
end

function Agent:maintain()
  self:observe()
  self:project()
  self.reg:publish(self.sample_k)
  -- Persistence is NOT here: save() runs on its own slower timer (save_ms),
  -- a periodic full flush of the whole RAM library, not a per-maint rewrite.
end

-- Sample-pull: read one known server's published /s/discovery.sample and fold
-- its addresses in. Paced by novelty (backoff): a pull that returns something
-- new keeps the gap at the floor; an empty/redundant pull, or no eligible
-- target, doubles the gap toward the ceiling, so a converged agent goes quiet
-- and a growing network snaps it back.
function Agent:pull_tick()
  if now_ms() < self.next_pull_at then return end
  self:pull()
end

-- Read one known server's published sample over a transient file client and
-- fold its addresses in. A read we cannot pay for (no account on that peer yet)
-- just fails; backoff postpones it and the cycle moves on.
function Agent:pull()
  local r = self.reg:next_pullable()
  if not r then return self:backoff(false) end
  local host = r.addr:match("^([^:]+):")
  if not host or not ces.file_client then return self:backoff(false) end
  local target = host .. ":" .. r.rpc_port
  ces.log("trace", "discovery: pull <- " .. r.addr)  -- per-op firehose, trace-only
  -- Smart funding with hysteresis + a cached balance. Hysteresis: keep a balance
  -- at the peer to pay for reads, but don't micro-top-up every pull — let it
  -- drain to HALF the floor, then refill to the full floor in one transfer.
  -- Cache: we keep a per-peer estimate of our program-account balance and decide
  -- from THAT, instead of a remote read every pull; we seed it with one real
  -- read, set it to the floor right after a top-up, and INVALIDATE it (force a
  -- fresh read next pull) whenever an op here fails — the signal it drifted.
  -- peer_min_credit is the floor (config, default 1.0 credit).
  --
  -- Funding and the balance read are LEDGER ops, so they target the peer's main
  -- port (r.addr) — NOT the rpc port the file client below uses. The file read
  -- is a CesPlex verb on the rpc port; the account it spends from lives on the
  -- main port. Mixing them up sends a ledger transfer at the CesPlex port, which
  -- never lands.
  if ces.request_funds and ces.program_pubkey then
    local refill_at = self.peer_min_credit / 2
    local bal = self.peer_bal[r.addr]
    if bal == nil and ces.remote_account_read then       -- cold: one real read to seed
      bal = ces.remote_account_read(r.addr, ces.program_pubkey())
      self.peer_bal[r.addr] = bal
    end
    if bal == nil or bal < refill_at then
      local g = ces.request_funds(r.addr, self.peer_min_credit - (bal or 0))
      if g and g > 0 then
        self.peer_bal[r.addr] = self.peer_min_credit     -- topped to the floor
        ces.log("trace", "discovery: topped up " .. g .. " at " .. r.addr ..
                " (had " .. tostring(bal) .. ", floor " .. self.peer_min_credit .. ")")
      else
        self.peer_bal[r.addr] = nil                      -- fund failed: re-read next time
      end
    end
  end
  local fc = ces.file_client(target, r.pubkey)
  if not fc then self.peer_bal[r.addr] = nil; return self:backoff(false) end
  local novel, failed = 0, false
  local st = fc:stat(registry.SAMPLE_PATH)
  if not st then
    failed = true
  elseif st.size and st.size > 0 then
    local n = st.size < registry.SAMPLE_READ_MAX and st.size or registry.SAMPLE_READ_MAX
    local data = fc:read(registry.SAMPLE_PATH, 0, n)
    if data and data ~= "" then
      novel = self.reg:ingest_sample(data, self.sample_k * 10)  -- cap junk floods
    else
      failed = true
    end
  end
  fc:close()
  if failed then self.peer_bal[r.addr] = nil end          -- op failed: re-sync next pull
  ces.log("trace", "discovery: got " .. novel .. " new host(s) from " .. r.addr)
  self:backoff(novel > 0)
end

function Agent:backoff(novel)
  if novel then
    self.pull_interval = self.pull_floor_ms
  else
    local doubled = self.pull_interval * 2
    self.pull_interval = doubled < self.pull_ceil_ms and doubled or self.pull_ceil_ms
  end
  self.next_pull_at = now_ms() + self.pull_interval
end

-- Promote validated servers into the active peer set up to the target; shed the
-- least valuable (non-reciprocating) when over.
function Agent:project()
  if not ces.peers then return end
  local me = ces.owner_pubkey and ces.owner_pubkey() or nil

  -- Key the active set by pubkey, not address: a peer is one server whatever
  -- address representation it is known by (a hostname vs the IP it resolves to), so
  -- it is never promoted twice or churned. Promotable entries always carry a pubkey.
  local outbound = {}
  for _, p in ipairs(ces.peers() or {}) do
    if p.outbound and p.pubkey and #p.pubkey == 32 then outbound[p.pubkey] = p end
  end
  local count = 0
  for _ in pairs(outbound) do count = count + 1 end

  if count < self.active_target and ces.add_peer then
    for _, r in ipairs(self.reg:promotable()) do
      if count >= self.active_target then break end
      if not outbound[r.pubkey] and r.pubkey ~= me then
        if ces.add_peer(r.pubkey, r.addr) then
          outbound[r.pubkey] = { address = r.addr, pubkey = r.pubkey, inbound = false }
          count = count + 1
        end
      end
    end
  end

  if count > self.active_target and ces.remove_peer then
    local over = count - self.active_target
    for _, p in pairs(outbound) do
      if over <= 0 then break end
      if p.pubkey and not p.inbound then
        if ces.remove_peer(p.pubkey) then over = over - 1 end
      end
    end
  end
end

-- The canonical one-line registry summary, shared by the periodic log and the
-- operator's "dump" command so they never drift apart.
function Agent:summary()
  local s = self.reg:stats()
  return string.format(
    "registry=%d alive=%d verified=%d heard=%d dark=%d sample=%d",
    s.total, s.alive, s.verified, s.heard, s.dark, s.sample)
end

-- Periodic heartbeat: TRACE, not INFO -- on an idle node it would otherwise log
-- the same line every minute forever. Enable the module's trace to see it; the
-- dashboard status and the dump command surface the same numbers on demand.
function Agent:log_summary()
  ces.log("trace", "discovery: " .. self:summary())
end

return M

end

__mods["admin"] = function()
-- admin - observability console for the discovery extension, served over the
-- /ces/lua/1 relay (cesh dial <pid>). It only answers operator/test queries and
-- never dials out, so it is not a mesh. Verbs: status, dump.

local M = {}
local Admin = {}
Admin.__index = Admin

function M.new(reg)
  return setmetatable({ reg = reg, buf = {} }, Admin)
end

function Admin:on_data(conn, data)
  local b = (self.buf[conn.id] or "") .. data
  while true do
    local line, rest = b:match("^(.-)\n(.*)$")
    if not line then break end
    b = rest
    self:cmd(conn, line)
  end
  self.buf[conn.id] = b
end

function Admin:on_close(conn)
  self.buf[conn.id] = nil
end

function Admin:cmd(conn, line)
  local c = line:match("^(%S+)")
  if c == "status" then
    local s = self.reg:stats()
    local outbound = 0
    for _, p in ipairs(ces.peers() or {}) do
      if p.outbound then outbound = outbound + 1 end
    end
    conn:write(string.format(
      "registry=%d alive=%d verified=%d dark=%d sample=%d outbound=%d\n",
      s.total, s.alive, s.verified, s.dark, s.sample, outbound))
  elseif c == "dump" then
    for _, addr in ipairs(self.reg:addrs()) do
      local r = self.reg:get(addr)
      conn:write(addr .. " " .. (r and r.state or "?") .. "\n")
    end
    conn:write("END\n")
  else
    conn:write("ERR unknown\n")
  end
end

function M.attach(reg)
  local a = M.new(reg)
  ces.conn.set_listener({
    on_data  = function(conn, data) a:on_data(conn, data) end,
    on_close = function(conn) a:on_close(conn) end,
  })
  return a
end

return M

end

__mods["panel"] = function()
-- panel - the mene admin panel (webadmin Extensions tab): live registry
-- composition, a sortable/paged table of known servers, the dump action, and a
-- typed config form that applies live AND persists /s/discovery.conf (via
-- ces.extension_admin.save_config). Uses the host-installed global `mene`;
-- build() is only called when the host provides it (see main.lua), so the
-- bundle still runs on mene-less hosts.

local M = {}

-- Bound the per-frame table so a huge registry cannot bloat the render frame;
-- the table pages client-side and a truncation note keeps the cap visible.
local MAX_ROWS = 200

-- The config keys the form round-trips, in file order. Matches build_opts /
-- config_defaults in main.lua.
local CFG_KEYS = {
  "seeds", "announce", "active_target", "crawl_ms", "maint_ms", "save_ms",
  "pull_floor_ms", "pull_ceil_ms", "sample_k", "probe_ms", "dead_after",
  "peer_min_credit", "announce_ms", "announce_budget",
}

-- Form value map (strings/bools) -> conf-file text, fixed key order.
local function conf_text(v)
  local out = {}
  for _, k in ipairs(CFG_KEYS) do
    local x = v[k]
    if type(x) == "boolean" then x = x and "1" or "0" end
    out[#out + 1] = k .. " = " .. tostring(x or "")
  end
  return table.concat(out, "\n") .. "\n"
end

-- build(reg, agent, apply_cfg, get_opts):
--   apply_cfg(c) applies a conf string-map live (main's on_config path);
--   get_opts()   returns the CURRENT typed opts (form initial values).
function M.build(reg, agent, apply_cfg, get_opts)
  local function num_field(name, label, value)
    return mene.field({ name = name, kind = "number", label = label,
                        value = value })
  end
  return mene.app{
    model = { last_msg = "" },
    view = function(m)
      local s = reg:stats()
      local o = get_opts()
      local outbound = 0
      for _, p in ipairs(ces.peers() or {}) do
        if p.outbound then outbound = outbound + 1 end
      end
      local rows, total = {}, 0
      for _, addr in ipairs(reg:addrs()) do
        total = total + 1
        if #rows < MAX_ROWS then
          local r = reg:get(addr)
          rows[#rows + 1] = {
            addr,
            tostring(r and r.state or "?"),
            (r and (r.rpc_port or 0) > 0) and tostring(r.rpc_port) or "-",
          }
        end
      end
      return mene.card({ title = "Discovery", subtitle = "network registry / crawler" },
        mene.grid({ cols = 4, gap = 12 },
          mene.stat({ label = "registry", value = s.total }),
          mene.stat({ label = "alive", value = s.alive,
                      tone = s.alive > 0 and "ok" or "warn" }),
          mene.stat({ label = "outbound peers", value = outbound }),
          mene.stat({ label = "pull gap",
                      value = tostring(math.floor((agent.pull_interval or 0) / 1000)) .. "s" })),
        mene.breakdown({ label = "registry by liveness state", parts = {
          { "alive",    s.alive,    "ok"   },
          { "verified", s.verified, "info" },
          { "heard",    s.heard,    "warn" },
          { "dark",     s.dark,     "err"  },
        } }),
        mene.section({ title = "known servers" },
          mene.table({ sortable = true, page_size = 10,
                       cols = { "address", "state", "rpc" }, rows = rows }),
          total > MAX_ROWS and mene.text({ tone = "muted" },
            "showing " .. MAX_ROWS .. " of " .. total) or nil),
        mene.row({ align = "end" },
          mene.button({ on = "dump" }, "Log registry summary")),
        mene.section({ title = "config (applies live + persists)" },
          mene.form({ on = "cfg" },
            mene.field({ name = "seeds", label = "seeds (host:port, comma-separated)",
                         value = table.concat(o.seeds or {}, ",") }),
            mene.field({ name = "announce",
                         label = "announce (our public address; empty = off)",
                         value = o.announce or "" }),
            mene.grid({ cols = 4, gap = 10 },
              num_field("active_target", "active target", o.active_target),
              num_field("crawl_ms", "crawl ms", o.crawl_ms),
              num_field("maint_ms", "maint ms", o.maint_ms),
              num_field("save_ms", "save ms", o.save_ms),
              num_field("pull_floor_ms", "pull floor ms", o.pull_floor_ms),
              num_field("pull_ceil_ms", "pull ceil ms", o.pull_ceil_ms),
              num_field("sample_k", "sample k", o.sample_k),
              num_field("probe_ms", "probe ms", o.probe_ms),
              num_field("dead_after", "dead after", o.dead_after),
              num_field("peer_min_credit", "peer min credit", o.peer_min_credit),
              num_field("announce_ms", "announce ms", o.announce_ms),
              num_field("announce_budget", "announce budget", o.announce_budget)),
            mene.row({ align = "end" },
              mene.submit({ kind = "primary" }, "Apply + save"))),
          m.last_msg ~= "" and mene.text({ tone = "muted" }, m.last_msg) or nil))
    end,
    update = function(ev, m)
      if ev.on == "dump" then
        ces.log("discovery: " .. agent:summary())
        m.last_msg = ""
      elseif ev.on == "cfg" and type(ev.value) == "table" then
        local c = {}
        for k, x in pairs(ev.value) do
          if type(x) == "boolean" then x = x and "1" or "0" end
          c[k] = tostring(x)
        end
        apply_cfg(c)
        if ces.extension_admin.save_config then
          ces.extension_admin.save_config(conf_text(c))
          m.last_msg = "config applied live + saved to /s/discovery.conf"
        else
          m.last_msg = "config applied live (host cannot persist; edit Config below to save)"
        end
      end
      return m
    end,
  }
end

return M

end

__mods["conf"] = function()
-- conf - load an operator-deployed config file for an L2 extension from the file
-- store. Format: "key = value" lines; "#" starts a comment; blank and malformed
-- lines are ignored. load() returns a string->string table ({} if the file is
-- absent or file ops are unavailable, so a program always has working defaults);
-- num() and list() coerce a value. The same extension reads the same path in
-- production and under test; only the deployed file differs.

local M = {}

function M.load(path)
  local out = {}
  if not ces.file_read or not ces.file_stat then return out end
  -- READ rejects a range past EOF, so size the read to the file (stat first).
  local st = ces.file_stat(path)
  if not st or not st.size or st.size == 0 then return out end
  local n = st.size < 65536 and st.size or 65536
  local data = ces.file_read(path, 0, n)
  if not data or data == "" then return out end
  for line in (data .. "\n"):gmatch("(.-)\n") do
    local s = line:gsub("#.*$", "")
    local k, v = s:match("^%s*([%w_]+)%s*=%s*(.-)%s*$")
    if k then out[k] = v end
  end
  return out
end

-- Numeric value of key k, or default.
function M.num(t, k, default)
  local n = t[k] and tonumber(t[k])
  return n or default
end

-- Comma-separated value of key k as a list of trimmed, non-empty strings ({}).
function M.list(t, k)
  local out = {}
  if t[k] then
    for item in (t[k] .. ","):gmatch("%s*([^,]-)%s*,") do
      if item ~= "" then out[#out + 1] = item end
    end
  end
  return out
end

return M

end

CES_MANIFEST = { name = "Discovery", version = "0.2", description = "Network registry crawler that keeps the peer table populated and gossips known servers via paid sample exchange." }
return require("main")