Phase 6: OpenAI-compatible guardrail proxy
ober
a31a13a4fb2952aa1e75612692e3bb4ea67663ef
--- a/build-binary.ss +++ b/build-binary.ss @@ -145,6 +145,10 @@ "lib/jcode/core/workflow-runner" "lib/jcode/provider/sampling" "lib/jcode/provider/provider" + "lib/jcode/proxy/convert" + "lib/jcode/proxy/handler" + "lib/jcode/core/slot-worker" + "lib/jcode/proxy/server" "lib/jcode/tool/registry" "lib/jcode/tool/file" "lib/jcode/tool/apply-patch" new file mode 100644 --- /dev/null +++ b/src/jcode/core/slot-worker.ss @@ -0,0 +1,133 @@ +;;; jcode slot worker — serialized workflow execution with priority + preempt +;;; +;;; Faithful port of forge's core/slot_worker.py SlotWorker, re-expressed in +;;; jcode's green threads (forge uses asyncio). It serializes run-workflow on a +;;; single inference slot (single-GPU constraint): each submission waits its +;;; turn, runs to completion, and returns the terminal value. +;;; +;;; * priority — an int; LOWER runs first. FIFO within equal priority (a +;;; monotonic counter breaks ties). Default 0 (pure FIFO). +;;; * preemption — if a submission has strictly higher priority (lower int) +;;; than the running task, the running task's cancel flag is +;;; set, so run-workflow raises &workflow-cancelled and the +;;; higher-priority task takes the slot. +;;; +;;; The worker holds the responder (forge's SlotWorker wraps a WorkflowRunner +;;; that holds the client); each submit drives run-workflow with that responder +;;; and a cancel? thunk reading the task's cancel flag. submit blocks the +;;; calling green thread on a result cell until the worker fills it. + +(export make-slot-worker slot-worker? + slot-worker-start! slot-worker-stop! + slot-submit! slot-cancel-current! + slot-worker-running-priority slot-worker-pending + slot-priority<? slot-should-preempt?) + +(import :std/misc/thread + (rename (only (chezscheme) make-mutex) (make-mutex chez-make-mutex)) + :jcode/core/workflow-runner) + +(def (assoc-ref alist key) (let ((p (assoc key alist))) (and p (cdr p)))) + +;; A queued task: #(priority seq workflow user-message opt result-cell). +;; result-cell is #(status value): status 'pending | 'ok | 'error. +(defstruct sworker + (responder mutex queue counter current-priority cancel-cell worker running)) + +(def (make-slot-worker responder) + "Construct a slot worker driving RESPONDER (the injected inference seam)." + (make-sworker responder (chez-make-mutex) '() 0 #f #f #f #f)) + +(def (slot-worker? x) (sworker? x)) +(def (slot-worker-running-priority w) (sworker-current-priority w)) +(def (slot-worker-pending w) + (with-mutex (sworker-mutex w) (length (sworker-queue w)))) + +;; ── Scheduling policy (pure — exposed for testing) ─────────────────── +(def (slot-priority<? p1 s1 p2 s2) + "True if task (p1,s1) should run before (p2,s2): lower priority first, + then lower sequence (FIFO) within equal priority." + (or (< p1 p2) (and (= p1 p2) (< s1 s2)))) + +(def (slot-should-preempt? current-priority new-priority) + "True if a new task at NEW-PRIORITY should preempt a running task at + CURRENT-PRIORITY (strictly higher priority = strictly lower int)." + (and current-priority (< new-priority current-priority) #t)) + +(def (queue-insert q item) + (let ((pri (vector-ref item 0)) (seq (vector-ref item 1))) + (let loop ((xs q) (acc '())) + (cond + ((null? xs) (reverse (cons item acc))) + ((slot-priority<? pri seq (vector-ref (car xs) 0) (vector-ref (car xs) 1)) + (append (reverse acc) (cons item xs))) + (else (loop (cdr xs) (cons (car xs) acc))))))) + +;; ── Worker loop + submission ───────────────────────────────────────── + +(def (slot-worker-loop w) + (let loop () + (when (sworker-running w) + (let ((item #f)) + (with-mutex (sworker-mutex w) + (unless (null? (sworker-queue w)) + (set! item (car (sworker-queue w))) + (sworker-queue-set! w (cdr (sworker-queue w))) + (sworker-current-priority-set! w (vector-ref item 0)) + (sworker-cancel-cell-set! w (vector #f)))) + (if item + (let ((cell (sworker-cancel-cell w)) + (rcell (vector-ref item 5)) + (wf (vector-ref item 2)) + (um (vector-ref item 3)) + (o (vector-ref item 4))) + (guard (e [#t (vector-set! rcell 1 e) (vector-set! rcell 0 'error)]) + (let ((res (run-workflow wf um (sworker-responder w) + (cons (cons 'cancel? (lambda () (vector-ref cell 0))) o)))) + (vector-set! rcell 1 res) + (vector-set! rcell 0 'ok))) + (with-mutex (sworker-mutex w) + (sworker-current-priority-set! w #f) + (sworker-cancel-cell-set! w #f)) + (loop)) + (begin (thread-sleep! 0.005) (loop))))))) + +(def (slot-worker-start! w) + "Start the worker green thread (idempotent)." + (unless (sworker-worker w) + (sworker-running-set! w #t) + (sworker-worker-set! w (spawn (lambda () (slot-worker-loop w)))))) + +(def (slot-worker-stop! w) + "Signal the worker loop to exit after the current task." + (sworker-running-set! w #f)) + +(def (slot-cancel-current! w) + "Cancel the currently running task, if any." + (with-mutex (sworker-mutex w) + (when (sworker-cancel-cell w) + (vector-set! (sworker-cancel-cell w) 0 #t)))) + +(def (slot-submit! w workflow user-message . opt) + "Submit WORKFLOW + USER-MESSAGE and block until it completes, returning the + terminal value (or re-raising whatever run-workflow raised). OPT is an + options assoc; 'priority (default 0) sets scheduling priority and the rest + (e.g. 'prompt-vars, 'max-iterations) pass through to run-workflow." + (let* ((o (if (pair? opt) (car opt) '())) + (priority (or (assoc-ref o 'priority) 0)) + (rcell (vector 'pending #f))) + (with-mutex (sworker-mutex w) + (let ((seq (sworker-counter w))) + (sworker-counter-set! w (+ seq 1)) + (sworker-queue-set! w + (queue-insert (sworker-queue w) + (vector priority seq workflow user-message o rcell))) + (when (and (slot-should-preempt? (sworker-current-priority w) priority) + (sworker-cancel-cell w)) + (vector-set! (sworker-cancel-cell w) 0 #t)))) + (let wait () + (let ((status (vector-ref rcell 0))) + (cond + ((eq? status 'ok) (vector-ref rcell 1)) + ((eq? status 'error) (raise (vector-ref rcell 1))) + (else (thread-sleep! 0.005) (wait))))))) new file mode 100644 --- /dev/null +++ b/src/jcode/proxy/convert.ss @@ -0,0 +1,219 @@ +;;; jcode proxy — OpenAI ↔ jcode conversion +;;; +;;; Faithful port of forge's proxy/convert.py. Converts between the OpenAI +;;; chat-completions wire format and jcode's internal types: +;;; * inbound — a parsed OpenAI messages array (a list of JSON-object +;;; hashes) → a list of jcode `message` structs. +;;; * outbound — wtool-call list / text → an OpenAI chat.completion object, +;;; or a list of chat.completion.chunk objects for SSE. +;;; +;;; JSON objects are jcode prelude hashes (make-hash-table / hash-put! / +;;; hash-ref), JSON arrays are lists, and JSON null is (void) — the exact +;;; representation :std/text/json produces and message.ss already speaks. A +;;; tool call's "arguments" is always rendered as a JSON STRING on the wire. + +(export openai->messages + tool-calls->openai + text-response->openai + tool-calls->sse-events + text->sse-events) + +(import :std/text/json + :std/misc/uuid + :std/misc/string + :jcode/core/message + :jcode/core/workflow) + +;; ── ID helpers (forge uses uuid4().hex[:N]) ────────────────────────── +(def (hex-id n) + (let ((s (string-filter-hex (uuid-string)))) + (substring s 0 (min n (string-length s))))) + +(def (string-filter-hex s) + ;; uuid-string carries dashes; strip them to mimic uuid4().hex. + (list->string + (filter (lambda (c) (not (char=? c #\-))) (string->list s)))) + +(def (chatcmpl-id) (string-append "chatcmpl-" (hex-id 12))) +(def (call-id) (string-append "call_" (hex-id 8))) + +(def (assoc->json-string args) + ;; wtool-call args are an assoc ((name . value) ...); the wire wants a JSON + ;; object string, matching forge's json.dumps(tc.args). + (let ((ht (make-hash-table))) + (for-each (lambda (kv) (hash-put! ht (car kv) (cdr kv))) args) + (json-object->string ht))) + +;; ── Inbound: OpenAI request → jcode messages ───────────────────────── + +(def (normalize-content content) + ;; OpenAI allows content as a list of blocks ([{"type":"text","text":..}]). + ;; Flatten to a plain string; #f/(void) → "". + (cond + ((or (not content) (eq? content (void))) "") + ((string? content) content) + ((list? content) + (string-join + (map (lambda (block) + (cond + ((string? block) block) + ((and (hash-table? block) (equal? (hash-ref block "type" #f) "text")) + (hash-ref block "text" "")) + (else ""))) + content) + "\n")) + (else ""))) + +(def (openai-tc->tool-call tc) + ;; tc is a JSON-object hash {"id","function":{"name","arguments"}}. + (let* ((func (hash-ref tc "function" #f)) + (name (if func (hash-ref func "name" "") "")) + (args (if func (hash-ref func "arguments" "{}") "{}")) + ;; forge json.loads-es string args; jcode stores arguments as the + ;; JSON string on the message (message.ss tool-call-arguments), so a + ;; string passes straight through; a hash is re-serialized. + (args-str (if (string? args) args (json-object->string args))) + (id (let ((i (hash-ref tc "id" #f))) (if i i (call-id))))) + (restore-tool-call id name args-str))) + +(def (openai->messages openai-messages) + "Convert a list of OpenAI message hashes to jcode `message` structs. + system/user/assistant(+tool_calls)/tool are recognized; unknown → user." + (map + (lambda (msg) + (let ((role (hash-ref msg "role" "user")) + (content (normalize-content (hash-ref msg "content" "")))) + (cond + ((equal? role "system") (make-system-message content)) + ((equal? role "assistant") + (let ((tcs (hash-ref msg "tool_calls" #f))) + (if (and tcs (pair? tcs)) + (make-assistant-message content (map openai-tc->tool-call tcs)) + (make-assistant-message content)))) + ((equal? role "tool") + (make-tool-result (hash-ref msg "tool_call_id" "") content)) + (else (make-user-message content))))) + openai-messages)) + +;; ── Outbound: jcode response → OpenAI format ───────────────────────── + +(def (tc->openai-entry tc) + (let ((entry (make-hash-table)) + (fn (make-hash-table))) + (hash-put! entry "id" (call-id)) + (hash-put! entry "type" "function") + (hash-put! fn "name" (wtool-call-tool tc)) + (hash-put! fn "arguments" (assoc->json-string (wtool-call-args tc))) + (hash-put! entry "function" fn) + entry)) + +(def (make-choice index message finish-reason) + (let ((ch (make-hash-table))) + (hash-put! ch "index" index) + (hash-put! ch "message" message) + (hash-put! ch "finish_reason" finish-reason) + ch)) + +(def (make-usage) + (let ((u (make-hash-table))) + (hash-put! u "prompt_tokens" 0) + (hash-put! u "completion_tokens" 0) + (hash-put! u "total_tokens" 0) + u)) + +(def (make-completion model choices) + (let ((c (make-hash-table))) + (hash-put! c "id" (chatcmpl-id)) + (hash-put! c "object" "chat.completion") + (hash-put! c "model" model) + (hash-put! c "choices" choices) + (hash-put! c "usage" (make-usage)) + c)) + +(def (tool-calls->openai tool-calls model) + "wtool-call list → an OpenAI chat.completion object (finish_reason + \"tool_calls\"). The first call's reasoning becomes message.content." + (let ((msg (make-hash-table))) + (hash-put! msg "role" "assistant") + (hash-put! msg "content" + (or (wtool-call-reasoning (car tool-calls)) (void))) + (hash-put! msg "tool_calls" (map tc->openai-entry tool-calls)) + (make-completion model (list (make-choice 0 msg "tool_calls"))))) + +(def (text-response->openai text model) + "A text answer → an OpenAI chat.completion object (finish_reason \"stop\")." + (let ((msg (make-hash-table))) + (hash-put! msg "role" "assistant") + (hash-put! msg "content" text) + (make-completion model (list (make-choice 0 msg "stop"))))) + +;; ── SSE streaming chunks (chat.completion.chunk objects) ───────────── + +(def (make-chunk id model delta finish-reason) + (let ((c (make-hash-table)) + (ch (make-hash-table))) + (hash-put! c "id" id) + (hash-put! c "object" "chat.completion.chunk") + (hash-put! c "model" model) + (hash-put! ch "index" 0) + (hash-put! ch "delta" delta) + (hash-put! ch "finish_reason" finish-reason) + (hash-put! c "choices" (list ch)) + c)) + +(def (delta . kvs) + (let ((d (make-hash-table))) + (let loop ((p kvs)) + (unless (null? p) (hash-put! d (car p) (cadr p)) (loop (cddr p)))) + d)) + +(def (tool-calls->sse-events tool-calls model) + "wtool-call list → a list of SSE chunk objects: optional reasoning delta, + one tool-call delta per call, then a finish_reason=tool_calls terminator." + (let* ((id (chatcmpl-id)) + (reasoning (wtool-call-reasoning (car tool-calls))) + (head (if reasoning + (list (make-chunk id model + (delta "role" "assistant" "content" reasoning) (void))) + '())) + (tc-events + (let loop ((tcs tool-calls) (i 0) (acc '())) + (if (null? tcs) (reverse acc) + (let ((entry (make-hash-table)) + (fn (make-hash-table))) + (hash-put! fn "name" (wtool-call-tool (car tcs))) + (hash-put! fn "arguments" (assoc->json-string (wtool-call-args (car tcs)))) + (hash-put! entry "index" i) + (hash-put! entry "id" (call-id)) + (hash-put! entry "type" "function") + (hash-put! entry "function" fn) + (loop (cdr tcs) (+ i 1) + (cons (make-chunk id model (delta "tool_calls" (list entry)) (void)) acc)))))) + (tail (list (make-chunk id model (delta) "tool_calls")))) + (append head tc-events tail))) + +(def (chunk-text s n) + ;; Split S into N-char chunks (forge text_to_sse chunk_size). + (let loop ((i 0) (acc '())) + (if (>= i (string-length s)) (reverse acc) + (let ((end (min (+ i n) (string-length s)))) + (loop end (cons (substring s i end) acc)))))) + +(def (text->sse-events text model . opt) + "Text → a list of content-delta SSE chunks then a finish_reason=stop + terminator. Optional CHUNK-SIZE (>0) splits TEXT for realistic streaming." + (let* ((id (chatcmpl-id)) + (chunk-size (if (pair? opt) (car opt) 0)) + (chunks (if (and (> chunk-size 0) (> (string-length text) chunk-size)) + (chunk-text text chunk-size) + (list text)))) + (append + (let loop ((cs chunks) (i 0) (acc '())) + (if (null? cs) (reverse acc) + (loop (cdr cs) (+ i 1) + (cons (make-chunk id model + (if (= i 0) (delta "role" "assistant" "content" (car cs)) + (delta "content" (car cs))) + (void)) + acc)))) + (list (make-chunk id model (delta) "stop"))))) new file mode 100644 --- /dev/null +++ b/src/jcode/proxy/handler.ss @@ -0,0 +1,126 @@ +;;; jcode proxy — /v1/chat/completions handler +;;; +;;; Faithful port of forge's proxy/handler.py handle_chat_completions, +;;; decoupled from forge's run_inference / LLMClient / ContextManager via a +;;; single injected BACKEND closure (the same seam Phase 5's runner uses): +;;; +;;; (backend messages tool-specs sampling) -> response +;;; +;;; where RESPONSE is a text-response (the model answered in prose, or no +;;; tools were offered) or a list of wtool-call. The backend may raise +;;; &tool-call-error when retries are exhausted; the handler then returns the +;;; last raw text to the client (forge's "let the client's own loop decide"). +;;; Production wires BACKEND to the jcode provider + validator + error-tracker +;;; retry loop; tests pass a scripted closure. +;;; +;;; The respond tool is injected when the request carries tools and is then +;;; stripped from the outbound response: a pure respond() call comes back as a +;;; normal text answer (finish_reason "stop"), keeping the model in +;;; tool-calling mode where guardrails apply without leaking respond to the +;;; client. Body / response objects are jcode prelude hashes. + +(export handle-chat-completions + extract-sampling extract-tool-specs) + +(import :std/misc/string + :jcode/core/message + :jcode/core/workflow + :jcode/core/errors + :jcode/guardrails/respond + :jcode/proxy/convert) + +;; Top-level OpenAI body fields plumbed through to the backend as sampling +;; overrides (llama-server / Ollama honor these; Anthropic ignores them). +(def *sampling-fields* + '("temperature" "top_p" "top_k" "min_p" "repeat_penalty" + "presence_penalty" "seed" "chat_template_kwargs" "model")) + +(def (extract-sampling body) + "Pull recognized sampling fields out of the inbound body. Returns a hash, or + #f when none are present (the 'use client instance state' path)." + (let ((out (make-hash-table)) (found #f)) + (for-each + (lambda (f) + (let ((v (hash-ref body f 'jcode-absent))) + (unless (eq? v 'jcode-absent) + (set! found #t) + (hash-put! out f v)))) + *sampling-fields*) + (and found out))) + +(def (extract-tool-specs request-tools) + "Extract tool-spec objects from the OpenAI tools array (type=function)." + (if (or (not request-tools) (not (pair? request-tools))) '() + (let loop ((ts request-tools) (acc '())) + (cond + ((null? ts) (reverse acc)) + ((equal? (hash-ref (car ts) "type" #f) "function") + (let* ((func (hash-ref (car ts) "function" #f)) + (name (if func (hash-ref func "name" "") "")) + (desc (if func (hash-ref func "description" "") "")) + (params (if func (hash-ref func "parameters" (make-hash-table)) (make-hash-table)))) + (loop (cdr ts) (cons (tool-spec-from-json-schema name desc params) acc)))) + (else (loop (cdr ts) acc)))))) + +(def (any-spec-named? specs name) + (let loop ((s specs)) + (cond + ((null? s) #f) + ((equal? (tool-spec-name (car s)) name) #t) + (else (loop (cdr s)))))) + +(def (arg-ref args key) (let ((p (assoc key args))) (and p (cdr p)))) + +(def (render text is-stream model) + (if is-stream (text->sse-events text model) (text-response->openai text model))) + +(def (handle-chat-completions body backend) + "Handle a parsed /v1/chat/completions BODY hash. Returns an OpenAI + chat.completion object, or — when stream is true — a list of + chat.completion.chunk objects for SSE." + (let* ((openai-messages (hash-ref body "messages" '())) + (request-tools (hash-ref body "tools" #f)) + (is-stream (eq? (hash-ref body "stream" #f) #t)) + (model-name (let ((m (hash-ref body "model" "forge"))) + (if (string? m) m "forge"))) + (sampling (extract-sampling body)) + (messages (openai->messages openai-messages)) + (tool-specs0 (extract-tool-specs request-tools)) + (tool-specs (if (and (pair? tool-specs0) + (not (any-spec-named? tool-specs0 respond-tool-name))) + (append tool-specs0 + (list (make-tool-spec respond-tool-name + respond-description + (respond-schema)))) + tool-specs0))) + (cond + ;; No tools → plain chat completion, forward to backend, return text. + ((null? tool-specs) + (let* ((resp (backend messages '() sampling)) + (text (if (text-response? resp) (text-response-content resp) ""))) + (render text is-stream model-name))) + (else + (let ((result + (guard (e [(tool-call-error? e) + ;; Retries exhausted — return the last raw text so the + ;; client's own agentic loop can decide what to do. + (cons 'text (or (tool-call-error-raw-response e) ""))]) + (cons 'ok (backend messages tool-specs sampling))))) + (if (eq? (car result) 'text) + (render (cdr result) is-stream model-name) + (let ((resp (cdr result))) + (cond + ((text-response? resp) + (render (text-response-content resp) is-stream model-name)) + (else + ;; resp is a list of wtool-call; strip respond() calls. + (let ((respond-calls (filter (lambda (tc) (equal? (wtool-call-tool tc) respond-tool-name)) resp)) + (other-calls (filter (lambda (tc) (not (equal? (wtool-call-tool tc) respond-tool-name))) resp))) + (cond + ((and (pair? respond-calls) (null? other-calls)) + (render (or (arg-ref (wtool-call-args (car respond-calls)) "message") "") + is-stream model-name)) + ((pair? other-calls) + (if is-stream (tool-calls->sse-events other-calls model-name) + (tool-calls->openai other-calls model-name))) + (else (render "" is-stream model-name))))))))))))) new file mode 100644 --- /dev/null +++ b/src/jcode/proxy/server.ss @@ -0,0 +1,248 @@ +;;; jcode proxy — HTTP server (OpenAI-compatible) +;;; +;;; Faithful port of the serving essence of forge's proxy/server.py + +;;; proxy/proxy.py: an HTTP endpoint that speaks the OpenAI chat-completions +;;; wire format, runs every request through the guardrailed handler, and +;;; returns either a chat.completion JSON object or an SSE stream. forge uses +;;; raw asyncio with a single-GPU inference worker; here the accept loop is +;;; sequential (one request at a time = the same single-slot serialization) +;;; and the request handling reuses jcode's green-threaded TCP ports. +;;; +;;; The routing core `proxy-dispatch` is pure — it takes the request method, +;;; path, and body string plus an injected BACKEND closure and returns a +;;; `presp` (status / content-type / body) with no socket I/O — so the full +;;; convert + handler + dispatch pipeline is unit-testable without a live +;;; provider or a real socket. `proxy-serve` wraps it in a tcp-listen accept +;;; loop; `make-provider-backend` adapts a jcode provider into the BACKEND +;;; seam so `jcode proxy` serves a real (local) model behind the guardrails. + +(export proxy-dispatch + make-presp presp? presp-status presp-content-type presp-body + sse-body proxy-serve make-provider-backend) + +(import :std/text/json + :std/net/tcp + :std/misc/string + :jcode/core/message + :jcode/core/workflow + :jcode/provider/provider + :jcode/proxy/handler) + +;; A serialized HTTP response: status code, Content-Type, body string. +(defstruct presp (status content-type body)) + +;; ── response builders ──────────────────────────────────────────────── + +(def *reason-phrases* + '((200 . "OK") (400 . "Bad Request") (404 . "Not Found") + (405 . "Method Not Allowed") (500 . "Internal Server Error"))) + +(def (reason-phrase code) + (let ((p (assoc code *reason-phrases*))) (if p (cdr p) "OK"))) + +(def (json-resp status obj) + (make-presp status "application/json" (json-object->string obj))) + +(def (error-resp status msg) + (let ((o (make-hash-table)) (e (make-hash-table))) + (hash-put! e "message" msg) + (hash-put! e "type" "invalid_request_error") + (hash-put! o "error" e) + (json-resp status o))) + +(def (health-resp) + (let ((o (make-hash-table))) + (hash-put! o "status" "ok") + (json-resp 200 o))) + +(def (models-resp) + (let ((o (make-hash-table)) (m (make-hash-table))) + (hash-put! m "id" "forge") + (hash-put! m "object" "model") + (hash-put! m "created" 0) + (hash-put! m "owned_by" "forge") + (hash-put! o "object" "list") + (hash-put! o "data" (list m)) + (json-resp 200 o))) + +(def (sse-body chunks) + "Serialize a list of chat.completion.chunk hashes as an SSE event stream, + one `data: <json>` event per chunk, terminated by `data: [DONE]`." + (string-append + (apply string-append + (map (lambda (c) (string-append "data: " (json-object->string c) "\n\n")) + chunks)) + "data: [DONE]\n\n")) + +(def (condition->msg e) + (guard (_ [#t "internal error"]) + (cond + ((string? e) e) + ((and (condition? e) (message-condition? e)) (condition-message e)) + (else (call-with-string-output-port (lambda (p) (display-condition e p))))))) + +;; ── routing core (pure: no socket I/O — unit-testable) ──────────────── + +(def (proxy-dispatch method path body backend) + "Route one HTTP request to a `presp`. BODY is the raw request body string. + GET /health and GET /v1/models are static; POST /v1/chat/completions + parses BODY as JSON and runs it through handle-chat-completions, returning + a chat.completion object or — when stream=true — an SSE event stream." + (cond + ((and (equal? method "GET") (equal? path "/health")) (health-resp)) + ((and (equal? method "GET") (equal? path "/v1/models")) (models-resp)) + ((and (equal? method "POST") (equal? path "/v1/chat/completions")) + (let ((parsed (guard (_ [#t 'parse-error]) (string->json-object body)))) + (cond + ((eq? parsed 'parse-error) (error-resp 400 "invalid JSON body")) + ((not (hash-table? parsed)) (error-resp 400 "body must be a JSON object")) + (else + (guard (e [#t (error-resp 500 (condition->msg e))]) + (let* ((is-stream (eq? (hash-ref parsed "stream" #f) #t)) + (result (handle-chat-completions parsed backend))) + (if is-stream + (make-presp 200 "text/event-stream" (sse-body result)) + (json-resp 200 result)))))))) + ((member path '("/health" "/v1/models" "/v1/chat/completions")) + (error-resp 405 "method not allowed")) + (else (error-resp 404 "not found")))) + +;; ── HTTP wire I/O ───────────────────────────────────────────────────── +;; Minimal HTTP/1.1: read request-line + headers + Content-Length body, then +;; write a single Connection: close response. Content-Length is interpreted +;; in characters — exact for the ASCII/escaped JSON OpenAI clients send. + +(def (strip-cr s) + (let ((n (string-length s))) + (if (and (> n 0) (char=? (string-ref s (- n 1)) #\return)) + (substring s 0 (- n 1)) + s))) + +(def (split-spaces s) + (let ((n (string-length s))) + (let loop ((i 0) (start 0) (acc '())) + (cond + ((>= i n) (reverse (if (> i start) (cons (substring s start i) acc) acc))) + ((char=? (string-ref s i) #\space) + (loop (+ i 1) (+ i 1) + (if (> i start) (cons (substring s start i) acc) acc))) + (else (loop (+ i 1) start acc)))))) + +(def (parse-content-length line) + (let ((low (string-downcase line))) + (if (string-prefix? "content-length:" low) + (string->number + (string-trim (substring line (string-length "content-length:") + (string-length line)))) + #f))) + +(def (read-n-chars in n) + (let loop ((acc '()) (remaining n)) + (if (<= remaining 0) + (apply string-append (reverse acc)) + (let ((chunk (get-string-n in remaining))) + (if (or (eof-object? chunk) (= (string-length chunk) 0)) + (apply string-append (reverse acc)) + (loop (cons chunk acc) (- remaining (string-length chunk)))))))) + +(def (read-http-request in) + "Read one HTTP request from IN. Returns (method path body) or #f on EOF." + (let ((reqline (get-line in))) + (if (eof-object? reqline) + #f + (let* ((parts (split-spaces (strip-cr reqline))) + (method (if (pair? parts) (car parts) "")) + (path (if (and (pair? parts) (pair? (cdr parts))) (cadr parts) ""))) + (let hloop ((clen 0)) + (let ((line (get-line in))) + (cond + ((eof-object? line) (list method path "")) + ((string=? (strip-cr line) "") + (list method path (if (> clen 0) (read-n-chars in clen) ""))) + (else + (hloop (or (parse-content-length (strip-cr line)) clen)))))))))) + +(def (write-http-response out resp) + (let* ((status (presp-status resp)) + (body (presp-body resp)) + (blen (bytevector-length (string->utf8 body)))) + (display (string-append "HTTP/1.1 " (number->string status) " " + (reason-phrase status) "\r\n") out) + (display (string-append "Content-Type: " (presp-content-type resp) "\r\n") out) + (display (string-append "Content-Length: " (number->string blen) "\r\n") out) + (display "Connection: close\r\n" out) + (display "\r\n" out) + (display body out) + (flush-output-port out))) + +(def (proxy-serve bind-addr port backend) + "Listen on BIND-ADDR:PORT and serve OpenAI chat-completions through BACKEND. + Sequential accept loop — one request at a time (single-slot serialization)." + (let ((srv (tcp-listen bind-addr port))) + (fprintf (current-error-port) "[INFO] proxy listening on ~a:~a~n" + bind-addr (tcp-server-port srv)) + (flush-output-port (current-error-port)) + (let accept-loop () + (let-values (((in out) (tcp-accept srv))) + (guard (_ [#t (void)]) + (let ((req (read-http-request in))) + (write-http-response out + (if req + (proxy-dispatch (car req) (cadr req) (caddr req) backend) + (error-resp 400 "bad request"))))) + (guard (_ [#t (void)]) (close-port in)) + (guard (_ [#t (void)]) (close-port out)) + (accept-loop))))) + +;; ── provider-backed BACKEND adapter ─────────────────────────────────── +;; Wraps a jcode provider as the (messages tool-specs sampling) -> response +;; seam the handler drives. Request-level SAMPLING is accepted by the proxy +;; but not threaded into provider-chat (which applies the provider's own +;; per-model sampling policy); this matches a guardrail proxy sitting in +;; front of an already-configured local model. + +(def (params->hash p) + (cond + ((hash-table? p) p) + ((and (pair? p) (pair? (car p))) + (let ((ht (make-hash-table))) + (for-each (lambda (kv) (hash-put! ht (car kv) (cdr kv))) p) + ht)) + (else (make-hash-table)))) + +(def (tool-spec->provider-tool spec) + (let ((entry (make-hash-table)) (fn (make-hash-table))) + (hash-put! fn "name" (tool-spec-name spec)) + (hash-put! fn "description" (tool-spec-description spec)) + (hash-put! fn "parameters" (params->hash (tool-spec-get-json-schema spec))) + (hash-put! entry "type" "function") + (hash-put! entry "function" fn) + entry)) + +(def (hash->assoc ht) + (map (lambda (k) (cons k (hash-ref ht k (void)))) (hash-keys ht))) + +(def (args-string->assoc s) + (let ((parsed (guard (_ [#t #f]) (string->json-object s)))) + (if (hash-table? parsed) (hash->assoc parsed) '()))) + +(def (make-provider-backend provider) + "Adapt a jcode PROVIDER into a proxy BACKEND closure: it builds the OpenAI + tools array from the tool-specs, calls provider-chat once, and returns a + text-response (no tool calls) or a wtool-call list (the model called + tools). The first call carries the assistant's reasoning as its content." + (lambda (messages tool-specs sampling) + (let* ((tools (map tool-spec->provider-tool tool-specs)) + (resp (provider-chat provider messages tools)) + (tcs (message-tool-calls resp))) + (if (and tcs (pair? tcs)) + (let loop ((ts tcs) (first #t) (acc '())) + (if (null? ts) + (reverse acc) + (loop (cdr ts) #f + (cons (make-wtool-call + (tool-call-name (car ts)) + (args-string->assoc (tool-call-arguments (car ts))) + (if first (message-content resp) #f)) + acc)))) + (make-text-response (or (message-content resp) "")))))) --- a/src/jcode/ui/cli.ss +++ b/src/jcode/ui/cli.ss @@ -30,6 +30,8 @@ :jcode/core/compaction-strategy :jcode/core/workflow :jcode/core/workflow-runner + :jcode/core/slot-worker + :jcode/proxy/server :jcode/mcp/client :jcode/tool/lsp :jcode/core/plugin @@ -97,6 +99,7 @@ ((equal? (car rest) "config") (config-command (cdr rest))) ((equal? (car rest) "keys") (keys-command (cdr rest))) ((equal? (car rest) "serve") (serve-main (cdr rest))) + ((equal? (car rest) "proxy") (proxy-main (cdr rest))) ((equal? (car rest) "relay") (relay-main (cdr rest))) ((equal? (car rest) "connect") (connect-main (cdr rest))) (else (one-shot-mode (string-join rest " ") opts))) @@ -345,11 +348,54 @@ EXAMPLES: (printf " self-test ran ~a iterations, terminal returned: ~a~n" n result)) (printf "Define workflows in jerboa .ss with make-workflow; run via run-workflow.~n"))) +;; /forge proxy — describe the OpenAI-compatible proxy and self-test the +;; dispatch pipeline with a scripted backend (no live model needed). +(def (forge-print-proxy) + (printf "OpenAI-compatible proxy: available (guardrails wrap every request).~n") + (printf " Start with jcode proxy --port 8080 [--bind ADDR]~n") + (printf " Routes GET /health GET /v1/models POST /v1/chat/completions~n") + (let* ((backend (lambda (messages tool-specs sampling) + (make-text-response "pong"))) + (health (proxy-dispatch "GET" "/health" "" backend)) + (chat (proxy-dispatch "POST" "/v1/chat/completions" + "{\"messages\":[{\"role\":\"user\",\"content\":\"hi\"}]}" backend)) + (unknown (proxy-dispatch "GET" "/nope" "" backend))) + (printf " self-test /health -> ~a, chat -> ~a, unknown -> ~a~n" + (presp-status health) (presp-status chat) (presp-status unknown)) + (printf " slot policy higher priority preempts running: ~a~n" + (slot-should-preempt? 5 1))) + (printf "respond() is injected when the client sends tools, then stripped from~n") + (printf "the reply so the model stays in tool-calling mode where guardrails apply.~n")) + +;; `jcode proxy` — serve the configured provider behind the guardrail proxy. +(def (proxy-main args) + (let loop ((args args) (port 8080) (bind "127.0.0.1")) + (cond + ((null? args) + (let ((backend (make-provider-backend (get-current-provider)))) + (fprintf (current-error-port) + "[INFO] guardrail proxy for provider ~a~n" + (or (current-provider-override) (config-provider))) + (flush-output-port (current-error-port)) + (proxy-serve bind port backend))) + ((and (equal? (car args) "--port") (pair? (cdr args))) + (let ((p (string->number (cadr args)))) + (if (and p (> p 0) (< p 65536)) + (loop (cddr args) p bind) + (begin + (fprintf (current-error-port) "[ERROR] invalid port: ~a~n" (cadr args)) + (exit 1))))) + ((and (equal? (car args) "--bind") (pair? (cdr args))) + (loop (cddr args) port (cadr args))) + (else + (fprintf (current-error-port) "[ERROR] unknown proxy option: ~a~n" (car args)) + (exit 1))))) + (def (handle-command input session-id) (let ((cmd (string-trim (substring input 1 (string-length input))))) (cond ((equal? cmd "help") - (display "\nCommands:\n /help Show this help\n /model [name] Show or set model\n /provider [name] Show or set provider\n /plan Switch to PLAN mode (read-only)\n /build Switch to BUILD mode (read+write)\n /mode Show current mode\n /mcp Toggle MCP tools on/off\n /tools List available tools\n /clear Start a new session\n /sessions List saved sessions\n /compact Show message count\n /undo [N] Revert last N checkpoint(s) (default 1)\n /checkpoints List recent shadow-git checkpoints\n /forge [on|off] Show or toggle forge guardrails\n /forge sampling <off|on|strict> Per-model sampling policy\n /forge workflow Describe + self-test the workflow engine\n /quit Exit\n\nMulti-line: end a line with \\ to continue on the next line.\n\n")) + (display "\nCommands:\n /help Show this help\n /model [name] Show or set model\n /provider [name] Show or set provider\n /plan Switch to PLAN mode (read-only)\n /build Switch to BUILD mode (read+write)\n /mode Show current mode\n /mcp Toggle MCP tools on/off\n /tools List available tools\n /clear Start a new session\n /sessions List saved sessions\n /compact Show message count\n /undo [N] Revert last N checkpoint(s) (default 1)\n /checkpoints List recent shadow-git checkpoints\n /forge [on|off] Show or toggle forge guardrails\n /forge sampling <off|on|strict> Per-model sampling policy\n /forge workflow Describe + self-test the workflow engine\n /forge proxy Describe + self-test the OpenAI-compatible proxy\n /quit Exit\n\nMulti-line: end a line with \\ to continue on the next line.\n\n")) ((equal? cmd "model") (printf "Provider: ~a~n" (or (current-provider-override) (config-provider))) (printf "Model: ~a~n" (or (current-model-override) (config-model))) @@ -438,6 +484,8 @@ EXAMPLES: (forge-print-status)) ((equal? cmd "forge workflow") (forge-print-workflow)) + ((equal? cmd "forge proxy") + (forge-print-proxy)) ((or (equal? cmd "forge on") (equal? cmd "forge enforce") (equal? cmd "forge enforce on")) (forge-respond-enforced? #t) --- a/test/run.ss +++ b/test/run.ss @@ -22,7 +22,12 @@ (jcode core workflow) (jcode core steps) (jcode guardrails step-enforcer) - (jcode core workflow-runner)) + (jcode core workflow-runner) + (jcode proxy convert) + (jcode proxy handler) + (jcode core slot-worker) + (jcode proxy server) + (std text json)) ;; ── Helpers ────────────────────────────────────────────────────── @@ -1009,6 +1014,160 @@ (raises-pred? (lambda () (run-workflow w "go" resp (list (cons 'cancel? (lambda () #t))))) workflow-cancelled-error?) #t)) +;; ════════════════════════════════════════════════════════════════ +;; Phase 6 — OpenAI-compatible proxy (convert / handler / slot-worker / +;; dispatch). Bodies are built from JSON-string literals via +;; string->json-object so the representation matches a real request. +;; ════════════════════════════════════════════════════════════════ + +(define (p6-choice0 c) (car (hashtable-ref c "choices" '()))) +(define (p6-finish c) (hashtable-ref (p6-choice0 c) "finish_reason" #f)) +(define (p6-msg c) (hashtable-ref (p6-choice0 c) "message" #f)) +(define (p6-content c) (hashtable-ref (p6-msg c) "content" #f)) +(define (p6-tcs c) (hashtable-ref (p6-msg c) "tool_calls" #f)) +(define (p6-last xs) (list-ref xs (- (length xs) 1))) +(define (p6-chunk-finish ch) + (hashtable-ref (car (hashtable-ref ch "choices" '())) "finish_reason" #f)) +(define (p6-specs-have? specs name) + (let loop ([s specs]) + (cond [(null? s) #f] + [(equal? (tool-spec-name (car s)) name) #t] + [else (loop (cdr s))]))) + +(section "=== proxy convert: openai->messages ===") +(let* ([omsgs (string->json-object "[{\"role\":\"system\",\"content\":\"sys\"},{\"role\":\"user\",\"content\":\"hi\"},{\"role\":\"assistant\",\"content\":\"reasoning\",\"tool_calls\":[{\"id\":\"call_1\",\"type\":\"function\",\"function\":{\"name\":\"search\",\"arguments\":\"{\\\"q\\\":\\\"x\\\"}\"}}]},{\"role\":\"tool\",\"tool_call_id\":\"call_1\",\"content\":\"result\"}]")] + [cmsgs (openai->messages omsgs)]) + (check! "msgs count" (length cmsgs) 4) + (check! "msg0 system" (message-role (car cmsgs)) "system") + (check! "msg0 content" (message-content (car cmsgs)) "sys") + (check! "msg1 user" (message-role (cadr cmsgs)) "user") + (check! "msg2 assistant" (message-role (caddr cmsgs)) "assistant") + (let ([tcs (message-tool-calls (caddr cmsgs))]) + (check-pred! "assistant 1 tool call" tcs (lambda (x) (and (pair? x) (= (length x) 1)))) + (check! "tool call name" (tool-call-name (car tcs)) "search") + (check! "tool call args preserved" (tool-call-arguments (car tcs)) "{\"q\":\"x\"}")) + (check! "msg3 tool role" (message-role (cadddr cmsgs)) "tool") + (check! "msg3 tool_call_id" (message-tool-call-id (cadddr cmsgs)) "call_1")) + +;; content as a block list flattens to a newline-joined string +(let ([blk (openai->messages (string->json-object "[{\"role\":\"user\",\"content\":[{\"type\":\"text\",\"text\":\"a\"},{\"type\":\"text\",\"text\":\"b\"}]}]"))]) + (check! "block content flattened" (message-content (car blk)) "a\nb")) + +(section "=== proxy convert: outbound ===") +(let ([tco (tool-calls->openai + (list (make-wtool-call "search" '(("q" . "jb")) "reasoning text") + (make-wtool-call "answer" '(("a" . "1")) #f)) "m1")]) + (check! "tco finish" (p6-finish tco) "tool_calls") + (check! "tco content = first reasoning" (p6-content tco) "reasoning text") + (let ([entries (p6-tcs tco)]) + (check! "tco 2 entries" (length entries) 2) + (let ([fn (hashtable-ref (car entries) "function" #f)]) + (check! "tco entry name" (hashtable-ref fn "name" #f) "search") + (check! "tco entry args roundtrip" + (hashtable-ref (string->json-object (hashtable-ref fn "arguments" #f)) "q" #f) "jb")))) +(let ([tro (text-response->openai "the answer" "m1")]) + (check! "tro finish stop" (p6-finish tro) "stop") + (check! "tro content" (p6-content tro) "the answer")) +(let ([sse (text->sse-events "hello" "m1")]) + (check-pred! "text sse is a list" sse list?) + (check! "text sse last finish stop" (p6-chunk-finish (p6-last sse)) "stop")) +(let ([sse (tool-calls->sse-events (list (make-wtool-call "search" '(("q" . "x")) "r")) "m1")]) + (check! "tool sse last finish tool_calls" (p6-chunk-finish (p6-last sse)) "tool_calls")) + +(section "=== proxy handler: extraction ===") +(check! "extract-sampling present" + (hashtable-ref (extract-sampling (string->json-object "{\"temperature\":0.6,\"top_p\":0.9}")) "temperature" #f) 0.6) +(check! "extract-sampling absent → #f" + (extract-sampling (string->json-object "{\"messages\":[]}")) #f) +(let ([especs (extract-tool-specs (string->json-object "[{\"type\":\"function\",\"function\":{\"name\":\"search\",\"description\":\"d\",\"parameters\":{\"type\":\"object\"}}}]"))]) + (check! "extract-tool-specs count" (length especs) 1) + (check! "extract-tool-specs name" (tool-spec-name (car especs)) "search")) + +(section "=== proxy handler: completion paths ===") +(let ([h (handle-chat-completions + (string->json-object "{\"messages\":[{\"role\":\"user\",\"content\":\"hi\"}]}") + (lambda (msgs specs samp) (make-text-response "plain answer")))]) + (check! "no-tools finish stop" (p6-finish h) "stop") + (check! "no-tools content" (p6-content h) "plain answer")) +(let ([h (handle-chat-completions + (string->json-object "{\"messages\":[{\"role\":\"user\",\"content\":\"q\"}],\"tools\":[{\"type\":\"function\",\"function\":{\"name\":\"search\",\"description\":\"d\",\"parameters\":{\"type\":\"object\"}}}]}") + (lambda (msgs specs samp) (list (make-wtool-call "search" '(("q" . "x")) "thinking"))))]) + (check! "tools finish tool_calls" (p6-finish h) "tool_calls") + (check! "tools entry name" + (hashtable-ref (hashtable-ref (car (p6-tcs h)) "function" #f) "name" #f) "search")) +;; respond is injected into the offered specs (and the original tool retained) +(let ([seen #f]) + (handle-chat-completions + (string->json-object "{\"messages\":[{\"role\":\"user\",\"content\":\"q\"}],\"tools\":[{\"type\":\"function\",\"function\":{\"name\":\"search\",\"description\":\"d\",\"parameters\":{}}}]}") + (lambda (msgs specs samp) (set! seen specs) (make-text-response "x"))) + (check! "respond injected into specs" (p6-specs-have? seen "respond") #t) + (check! "original tool retained" (p6-specs-have? seen "search") #t)) +;; a pure respond() call is stripped and returned as a normal text answer +(let ([h (handle-chat-completions + (string->json-object "{\"messages\":[{\"role\":\"user\",\"content\":\"q\"}],\"tools\":[{\"type\":\"function\",\"function\":{\"name\":\"search\",\"description\":\"d\",\"parameters\":{}}}]}") + (lambda (msgs specs samp) (list (make-wtool-call "respond" '(("message" . "final words")) #f))))]) + (check! "respond-only finish stop" (p6-finish h) "stop") + (check! "respond-only content = message" (p6-content h) "final words")) +;; retries exhausted (&tool-call-error) → return last raw text for the client +(let ([h (handle-chat-completions + (string->json-object "{\"messages\":[{\"role\":\"user\",\"content\":\"q\"}],\"tools\":[{\"type\":\"function\",\"function\":{\"name\":\"search\",\"description\":\"d\",\"parameters\":{}}}]}") + (lambda (msgs specs samp) (raise-tool-call-error "exhausted" "raw text fallback")))]) + (check! "tool-call-error finish stop" (p6-finish h) "stop") + (check! "tool-call-error returns raw" (p6-content h) "raw text fallback")) +;; stream:true → handler returns a list of chunk objects +(check-pred! "stream handler returns chunk list" + (handle-chat-completions + (string->json-object "{\"messages\":[{\"role\":\"user\",\"content\":\"hi\"}],\"stream\":true}") + (lambda (msgs specs samp) (make-text-response "streamed"))) + list?) + +(section "=== slot-worker: scheduling policy ===") +(check! "priority lower int first" (slot-priority<? 0 5 1 0) #t) +(check! "priority equal → lower seq" (slot-priority<? 0 1 0 2) #t) +(check! "priority equal → higher seq not first" (slot-priority<? 0 2 0 1) #f) +(check! "priority higher int not first" (slot-priority<? 1 0 0 5) #f) +(check! "preempt: strictly higher" (slot-should-preempt? 5 1) #t) +(check! "no preempt: lower" (slot-should-preempt? 1 5) #f) +(check! "no preempt: equal" (slot-should-preempt? 3 3) #f) +(check! "no preempt: idle slot" (slot-should-preempt? #f 1) #f) +(let ([w (make-slot-worker (scripted-responder '()))]) + (check! "fresh worker is slot-worker?" (slot-worker? w) #t) + (check! "fresh worker pending 0" (slot-worker-pending w) 0) + (check! "fresh worker no running priority" (slot-worker-running-priority w) #f)) +;; end-to-end: start, submit a one-step workflow, get its terminal value +(let* ([wf (make-workflow "r" "d" + (list (make-tool-def (make-tool-spec "done" "d" '()) (lambda (a) "DONE-VAL") '())) + '() "done" "p")] + [w (make-slot-worker (scripted-responder (list (list (make-wtool-call "done" '() #f)))))]) + (slot-worker-start! w) + (let ([result (slot-submit! w wf "go" (list (cons 'max-iterations 4)))]) + (slot-worker-stop! w) + (check! "slot-submit! returns terminal value" result "DONE-VAL"))) + +(section "=== proxy dispatch: routing ===") +(define p6-backend (lambda (msgs specs samp) (make-text-response "pong"))) +(let ([r (proxy-dispatch "GET" "/health" "" p6-backend)]) + (check! "health status 200" (presp-status r) 200) + (check-pred! "health body has ok" (presp-body r) (lambda (b) (str-contains? b "ok")))) +(let ([r (proxy-dispatch "GET" "/v1/models" "" p6-backend)]) + (check! "models status 200" (presp-status r) 200) + (check-pred! "models body has forge" (presp-body r) (lambda (b) (str-contains? b "forge")))) +(let ([r (proxy-dispatch "POST" "/v1/chat/completions" + "{\"messages\":[{\"role\":\"user\",\"content\":\"hi\"}]}" p6-backend)]) + (check! "chat status 200" (presp-status r) 200) + (check! "chat content-type json" (presp-content-type r) "application/json") + (check! "chat finish stop" (p6-finish (string->json-object (presp-body r))) "stop")) +(let ([r (proxy-dispatch "POST" "/v1/chat/completions" + "{\"messages\":[{\"role\":\"user\",\"content\":\"hi\"}],\"stream\":true}" p6-backend)]) + (check! "stream content-type sse" (presp-content-type r) "text/event-stream") + (check-pred! "stream body has [DONE]" (presp-body r) (lambda (b) (str-contains? b "data: [DONE]")))) +(let ([r (proxy-dispatch "POST" "/v1/chat/completions" "not json at all" p6-backend)]) + (check! "bad json → 400" (presp-status r) 400)) +(let ([r (proxy-dispatch "GET" "/nope" "" p6-backend)]) + (check! "unknown route → 404" (presp-status r) 404))