diff --git a/agent-contract/inner/edge-config-runtime-refresh.md b/agent-contract/inner/edge-config-runtime-refresh.md index 9f18da6..ffb3b6e 100644 --- a/agent-contract/inner/edge-config-runtime-refresh.md +++ b/agent-contract/inner/edge-config-runtime-refresh.md @@ -37,6 +37,7 @@ tracked config에는 public 예시와 기본 구조만 두고, 실제 endpoint/c - `nodes[].providers[].adapter`는 같은 Node 안의 enabled adapter instance key를 참조해야 한다. Exact instance key를 우선하고, legacy type-name route는 같은 type의 enabled instance가 정확히 하나일 때만 허용한다. `category: cli` resource는 enabled CLI adapter가 필요하다. - `nodes[].providers[].enabled`: 생략 또는 `true` → provider pool dispatch 후보에 포함. `false` → dispatch pool에서 제외. 비활성화된 provider는 status snapshot에 `status=disabled`, `health=disabled`, `capacity=0`으로 표시된다. adapter process lifecycle 변경 없음. config refresh 시 `enabled` 토글은 live-apply(restart 불필요)로 분류된다. disabled provider의 adapter reference check는 skip되지만 structural validation(type, category, models, numeric bounds)은 수행된다. - `nodes[].providers[].priority`: provider-pool dispatch tie-breaker다. 기본값은 `0`이고 음수는 validation error다. dispatch는 `in_flight < capacity` 후보 중 가장 낮은 `in_flight`를 먼저 선택하며, `in_flight`가 같은 후보에서만 낮은 숫자의 `priority`를 우선한다. `in_flight`와 `priority`가 모두 같으면 기존 순환을 유지한다. priority 변경은 live-apply(restart 불필요)로 분류된다. +- `nodes[].providers[].queue_timeout_ms`와 adapter instance `queue_timeout_ms`는 provider-pool 대기열 timeout이다. queue policy가 구성된 상태에서 값이 `0`이면 queue timeout을 두지 않고, caller context cancellation 또는 연결 종료로만 대기 요청을 중단한다. provider가 queue policy를 전혀 제공하지 않으면 runtime 기본 queue timeout을 사용한다. - legacy single-instance adapter 설정은 load 시 named instance slice로 normalize된다. - `NodeConfigPayload`는 Edge가 Node에 내려주는 실행 adapter/runtime payload다. - refresh 결과는 `applied`, `restart_required`, `rejected`를 구분하고, changed node/provider/model/report slice는 안정적으로 non-nil이어야 한다. diff --git a/agent-contract/outer/openai-compatible-api.md b/agent-contract/outer/openai-compatible-api.md index 6011190..a9ea6d9 100644 --- a/agent-contract/outer/openai-compatible-api.md +++ b/agent-contract/outer/openai-compatible-api.md @@ -207,7 +207,7 @@ Provider pool model catalog의 `models[]` entry가 generation policy를 제공 - `default_max_tokens`: caller가 출력 token limit을 생략했을 때 `max_tokens` 또는 `max_output_tokens`로 주입한다. - `min_max_tokens`: caller가 너무 작은 출력 token limit을 보냈을 때 해당 값까지 올린다. caller 값이 더 크면 보존한다. -- `default_thinking_token_budget`: caller가 `thinking_token_budget`을 생략했고 strict output이 reasoning을 금지하지 않을 때 내부 실행 입력에 주입한다. +- `default_thinking_token_budget`: caller가 `thinking_token_budget`을 생략했을 때 내부 실행 입력에 주입한다. provider-pool catalog에 이 값이 있으면 Edge는 `think=true`도 함께 주입해 vLLM/vLLM-MLX 계열 adapter가 `chat_template_kwargs.enable_thinking=true`로 전달하도록 한다. Conflict 정책: @@ -220,6 +220,7 @@ Conflict 정책: Strict output 모드: - strict output가 활성화되면 `think=true`가 명시되지 않은 요청은 내부 실행 입력에서 `think=false`로 낮춘다. +- 단, provider-pool `models[]` entry의 `default_thinking_token_budget`가 적용되는 모델은 catalog의 thinking policy가 우선한다. 이 경우 요청이 `think=false` 또는 `reasoning_effort=none`을 명시하지 않았다면 strict output에서도 `think=true`와 `thinking_token_budget`을 내부 실행 입력에 넣는다. `tools`가 있는 Chat Completions 요청에서 provider route(`openai_compat`, `vllm`, `ollama`, provider pool)는 forced tool 선택 객체와 `"none"` 같은 명시적 `tool_choice`를 backend에 전달한다. 단, `"auto"`는 OpenAI-compatible 기본값과 같으므로 provider request에서는 생략한다. 일부 vLLM 계열 backend는 explicit/default `"auto"`를 `--enable-auto-tool-choice`/`--tool-call-parser` 없이 400으로 거부한다. 이 400이 발생하고 요청 tool이 정확히 1개이면 Node adapter는 해당 tool에 대한 forced `tool_choice`로 1회 재시도한다. forced tool도 `--tool-call-parser` 요구로 거부되거나 여러 tool이라 forced를 고를 수 없으면, Node adapter는 `tools`/`tool_choice`를 제거하고 text tool-call system instruction을 leading system message에 병합해 1회 재시도하며 완료 metadata에 `openai_text_tool_fallback: "true"`를 싣는다. provider가 native OpenAI-compatible `tool_calls`를 반환하면 Node는 내부 `RunEvent.metadata["openai_tool_calls"]` JSON으로 보존하고, Edge는 이를 OpenAI-compatible `message.tool_calls` 또는 stream `delta.tool_calls`로 반환하며 `finish_reason: "tool_calls"`를 사용한다. diff --git a/agent-test/dev-corp/edge-smoke.md b/agent-test/dev-corp/edge-smoke.md index e544f5d..092d89e 100644 --- a/agent-test/dev-corp/edge-smoke.md +++ b/agent-test/dev-corp/edge-smoke.md @@ -93,6 +93,8 @@ Mac Studio의 `http://192.168.2.3:8005/v1` mlx-vlm DiffusionGemma endpoint는 - DGX Spark 01은 FlashInfer FP4 JIT가 `ninja`를 PATH에서 호출하므로 start script에 `PATH=/home/digitalcommerce_dgx_spark_01/vllm_env/bin:$PATH`를 둔다. - Mac Studio provider runtime은 capacity `5`, context window `262144`, requested KV `262144x3` 기준으로 `--continuous-batching`, `--max-num-seqs 5`, `--prefill-batch-size 5`, `--completion-batch-size 5`, `--chunked-prefill-tokens 1024`, `--max-request-tokens 262144`, `--max-kv-size 786432`, `--reasoning-parser gemma4`을 적용했고 `/health`, `/v1/models`, 직접 동시성 `1..5` chat completion benchmark 통과를 확인했다. - 2026-07-02 재확인 기준 DGX Spark 01은 mac-mini 내부 endpoint `http://192.168.2.2:8002`에서 `/health` 200과 `/v1/models`를 통과했다. DGX Spark 02는 mac-mini에서 SSH, provider `/health`, `/v1/models`가 회복됐고 직접 provider benchmark를 통과했다. Edge OpenAI-compatible capacity smoke는 Control Plane status snapshot 기준으로 재검증 완료됐다. +- Gemma provider-pool `gemma4:26b`는 `default_thinking_token_budget: 1024` 기준으로 thinking을 기본 활성화한다. Edge strict output이 켜져 있어도 provider-pool catalog의 thinking policy가 우선하며, vLLM/vLLM-MLX adapter에는 `chat_template_kwargs.enable_thinking=true`로 전달되는지 확인한다. +- Gemma provider-pool의 IOP queue timeout은 두지 않는다. config 값은 `queue_timeout_ms=0`이며, caller context cancellation 또는 연결 종료로만 queued request를 중단한다. OpenAI run timeout과 backend request timeout은 long reasoning/long-context 요청을 위해 각각 `openai.timeout_sec=1800`, `request_timeout_ms=1800000`으로 둔다. ## 2026-07-02 dev-corp native runtime 상태 @@ -136,7 +138,7 @@ Mac Studio의 `http://192.168.2.3:8005/v1` mlx-vlm DiffusionGemma endpoint는 - OpenAI-compatible smoke에서 `/healthz`, `/v1/models`, `/v1/responses`가 기대 상태로 응답한다. - dev-corp capacity smoke는 `/v1/responses`와 `/v1/chat/completions` 각각에 provider capacity 총합 + 1개 동시 요청을 보낸다. Edge config에서 기본 후보 capacity `4 + 4 + 5 = 13`을 확정한 경우 Control Plane status의 `provider_snapshots`에서 총 `in_flight=13`, `queued>=1` 관측을 기준으로 한다. - capacity smoke 완료 후 대상 provider의 `in_flight=0`, `queued=0` 회복을 확인한다. -- Gemma 계열 provider-pool smoke는 reasoning/tool-parser 관련 텍스트가 포함될 수 있다. exact-output match를 기본 판정으로 쓰지 않는다. +- Gemma 계열 provider-pool smoke는 thinking enabled 기준이며 reasoning/tool-parser 관련 텍스트가 포함될 수 있다. exact-output match를 기본 판정으로 쓰지 않는다. ## 기준 출력 예시 diff --git a/agent-test/dev-corp/inventory.yaml b/agent-test/dev-corp/inventory.yaml index 3b4ae1b..dd4ff27 100644 --- a/agent-test/dev-corp/inventory.yaml +++ b/agent-test/dev-corp/inventory.yaml @@ -1,6 +1,6 @@ test_env: dev-corp profile: dev-corp-provider-pool -last_updated_at: "2026-07-02" +last_updated_at: "2026-07-06" source: remote_runner: @@ -8,7 +8,7 @@ source: repo_root: /Users/fe/agent-work/iop-dev-corp setup_required: true setup_status: deployed - current_observation: /Users/fe/agent-work/iop-dev-corp checkout and build/dev-corp-runtime runtime were created on 2026-06-25, then clean-synced/rebuilt at source ref 5b1255a on 2026-07-02; /Users/fe/iop-field remains unrelated legacy field state + current_observation: /Users/fe/agent-work/iop-dev-corp checkout and build/dev-corp-runtime runtime were created on 2026-06-25, clean-synced/rebuilt at source ref 2d08d24 on 2026-07-06, then validated with the provider-pool thinking-policy patch and gemma4:26b default_thinking_token_budget config update; /Users/fe/iop-field remains unrelated legacy field state clean_sync: - git fetch origin main - git reset --hard origin/main @@ -50,8 +50,20 @@ model: alias_status: candidate alias_policy: expose one IOP model alias and map provider-specific served_model values per provider provider_capacity_total: 13 - provider_capacity_status: verified_with_control_plane_provider_snapshots_2026_07_02 + provider_capacity_status: verified_with_control_plane_provider_snapshots_2026_07_06 context_window_max: 262144 + default_thinking_token_budget: 1024 + reasoning_policy: bounded_thinking_enabled + think_policy: + default: enabled + strict_output_behavior: provider_pool_catalog_default_overrides_strict_disable + adapter_mapping: vllm_and_vllm_mlx_use_chat_template_kwargs_enable_thinking_true + timeout_policy: + openai_timeout_sec: 1800 + provider_queue_timeout_ms: 0 + provider_queue_timeout_policy: disabled_no_iop_queue_timeout + backend_request_timeout_ms: 1800000 + note: provider-pool long reasoning and long-context requests do not use an IOP queue timeout; backend/OpenAI run budgets remain 30 minutes runtime_capacity_targets: dgx_spark: capacity: 4 @@ -70,15 +82,18 @@ model: - /v1/chat/completions concurrent_requests: capacity_plus_one expected_total_in_flight: 13 - expected_total_in_flight_status: verified_2026_07_02 + expected_total_in_flight_status: verified_2026_07_06 expected_min_queued: 1 extended_concurrent_requests: 16 extended_expected_min_queued: 3 prompt_policy: long_reasoning_allowed + think_policy: enabled_for_provider_pool exact_output_match: false latest_capacity_verification: - date: "2026-07-02" - source_ref: 5b1255abc5ad530ebfd7a65deecdf5575425a34c + date: "2026-07-06" + source_ref: 2d08d240d48a954df6847df2ba08fb8b39fc3f7d + edge_runtime_patch: provider_pool_thinking_policy_enabled_under_strict_output + default_thinking_token_budget: 1024 control_plane_enabled: true control_plane_status_url: http://127.0.0.1:18002/edges/dev-corp-edge/status provider_snapshot_baseline: @@ -99,6 +114,8 @@ model: responses: concurrent_requests: 14 ok: 14 + timeout_errors: 0 + http_errors: 0 max_total_in_flight: 13 max_total_queued: 3 provider_max: @@ -115,6 +132,8 @@ model: chat_completions: concurrent_requests: 14 ok: 14 + timeout_errors: 0 + http_errors: 0 max_total_in_flight: 13 max_total_queued: 3 provider_max: @@ -148,7 +167,7 @@ model: - DGX Spark 01 FlashInfer FP4 JIT invokes ninja from PATH; start script now prefixes /home/digitalcommerce_dgx_spark_01/vllm_env/bin - Mac Studio vLLM-MLX pyexpat required DYLD_LIBRARY_PATH=/opt/homebrew/opt/expat/lib verification: - last_rechecked_at: "2026-07-02" + last_rechecked_at: "2026-07-06" mac_studio_health: passed dgx01_health: passed dgx02_health: passed @@ -156,7 +175,7 @@ model: dgx01: mac-mini http://192.168.2.2:8002 returned /health 200 and /v1/models exposed gemma-4-26B-A4B-it-NVFP4 after vllm_server_8002 restart; startup log reported GPU KV cache size 834,507 tokens and 3.18x concurrency for 262,144-token requests dgx02: node-local http://127.0.0.1:8004 and mac-mini http://192.168.2.4:8004 returned /health 200, /v1/models exposed gemma-4-26B-A4B-it-NVFP4, and direct concurrency 1..4 chat completion benchmark passed mac_studio: node-local http://127.0.0.1:8004 and mac-mini http://192.168.2.3:8004 returned /health 200, /v1/models exposed mlx-community/gemma-4-26b-a4b-it-nvfp4, and direct concurrency 1..5 chat completion benchmark passed - edge_openai_smoke: passed_with_control_plane_capacity_snapshot_2026_07_02 + edge_openai_smoke: passed_with_control_plane_capacity_snapshot_2026_07_06 nodes: - id: corp-dgx-spark-01-vllm-node @@ -177,6 +196,8 @@ nodes: capacity_status: configured_health_passed_direct_smoke_and_edge_capacity_smoke last_runtime_observation: restarted with gpu_memory_utilization 0.40 on 2026-07-02; mac-mini http://192.168.2.2:8002 returned /health 200 and /v1/models; startup log reported GPU KV cache size 834,507 tokens and 3.18x concurrency for 262,144-token requests capacity_basis: dev-corp target baseline; vLLM --max-num-seqs 4, --max-model-len 262144, --gpu-memory-utilization 0.40; observed KV cache satisfies requested 262144x2 minimum + queue_timeout_ms: 0 + request_timeout_ms: 1800000 edge_connectivity: mode: reverse_ssh_tunnel reason: current node.yaml uses tunnel-local Edge addr in the native provider-pool runtime @@ -215,6 +236,8 @@ nodes: context_window_max: 262144 kv_size_tokens_effective: 834507 full_context_concurrency_observed: 3.18 + default_chat_template_kwargs: + enable_thinking: true - id: corp-dgx-spark-02-vllm-node alias: corp-dgx-spark-02-vllm @@ -234,6 +257,8 @@ nodes: capacity_status: configured_health_passed_direct_smoke_and_edge_capacity_smoke last_runtime_observation: Docker vllm-gemma4 running with gpu_memory_utilization 0.40; node-local and mac-mini /health passed, /v1/models exposed gemma-4-26B-A4B-it-NVFP4, and direct concurrency 1..4 chat completion benchmark passed; startup log reported GPU KV cache size 860,222 tokens and 3.28x concurrency for 262,144-token requests capacity_basis: dev-corp target baseline; Docker publishes host 8004 to container 8004; vLLM --max-num-seqs 4, --max-model-len 262144, --gpu-memory-utilization 0.40; observed KV cache satisfies requested 262144x2 minimum + queue_timeout_ms: 0 + request_timeout_ms: 1800000 edge_connectivity: mode: reverse_ssh_tunnel reason: node02 cannot reach mac-mini 172.24.63.178:18085/18086 directly from 192.168.2.4 @@ -270,6 +295,8 @@ nodes: context_window_max: 262144 kv_size_tokens_effective: 860222 full_context_concurrency_observed: 3.28 + default_chat_template_kwargs: + enable_thinking: true - id: corp-mac-studio-mlx-vllm-node alias: corp-mac-studio-mlx-vllm @@ -290,6 +317,8 @@ nodes: capacity_status: configured_health_passed_direct_smoke_and_edge_capacity_smoke last_runtime_observation: screen vllm_mlx_8004 running with max_num_seqs 5, max_request_tokens 262144, max_kv_size 786432; node-local and mac-mini /health passed, /v1/models exposed mlx-community/gemma-4-26b-a4b-it-nvfp4, and direct concurrency 1..5 chat completion benchmark passed capacity_basis: vllm-mlx --max-num-seqs 5 with requested KV 262144x3 mapped to --max-kv-size 786432 + queue_timeout_ms: 0 + request_timeout_ms: 1800000 runtime: host: dc-devui-MacStudio.local os: macOS 26.2 @@ -320,6 +349,8 @@ nodes: context_window_max: 262144 kv_size_tokens_effective: 786432 reasoning_parser: gemma4 + default_chat_template_kwargs: + enable_thinking: true secondary_provider_candidate: id: corp-mac-studio-mlx-vlm-diffusiongemma type: mlx-vlm diff --git a/agent-test/dev-corp/node-smoke.md b/agent-test/dev-corp/node-smoke.md index 356a381..8e11248 100644 --- a/agent-test/dev-corp/node-smoke.md +++ b/agent-test/dev-corp/node-smoke.md @@ -92,6 +92,8 @@ Mac Studio의 secondary `http://192.168.2.3:8005/v1` endpoint는 기본 Node/pro - DGX Spark에서 explicit `--max-num-batched-tokens 524288`를 사용하는 경우 first profile 중 FP4 MoE 커널 한계에 걸릴 수 있어 `VLLM_MAX_TOKENS_PER_EXPERT_FP4_MOE=4194304` 보정이 필요하다. - DGX Spark 01은 FlashInfer FP4 JIT가 first compile 중 `ninja`를 PATH에서 찾으므로 `/home/digitalcommerce_dgx_spark_01/vllm_env/bin` PATH prefix가 필요하다. - Mac Studio에는 capacity `5`, context window `262144`, requested KV `262144x3` 기준을 `--max-num-seqs 5`, `--max-request-tokens 262144`, `--max-kv-size 786432`로 반영했고 `/health`, `/v1/models`, 직접 동시성 `1..5` chat completion benchmark 통과를 확인했다. +- Gemma provider-pool은 `default_thinking_token_budget: 1024` 기준으로 thinking을 기본 활성화한다. vLLM과 vLLM-MLX provider 모두 Edge/Node adapter 요청에서 `chat_template_kwargs.enable_thinking=true`가 적용되는 방향으로 검증한다. +- Gemma provider-pool의 provider queue timeout은 두지 않는다. config 값은 `queue_timeout_ms=0`이며, backend request timeout은 30분 기준인 `request_timeout_ms=1800000`으로 맞춘다. - 2026-07-02 재확인 기준 DGX Spark 02는 mac-mini에서 SSH, provider `/health`, `/v1/models`가 회복됐고 직접 provider benchmark를 통과했다. Edge OpenAI-compatible capacity smoke는 Control Plane status snapshot 기준으로 재검증 완료됐다. ## 2026-07-02 연결/용량 검증 상태 diff --git a/apps/edge/internal/openai/chat_handler.go b/apps/edge/internal/openai/chat_handler.go index 4879e14..087b78d 100644 --- a/apps/edge/internal/openai/chat_handler.go +++ b/apps/edge/internal/openai/chat_handler.go @@ -181,8 +181,13 @@ func applyModelCatalogGenerationPolicyToChat(req *chatCompletionRequest, entry c return } applyOutputTokenPolicy(&req.MaxTokens, req.MaxCompletionTokens, entry.DefaultMaxTokens, entry.MinMaxTokens) - if entry.DefaultThinkingTokenBudget > 0 && req.ThinkingTokenBudget == nil && chatRequestAllowsDefaultThinkingBudget(*req, strictOutput) { - req.ThinkingTokenBudget = intPtr(entry.DefaultThinkingTokenBudget) + if entry.DefaultThinkingTokenBudget > 0 && chatRequestAllowsDefaultThinkingBudget(*req) { + if req.ThinkingTokenBudget == nil { + req.ThinkingTokenBudget = intPtr(entry.DefaultThinkingTokenBudget) + } + if strictOutput && req.Think == nil { + req.Think = boolPtr(true) + } } } @@ -216,10 +221,7 @@ func applyOutputTokenPolicy(primary **int, fallback *int, defaultTokens, minToke } } -func chatRequestAllowsDefaultThinkingBudget(req chatCompletionRequest, strictOutput bool) bool { - if strictOutput && (req.Think == nil || !*req.Think) { - return false - } +func chatRequestAllowsDefaultThinkingBudget(req chatCompletionRequest) bool { if req.Think != nil && !*req.Think { return false } @@ -233,6 +235,10 @@ func intPtr(v int) *int { return &v } +func boolPtr(v bool) *bool { + return &v +} + func chatRunMetadata(runMeta map[string]string, req chatCompletionRequest, outputPolicy strictOutputPolicy) map[string]string { if runMeta == nil { runMeta = make(map[string]string) diff --git a/apps/edge/internal/openai/responses_handler.go b/apps/edge/internal/openai/responses_handler.go index 459fefa..f1f544c 100644 --- a/apps/edge/internal/openai/responses_handler.go +++ b/apps/edge/internal/openai/responses_handler.go @@ -65,7 +65,7 @@ func (s *Server) handleResponses(w http.ResponseWriter, r *http.Request) { var defaultThinkingTokenBudget int if catalogEntry := s.findProviderPoolEntry(req.Model); catalogEntry != nil { applyModelCatalogGenerationPolicyToResponses(&req, *catalogEntry) - if !outputPolicy.Strict && catalogEntry.DefaultThinkingTokenBudget > 0 { + if catalogEntry.DefaultThinkingTokenBudget > 0 { defaultThinkingTokenBudget = catalogEntry.DefaultThinkingTokenBudget } } @@ -74,10 +74,11 @@ func (s *Server) handleResponses(w http.ResponseWriter, r *http.Request) { runMeta["openai_stream"] = fmt.Sprintf("%t", req.Stream) runMeta["strict_output"] = fmt.Sprintf("%t", outputPolicy.Strict) input := map[string]any{"prompt": prompt} - if outputPolicy.Strict { - input["think"] = false - } else if defaultThinkingTokenBudget > 0 { + if defaultThinkingTokenBudget > 0 { + input["think"] = true input["thinking_token_budget"] = defaultThinkingTokenBudget + } else if outputPolicy.Strict { + input["think"] = false } if options := req.providerOptions(); len(options) > 0 { input["options"] = options diff --git a/apps/edge/internal/openai/server_test.go b/apps/edge/internal/openai/server_test.go index 8e2051b..fae6df3 100644 --- a/apps/edge/internal/openai/server_test.go +++ b/apps/edge/internal/openai/server_test.go @@ -3730,6 +3730,37 @@ func TestChatCompletionsProviderPoolAppliesGenerationPolicy(t *testing.T) { } } +func TestChatCompletionsProviderPoolThinkingPolicyOverridesStrictOutputDisable(t *testing.T) { + fake := &fakeRunService{events: bufferedRunEvents( + &iop.RunEvent{Type: "delta", Delta: "ok"}, + &iop.RunEvent{Type: "complete"}, + )} + catalog := []config.ModelCatalogEntry{{ + ID: "ornith:35b", + DefaultThinkingTokenBudget: 8192, + Providers: map[string]string{"prov-vllm": "Ornith-1.0-35B"}, + }} + srv := NewServer(config.EdgeOpenAIConf{StrictOutput: true}, fake, nil) + srv.SetModelCatalog(catalog) + + req := httptest.NewRequest(http.MethodPost, "/v1/chat/completions", strings.NewReader(`{ + "model":"ornith:35b", + "messages":[{"role":"user","content":"hello"}] + }`)) + w := httptest.NewRecorder() + srv.handleChatCompletions(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status: got %d body=%s", w.Code, w.Body.String()) + } + if fake.req.Input["think"] != true { + t.Fatalf("provider-pool thinking policy should keep thinking enabled under strict output: %+v", fake.req.Input) + } + if fake.req.Input["thinking_token_budget"].(int) != 8192 { + t.Fatalf("thinking_token_budget policy not applied: %+v", fake.req.Input) + } +} + func TestChatCompletionsProviderPoolPreservesLargerGenerationPolicyValues(t *testing.T) { fake := &fakeRunService{events: bufferedRunEvents( &iop.RunEvent{Type: "delta", Delta: "ok"}, @@ -3876,6 +3907,37 @@ func TestResponsesProviderPoolAppliesGenerationPolicy(t *testing.T) { } } +func TestResponsesProviderPoolThinkingPolicyOverridesStrictOutputDisable(t *testing.T) { + fake := &fakeRunService{events: bufferedRunEvents( + &iop.RunEvent{Type: "delta", Delta: "ok"}, + &iop.RunEvent{Type: "complete", Usage: &iop.Usage{InputTokens: 1, OutputTokens: 1}}, + )} + catalog := []config.ModelCatalogEntry{{ + ID: "ornith:35b", + DefaultThinkingTokenBudget: 8192, + Providers: map[string]string{"prov-vllm": "Ornith-1.0-35B"}, + }} + srv := NewServer(config.EdgeOpenAIConf{TimeoutSec: 5, StrictOutput: true}, fake, nil) + srv.SetModelCatalog(catalog) + + req := httptest.NewRequest(http.MethodPost, "/v1/responses", strings.NewReader(`{ + "model":"ornith:35b", + "input":"hello" + }`)) + w := httptest.NewRecorder() + srv.handleResponses(w, req) + + if w.Code != http.StatusOK { + t.Fatalf("status: got %d body=%s", w.Code, w.Body.String()) + } + if fake.req.Input["think"] != true { + t.Fatalf("provider-pool thinking policy should keep thinking enabled under strict output: %+v", fake.req.Input) + } + if fake.req.Input["thinking_token_budget"].(int) != 8192 { + t.Fatalf("thinking_token_budget policy not applied: %+v", fake.req.Input) + } +} + // TestChatCompletionsProviderPoolFallsBackToLegacyRoute verifies that when the // request model does not match the catalog, the legacy model_routes path is used. func TestChatCompletionsProviderPoolFallsBackToLegacyRoute(t *testing.T) { diff --git a/apps/edge/internal/service/model_queue.go b/apps/edge/internal/service/model_queue.go index d55b4c8..a87d7d6 100644 --- a/apps/edge/internal/service/model_queue.go +++ b/apps/edge/internal/service/model_queue.go @@ -56,8 +56,9 @@ func (c *candidateNode) slotKey() string { } type groupPolicy struct { - maxQueue int - queueTimeout time.Duration + maxQueue int + queueTimeout time.Duration + queueTimeoutSet bool } type inflightRec struct { @@ -150,19 +151,30 @@ func isTerminalRunEvent(e *iop.RunEvent) bool { return t == "complete" || t == "error" || t == "cancelled" } +// normalizeGroupPolicy applies queue defaults while preserving an explicit +// zero queue timeout as "no timeout". +func normalizeGroupPolicy(policy groupPolicy) groupPolicy { + if policy.maxQueue <= 0 { + policy.maxQueue = defaultGroupMaxQueue + } + if policy.queueTimeout > 0 { + policy.queueTimeoutSet = true + } + if !policy.queueTimeoutSet { + policy.queueTimeout = defaultQueueTimeout + policy.queueTimeoutSet = true + } + return policy +} + // getOrCreateGroupLocked returns an existing group or creates one with the -// given policy (applying defaults for zero fields). Policy is only applied +// given policy (applying defaults for unset fields). Policy is only applied // at creation; existing groups retain their current policy until an explicit // updateGroupPolicyLocked call (used for config-refresh policy propagation). func (m *modelQueueManager) getOrCreateGroupLocked(key string, policy groupPolicy) *modelQueueGroup { g, ok := m.groups[key] if !ok { - if policy.maxQueue <= 0 { - policy.maxQueue = defaultGroupMaxQueue - } - if policy.queueTimeout <= 0 { - policy.queueTimeout = defaultQueueTimeout - } + policy = normalizeGroupPolicy(policy) g = &modelQueueGroup{ key: key, policy: policy, @@ -176,7 +188,7 @@ func (m *modelQueueManager) getOrCreateGroupLocked(key string, policy groupPolic // updateGroupPolicyLocked refreshes an existing group's queue policy so that // provider max_queue / queue_timeout_ms changes applied via config refresh take -// effect on subsequent admissions. Only non-zero policy fields override the +// effect on subsequent admissions. Only configured policy fields override the // current values; in-flight counts and queued items are preserved untouched. func (m *modelQueueManager) updateGroupPolicyLocked(g *modelQueueGroup, policy groupPolicy) { if g == nil { @@ -187,6 +199,10 @@ func (m *modelQueueManager) updateGroupPolicyLocked(g *modelQueueGroup, policy g } if policy.queueTimeout > 0 { g.policy.queueTimeout = policy.queueTimeout + g.policy.queueTimeoutSet = true + } else if policy.queueTimeoutSet { + g.policy.queueTimeout = policy.queueTimeout + g.policy.queueTimeoutSet = true } } @@ -328,10 +344,14 @@ func (m *modelQueueManager) admitWithReason(ctx context.Context, groupKey, adapt } } + var deadline time.Time + if group.policy.queueTimeout > 0 { + deadline = time.Now().Add(group.policy.queueTimeout) + } item := &queueItem{ candidates: candidates, waitCh: make(chan admitResult, 1), - deadline: time.Now().Add(group.policy.queueTimeout), + deadline: deadline, long: long, reason: reason, } @@ -339,6 +359,35 @@ func (m *modelQueueManager) admitWithReason(ctx context.Context, groupKey, adapt timeout := group.policy.queueTimeout m.mu.Unlock() + if timeout <= 0 { + select { + case res := <-item.waitCh: + return res.candidate, item.reason, res.err + case <-ctx.Done(): + m.mu.Lock() + m.removeItemLocked(groupKey, item) + m.mu.Unlock() + // Handle race: dispatch may have sent to waitCh just before cancellation. + select { + case res := <-item.waitCh: + if res.candidate != nil { + slot := res.candidate.slotKey() + var nodeID, providerID string + if colonIdx := findLastColon(slot); colonIdx > 0 { + nodeID = slot[:colonIdx] + providerID = slot[colonIdx+1:] + } else { + nodeID = slot + } + longSlot := item.long && res.candidate.longContextCapacity > 0 + m.releaseSlotWithLong(groupKey, nodeID, providerID, longSlot) + } + default: + } + return nil, "", ctx.Err() + } + } + timer := time.NewTimer(timeout) defer timer.Stop() @@ -547,7 +596,7 @@ func (m *modelQueueManager) tryDispatchLocked(group *modelQueueGroup) { for i := 0; i < len(group.queue); i++ { item := group.queue[i] - if time.Now().After(item.deadline) { + if !item.deadline.IsZero() && time.Now().After(item.deadline) { group.queue = append(group.queue[:i], group.queue[i+1:]...) select { case item.waitCh <- admitResult{err: fmt.Errorf("model group %q: %w", group.key, errQueueTimeout)}: diff --git a/apps/edge/internal/service/model_queue_test.go b/apps/edge/internal/service/model_queue_test.go index eed7b77..89a1176 100644 --- a/apps/edge/internal/service/model_queue_test.go +++ b/apps/edge/internal/service/model_queue_test.go @@ -495,6 +495,67 @@ func TestModelQueueUsesProviderQueuePolicy(t *testing.T) { t.Fatalf("timed out too fast: %v", elapsed) } }) + + t.Run("zeroQueueTimeoutWaitsForContextCancellation", func(t *testing.T) { + entry := &edgenode.NodeEntry{NodeID: "node-pq3"} + cands := []candidateNode{{entry: entry, capacity: 1}} + policy := groupPolicy{maxQueue: 16, queueTimeoutSet: true} + + m := newModelQueueManager(nil) + + m.mu.Lock() + g := m.getOrCreateGroupLocked("g-pq-no-timeout", policy) + g.inflight["node-pq3"] = 1 + m.mu.Unlock() + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) + defer cancel() + + start := time.Now() + _, err := m.admit(ctx, "g-pq-no-timeout", "", "", cands, policy) + elapsed := time.Since(start) + + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("expected context deadline, got: %v", err) + } + if errors.Is(err, errQueueTimeout) { + t.Fatalf("queue timeout should be disabled, got: %v", err) + } + if elapsed < 25*time.Millisecond { + t.Fatalf("returned before context deadline: %v", elapsed) + } + }) +} + +func TestGroupPolicyFromStoreZeroQueueTimeoutDisablesTimeout(t *testing.T) { + store := edgenode.NewNodeStore() + store.Add(&edgenode.NodeRecord{ + ID: "node-policy-zero-timeout", + Adapters: config.AdaptersConf{ + OpenAICompatInstances: []config.OpenAICompatInstanceConf{ + { + Name: "provider-zero-timeout", + Enabled: true, + Capacity: 1, + MaxQueue: 16, + QueueTimeoutMS: 0, + }, + }, + }, + }) + entries := []*edgenode.NodeEntry{{NodeID: "node-policy-zero-timeout"}} + + policy := groupPolicyFromStore(store, entries, "provider-zero-timeout", "") + + if policy.maxQueue != 16 { + t.Fatalf("maxQueue: got %d, want 16", policy.maxQueue) + } + if !policy.queueTimeoutSet { + t.Fatalf("queueTimeoutSet: got false, want true") + } + if policy.queueTimeout != 0 { + t.Fatalf("queueTimeout: got %v, want no timeout", policy.queueTimeout) + } } // TestModelQueueProviderInflightSelectionBeatsPriority verifies that in_flight diff --git a/apps/edge/internal/service/run_dispatch.go b/apps/edge/internal/service/run_dispatch.go index f947c49..e8d3f95 100644 --- a/apps/edge/internal/service/run_dispatch.go +++ b/apps/edge/internal/service/run_dispatch.go @@ -443,13 +443,10 @@ func groupPolicyFromRequestOrStore(req SubmitRunRequest, store *edgenode.NodeSto if maxQueue <= 0 { maxQueue = defaultGroupMaxQueue } - queueTimeout := defaultQueueTimeout - if req.QueueTimeoutMS > 0 { - queueTimeout = time.Duration(req.QueueTimeoutMS) * time.Millisecond - } return groupPolicy{ - maxQueue: maxQueue, - queueTimeout: queueTimeout, + maxQueue: maxQueue, + queueTimeout: time.Duration(req.QueueTimeoutMS) * time.Millisecond, + queueTimeoutSet: true, } } return groupPolicyFromStore(store, entries, req.Adapter, req.Target) @@ -469,17 +466,15 @@ func groupPolicyFromStore(store *edgenode.NodeStore, entries []*edgenode.NodeEnt } if res.maxQueue > 0 || res.queueTimeoutMS > 0 { p := groupPolicy{ - maxQueue: positiveOr(res.maxQueue, defaultGroupMaxQueue), - queueTimeout: time.Duration(positiveOr(res.queueTimeoutMS, 0)) * time.Millisecond, - } - if p.queueTimeout <= 0 { - p.queueTimeout = defaultQueueTimeout + maxQueue: positiveOr(res.maxQueue, defaultGroupMaxQueue), + queueTimeout: time.Duration(res.queueTimeoutMS) * time.Millisecond, + queueTimeoutSet: true, } return p } } } - return groupPolicy{maxQueue: defaultGroupMaxQueue, queueTimeout: defaultQueueTimeout} + return groupPolicy{maxQueue: defaultGroupMaxQueue, queueTimeout: defaultQueueTimeout, queueTimeoutSet: true} } // providerCanServe checks whether the provider advertises the served model in @@ -640,8 +635,9 @@ func (s *Service) resolveProviderPoolCandidates(req SubmitRunRequest, store *edg }) if !policySet && (prov.MaxQueue > 0 || prov.QueueTimeoutMS > 0) { policy = groupPolicy{ - maxQueue: positiveOr(prov.MaxQueue, defaultGroupMaxQueue), - queueTimeout: time.Duration(positiveOr(prov.QueueTimeoutMS, int(defaultQueueTimeout/time.Millisecond))) * time.Millisecond, + maxQueue: positiveOr(prov.MaxQueue, defaultGroupMaxQueue), + queueTimeout: time.Duration(prov.QueueTimeoutMS) * time.Millisecond, + queueTimeoutSet: true, } policySet = true } @@ -652,7 +648,7 @@ func (s *Service) resolveProviderPoolCandidates(req SubmitRunRequest, store *edg return nil, groupPolicy{}, fmt.Errorf("no connected nodes support provider pool model %q", req.ModelGroupKey) } if !policySet { - policy = groupPolicy{maxQueue: defaultGroupMaxQueue, queueTimeout: defaultQueueTimeout} + policy = groupPolicy{maxQueue: defaultGroupMaxQueue, queueTimeout: defaultQueueTimeout, queueTimeoutSet: true} } return candidates, policy, nil } diff --git a/packages/go/config/config.go b/packages/go/config/config.go index 36535d0..5b295d9 100644 --- a/packages/go/config/config.go +++ b/packages/go/config/config.go @@ -163,7 +163,8 @@ type NodeProviderConf struct { LongContextCapacity int `mapstructure:"long_context_capacity" yaml:"long_context_capacity,omitempty"` // MaxQueue is the maximum queue depth. MaxQueue int `mapstructure:"max_queue" yaml:"max_queue,omitempty"` - // QueueTimeoutMS is the queue timeout in milliseconds. + // QueueTimeoutMS is the queue timeout in milliseconds. Zero means no queue + // timeout when a queue policy is configured. QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms,omitempty"` // Priority is the provider dispatch priority. Lower values are preferred // when in_flight counts are equal. Must be non-negative; zero is the @@ -481,7 +482,7 @@ type OllamaInstanceConf struct { ContextSize int `mapstructure:"context_size" yaml:"context_size"` Capacity int `mapstructure:"capacity" yaml:"capacity"` MaxQueue int `mapstructure:"max_queue" yaml:"max_queue"` - QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` + QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` // zero means no queue timeout RequestTimeoutMS int `mapstructure:"request_timeout_ms" yaml:"request_timeout_ms"` } @@ -492,7 +493,7 @@ type VllmInstanceConf struct { Endpoint string `mapstructure:"endpoint" yaml:"endpoint"` Capacity int `mapstructure:"capacity" yaml:"capacity"` MaxQueue int `mapstructure:"max_queue" yaml:"max_queue"` - QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` + QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` // zero means no queue timeout RequestTimeoutMS int `mapstructure:"request_timeout_ms" yaml:"request_timeout_ms"` } @@ -503,7 +504,7 @@ type OpenAICompatConf struct { Headers map[string]string `mapstructure:"headers" yaml:"headers"` Capacity int `mapstructure:"capacity" yaml:"capacity"` MaxQueue int `mapstructure:"max_queue" yaml:"max_queue"` - QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` + QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` // zero means no queue timeout RequestTimeoutMS int `mapstructure:"request_timeout_ms" yaml:"request_timeout_ms"` } @@ -515,7 +516,7 @@ type OpenAICompatInstanceConf struct { Headers map[string]string `mapstructure:"headers" yaml:"headers"` Capacity int `mapstructure:"capacity" yaml:"capacity"` MaxQueue int `mapstructure:"max_queue" yaml:"max_queue"` - QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` + QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` // zero means no queue timeout RequestTimeoutMS int `mapstructure:"request_timeout_ms" yaml:"request_timeout_ms"` } @@ -525,7 +526,7 @@ type OllamaConf struct { ContextSize int `mapstructure:"context_size" yaml:"context_size"` Capacity int `mapstructure:"capacity" yaml:"capacity"` MaxQueue int `mapstructure:"max_queue" yaml:"max_queue"` - QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` + QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` // zero means no queue timeout RequestTimeoutMS int `mapstructure:"request_timeout_ms" yaml:"request_timeout_ms"` } @@ -534,7 +535,7 @@ type VllmConf struct { Endpoint string `mapstructure:"endpoint" yaml:"endpoint"` Capacity int `mapstructure:"capacity" yaml:"capacity"` MaxQueue int `mapstructure:"max_queue" yaml:"max_queue"` - QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` + QueueTimeoutMS int `mapstructure:"queue_timeout_ms" yaml:"queue_timeout_ms"` // zero means no queue timeout RequestTimeoutMS int `mapstructure:"request_timeout_ms" yaml:"request_timeout_ms"` }