im.ttalk.agent.provider.http.stream-client — im.ttalk/clj-agent-provider 0.5.447

provider 的真流式 —— 传输是 im.ttalk.http/stream-lines（clj-http，designs/http-module.md； 最初为替代 http-kit 的伪流式而写，修 BUG2）。本 ns 只留 provider 语义：parse-fn / process-fn、canonical error （超时 / 非 2xx / IO）、模型名核对、限流头、重试与取消登记。

与 http-kit 客户端的根本区别：http-kit 的 :as :stream 等整段响应收完才兑现 promise（伪流式）；stream-lines 用 BodyHandlers/fromLineSubscriber（响应式 Flow.Subscriber）， 每行 SSE 到达即回调——真增量、非阻塞、可取消、原生背压。

与现有 stream 处理器无缝衔接：parse-fn / process-fn 复用 stream.anthropic / stream.openai 的纯函数（parse-sse-line / process-event），传输层只负责把行喂进来。

回调契约（与 http-kit 版 post-stream-async 一致）：

• :on-token (fn [token-data] ...) 每个 token/事件产出
• :on-complete (fn [final-state] ...) 流正常结束
• :on-error (fn [{:keys [error cause status]}] ...) 连接/解析/非 2xx 失败

返回 {:future CompletableFuture :cancel (fn [])}：

• future 在流结束（或失败）时完成；
• cancel 取消上游 HTTP（客户端断连/超时/用户停止时调用，不再消耗 token）。

default-client clj

默认共享 HttpClient——与非流式 http.client 同一实例（同一连接池 / 虚拟线程 executor）， 避免两套独立连接池。

post-stream-async clj

(post-stream-async url
                   {:keys [headers body timeout parse-fn process-fn
                           initial-state on-token on-complete on-error
                           on-headers provider client first-byte-ms idle-ms
                           check-model]
                    :or {initial-state {}}})

真流式 POST。opts:

• :headers :body :timeout
• :parse-fn :process-fn :initial-state （复用 stream 处理器）
• :on-token :on-complete :on-error （终止回调只触发一次：watchdog / 模型核对 / 取消竞争时先到者为准）
• :provider （错误归一化用，可选）
• :client （自定义 HttpClient，可选） designs/provider-reliability.md §4 / §6 新增（均可选，不给即旧行为）：
• :first-byte-ms 从发出请求到第一行 SSE 到达的上限（缺省 = :idle-ms）
• :idle-ms 相邻两行之间的上限（SSE 注释心跳行也算活动） 超时 → on-error 收 {:type :timeout-error :retryable? true :code :first-byte-timeout | :idle-timeout}，并取消上游
• :check-model {:requested "m" :model-fn (fn [state] model)}：每行处理后若尚未核对且 model-fn 给出非空模型名 → 比较；不一致 → on-error 收 :model-mismatch，该行 token 不投递
• :on-headers (fn [status headers]) 响应头到达时调用（限流快照 / 错误带头）

非 2xx：错误体 + 响应头经 http.client/response->error 归一（业务码表、Retry-After）。

返回 {:future :cancel}。

post-stream-deferred clj

(post-stream-deferred
  url
  {:keys [make-initial-state build-response provider on-token retry] :as opts})

post-stream-async 的异步包装：不阻塞，返回 deferred<最终响应>。

opts 与语义（含 :retry 的「未出 token 才重试」、取消返回空响应、 :provider-error 兜底）与 post-stream-sync 逐字相同——两者共用同一批 判据，差别只有两处：等流结束用 flt/fbind 而不是 @future；重试退避用 async/delayed 而不是 Thread/sleep（异步路径上 sleep 会睡掉传输层 executor 的线程）。

取消登记的线程：streaming/*register-cancel* 是动态 var，只在调用线程 可见。故本函数在进入时就把它取出来闭包住，重试 attempt（跑在完成线程上） 照样登记得上——同步版靠「整个 loop 都在调用线程」天然成立，异步版必须显式 把它带过去。见 docs/async-chat-model-design.md §6.1。

post-stream-sync clj

(post-stream-sync
  url
  {:keys [make-initial-state build-response provider on-token retry] :as opts})

post-stream-async 的同步包装：阻塞当前线程直至流结束，返回最终响应（保持同步签名）。

之前 openai_compat / anthropic / dashscope 三处各自手写同一套 「promise 对 + 包装 cancel + cond 分派」编排，现统一收敛于此。

在 post-stream-async 的 opts 之外新增/替代：

• :make-initial-state (fn [] state) 初始状态构造器（取消时再次调用以构建空响应）
• :build-response (fn [final-state]) 流结束时构建最终响应
• （post-stream-async 的 :first-byte-ms / :idle-ms / :check-model / :on-headers 原样透传； 成功响应按响应头挂 :rate-limit，designs/provider-reliability.md §4 / §6）
• :retry 建链阶段重试（opt-in，同 http.retry/maybe-with-retry 约定： true → 默认配置；map → 合并 default-retry-opts）。 仅当失败可重试（canonical error :retryable?）且尚未向 on-token 流出任何 token 时退避重试——token 已出说明调用方 已观察到部分输出，重试会重复内容，此时按原样抛错。 流式错误体不带响应头，故不支持 Retry-After，只走指数退避。 （:initial-state / :on-complete / :on-error 由本函数接管，调用方不再传。）

取消语义：包装的 cancel 登记到 im.ttalk.agent.streaming 的在途注册表（每次 attempt 重新登记，替换为最新 cancel）；被调用时先标记本地 cancelled? 再取消上游。 取消会让 java.net.http 触发 onError（连接中止）或让 future 抛 CancellationException ——故 @future 宽 catch，且 cond 里 cancelled? 优先于 err。取消返回空响应，不抛错、 不重试。

失败：err 已兑现 → 按 D5 canonical error 抛出；流结束但无结果 → :provider-error。