fix(kimi): resolve the server port from the live lock, not a stale log record (#39)

.collab/runbooks/c2c-env-vars.md

@@ -356,6 +356,33 @@ hook. Same removal cycle as the parent env var.
 Seconds the hook will block on `c2c await-reply` before falling closed
 (default 120). Not deprecated; tunable independently.
 
+### `C2C_KIMI_SERVER_PORT`
+
+Overrides the port c2c uses to reach the local Kimi Code REST server. It is
+one candidate in a precedence chain, not the whole answer (#39):
+
+1. `C2C_KIMI_DELIVER_FIXTURE_BASE_URL` (tests only; requires
+   `C2C_KIMI_DELIVER_FIXTURE=1`)
+2. kimi's live server lock — `~/.kimi-code/server/lock`, or
+   `~/.kimi-code/server/instances/*.json` under kimi's `multi_server` flag.
+   Used only when the recorded pid is alive.
+3. `C2C_KIMI_SERVER_PORT`
+4. the last `"msg":"server listening"` record in
+   `~/.kimi-code/server/server.log` — **only if a TCP probe says it is live**
+5. default port `58627`
+
+The log record used to sit at the top of this chain. Modern kimi-code appends
+it only on a *cold* start (a warm start logs plain text: `server already
+running (pid=…, port=…)`), so it aged into a permanently dead port and
+silently broke 100% of Kimi delivery — and `C2C_KIMI_SERVER_PORT` could not
+rescue it, because the log scrape pre-empted the fallback it fed.
+
+### `C2C_KIMI_PROBE_TIMEOUT`
+
+Seconds the TCP liveness probe waits when validating a candidate Kimi server
+address (default `0.5`). Only the log-scraped candidate is probed; the lock
+file is validated by pid-liveness instead.
+
 ---
 
 ## E2E / Relay

CLAUDE.md

@@ -182,6 +182,10 @@ app-server, OpenCode the plugin API). That is an upstream ask, not our bug.
   (pidfile/sidfile) while inboxes are **session-id**-keyed, so it records its
   binding in `<alias>.sid` and re-keys when a real session id appears (#9).
   `c2c doctor hooks --rearm` re-arms DEAF sessions (inbox > 0, no notifier).
+  **Server address resolution (#39)**: live lock (`~/.kimi-code/server/lock`,
+  pid-checked) → `C2C_KIMI_SERVER_PORT` → liveness-probed `server.log` record →
+  default 58627. Never trust `server.log` unprobed — kimi writes the
+  `"server listening"` record on cold start only, so it ages into a dead port.
   Legacy notification-store runbook:
   `.collab/runbooks/kimi-notification-store-delivery.md` (deprecated).
 - **OpenCode**: SIGUSR1 to the *inner* OpenCode pid (not the outer wrapper)

ocaml/c2c_kimi_deliver.ml

@@ -73,13 +73,198 @@ let server_listening_url () =
          in
          scan None)
 
+(* ─── Live-server discovery (#39) ────────────────────────────────────────── *)
+
+(* Kimi Code's single-instance lock: <kimi home>/server/lock, JSON of the shape
+     {"pid":721503,"started_at":"2026-07-19T08:50:28.865Z",
+      "host":"127.0.0.1","port":58627,"host_version":"…","entry":"…"}
+   It is created with O_EXCL when a server binds, its `port` is rewritten via
+   updatePort(), and it is unlinked on clean shutdown. A lock whose recorded
+   pid is dead is stale (kimi itself treats it as takeable), so a pid-liveness
+   probe is the authoritative "is this the running server?" check. *)
+let server_lock_path () = kimi_code_home () // "server" // "lock"
+
+(* With kimi's experimental `multi_server` flag the exclusive lock is replaced
+   by one self-describing file per instance under server/instances/. Same idea,
+   different shape: {"server_id":…,"pid":…,"host":…,"port":…,
+   "started_at":<ms>,"heartbeat_at":<ms>}. Stale entries are swept lazily by
+   kimi, so we pid-check here too. *)
+let server_instances_dir () = kimi_code_home () // "server" // "instances"
+
+let pid_alive pid =
+  if pid <= 0 then false
+  else
+    try Unix.kill pid 0; true with
+    | Unix.Unix_error (Unix.ESRCH, _, _) -> false
+    | Unix.Unix_error (Unix.EPERM, _, _) -> true
+    | _ -> true
+
+let read_file_opt path =
+  if not (Sys.file_exists path) then None
+  else
+    try
+      let ic = open_in_bin path in
+      Fun.protect ~finally:(fun () -> close_in ic) (fun () ->
+        Some (really_input_string ic (in_channel_length ic)))
+    with _ -> None
+
+(* Extract (pid, host, port) from a lock/instance JSON blob. Returns None on
+   anything malformed — callers fall through to the next candidate rather than
+   raising, so a corrupt lock can never break delivery. *)
+let parse_lock_json raw =
+  try
+    let json = Yojson.Safe.from_string raw in
+    let open Yojson.Safe.Util in
+    let pid = json |> member "pid" |> to_int_option in
+    let port = json |> member "port" |> to_int_option in
+    let host =
+      match json |> member "host" |> to_string_option with
+      | Some h when String.trim h <> "" -> String.trim h
+      | _ -> "127.0.0.1"
+    in
+    match pid, port with
+    | Some pid, Some port when port > 0 -> Some (pid, host, port)
+    | _ -> None
+  with _ -> None
+
+let url_of_host_port host port =
+  (* An IPv6 literal needs brackets in a URL authority. *)
+  let host = if String.contains host ':' then "[" ^ host ^ "]" else host in
+  Printf.sprintf "http://%s:%d" host port
+
+let lock_file_url () =
+  match read_file_opt (server_lock_path ()) with
+  | None -> None
+  | Some raw ->
+      (match parse_lock_json raw with
+       | Some (pid, host, port) when pid_alive pid -> Some (url_of_host_port host port)
+       | _ -> None)
+
+let instance_registry_url () =
+  let dir = server_instances_dir () in
+  match (try Some (Sys.readdir dir) with _ -> None) with
+  | None -> None
+  | Some names ->
+      let live =
+        Array.to_list names
+        |> List.filter (fun n -> Filename.check_suffix n ".json")
+        |> List.filter_map (fun n ->
+               match read_file_opt (dir // n) with
+               | None -> None
+               | Some raw ->
+                   (match parse_lock_json raw with
+                    | Some (pid, host, port) when pid_alive pid -> Some (host, port)
+                    | _ -> None))
+      in
+      (match live with
+       | [] -> None
+       | (host, port) :: _ -> Some (url_of_host_port host port))
+
+let live_lock_url () =
+  match lock_file_url () with
+  | Some url -> Some url
+  | None -> instance_registry_url ()
+
+(* Cheap TCP liveness probe for an http://host:port base URL. Used only for
+   candidates we do not otherwise trust (the historical log scrape), so that a
+   dead address is skipped instead of being retried forever. *)
+let probe_timeout () =
+  match Sys.getenv_opt "C2C_KIMI_PROBE_TIMEOUT" with
+  | Some s -> (match float_of_string_opt s with Some f when f > 0.0 -> f | _ -> 0.5)
+  | None -> 0.5
+
+let host_port_of_url url =
+  match String.index_opt url ':' with
+  | None -> None
+  | Some _ ->
+      let rest =
+        match String.index_opt url '/' with
+        | Some i when i + 1 < String.length url && url.[i + 1] = '/' ->
+            String.sub url (i + 2) (String.length url - i - 2)
+        | _ -> url
+      in
+      let authority =
+        match String.index_opt rest '/' with
+        | Some i -> String.sub rest 0 i
+        | None -> rest
+      in
+      (match String.rindex_opt authority ':' with
+       | None -> None
+       | Some i ->
+           let host = String.sub authority 0 i in
+           let port_s = String.sub authority (i + 1) (String.length authority - i - 1) in
+           let host =
+             let n = String.length host in
+             if n >= 2 && host.[0] = '[' && host.[n - 1] = ']' then String.sub host 1 (n - 2)
+             else host
+           in
+           (match int_of_string_opt port_s with
+            | Some p when p > 0 && host <> "" -> Some (host, p)
+            | _ -> None))
+
+let address_is_live url =
+  match host_port_of_url url with
+  | None -> false
+  | Some (host, port) ->
+      let addrs =
+        try
+          Unix.getaddrinfo host (string_of_int port)
+            [ Unix.AI_SOCKTYPE Unix.SOCK_STREAM ]
+        with _ -> []
+      in
+      let timeout = probe_timeout () in
+      let try_addr ai =
+        let fd = Unix.socket ai.Unix.ai_family ai.Unix.ai_socktype 0 in
+        Fun.protect
+          ~finally:(fun () -> try Unix.close fd with _ -> ())
+          (fun () ->
+            try
+              Unix.set_nonblock fd;
+              (try Unix.connect fd ai.Unix.ai_addr with
+               | Unix.Unix_error (Unix.EINPROGRESS, _, _) -> ());
+              (match Unix.select [] [ fd ] [] timeout with
+               | _, [ _ ], _ -> Unix.getsockopt_error fd = None
+               | _ -> false)
+            with _ -> false)
+      in
+      List.exists try_addr addrs
+
+(* Resolution precedence (#39):
+     0. fixture override               — tests only, gated by C2C_KIMI_DELIVER_FIXTURE
+     1. live lock / instance registry  — ground truth for the process actually
+                                         listening right now; pid-liveness checked
+     2. $C2C_KIMI_SERVER_PORT          — explicit operator intent
+     3. server.log "server listening"  — historical and append-only; modern
+                                         kimi only writes it on a COLD start, so
+                                         it ages into a dead port. Usable only
+                                         after a liveness probe.
+     4. default port 58627
+   The log used to sit at the TOP of this list, which is what broke delivery
+   permanently on hosts whose server had warm-started onto a different port. *)
 let server_base_url () =
   match fixture_enabled (), Sys.getenv_opt "C2C_KIMI_DELIVER_FIXTURE_BASE_URL" with
   | true, Some u when String.trim u <> "" -> Some (String.trim u)
   | _ ->
-      (match server_listening_url () with
+      let env_port =
+        match Sys.getenv_opt "C2C_KIMI_SERVER_PORT" with
+        | Some p when p <> "" -> Some (Printf.sprintf "http://127.0.0.1:%s" p)
+        | _ -> None
+      in
+      let probed_log () =
+        match server_listening_url () with
+        | Some url when address_is_live url -> Some url
+        | _ -> None
+      in
+      (match live_lock_url () with
        | Some url -> Some url
-       | None -> Some (Printf.sprintf "http://127.0.0.1:%s" (default_port ())))
+       | None ->
+           (match env_port with
+            | Some url -> Some url
+            | None ->
+                (match probed_log () with
+                 | Some url -> Some url
+                 | None ->
+                     Some (Printf.sprintf "http://127.0.0.1:%s" (default_port ())))))
 
 let session_index_path () = kimi_code_home () // "session_index.jsonl"
 

ocaml/c2c_kimi_deliver.mli

@@ -2,6 +2,13 @@
 
 val server_token_path : unit -> string
 val server_base_url : unit -> string option
+(** [server_base_url ()] resolves the base URL of the local Kimi Code server.
+
+    Precedence (#39): fixture override → live lock file / instance registry →
+    [$C2C_KIMI_SERVER_PORT] → liveness-probed server.log record → default port
+    58627. The log record is last-resort and probed because modern kimi-code
+    only appends it on a COLD start, so it ages into a dead port and used to
+    pre-empt every correct source. *)
 val read_server_token : unit -> string option
 val fixture_enabled : unit -> bool
 
@@ -23,7 +30,24 @@ val session_dir_for_session_id : session_id:string -> string option
 
 val server_listening_url : unit -> string option
 (** [server_listening_url ()] scans ~/.kimi-code/server/server.log for the
-    latest "server listening" log line and returns the bound address. *)
+    latest "server listening" log line and returns the bound address.
+
+    HISTORICAL ONLY (#39): the log is append-only and the record is written on
+    a cold start, so the address may name a long-dead server. Never use it
+    without [address_is_live]. *)
+
+val server_lock_path : unit -> string
+(** Path to kimi's single-instance server lock, ~/.kimi-code/server/lock. *)
+
+val live_lock_url : unit -> string option
+(** [live_lock_url ()] returns the base URL recorded by the currently-running
+    Kimi server, from the exclusive lock file or (under kimi's [multi_server]
+    flag) the server/instances registry. [None] when the file is missing,
+    malformed, or records a dead pid. Never raises. *)
+
+val address_is_live : string -> bool
+(** [address_is_live url] TCP-connects to the host:port of [url] with a short
+    timeout ([$C2C_KIMI_PROBE_TIMEOUT], default 0.5s). Never raises. *)
 
 val submit_prompt : session_id:string -> body:string -> (int, string) result
 (** [submit_prompt ~session_id ~body] POSTs a text prompt to

ocaml/c2c_kimi_notifier.ml

@@ -257,27 +257,90 @@ let kimi_server_is_responding () =
                Lwt.return (code = 200))
            with _ -> false)
 
+(* #39: consecutive failed start attempts, and the earliest time we are willing
+   to try `kimi server run` again. Before #39 this path re-spawned + waited 10s
+   on every poll (~every 2s) against an address it had never validated, so a
+   stale-port resolution produced an identical ECONNREFUSED line forever with
+   no actionable signal. *)
+let ensure_failures = ref 0
+let next_start_attempt = ref 0.0
+
+(* Escalating backoff so a persistently unreachable server costs one attempt
+   per 10s → 20s → … → 5min instead of one per poll. *)
+let start_backoff_s n =
+  let base = 10.0 *. (2.0 ** float_of_int (max 0 (n - 1))) in
+  if base > 300.0 then 300.0 else base
+
+(* Threshold at which the log line changes from "transient" to "actionable" —
+   past this, retrying is not going to fix it and the operator needs to know
+   WHICH address we are failing against. *)
+let ensure_failures_actionable = 3
+
 (* Start the Kimi server if it is not already running.  The command is
    idempotent: if a server is already bound it prints the existing URL and
-   exits.  We only start a real server outside of fixture/test mode. *)
+   exits.  We only start a real server outside of fixture/test mode.
+
+   Non-fatal by construction: on any failure we simply return, delivery reports
+   an error, and the message stays in the broker inbox for the next poll. Each
+   call re-resolves the base URL via [C2c_kimi_deliver.server_base_url] rather
+   than reusing a cached address, so a server that moved ports is picked up on
+   the next attempt (#39). *)
 let ensure_kimi_server_running () =
   if C2c_kimi_deliver.fixture_enabled () then ()
-  else if not (kimi_server_is_responding ()) then begin
-    Printf.eprintf
-      "[kimi-notifier] Kimi server not responding; starting it...\n%!";
-    let cmd = "kimi server run --keep-alive >/dev/null 2>&1" in
-    ignore (Unix.system cmd);
-    let deadline = Unix.gettimeofday () +. 10.0 in
-    let rec wait () =
-      if Unix.gettimeofday () > deadline then
-        Printf.eprintf
-          "[kimi-notifier] timed out waiting for Kimi server\n%!"
-      else if kimi_server_is_responding () then ()
-      else (
-        Unix.sleepf 0.2;
-        wait ())
-    in
-    wait ()
+  else if kimi_server_is_responding () then begin
+    if !ensure_failures > 0 then
+      Printf.eprintf "[kimi-notifier] Kimi server reachable again at %s\n%!"
+        (Option.value ~default:"(unknown)" (C2c_kimi_deliver.server_base_url ()));
+    ensure_failures := 0;
+    next_start_attempt := 0.0
+  end
+  else begin
+    let now = Unix.gettimeofday () in
+    if now < !next_start_attempt then
+      (* Backing off: do not spawn another server, do not block the poll loop.
+         Delivery will fail cleanly and the message stays queued. *)
+      ()
+    else begin
+      let base =
+        Option.value ~default:"(unresolved)" (C2c_kimi_deliver.server_base_url ())
+      in
+      Printf.eprintf
+        "[kimi-notifier] Kimi server not responding at %s; starting it...\n%!" base;
+      let cmd = "kimi server run --keep-alive >/dev/null 2>&1" in
+      ignore (Unix.system cmd);
+      let deadline = Unix.gettimeofday () +. 10.0 in
+      let rec wait () =
+        if kimi_server_is_responding () then true
+        else if Unix.gettimeofday () > deadline then false
+        else (
+          Unix.sleepf 0.2;
+          wait ())
+      in
+      if wait () then begin
+        ensure_failures := 0;
+        next_start_attempt := 0.0
+      end
+      else begin
+        incr ensure_failures;
+        let n = !ensure_failures in
+        let backoff = start_backoff_s n in
+        next_start_attempt := Unix.gettimeofday () +. backoff;
+        if n >= ensure_failures_actionable then
+          Printf.eprintf
+            "[kimi-notifier] Kimi server still unreachable at %s after %d \
+             attempts. c2c resolved that address from kimi's server lock \
+             (%s), $C2C_KIMI_SERVER_PORT, or the server.log record. Check the \
+             real port with `ss -ltnp | grep kimi` and export \
+             C2C_KIMI_SERVER_PORT=<port> if it disagrees. Mail stays queued; \
+             retrying in %.0fs.\n%!"
+            base n (C2c_kimi_deliver.server_lock_path ()) backoff
+        else
+          Printf.eprintf
+            "[kimi-notifier] timed out waiting for Kimi server at %s \
+             (attempt %d); retrying in %.0fs\n%!"
+            base n backoff
+      end
+    end
   end
 
 (* Resolve the Kimi session id for the current working directory by reading