session.clj

(ns github.copilot-sdk.session
  "CopilotSession - session operations using centralized client state.
   
   All session state is stored in the client's :state atom under:
   - [:sessions session-id] -> {:tool-handlers {} :permission-handler nil :destroyed? false :workspace-path nil}
   - [:session-io session-id] -> {:event-chan :event-mult}
   
   Functions take client + session-id, accessing state through the client."
  (:require [clojure.core.async :as async :refer [go go-loop <! >! >!! <!! chan close! put! alts!! mult tap untap]]
            [clojure.core.async.impl.channels :as channels]
            [clojure.spec.alpha :as s]
            [clojure.data.json :as json]
            [github.copilot-sdk.protocol :as proto]
            [github.copilot-sdk.logging :as log]
            [github.copilot-sdk.specs :as specs]
            [github.copilot-sdk.util :as util]))

;; -----------------------------------------------------------------------------
;; State accessors - all state lives in client's atom
;; -----------------------------------------------------------------------------

(defn- session-state [client session-id]
  (get-in @(:state client) [:sessions session-id]))

(defn- session-io [client session-id]
  (get-in @(:state client) [:session-io session-id]))

(defn- update-session! [client session-id f & args]
  (apply swap! (:state client) update-in [:sessions session-id] f args))

(defn- connection-io [client]
  (:connection-io @(:state client)))

;; -----------------------------------------------------------------------------
;; Session record - lightweight handle returned to users
;; Contains only immutable data + reference to client
;; -----------------------------------------------------------------------------

(defrecord CopilotSession
           [session-id
            client])     ; reference to owning client

;; -----------------------------------------------------------------------------
;; Internal functions
;; -----------------------------------------------------------------------------

(defn create-session
  "Create a new session. Internal use - called by client.
   Initializes session state in client's atom and returns a CopilotSession handle.
   If :on-event is provided, taps a subscriber that forwards events to the handler
   on a dedicated thread. Uses a sliding buffer, so events may be dropped under
   extreme backpressure if the handler cannot keep up with the event rate."
  [client session-id {:keys [tools commands on-permission-request on-user-input-request hooks workspace-path on-event config]}]
  (log/debug "Creating session: " session-id)
  (let [event-chan (chan (async/sliding-buffer 4096))
        event-mult (mult event-chan)
        send-lock (doto (chan 1) (>!! :token))
        tool-handlers (into {} (map (fn [t] [(:tool-name t) (:tool-handler t)]) tools))
        command-handlers (into {} (map (fn [c] [(:command-name c) (:command-handler c)]) commands))]
    ;; Store session state and IO in client's atom
    (swap! (:state client)
           (fn [state]
             (-> state
                  (assoc-in [:sessions session-id]
                            {:tool-handlers tool-handlers
                             :command-handlers command-handlers
                             :permission-handler on-permission-request
                             :user-input-handler on-user-input-request
                             :hooks hooks
                             :capabilities {}
                             :destroyed? false
                             :workspace-path workspace-path
                             :config config})
                 (assoc-in [:session-io session-id]
                           {:event-chan event-chan
                            :event-mult event-mult
                            :send-lock send-lock}))))
    ;; If an on-event handler is provided, tap and forward events to it.
    ;; Uses async/thread to avoid blocking core.async dispatch threads,
    ;; since user handlers may perform blocking I/O.
    ;; The handler channel uses a sliding buffer — if the handler cannot keep up
    ;; with the event rate, oldest unprocessed events are silently dropped.
    (when on-event
      (let [handler-ch (chan (async/sliding-buffer 1024))]
        (tap event-mult handler-ch)
        (async/thread
          (loop []
            (if-let [event (<!! handler-ch)]
              (do
                (try
                  (on-event event)
                  (catch Throwable t
                    (log/warn t "on-event handler threw"
                              {:session-id session-id
                               :event-type (:type event)})))
                (recur))
              ;; Channel closed — session torn down
              nil)))))
    (log/debug "Session created: " session-id)
    ;; Return lightweight handle
    (->CopilotSession session-id client)))

(defn set-workspace-path!
  "Update the workspace path in session state. Called after RPC response."
  [client session-id workspace-path]
  (when workspace-path
    (swap! (:state client) assoc-in [:sessions session-id :workspace-path] workspace-path)))

(defn set-capabilities!
  "Update session capabilities from the create/resume RPC response.
   Called after session.create or session.resume succeeds."
  [client session-id caps]
  (swap! (:state client) assoc-in [:sessions session-id :capabilities] (or caps {})))

(defn register-transform-callbacks!
  "Store system message transform callbacks on a session.
   Callbacks is a map of wire section ID strings to 1-arity functions
   that receive current content and return transformed content."
  [client session-id callbacks]
  (when callbacks
    (swap! (:state client) assoc-in [:sessions session-id :transform-callbacks] callbacks)))

(defn handle-system-message-transform
  "Handle a systemMessage.transform RPC request from the CLI runtime.
   Dispatches each section to its registered transform callback.
   On callback error, returns the original content (graceful fallback).
   
   Uses string keys in the response to preserve the original wire-format
   section IDs (e.g. \"tool_efficiency\", not \"tool-efficiency\")."
  [client session-id sections]
  (let [callbacks (get-in @(:state client) [:sessions session-id :transform-callbacks])]
    {:sections
     (reduce-kv
      (fn [acc section-id {:keys [content]}]
        (let [;; Convert incoming kebab-case keyword back to wire string ID
              ;; e.g. :tool-efficiency -> "tool_efficiency"
              wire-id (util/section-kw->wire-id section-id)
              callback (get callbacks wire-id)]
          ;; Use wire string as response key to preserve original format
          (assoc acc wire-id
                 {:content
                  (if callback
                    (try
                      (callback content)
                      (catch Throwable t
                        (log/warn t "systemMessage.transform callback failed"
                                  {:session-id session-id :section wire-id})
                        content))
                    content)})))
      {}
      sections)}))

(defn remove-session!
  "Remove a session from client state. Called on RPC failure during pre-registration."
  [client session-id]
  (when-let [{:keys [event-chan]} (get-in @(:state client) [:session-io session-id])]
    (close! event-chan))
  (swap! (:state client) (fn [s]
                           (-> s
                               (update :sessions dissoc session-id)
                               (update :session-io dissoc session-id)))))

(defn dispatch-event!
  "Dispatch an event to all subscribers via the mult. Called by client notification router.
   Events are dropped (with warning) if the session event buffer is full."
  [client session-id event]
  (let [normalized-event (update event :type util/event-type->keyword)]
    (log/debug "Dispatching event to session " session-id ": type=" (:type normalized-event))
    (when-not (:destroyed? (session-state client session-id))
      (when-let [{:keys [event-chan]} (session-io client session-id)]
        (when-not (async/offer! event-chan normalized-event)
          (log/warn "Dropping event for session " session-id
                    " type=" (:type normalized-event) " (event buffer full)"))))))

(defn- normalize-tool-result
  "Normalize a tool result to the wire format."
  [result]
  (cond
    (nil? result)
    {:text-result-for-llm "Tool returned no result"
     :result-type "failure"
     :error "tool returned no result"
     :tool-telemetry {}}

    ;; Already a result object (duck-type check)
    (and (map? result) (:text-result-for-llm result) (:result-type result))
    result

    ;; Backward compatibility for camelCase result maps
    (and (map? result) (:textResultForLlm result) (:resultType result))
    (util/wire->clj result)

    ;; String result
    (string? result)
    {:text-result-for-llm result
     :result-type "success"
     :tool-telemetry {}}

    ;; Any other value - JSON encode
    :else
    {:text-result-for-llm (json/write-str result)
     :result-type "success"
     :tool-telemetry {}}))

(defn- channel?
  "Check if x is a core.async channel."
  [x]
  (instance? clojure.core.async.impl.channels.ManyToManyChannel x))

(defn handle-tool-call!
  "Handle an incoming tool call request. Returns a channel with the result wrapper."
  [client session-id tool-call-id tool-name arguments & {:keys [traceparent tracestate]}]
  (async/thread-call
   (fn []
     (let [handler (get-in (session-state client session-id) [:tool-handlers tool-name])
           timeout-ms (or (:tool-timeout-ms (:options client)) 120000)]
       (if-not handler
         {:result {:text-result-for-llm (str "Tool '" tool-name "' is not supported by this client instance.")
                   :result-type "failure"
                   :error (str "tool '" tool-name "' not supported")
                   :tool-telemetry {}}}
         (try
           (let [invocation (cond-> {:session-id session-id
                                     :tool-call-id tool-call-id
                                     :tool-name tool-name
                                     :arguments arguments}
                              traceparent (assoc :traceparent traceparent)
                              tracestate (assoc :tracestate tracestate))
                 result (handler arguments invocation)
                 result (if (channel? result)
                          (let [timeout-ch (async/timeout timeout-ms)
                                [value ch] (alts!! [result timeout-ch])]
                            (if (= ch timeout-ch)
                              (throw (ex-info "Tool timeout" {:timeout-ms timeout-ms
                                                              :tool-name tool-name
                                                              :tool-call-id tool-call-id}))
                              value))
                          result)]
             {:result (normalize-tool-result result)})
           (catch Exception e
             {:result {:text-result-for-llm "Invoking this tool produced an error. Detailed information is not available."
                       :result-type "failure"
                       :error (ex-message e)
                       :tool-telemetry {}}})))))
   :mixed))

(defn handle-permission-request!
  "Handle an incoming permission request. Returns a channel with the result.
   When the handler returns `{:kind :no-result}`, the result is
   `{:result :no-result}` — callers must check for this sentinel:
   - **v3 (broadcast path):** skip the `handlePendingPermissionRequest` RPC
     entirely so the extension does not answer this permission request.
   - **v2 (request-handler path):** propagate as a JSON-RPC internal error
     (code -32603) so the CLI knows the request was not handled."
  [client session-id request]
  (async/thread-call
   (fn []
     (let [handler (:permission-handler (session-state client session-id))]
       (if-not handler
         {:result {:kind :denied-no-approval-rule-and-could-not-request-from-user}}
         (try
           (let [result (handler request {:session-id session-id})
                 ;; If handler returns a channel, await it
                 result (if (channel? result)
                          (<!! result)
                          result)]
             (cond
               ;; no-result: extension doesn't answer this permission request
               (and (map? result) (= :no-result (:kind result)))
               {:result :no-result}

               (and (map? result) (contains? result :kind))
               {:result result}

               ;; Wrapped form: {:result {:kind ...}}
               (and (map? result) (contains? result :result)
                    (map? (:result result)) (= :no-result (:kind (:result result))))
               {:result :no-result}

               (and (map? result) (contains? result :result)
                    (map? (:result result)) (contains? (:result result) :kind))
                result

                :else
                (do
                  (log/warn "Invalid permission response for session " session-id ": " result)
                  {:result {:kind :denied-no-approval-rule-and-could-not-request-from-user}})))
             (catch Exception e
               (log/error "Permission handler error for session " session-id ": " (ex-message e))
               {:result {:kind :denied-no-approval-rule-and-could-not-request-from-user}})))))
     :io))

(defn handle-user-input-request!
  "Handle an incoming user input request (ask_user). Returns a channel with the result.
   PR #269 feature.
   
   The handler should return a map with :answer (string) and optionally :was-freeform (boolean).
   For backwards compatibility, :response is also accepted as an alias for :answer."
  [client session-id request]
  (async/thread-call
   (fn []
     (let [handler (:user-input-handler (session-state client session-id))]
       (if-not handler
         {:error {:code -32001 :message "User input requested but no handler registered"}}
         (try
            (let [result (handler request {:session-id session-id})
                  ;; If handler returns a channel, await it
                  result (if (channel? result)
                           (<!! result)
                           result)
                  ;; Normalize result to expected wire format
                  ;; Accept :answer or :response, default was-freeform to true if not specified
                  answer (or (:answer result) (:response result))
                  was-freeform (if (contains? result :was-freeform)
                                 (:was-freeform result)
                                 true)]
              (if (and (string? answer) (not (empty? answer)))
                {:result {:answer answer :was-freeform was-freeform}}
                (do
                  (log/warn "Invalid user input response for session " session-id ": " result)
                  {:error {:code -32001 :message "User input handler returned invalid answer"}})))
           (catch Exception e
             (log/error "User input handler error for session " session-id ": " (ex-message e))
             {:error {:code -32001 :message (str "User input handler error: " (ex-message e))}})))))
   :io))

(defn handle-hooks-invoke!
  "Handle an incoming hooks invocation. Returns a channel with the result.
   PR #269 feature."
  [client session-id hook-type input]
  (async/thread-call
   (fn []
     (let [hooks (:hooks (session-state client session-id))]
       (if-not hooks
         {:result nil}
         (let [;; Map hook type strings to handler keywords
               handler-key (case hook-type
                             "preToolUse" :on-pre-tool-use
                             "postToolUse" :on-post-tool-use
                             "userPromptSubmitted" :on-user-prompt-submitted
                             "sessionStart" :on-session-start
                             "sessionEnd" :on-session-end
                             "errorOccurred" :on-error-occurred
                             nil)
               handler (when handler-key (get hooks handler-key))]
           (if-not handler
             {:result nil}
             (try
               (let [result (handler input {:session-id session-id})
                     ;; If handler returns a channel, await it
                     result (if (channel? result)
                              (<!! result)
                              result)]
                 {:result result})
               (catch Exception e
                 (log/error "Hook handler error for session " session-id ", hook " hook-type ": " (ex-message e))
                 {:result nil})))))))
   :io))

(defn handle-command!
  "Handle an incoming command.execute event. Returns a channel with the result.
   Context map passed to handler mirrors TypeScript's CommandContext:
   {:session-id :command-name :command :args}"
  [client session-id command-name command args]
  (async/thread-call
   (fn []
     (let [handler (get-in (session-state client session-id) [:command-handlers command-name])]
       (if-not handler
         {:error (str "Unknown command: " command-name)}
         (try
           (let [ctx {:session-id session-id
                      :command-name command-name
                      :command command
                      :args args}
                 result (handler ctx)
                 ;; If handler returns a channel, await it
                 _ (when (channel? result) (<!! result))]
             {:ok true})
           (catch Exception e
             (log/error "Command handler error for session " session-id ", command " command-name ": " (ex-message e))
             {:error (ex-message e)})))))
   :io))

;; -----------------------------------------------------------------------------
;; Public API - functions that take CopilotSession handle
;; -----------------------------------------------------------------------------

(defn config
  "Get the session configuration that was used to create this session.
   Returns the user-provided config. Note: This reflects what was requested,
   not necessarily what the server is using. The session.start event contains
   the actual selectedModel if validation is needed."
  [session]
  (let [{:keys [session-id client]} session]
    (:config (session-state client session-id))))

(defn send!
  "Send a message to the session.
   Returns the message ID immediately (fire-and-forget).
   
   Options:
   - :prompt       - The message text (required)
   - :attachments  - Vector of attachments (file/directory/selection)
   - :mode         - :enqueue (default) or :immediate"
  [session opts]
  (when-not (s/valid? ::specs/send-options opts)
    (throw (ex-info "Invalid send options"
                    {:opts opts
                     :explain (s/explain-data ::specs/send-options opts)})))
  (let [{:keys [session-id client]} session]
    (log/debug "send! called for session " session-id " with prompt: " (subs (str (:prompt opts)) 0 (min 50 (count (str (:prompt opts))))) "...")
    (when (:destroyed? (session-state client session-id))
      (throw (ex-info "Session has been disconnected" {:session-id session-id})))
    (let [conn (connection-io client)
          wire-attachments (when (:attachments opts)
                             (util/attachments->wire (:attachments opts)))
          trace-ctx (when-let [provider (:on-get-trace-context client)]
                      (try (let [ctx (provider)]
                             (when (map? ctx)
                               (select-keys ctx [:traceparent :tracestate])))
                           (catch Throwable _ nil)))
          params (cond-> {:session-id session-id
                          :prompt (:prompt opts)}
                   trace-ctx (merge trace-ctx)
                   wire-attachments (assoc :attachments wire-attachments)
                   (:mode opts) (assoc :mode (name (:mode opts))))
          result (proto/send-request! conn "session.send" params)
          msg-id (:message-id result)]
      (log/debug "send! completed for session " session-id " message-id=" msg-id)
      msg-id)))

(defn send-and-wait!
  "Send a message and wait until the session becomes idle.
   Returns the final assistant message event, or nil if none received.
   Serialized per session to avoid mixing concurrent sends.
   
   Options: same as send!
   
   Additional options:
   - :timeout-ms   - Timeout in milliseconds (default: 300000)"
  ([session opts]
   (send-and-wait! session opts 300000))
  ([session opts timeout-ms]
   (let [{:keys [session-id client]} session]
     (log/debug "send-and-wait! called for session " session-id)
     (when (:destroyed? (session-state client session-id))
       (throw (ex-info "Session has been disconnected" {:session-id session-id})))

     (let [event-ch (chan 1024)
           last-assistant-msg (atom nil)
           {:keys [event-mult send-lock]} (session-io client session-id)]
        ;; Acquire channel-based lock (blocks calling thread)
        (<!! send-lock)

        (try
         ;; Tap the mult BEFORE sending - ensures we don't miss events
         (log/debug "send-and-wait! tapping event mult for session " session-id)
         (tap event-mult event-ch)

         ;; Send the message
         (log/debug "send-and-wait! sending message")
         (send! session opts)

         ;; Wait for events with single deadline timeout
         (log/debug "send-and-wait! waiting for result with timeout " timeout-ms "ms")
         (let [deadline-ch (async/timeout timeout-ms)]
           (loop []
             (let [[event ch] (alts!! [event-ch deadline-ch])]
               (cond
                 (= ch deadline-ch)
                 (do
                   (log/error "send-and-wait! timeout after " timeout-ms "ms for session " session-id)
                   (throw (ex-info (str "Timeout after " timeout-ms "ms waiting for session.idle")
                                   {:timeout-ms timeout-ms})))

                 (nil? event)
                 (do
                   (log/debug "send-and-wait! event channel closed for session " session-id)
                   (throw (ex-info "Event channel closed unexpectedly" {})))

                 (= :copilot/assistant.message (:type event))
                 (do
                   (log/debug "send-and-wait! got assistant.message, continuing to wait for idle")
                   (reset! last-assistant-msg event)
                   (recur))

                 (= :copilot/session.idle (:type event))
                 (do
                   (log/debug "send-and-wait! got session.idle, returning result for session " session-id)
                   @last-assistant-msg)

                 (= :copilot/session.error (:type event))
                 (do
                   (log/error "send-and-wait! got session.error for session " session-id)
                   (throw (ex-info (get-in event [:data :message] "Session error")
                                   {:event event})))

                 :else
                 (do
                   (log/debug "send-and-wait! ignoring event type: " (:type event))
                   (recur))))))

         (finally
           (log/debug "send-and-wait! cleaning up subscription")
           (untap event-mult event-ch)
           (close! event-ch)
           (put! send-lock :token)))))))

(defn- send-async*
  "Send a message and return {:message-id :events-ch}."
  ([session opts]
   (send-async* session opts nil))
  ([session opts timeout-ms]
   (let [{:keys [session-id client]} session]
     (when (:destroyed? (session-state client session-id))
       (throw (ex-info "Session has been disconnected" {:session-id session-id})))

     (let [out-ch (chan 1024)
           event-ch (chan 1024)
           {:keys [event-mult send-lock]} (session-io client session-id)
           released? (atom false)
           release-lock! (fn []
                           (when (compare-and-set! released? false true)
                             (put! send-lock :token)))
           deadline-ch (when timeout-ms (async/timeout timeout-ms))
           timeout-event {:type :copilot/session.error
                          :data {:message (str "Timeout after " timeout-ms "ms waiting for session.idle")
                                 :timeout-ms timeout-ms}}
           emit! (fn [event]
                   (when-not (async/offer! out-ch event)
                     (log/debug "Dropping event for session " session-id " due to full async buffer")))]
        ;; Acquire channel-based lock (blocks calling thread)
        (<!! send-lock)

       ;; Tap the mult for events, then send
       (try
         (tap event-mult event-ch)
         (let [message-id (send! session opts)]
           (go-loop []
             (let [[event ch] (if deadline-ch
                                (async/alts! [event-ch deadline-ch])
                                [(<! event-ch) event-ch])]
               (cond
                 (and deadline-ch (= ch deadline-ch))
                 (do
                   (emit! timeout-event)
                   (untap event-mult event-ch)
                   (close! event-ch)
                   (close! out-ch)
                   (release-lock!))

                 (nil? event)
                 (do
                   (untap event-mult event-ch)
                   (close! out-ch)
                   (release-lock!))

                 (= :copilot/session.idle (:type event))
                 (do
                   (emit! event)
                   (untap event-mult event-ch)
                   (close! event-ch)
                   (close! out-ch)
                   (release-lock!))

                 (= :copilot/session.error (:type event))
                 (do
                   (emit! event)
                   (untap event-mult event-ch)
                   (close! event-ch)
                   (close! out-ch)
                   (release-lock!))

                 :else
                 (do
                   (emit! event)
                   (recur)))))
           {:message-id message-id
            :events-ch out-ch})
         (catch Exception e
           (untap event-mult event-ch)
           (close! event-ch)
           (close! out-ch)
           (release-lock!)
           (throw e)))))))

(defn- <send-async*
  "Fully non-blocking send pipeline for use in go blocks.
   Acquires lock, sends message, and processes events — all via parking channel ops.
   Returns events-ch immediately; events flow once the go block completes setup."
  [session opts timeout-ms]
  (let [{:keys [session-id client]} session]
    (when (:destroyed? (session-state client session-id))
      (throw (ex-info "Session has been disconnected" {:session-id session-id})))
    (let [out-ch (chan 1024)
          event-ch (chan 1024)
          {:keys [event-mult send-lock]} (session-io client session-id)
          released? (atom false)
          release-lock! (fn []
                          (when (compare-and-set! released? false true)
                            (put! send-lock :token)))
          deadline-ch (when timeout-ms (async/timeout timeout-ms))
          timeout-event {:type :copilot/session.error
                         :data {:message (str "Timeout after " timeout-ms "ms waiting for session.idle")
                                :timeout-ms timeout-ms}}
          emit! (fn [event]
                  (when-not (async/offer! out-ch event)
                    (log/debug "Dropping event for session " session-id " due to full async buffer")))
          cleanup! (fn []
                     (untap event-mult event-ch)
                     (close! event-ch)
                     (close! out-ch)
                     (release-lock!))]
      (go
        (if-not (<! send-lock) ;; park for lock (nil = channel closed)
          (do (close! event-ch) (close! out-ch))
          (try
            (tap event-mult event-ch)
            ;; Send message via channel-based RPC (no blocking)
            (let [conn (connection-io client)
                  wire-attachments (when (:attachments opts)
                                     (util/attachments->wire (:attachments opts)))
                  trace-ctx (when-let [provider (:on-get-trace-context client)]
                              (try (let [ctx (provider)]
                                     (when (map? ctx)
                                       (select-keys ctx [:traceparent :tracestate])))
                                   (catch Throwable _ nil)))
                  params (cond-> {:session-id session-id
                                  :prompt (:prompt opts)}
                           trace-ctx (merge trace-ctx)
                           wire-attachments (assoc :attachments wire-attachments)
                           (:mode opts) (assoc :mode (name (:mode opts))))
                  response-ch (proto/send-request conn "session.send" params)
                  [result port] (if deadline-ch
                                  (async/alts! [response-ch deadline-ch])
                                  [(<! response-ch) response-ch])]
              (cond
                ;; Timeout during send
                (and deadline-ch (= port deadline-ch))
                (do (emit! timeout-event) (cleanup!))

                ;; RPC error or channel closed
                (or (nil? result) (:error result))
                (do
                  (when (:error result)
                    (log/error "Async send RPC error: " (get-in result [:error :message])))
                  (cleanup!))

                ;; Success — process events
                :else
                (loop []
                  (let [[event ch] (if deadline-ch
                                     (async/alts! [event-ch deadline-ch])
                                     [(<! event-ch) event-ch])]
                    (cond
                      (and deadline-ch (= ch deadline-ch))
                      (do (emit! timeout-event) (cleanup!))

                      (nil? event)
                      (do (untap event-mult event-ch) (close! out-ch) (release-lock!))

                      (#{:copilot/session.idle :copilot/session.error} (:type event))
                      (do (emit! event) (cleanup!))

                      :else
                      (do (emit! event) (recur)))))))
            (catch Exception e
              (log/error "<send-async* error for session " session-id ": " (ex-message e))
              (cleanup!)))))
      out-ch)))

(defn send-async
  "Send a message and return a channel that receives events until session.idle.
   The channel closes after session.idle or session.error.
   Serialized per session to avoid mixing concurrent sends.
   Safe for use inside go blocks — no blocking operations.
   
   Options:
   - :timeout-ms   - Timeout in milliseconds (default: 300000, set to nil to disable)"
  [session opts]
  (let [timeout-ms (if (contains? opts :timeout-ms) (:timeout-ms opts) 300000)
        opts (dissoc opts :timeout-ms)]
    (<send-async* session opts timeout-ms)))

(defn <send!
  "Send a message and return a channel that delivers the final content string.
   This is the async equivalent of send-and-wait! - use inside go blocks.
   
   Options:
   - :timeout-ms   - Timeout in milliseconds (default: 300000, set to nil to disable)
   
   The returned channel delivers a single value (the response content) then closes."
  [session opts]
  (let [timeout-ms (if (contains? opts :timeout-ms) (:timeout-ms opts) 300000)
        events-ch (send-async session (assoc opts :timeout-ms timeout-ms))
        out-ch (chan (async/sliding-buffer 1))]
    (go
      (loop [last-content nil]
        (when-let [event (<! events-ch)]
          (cond
            (= :copilot/assistant.message (:type event))
            (recur (get-in event [:data :content]))

            (#{:copilot/session.idle :copilot/session.error} (:type event))
            (when last-content
              (async/offer! out-ch last-content))

            :else
            (recur last-content))))
      (close! out-ch))
    out-ch))

(defn send-async-with-id
  "Send a message and return {:message-id :events-ch}."
  [session opts]
  (let [timeout-ms (if (contains? opts :timeout-ms) (:timeout-ms opts) 300000)
        opts (dissoc opts :timeout-ms)]
    (send-async* session opts timeout-ms)))

(defn abort!
  "Abort the currently processing message in this session."
  [session]
  (let [{:keys [session-id client]} session]
    (when (:destroyed? (session-state client session-id))
      (throw (ex-info "Session has been disconnected" {:session-id session-id})))
    (let [conn (connection-io client)]
      (proto/send-request! conn "session.abort" {:session-id session-id})
      nil)))

(defn get-messages
  "Get all events/messages from this session's history."
  [session]
  (let [{:keys [session-id client]} session]
    (when (:destroyed? (session-state client session-id))
      (throw (ex-info "Session has been disconnected" {:session-id session-id})))
    (let [conn (connection-io client)
          result (proto/send-request! conn "session.getMessages" {:session-id session-id})]
      (mapv #(update % :type util/event-type->keyword) (:events result)))))

(defn disconnect!
  "Disconnects the session and releases in-memory resources (event handlers,
   tool handlers, permission handler). Session data on disk (conversation
   history, planning state, artifacts) is preserved for later resumption
   via `resume-session`. To permanently remove all session data, use
   `delete-session!` instead.
   Can be called with either a CopilotSession handle or (client, session-id)."
  ([session]
   (disconnect! (:client session) (:session-id session)))
  ([client session-id]
   (log/debug "Disconnecting session: " session-id)
   (when-not (:destroyed? (session-state client session-id))
     (let [conn (connection-io client)]
       ;; Try to notify server, but don't block forever if connection is broken
       (try
         (proto/send-request! conn "session.destroy" {:session-id session-id} 5000)
         (catch Exception _
           ;; Ignore errors - we're cleaning up anyway
           nil))
       ;; Atomically update state — clear handlers and closures to aid GC
       (update-session! client session-id assoc
                        :destroyed? true
                        :tool-handlers {}
                        :permission-handler nil
                        :user-input-handler nil
                        :hooks {}
                        :config nil)
       ;; Close the event source channel - this propagates to all tapped channels
       (when-let [{:keys [event-chan]} (session-io client session-id)]
         (close! event-chan))
       (log/debug "Session disconnected: " session-id)
       nil))))

(defn destroy!
  "Deprecated: Use disconnect! instead. This function will be removed in a future release.
   Disconnects the session and releases in-memory resources.
   Session data on disk is preserved for later resumption."
  ([session]
   (disconnect! session))
  ([client session-id]
   (disconnect! client session-id)))

(defn events
  "Get the event mult for this session. Use tap to subscribe:
   
   (let [ch (chan 100)]
     (tap (events session) ch)
     (go-loop []
       (when-let [event (<! ch)]
         (println event)
         (recur))))
   
   Remember to untap and close your channel when done."
  [session]
  (let [{:keys [session-id client]} session]
    (:event-mult (session-io client session-id))))

(defn subscribe-events
  "Subscribe to session events. Returns a channel that receives events.
   
   The channel will receive nil (close) when the session is disconnected.
   For explicit cleanup before session disconnection, call unsubscribe-events.
   
   Drop behavior: If this subscriber's channel buffer is full when mult tries
   to deliver an event, that specific event is silently dropped for this
   subscriber only. Other subscribers with available buffer space still receive
   the event. The returned channel has a buffer of 1024 events which should be
   sufficient for most use cases.
   
   This is a convenience wrapper around (tap (events session) ch)."
  [session]
  (let [ch (chan 1024)
        {:keys [session-id client]} session
        {:keys [event-mult]} (session-io client session-id)]
    (tap event-mult ch)
    ch))

(defn events->chan
  "Subscribe to session events with options.

   Options:
   - :buffer - Channel buffer size (default 1024)
   - :xf     - Transducer applied to events

   Drop behavior: If this subscriber's channel buffer is full when mult tries
   to deliver an event, that specific event is silently dropped for this
   subscriber only. Other subscribers with available buffer space still receive
   the event."
  ([session]
   (events->chan session {}))
  ([session {:keys [buffer xf] :or {buffer 1024}}]
   (let [{:keys [session-id client]} session
         {:keys [event-mult]} (session-io client session-id)
         ch (if xf (chan buffer xf) (chan buffer))]
     (tap event-mult ch)
     ch)))

(defn unsubscribe-events
  "Unsubscribe a channel from session events."
  [session ch]
  (let [{:keys [session-id client]} session
        {:keys [event-mult]} (session-io client session-id)]
    (untap event-mult ch)
    (close! ch)))

(defn session-id
  "Get the session ID."
  [session]
  (:session-id session))

(defn workspace-path
  "Get the session workspace path when provided by the CLI."
  [session]
  (let [{:keys [session-id client]} session]
    (:workspace-path (session-state client session-id))))

(defn get-current-model
  "Get the current model for this session.
   Returns the model ID string, or nil if none set."
  [session]
  (let [{:keys [session-id client]} session
        conn (connection-io client)
        result (proto/send-request! conn "session.model.getCurrent"
                                    {:sessionId session-id})]
    (:model-id result)))

(defn switch-model!
  "Switch the model for this session.
   The new model takes effect for the next message. Conversation history is preserved.

   Optional opts map:
   - :reasoning-effort - Reasoning effort level for the new model (\"low\", \"medium\", \"high\", \"xhigh\")

   Returns the new model ID string, or nil."
  ([session model-id] (switch-model! session model-id nil))
  ([session model-id opts]
   (let [{:keys [session-id client]} session
         conn (connection-io client)
         params (cond-> {:sessionId session-id
                         :modelId model-id}
                  (:reasoning-effort opts) (assoc :reasoningEffort (:reasoning-effort opts)))
         result (proto/send-request! conn "session.model.switchTo" params)]
     (:model-id result))))

(defn set-model!
  "Alias for switch-model!. Matches the upstream SDK's setModel() API.
   See switch-model! for details."
  ([session model-id] (switch-model! session model-id nil))
  ([session model-id opts] (switch-model! session model-id opts)))

(defn log!
  "Log a message to the session timeline.
   Options (optional map):
   - :level      - \"info\", \"warning\", or \"error\" (default: \"info\")
   - :ephemeral? - when true, message is not persisted to disk (default: false)
   Returns the event ID string."
  ([session message] (log! session message nil))
  ([session message opts]
   (let [{:keys [session-id client]} session
         conn (connection-io client)
         params (cond-> {:sessionId session-id :message message}
                  (:level opts) (assoc :level (:level opts))
                  (:ephemeral? opts) (assoc :ephemeral (:ephemeral? opts)))
         result (proto/send-request! conn "session.log" params)]
     (:event-id result))))

;; =============================================================================
;; Low-level RPC methods (session.rpc.*)
;;
;; These are thin wrappers around the CLI's JSON-RPC methods. They are emerging
;; APIs that don't yet have friendly high-level wrappers in the upstream SDK.
;; Some are marked experimental and may change.
;; =============================================================================

;; -- Skills ------------------------------------------------------------------

(defn ^:experimental skills-list
  "List all skills available to the session.
   Returns a map with :skills (vector of skill info maps)."
  [session]
  (let [{:keys [session-id client]} session
        conn (connection-io client)]
    (util/wire->clj
     (proto/send-request! conn "session.skills.list" {:sessionId session-id}))))

(defn ^:experimental skills-enable!
  "Enable a skill by name."
  [session skill-name]
  (let [{:keys [session-id client]} session
        conn (connection-io client)]
    (proto/send-request! conn "session.skills.enable"
                         {:sessionId session-id :name skill-name})))

(defn ^:experimental skills-disable!
  "Disable a skill by name."
  [session skill-name]
  (let [{:keys [session-id client]} session
        conn (connection-io client)]
    (proto/send-request! conn "session.skills.disable"
                         {:sessionId session-id :name skill-name})))

(defn ^:experimental skills-reload!
  "Reload all skills."
  [session]
  (let [{:keys [session-id client]} session
        conn (connection-io client)]
    (proto/send-request! conn "session.skills.reload" {:sessionId session-id})))

;; -- MCP Servers -------------------------------------------------------------

(defn ^:experimental mcp-list
  "List all MCP servers configured for the session.
   Returns a map with :servers (vector of server info maps)."
  [session]
  (let [{:keys [session-id client]} session
        conn (connection-io client)]
    (util/wire->clj
     (proto/send-request! conn "session.mcp.list" {:sessionId session-id}))))

(defn ^:experimental mcp-enable!
  "Enable an MCP server by name."
  [session server-name]
  (let [{:keys [session-id client]} session
        conn (connection-io client)]
    (proto/send-request! conn "session.mcp.enable"
                         {:sessionId session-id :serverName server-name})))

(defn ^:experimental mcp-disable!
  "Disable an MCP server by name."
  [session server-name]
  (let [{:keys [session-id client]} session
        conn (connection-io client)]
    (proto/send-request! conn "session.mcp.disable"
                         {:sessionId session-id :serverName server-name})))

(defn ^:experimental mcp-reload!
  "Reload all MCP servers."
  [session]
  (let [{:keys [session-id client]} session
        conn (connection-io client)]
    (proto/send-request! conn "session.mcp.reload" {:sessionId session-id})))

;; -- Extensions --------------------------------------------------------------

(defn ^:experimental extensions-list
  "List all extensions for the session.
   Returns a map with :extensions (vector of extension info maps)."
  [session]
  (let [{:keys [session-id client]} session
        conn (connection-io client)]
    (util/wire->clj
     (proto/send-request! conn "session.extensions.list" {:sessionId session-id}))))

(defn ^:experimental extensions-enable!
  "Enable an extension by its source-qualified ID."
  [session extension-id]
  (let [{:keys [session-id client]} session
        conn (connection-io client)]
    (proto/send-request! conn "session.extensions.enable"
                         {:sessionId session-id :id extension-id})))

(defn ^:experimental extensions-disable!
  "Disable an extension by its source-qualified ID."
  [session extension-id]
  (let [{:keys [session-id client]} session
        conn (connection-io client)]
    (proto/send-request! conn "session.extensions.disable"
                         {:sessionId session-id :id extension-id})))

(defn ^:experimental extensions-reload!
  "Reload all extensions."
  [session]
  (let [{:keys [session-id client]} session
        conn (connection-io client)]
    (proto/send-request! conn "session.extensions.reload" {:sessionId session-id})))

;; -- Plugins -----------------------------------------------------------------

(defn ^:experimental plugins-list
  "List all plugins for the session.
   Returns a map with :plugins (vector of plugin info maps)."
  [session]
  (let [{:keys [session-id client]} session
        conn (connection-io client)]
    (util/wire->clj
     (proto/send-request! conn "session.plugins.list" {:sessionId session-id}))))

;; -- Compaction --------------------------------------------------------------

(defn ^:experimental compaction-compact!
  "Trigger manual compaction of the session context."
  [session]
  (let [{:keys [session-id client]} session
        conn (connection-io client)]
    (util/wire->clj
     (proto/send-request! conn "session.compaction.compact" {:sessionId session-id}))))

;; -- Shell -------------------------------------------------------------------

(defn ^:experimental shell-exec!
  "Execute a shell command in the session.
   Returns the execution result."
  [session command]
  (let [{:keys [session-id client]} session
        conn (connection-io client)]
    (util/wire->clj
     (proto/send-request! conn "session.shell.exec"
                          {:sessionId session-id :command command}))))

(defn ^:experimental shell-kill!
  "Kill a running shell process by process ID."
  [session process-id]
  (let [{:keys [session-id client]} session
        conn (connection-io client)]
    (proto/send-request! conn "session.shell.kill"
                         {:sessionId session-id :processId process-id})))

;; -- UI Elicitation ----------------------------------------------------------

(defn capabilities
  "Return the host capabilities reported when the session was created or resumed.
   Returns a map with optional :ui key. Check :ui/:elicitation before calling UI methods.

   Example:
     (when (get-in (session/capabilities session) [:ui :elicitation])
       (session/ui-confirm! session \"Deploy to production?\"))"
  [session]
  (let [{:keys [session-id client]} session]
    (:capabilities (session-state client session-id))))

(defn ^:experimental ui-elicitation!
  "Request structured user input via an elicitation prompt.
   params is a map with elicitation configuration (schema, message, etc.).
   Throws if the host does not support elicitation (check capabilities first)."
  [session params]
  (let [{:keys [session-id client]} session
        conn (connection-io client)]
    (util/wire->clj
     (proto/send-request! conn "session.ui.elicitation"
                          (assoc (util/clj->wire params) :sessionId session-id)))))

(defn ^:experimental ui-confirm!
  "Show a confirmation dialog and return true if the user confirmed, false otherwise.
   Returns false if the user declines or cancels.
   Throws if the host does not support elicitation (check (capabilities session) first)."
  [session message]
  (let [result (ui-elicitation! session
                                {:message message
                                 :requested-schema
                                 {:type "object"
                                  :properties {:confirmed {:type "boolean" :default true}}
                                  :required ["confirmed"]}})]
    (and (= "accept" (:action result))
         (true? (get-in result [:content :confirmed])))))

(defn ^:experimental ui-select!
  "Show a selection dialog with the given string options.
   Returns the selected value, or nil if the user declines or cancels.
   Throws if the host does not support elicitation (check (capabilities session) first)."
  [session message options]
  (let [result (ui-elicitation! session
                                {:message message
                                 :requested-schema
                                 {:type "object"
                                  :properties {:selection {:type "string" :enum options}}
                                  :required ["selection"]}})]
    (when (= "accept" (:action result))
      (get-in result [:content :selection]))))

(defn ^:experimental ui-input!
  "Show a text input dialog.
   Returns the entered text, or nil if the user declines or cancels.
   opts (optional map):
   - :title       - Label for the input field
   - :description - Descriptive text shown below the field
   - :min-length  - Minimum character length
   - :max-length  - Maximum character length
   - :format      - Semantic format hint: \"email\", \"uri\", \"date\", \"date-time\"
   - :default     - Default value pre-populated in the field
   Throws if the host does not support elicitation (check (capabilities session) first)."
  ([session message] (ui-input! session message nil))
  ([session message opts]
   (let [field (cond-> {:type "string"}
                 (:title opts) (assoc :title (:title opts))
                 (:description opts) (assoc :description (:description opts))
                 (:min-length opts) (assoc :minLength (:min-length opts))
                 (:max-length opts) (assoc :maxLength (:max-length opts))
                 (:format opts) (assoc :format (:format opts))
                 (contains? opts :default) (assoc :default (:default opts)))
         result (ui-elicitation! session
                                 {:message message
                                  :requested-schema
                                  {:type "object"
                                   :properties {:value field}
                                   :required ["value"]}})]
     (when (= "accept" (:action result))
       (get-in result [:content :value])))))