From 4684390bc65890d473ade87ca6e1d3f82c2bd63f Mon Sep 17 00:00:00 2001 From: zensi-dev <47974166+zensi-dev@users.noreply.github.com> Date: Sat, 18 Jul 2026 22:07:07 +0530 Subject: [PATCH] feat(acp): add Grok CLI driver (#1745) --- apps/backend/internal/agentctl/AGENTS.md | 17 +- .../internal/agentctl/acpcompat/grok.go | 328 ++++++++++ .../internal/agentctl/acpcompat/grok_test.go | 33 + .../server/adapter/transport/acp/adapter.go | 27 +- .../adapter/transport/acp/adapter_prompt.go | 2 +- .../adapter/transport/acp/adapter_session.go | 244 ++++++-- .../adapter/transport/acp/adapter_updates.go | 61 +- .../server/adapter/transport/acp/dialect.go | 118 ++++ .../adapter/transport/acp/dialect_grok.go | 202 ++++++ .../transport/acp/dialect_grok_test.go | 585 ++++++++++++++++++ .../adapter/transport/acp/usage_test.go | 17 + .../agentctl/server/utility/acp_executor.go | 56 +- .../server/utility/acp_executor_test.go | 63 +- .../0044-acp-agent-compatibility-dialects.md | 43 ++ docs/decisions/INDEX.md | 1 + 15 files changed, 1735 insertions(+), 62 deletions(-) create mode 100644 apps/backend/internal/agentctl/acpcompat/grok.go create mode 100644 apps/backend/internal/agentctl/acpcompat/grok_test.go create mode 100644 apps/backend/internal/agentctl/server/adapter/transport/acp/dialect.go create mode 100644 apps/backend/internal/agentctl/server/adapter/transport/acp/dialect_grok.go create mode 100644 apps/backend/internal/agentctl/server/adapter/transport/acp/dialect_grok_test.go create mode 100644 docs/decisions/0044-acp-agent-compatibility-dialects.md diff --git a/apps/backend/internal/agentctl/AGENTS.md b/apps/backend/internal/agentctl/AGENTS.md index 033051a6a3..cdecb62a16 100644 --- a/apps/backend/internal/agentctl/AGENTS.md +++ b/apps/backend/internal/agentctl/AGENTS.md @@ -35,7 +35,22 @@ Protocol adapters in `server/adapter/transport/` normalize different agent CLIs: - `process.Manager` owns subprocess, wires stdio to adapter - Factory pattern in `server/adapter/factory.go` selects adapter by agent type -The `acp` transport is split by concern across `adapter_*.go` files: `adapter.go` (core/lifecycle), `adapter_session.go` (initialize/new/load/resume), `adapter_prompt.go` (prompt/cancel), `adapter_updates.go` (`session/update` notification fan-out), `adapter_tools.go` (`convertToolCallUpdate` / `convertToolCallResultUpdate` → normalized payloads), `adapter_permissions.go`, and `adapter_helpers.go`. Tool-call conversion lives in `adapter_tools.go`, not `adapter.go`. +The `acp` transport is split by concern across `adapter_*.go` files: `adapter.go` (core/lifecycle), `adapter_session.go` (initialize/new/load/resume), `adapter_prompt.go` (prompt/cancel), `adapter_updates.go` (`session/update` notification fan-out), `adapter_tools.go` (`convertToolCallUpdate` / `convertToolCallResultUpdate` -> normalized payloads), `adapter_permissions.go`, and `adapter_helpers.go`. Agent-specific ACP extensions use the package-private `acpDialect` function table in `dialect.go`; keep observed wire translation in `dialect_.go`. Dialect hooks return normalized data or request descriptions and never receive `*Adapter` or execute RPCs. Shared capability normalization used by both live sessions and utility probes belongs in `internal/agentctl/acpcompat/`. Tool-call conversion lives in `adapter_tools.go`, not `adapter.go`. See ADR-0043. + +### Grok ACP dialect (`dialect_grok.go`) + +Grok exposes some model, config, and usage surfaces through implementation-specific ACP metadata. Its dialect normalizes them in the backend so shared ACP paths and the frontend stay provider-agnostic: + +| Grok wire | Kandev shape | +|---|---| +| Legacy `models` catalog | model `ConfigOption` used by generic frontend selectors | +| Model `_meta.supportsReasoningEffort` / `reasoningEfforts` / `reasoningEffort` | gated `reasoning_effort` select with category `thought_level` | +| `SetConfigOption(model, …)` or `(reasoning_effort, …)` | Grok has **no** `session/set_config_option`. Rewrite both to `session/set_model` (effort keeps current modelId + `_meta.reasoningEffort`; model switch may carry prior effort in meta when still supported). Frontend always uses set_config_option when a model ConfigOption exists. | +| Mid-session `MODEL_SWITCH_INCOMPATIBLE_AGENT` | return Grok's actionable error; user starts a new session explicitly | +| Notification `_meta.totalTokens` + model `_meta.totalContextTokens` | `EventTypeContextWindow` (used/size/remaining). Compaction may decrease used; clear used on model switch. | +| Prompt `_meta.usage` | input/output/cache/total plus `reasoningTokens` as `PromptUsage.ThoughtTokens`; sibling flat fields are not whole-turn totals | + +Grok ACP currently exposes neither per-turn cost nor subscription quota/reset values. Do not estimate them or treat the optional subscription tier label as usage data. ## ACP Protocol diff --git a/apps/backend/internal/agentctl/acpcompat/grok.go b/apps/backend/internal/agentctl/acpcompat/grok.go new file mode 100644 index 0000000000..b47f61c2bc --- /dev/null +++ b/apps/backend/internal/agentctl/acpcompat/grok.go @@ -0,0 +1,328 @@ +// Package acpcompat normalizes observed agent-specific ACP extensions into +// provider-neutral Kandev session shapes. +package acpcompat + +import ( + "fmt" + "math" + "strings" + + "github.com/kandev/kandev/internal/agentctl/types/streams" +) + +const ( + GrokAgentID = "grok-acp" + + GrokReasoningEffortOptionID = "reasoning_effort" + GrokReasoningEffortMetaKey = "reasoningEffort" + + grokReasoningCategory = "thought_level" + grokReasoningName = "Reasoning Effort" + grokSupportsReasoning = "supportsReasoningEffort" + grokReasoningEfforts = "reasoningEfforts" + grokTotalContextTokens = "totalContextTokens" + grokTotalTokens = "totalTokens" + modelOptionID = "model" +) + +// Model is the provider-neutral subset of an ACP model entry needed by +// compatibility normalization. +type Model struct { + ID string + Name string + Meta map[string]any +} + +// ContextWindow is a context-usage sample derived from agent metadata. +type ContextWindow struct { + Used int64 + Size int64 +} + +// NormalizeSessionConfig merges observed Grok model metadata into canonical +// session options without discarding typed or _meta options already parsed by +// the caller. Other agents pass through unchanged. +func NormalizeSessionConfig( + agentID string, + base []streams.ConfigOption, + models []Model, + currentModelID string, +) []streams.ConfigOption { + if agentID != GrokAgentID { + return base + } + out := append([]streams.ConfigOption(nil), base...) + if currentModelID == "" { + currentModelID = currentModelFromConfig(out) + } + if currentModelID == "" && len(models) == 1 { + currentModelID = models[0].ID + } + if !hasModelOption(out) { + if modelOption := buildModelOption(models, currentModelID); modelOption != nil { + out = append(out, *modelOption) + } + } + if !hasOption(out, GrokReasoningEffortOptionID) { + if effort := buildReasoningOption(models, currentModelID, ""); effort != nil { + out = append(out, *effort) + } + } + return out +} + +// ConfigAfterModelChange rebuilds model-dependent Grok options after a model +// switch. Other agents pass through unchanged. +func ConfigAfterModelChange( + agentID string, + config []streams.ConfigOption, + models []Model, + modelID string, +) []streams.ConfigOption { + if agentID != GrokAgentID { + return config + } + previousEffort := "" + out := make([]streams.ConfigOption, 0, len(config)) + for _, option := range config { + if option.ID == GrokReasoningEffortOptionID { + previousEffort = option.CurrentValue + continue + } + out = append(out, option) + } + if effort := buildReasoningOption(models, modelID, previousEffort); effort != nil { + out = append(out, *effort) + } + return out +} + +// ValidateGrokReasoningEffort resolves a UI selection to Grok's canonical +// wire value for the active model. +func ValidateGrokReasoningEffort(models []Model, currentModelID, effort string) (string, error) { + if currentModelID == "" { + return "", fmt.Errorf("no active model to apply reasoning effort") + } + info := findModel(models, currentModelID) + if info == nil { + return "", fmt.Errorf("unknown model %q; cannot set reasoning effort", currentModelID) + } + if !getBool(info.Meta, grokSupportsReasoning) { + return "", fmt.Errorf("model %q does not support reasoning effort", currentModelID) + } + options, _ := parseReasoningEfforts(info.Meta) + if len(options) == 0 { + return "", fmt.Errorf("model %q does not advertise reasoning effort values", currentModelID) + } + if value := matchEffort(effort, options); value != "" { + return value, nil + } + offered := make([]string, 0, len(options)) + for _, option := range options { + offered = append(offered, option.Value) + } + return "", fmt.Errorf("unsupported reasoning effort %q for model %q; use one of: %s", + effort, currentModelID, strings.Join(offered, ", ")) +} + +// GrokContextWindow reads Grok's notification total and active model context +// size as one immutable sample. +func GrokContextWindow( + meta map[string]any, + models []Model, + config []streams.ConfigOption, +) (ContextWindow, bool) { + used, ok := nonNegativeInt64(meta, grokTotalTokens) + if !ok { + return ContextWindow{}, false + } + model := findModel(models, currentModelFromConfig(config)) + if model == nil { + return ContextWindow{}, false + } + size, ok := positiveInt64(model.Meta, grokTotalContextTokens) + if !ok { + return ContextWindow{}, false + } + return ContextWindow{Used: used, Size: size}, true +} + +// GrokReasoningTokens reads Grok's nested prompt usage extension. +func GrokReasoningTokens(meta map[string]any) int64 { + usage, _ := meta["usage"].(map[string]any) + value, _ := nonNegativeInt64(usage, "reasoningTokens") + return value +} + +func buildModelOption(models []Model, currentModelID string) *streams.ConfigOption { + values := make([]streams.ConfigOptionValue, 0, len(models)) + for _, model := range models { + name := model.Name + if name == "" { + name = model.ID + } + values = append(values, streams.ConfigOptionValue{Value: model.ID, Name: name}) + } + if len(values) == 0 { + return nil + } + return &streams.ConfigOption{ + Type: "select", + ID: modelOptionID, + Category: modelOptionID, + Name: "Model", + CurrentValue: currentModelID, + Options: values, + } +} + +func buildReasoningOption(models []Model, modelID, selected string) *streams.ConfigOption { + model := findModel(models, modelID) + if model == nil || !getBool(model.Meta, grokSupportsReasoning) { + return nil + } + options, defaultEffort := parseReasoningEfforts(model.Meta) + if len(options) == 0 { + return nil + } + current := resolveEffort(selected, options, defaultEffort, model.Meta) + return &streams.ConfigOption{ + Type: "select", + ID: GrokReasoningEffortOptionID, + Category: grokReasoningCategory, + Name: grokReasoningName, + CurrentValue: current, + Options: options, + } +} + +func parseReasoningEfforts(meta map[string]any) ([]streams.ConfigOptionValue, string) { + items, ok := meta[grokReasoningEfforts].([]any) + if !ok { + return nil, "" + } + var out []streams.ConfigOptionValue + defaultEffort := "" + for _, item := range items { + switch value := item.(type) { + case string: + value = strings.TrimSpace(value) + if value != "" { + out = append(out, streams.ConfigOptionValue{Value: value, Name: humanize(value)}) + } + case map[string]any: + wireValue := getString(value, "value") + if wireValue == "" { + wireValue = getString(value, "id") + } + if wireValue == "" { + continue + } + name := getString(value, "label") + if name == "" { + name = humanize(wireValue) + } + out = append(out, streams.ConfigOptionValue{Value: wireValue, Name: name}) + if getBool(value, "default") { + defaultEffort = wireValue + } + } + } + return out, defaultEffort +} + +func resolveEffort(selected string, options []streams.ConfigOptionValue, fallback string, meta map[string]any) string { + for _, candidate := range []string{selected, getString(meta, GrokReasoningEffortMetaKey), fallback, "medium"} { + if value := matchEffort(candidate, options); value != "" { + return value + } + } + return options[0].Value +} + +func matchEffort(token string, options []streams.ConfigOptionValue) string { + for _, option := range options { + if option.Value == token || option.Name == token { + return option.Value + } + } + return "" +} + +func currentModelFromConfig(options []streams.ConfigOption) string { + for _, option := range options { + if option.ID == modelOptionID || option.Category == modelOptionID { + return option.CurrentValue + } + } + return "" +} + +func hasModelOption(options []streams.ConfigOption) bool { + for _, option := range options { + if option.ID == modelOptionID || option.Category == modelOptionID { + return true + } + } + return false +} + +func hasOption(options []streams.ConfigOption, id string) bool { + for _, option := range options { + if option.ID == id { + return true + } + } + return false +} + +func findModel(models []Model, id string) *Model { + for i := range models { + if models[i].ID == id { + return &models[i] + } + } + return nil +} + +func humanize(value string) string { + if value == "" { + return "" + } + return strings.ToUpper(value[:1]) + value[1:] +} + +func getString(meta map[string]any, key string) string { + value, _ := meta[key].(string) + return value +} + +func getBool(meta map[string]any, key string) bool { + value, _ := meta[key].(bool) + return value +} + +func positiveInt64(meta map[string]any, key string) (int64, bool) { + value, ok := nonNegativeInt64(meta, key) + return value, ok && value > 0 +} + +func nonNegativeInt64(meta map[string]any, key string) (int64, bool) { + if meta == nil { + return 0, false + } + switch value := meta[key].(type) { + case float64: + if value < 0 || math.IsNaN(value) || math.IsInf(value, 0) || + value != math.Trunc(value) || value >= math.Exp2(63) { + return 0, false + } + return int64(value), true + case int64: + return value, value >= 0 + case int: + return int64(value), value >= 0 + default: + return 0, false + } +} diff --git a/apps/backend/internal/agentctl/acpcompat/grok_test.go b/apps/backend/internal/agentctl/acpcompat/grok_test.go new file mode 100644 index 0000000000..da214e3804 --- /dev/null +++ b/apps/backend/internal/agentctl/acpcompat/grok_test.go @@ -0,0 +1,33 @@ +package acpcompat + +import ( + "math" + "testing" +) + +func TestNonNegativeInt64RejectsInvalidFloatTokenCounts(t *testing.T) { + tests := []struct { + name string + value any + want int64 + ok bool + }{ + {name: "zero", value: float64(0), want: 0, ok: true}, + {name: "whole number", value: float64(42), want: 42, ok: true}, + {name: "largest representable below int64 limit", value: math.Nextafter(math.Exp2(63), 0), want: 9_223_372_036_854_774_784, ok: true}, + {name: "negative", value: float64(-1)}, + {name: "fractional", value: 1.5}, + {name: "nan", value: math.NaN()}, + {name: "positive infinity", value: math.Inf(1)}, + {name: "int64 overflow", value: math.Exp2(63)}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, ok := nonNegativeInt64(map[string]any{"tokens": tt.value}, "tokens") + if ok != tt.ok || got != tt.want { + t.Fatalf("nonNegativeInt64() = (%d, %t), want (%d, %t)", got, ok, tt.want, tt.ok) + } + }) + } +} diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go index 715686d2f8..96fcfde9d4 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter.go @@ -228,6 +228,15 @@ type Adapter struct { // the options list when the model is changed. availableConfigOptions []streams.ConfigOption + dialect acpDialect + + // Session configuration changes are serialized across model and option + // RPCs. configGeneration is incremented when a change begins so an older + // completion cannot overwrite a newer selection. + configChangeMu sync.Mutex + configGeneration uint64 + contextSamples map[string]contextWindowSample + // Synchronization mu sync.RWMutex closed bool @@ -293,12 +302,14 @@ func NewAdapter(cfg *shared.Config, log *logger.Logger) *Adapter { logger: l, agentID: cfg.AgentID, normalizer: NewNormalizer(cfg.AgentID), + dialect: newACPDialect(cfg.AgentID), updatesCh: make(chan AgentEvent, 100), notifQueue: make(chan notifWork, notifQueueCapacity), activeToolCalls: make(map[string]*streams.NormalizedPayload), activeMonitors: make(map[string]map[string]string), pendingWakeups: make(map[string]*pendingWakeup), usageBySession: make(map[string]*usageTracker), + contextSamples: make(map[string]contextWindowSample), attachMgr: shared.NewAttachmentManager(cfg.WorkDir, l.Zap()), promptGate: make(chan struct{}, 1), asyncTurnFinalizers: make(map[string]*asyncTurnFinalizer), @@ -473,20 +484,22 @@ func (a *Adapter) SetPermissionHandler(handler PermissionHandler) { func (a *Adapter) sendUpdate(event AgentEvent) { a.mu.RLock() defer a.mu.RUnlock() - a.sendUpdateLocked(event) + if !a.sendUpdateLocked(event) && !a.closed { + a.logger.Warn("updates channel full, dropping event", zap.String("type", event.Type)) + } } -// sendUpdateLocked queues an event while the caller holds a.mu for reading or -// writing. Session-bound emitters use it to make their session check, cache -// mutation, and convergence event atomic with session replacement. -func (a *Adapter) sendUpdateLocked(event AgentEvent) { +// sendUpdateLocked enqueues an event without acquiring a.mu. Callers must +// hold a.mu for reading or writing. +func (a *Adapter) sendUpdateLocked(event AgentEvent) bool { if a.closed { - return + return false } select { case a.updatesCh <- event: + return true default: - a.logger.Warn("updates channel full, dropping event", zap.String("type", event.Type)) + return false } } diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt.go index 3ec4fae97f..81ef2bf5c7 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_prompt.go @@ -183,7 +183,7 @@ func (a *Adapter) sendPrompt( zap.String("session_id", sessionID), zap.String("stop_reason", stopReason)) a.cancelAsyncTurnComplete(sessionID) - usage := extractUsage(&resp) + usage := a.dialect.promptUsage(extractUsage(&resp), resp.Meta) // codex-acp emits no per-turn usage frame, only cumulative // usage_update.used. Fall back to the inferred delta so the office // cost subscriber sees at least an approximate input count. The diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_session.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_session.go index 1d2ceb94f7..75e8a07c17 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_session.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_session.go @@ -55,6 +55,8 @@ func (a *Adapter) NewSession(ctx context.Context, mcpServers []types.McpServer) a.mu.Lock() a.sessionID = string(resp.SessionId) sessionID := a.sessionID + a.configGeneration++ + clear(a.contextSamples) // Reset session-scoped model caches before computing the new session's // state so a session without a model surface can't reuse the previous // session's models / configOptions for validation in SetModel. @@ -134,14 +136,6 @@ func hasModelConfigOption(options []streams.ConfigOption) bool { return false } -func sessionConfigOptions(meta map[string]any, acpConfigOptions []acp.SessionConfigOption) []streams.ConfigOption { - configOptions := convertACPConfigOptions(acpConfigOptions) - if len(configOptions) > 0 { - return configOptions - } - return extractConfigOptions(meta) -} - // effectiveMcpCapabilities applies the adapter's AssumeMcpSse/AssumeMcpHttp // config overrides on top of the capabilities advertised by the agent during // the ACP handshake. Some agents (e.g. Auggie) support SSE/HTTP MCP transports @@ -320,6 +314,8 @@ func (a *Adapter) LoadSession(ctx context.Context, sessionID string, mcpServers a.mu.Lock() a.sessionID = sessionID + a.configGeneration++ + clear(a.contextSamples) // Reset session-scoped model caches so a load that lands on a session // without a model surface can't reuse the previous session's data. a.availableModels = nil @@ -426,9 +422,9 @@ func (a *Adapter) emitInitialModeState(modes *acp.SessionModeState) { // emitSessionModels emits a session_models event from the session response. func (a *Adapter) emitSessionModels(sessionID string, models *sessionModelState, meta map[string]any, acpConfigOptions []acp.SessionConfigOption) { currentModelID := models.CurrentModelId - // Prefer typed config options from the response; fall back to _meta - // extraction for older agents. - configOptions := sessionConfigOptions(meta, acpConfigOptions) + configOptions := a.dialect.sessionConfigOptions( + meta, acpConfigOptions, models.AvailableModels, currentModelID, + ) // Fallback: if the SDK didn't parse currentModelId (some agents omit it), // take the model-shaped configOption's CurrentValue verbatim. We @@ -473,7 +469,13 @@ func (a *Adapter) emitSessionModels(sessionID string, models *sessionModelState, // — this prevents a downstream consumer that reads ConfigOptions[model] // .CurrentValue (codex-style agents surface the current model there) from // disagreeing with the CurrentModelID emitted on the same event. -func (a *Adapter) emitSetModelEvent(sessionID, modelID string, cachedModels []modelInfo, cachedConfig []streams.ConfigOption) { +func (a *Adapter) emitSetModelEvent( + sessionID string, + modelID string, + cachedModels []modelInfo, + cachedConfig []streams.ConfigOption, + expectedGeneration ...uint64, +) { outConfig := cachedConfig if len(cachedConfig) > 0 { // Shallow copy: only CurrentValue (a string) is rewritten below, so @@ -488,7 +490,24 @@ func (a *Adapter) emitSetModelEvent(sessionID, modelID string, cachedModels []mo } } } - + outConfig = a.dialect.modelConfigAfterChange(outConfig, cachedModels, modelID) + a.mu.Lock() + if a.sessionID != sessionID || + (len(expectedGeneration) > 0 && a.configGeneration != expectedGeneration[0]) { + a.mu.Unlock() + return + } + if len(outConfig) > 0 { + // Refresh the cached config options so a subsequent SetConfigOption + // (e.g. user toggles reasoning effort after switching model) doesn't + // reuse the stale model CurrentValue from session/new and clobber the + // just-applied model in the convergence event. + a.availableConfigOptions = outConfig + } + if tracker := a.usageBySession[sessionID]; tracker != nil { + tracker.maxSize = 0 + } + delete(a.contextSamples, sessionID) event := AgentEvent{ Type: streams.EventTypeSessionModels, SessionID: sessionID, @@ -496,13 +515,17 @@ func (a *Adapter) emitSetModelEvent(sessionID, modelID string, cachedModels []mo SessionModels: convertSessionModels(cachedModels), ConfigOptions: outConfig, } - if !a.emitConfigUpdateForSession(sessionID, outConfig, event) { - return - } + sent := a.sendUpdateLocked(event) + closed := a.closed + a.mu.Unlock() + a.logger.Info("emitting session_models convergence event after SetModel", zap.String("session_id", sessionID), zap.String("model_id", modelID), ) + if !sent && !closed { + a.logger.Warn("updates channel full, dropping event", zap.String("type", event.Type)) + } } // currentModelFromConfig returns the CurrentValue of the model-shaped @@ -567,6 +590,53 @@ func (a *Adapter) SetModel(ctx context.Context, modelID string) error { if conn == nil { return fmt.Errorf("adapter not initialized") } + return a.setModelWithConn(ctx, conn, sessionID, modelID, available, cachedConfig) +} + +func (a *Adapter) setModelWithConn( + ctx context.Context, + conn sessionmodel.SDKConn, + sessionID string, + modelID string, + available []modelInfo, + cachedConfig []streams.ConfigOption, +) error { + generation := a.beginConfigChange() + a.configChangeMu.Lock() + defer a.configChangeMu.Unlock() + if !a.isCurrentConfigChange(sessionID, generation) { + return nil + } + + rpc, err := a.dialect.modelRequest(dialectConfigChange{ + sessionID: sessionID, + value: modelID, + models: available, + config: cachedConfig, + }) + if err != nil { + return err + } + if rpc != nil { + _, err = conn.UnstableSetSessionModel(ctx, rpc.request) + if !a.isCurrentConfigChange(sessionID, generation) { + return nil + } + if err != nil { + return rpc.formatError(err) + } + a.finalizeSetModel( + sessionmodel.MethodSetModel, + sessionID, + modelID, + available, + cachedConfig, + configOptionIDModel, + nil, + generation, + ) + return nil + } // Validate model exists in the agent's available models (if known). if len(available) > 0 { @@ -576,10 +646,13 @@ func (a *Adapter) SetModel(ctx context.Context, modelID string) error { } method, responseConfig, configID, err := applySessionModelWithConfigOptions(ctx, conn, sessionID, modelID, cachedConfig) + if !a.isCurrentConfigChange(sessionID, generation) { + return nil + } if err != nil { return fmt.Errorf("set session model failed via %s: %w", method, err) } - a.finalizeSetModel(method, sessionID, modelID, available, cachedConfig, configID, responseConfig) + a.finalizeSetModel(method, sessionID, modelID, available, cachedConfig, configID, responseConfig, generation) return nil } @@ -604,16 +677,33 @@ func (a *Adapter) finalizeSetModel( cachedConfig []streams.ConfigOption, configID string, responseConfig []acp.SessionConfigOption, + expectedGeneration ...uint64, ) { if method == sessionmodel.MethodNone { return } - a.resetContextWindowMaxSize(sessionID) if len(responseConfig) > 0 { - a.emitAuthoritativeConfigOptions(sessionID, configID, responseConfig, available) + a.emitAuthoritativeConfigOptions(sessionID, configID, responseConfig, available, expectedGeneration...) return } - a.emitSetModelEvent(sessionID, modelID, available, cachedConfig) + a.emitSetModelEvent(sessionID, modelID, available, cachedConfig, expectedGeneration...) +} + +func (a *Adapter) beginConfigChange() uint64 { + a.mu.Lock() + defer a.mu.Unlock() + a.configGeneration++ + return a.configGeneration +} + +func (a *Adapter) isCurrentConfigChange(sessionID string, generation uint64) bool { + a.mu.RLock() + defer a.mu.RUnlock() + return a.configChangeIsCurrentLocked(sessionID, generation) +} + +func (a *Adapter) configChangeIsCurrentLocked(sessionID string, generation uint64) bool { + return a.sessionID == sessionID && a.configGeneration == generation } func applySessionModel( @@ -713,6 +803,48 @@ func (a *Adapter) SetConfigOption(ctx context.Context, configID, value string) e return fmt.Errorf("no active session: call NewSession before SetConfigOption") } + generation := a.beginConfigChange() + a.configChangeMu.Lock() + defer a.configChangeMu.Unlock() + if !a.isCurrentConfigChange(sessionID, generation) { + return nil + } + + rpc, err := a.dialect.configRequest(dialectConfigChange{ + sessionID: sessionID, + configID: configID, + value: value, + models: cachedModels, + config: cachedConfig, + }) + if err != nil { + return err + } + if rpc != nil { + _, err = conn.UnstableSetSessionModel(ctx, rpc.request) + if !a.isCurrentConfigChange(sessionID, generation) { + return nil + } + if err != nil { + return rpc.formatError(err) + } + if isModelConfigID(configID, cachedConfig) { + a.finalizeSetModel( + sessionmodel.MethodSetModel, + sessionID, + value, + cachedModels, + cachedConfig, + configID, + nil, + generation, + ) + } else { + a.emitSetConfigOptionEvent(sessionID, configID, value, cachedModels, cachedConfig, generation) + } + return nil + } + resp, err := conn.SetSessionConfigOption(ctx, acp.SetSessionConfigOptionRequest{ ValueId: &acp.SetSessionConfigOptionValueId{ SessionId: acp.SessionId(sessionID), @@ -720,6 +852,9 @@ func (a *Adapter) SetConfigOption(ctx context.Context, configID, value string) e Value: acp.SessionConfigValueId(value), }, }) + if !a.isCurrentConfigChange(sessionID, generation) { + return nil + } if err != nil { return fmt.Errorf("set session config option failed: %w", err) } @@ -730,13 +865,13 @@ func (a *Adapter) SetConfigOption(ctx context.Context, configID, value string) e return nil } if len(resp.ConfigOptions) > 0 { - a.emitAuthoritativeConfigOptions(sessionID, configID, resp.ConfigOptions, cachedModels) + a.emitAuthoritativeConfigOptions(sessionID, configID, resp.ConfigOptions, cachedModels, generation) return nil } if isModelConfigID(configID, cachedConfig) { - a.emitSetModelEvent(sessionID, value, cachedModels, cachedConfig) + a.emitSetModelEvent(sessionID, value, cachedModels, cachedConfig, generation) } else { - a.emitSetConfigOptionEvent(sessionID, configID, value, cachedModels, cachedConfig) + a.emitSetConfigOptionEvent(sessionID, configID, value, cachedModels, cachedConfig, generation) } return nil } @@ -746,9 +881,10 @@ func (a *Adapter) emitAuthoritativeConfigOptions( configID string, options []acp.SessionConfigOption, cachedModels []modelInfo, + expectedGeneration ...uint64, ) { configOptions := convertACPConfigOptions(options) - a.emitConfigUpdateForSession(sessionID, configOptions, AgentEvent{ + event := AgentEvent{ Type: streams.EventTypeSessionModels, SessionID: sessionID, CurrentModelID: currentModelFromConfig(configOptions), @@ -758,29 +894,43 @@ func (a *Adapter) emitAuthoritativeConfigOptions( "config_options_source": "provider_response", "config_options_config_id": configID, }, - }) -} - -func (a *Adapter) emitConfigUpdateForSession( - sessionID string, - configOptions []streams.ConfigOption, - event AgentEvent, -) bool { + } a.mu.Lock() - defer a.mu.Unlock() + if len(expectedGeneration) > 0 && !a.configChangeIsCurrentLocked(sessionID, expectedGeneration[0]) { + a.mu.Unlock() + return + } if a.sessionID != sessionID || a.closed { - return false + a.mu.Unlock() + return } a.availableConfigOptions = configOptions - a.sendUpdateLocked(event) - return true + if isModelConfigID(configID, configOptions) { + if tracker := a.usageBySession[sessionID]; tracker != nil { + tracker.maxSize = 0 + } + delete(a.contextSamples, sessionID) + } + sent := a.sendUpdateLocked(event) + closed := a.closed + a.mu.Unlock() + if !sent && !closed { + a.logger.Warn("updates channel full, dropping event", zap.String("type", event.Type)) + } } // emitSetConfigOptionEvent emits a session_models convergence event after a // non-model SetConfigOption RPC succeeds. The frontend uses this to keep the // option dropdowns in sync with the agent without waiting for an agent-driven // ConfigOptionUpdate. -func (a *Adapter) emitSetConfigOptionEvent(sessionID, configID, value string, cachedModels []modelInfo, cachedConfig []streams.ConfigOption) { +func (a *Adapter) emitSetConfigOptionEvent( + sessionID string, + configID string, + value string, + cachedModels []modelInfo, + cachedConfig []streams.ConfigOption, + expectedGeneration ...uint64, +) { if len(cachedConfig) == 0 { // In normal operation session/new populates availableConfigOptions // before the frontend can fire a SetConfigOption. Hitting this branch @@ -815,6 +965,16 @@ func (a *Adapter) emitSetConfigOptionEvent(sessionID, configID, value string, ca zap.String("config_id", configID), ) } + // Refresh the cached config options so consecutive SetConfigOption + // calls (or a follow-up SetModel) read the latest CurrentValues + // instead of the stale session/new snapshot. + a.mu.Lock() + if a.sessionID != sessionID || + (len(expectedGeneration) > 0 && a.configGeneration != expectedGeneration[0]) { + a.mu.Unlock() + return + } + a.availableConfigOptions = outConfig event := AgentEvent{ Type: streams.EventTypeSessionModels, SessionID: sessionID, @@ -822,14 +982,18 @@ func (a *Adapter) emitSetConfigOptionEvent(sessionID, configID, value string, ca SessionModels: convertSessionModels(cachedModels), ConfigOptions: outConfig, } - if !a.emitConfigUpdateForSession(sessionID, outConfig, event) { - return - } + sent := a.sendUpdateLocked(event) + closed := a.closed + a.mu.Unlock() + a.logger.Info("emitting session_models convergence event after SetConfigOption", zap.String("session_id", sessionID), zap.String("config_id", configID), zap.String("value", value), ) + if !sent && !closed { + a.logger.Warn("updates channel full, dropping event", zap.String("type", event.Type)) + } } // isModelConfigID reports whether configID identifies the model-shaped diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_updates.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_updates.go index 9dca743c61..d21543b326 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_updates.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/adapter_updates.go @@ -150,8 +150,12 @@ func (a *Adapter) handleACPUpdate(n acp.SessionNotification) { sessionID := string(n.SessionId) - event := a.convertNotification(n) - if event == nil { + suppressed := a.dialect.suppresses(n) + var event *AgentEvent + if !suppressed { + event = a.convertNotification(n) + } + if event == nil && !suppressed { // Try untyped updates not yet supported by the ACP SDK. event = a.tryConvertUntypedUpdate(rawData, sessionID) } @@ -161,11 +165,55 @@ func (a *Adapter) handleACPUpdate(n acp.SessionNotification) { event.Type, rawData, event) a.sendUpdate(*event) a.maybeScheduleAsyncTurnComplete(*event) - } else if updateJSON, err := json.Marshal(n.Update); err == nil { - a.logger.Warn("unhandled ACP session notification", - zap.String("session_id", sessionID), - zap.String("update_json", string(updateJSON))) + } else if !suppressed { + if updateJSON, err := json.Marshal(n.Update); err == nil { + a.logger.Warn("unhandled ACP session notification", + zap.String("session_id", sessionID), + zap.String("update_json", string(updateJSON))) + } } + + if !isLoading { + if supplemental := a.emitDialectContextWindow(sessionID, n.Meta); supplemental != nil { + shared.LogNormalizedEvent(shared.ProtocolACP, a.agentID, sessionID, supplemental) + } + } +} + +// emitDialectContextWindow derives and enqueues an agent-specific context +// sample while holding the same lock that protects the active session/model. +// This keeps a model switch from interleaving between size selection and event +// delivery. +func (a *Adapter) emitDialectContextWindow(sessionID string, meta map[string]any) *AgentEvent { + a.mu.Lock() + if a.closed || sessionID != a.sessionID { + a.mu.Unlock() + return nil + } + sample, ok := a.dialect.contextSample(meta, a.availableModels, a.availableConfigOptions) + if !ok || a.contextSamples[sessionID] == sample { + a.mu.Unlock() + return nil + } + remaining := max(sample.size-sample.used, 0) + event := AgentEvent{ + Type: streams.EventTypeContextWindow, + SessionID: sessionID, + ContextWindowSize: sample.size, + ContextWindowUsed: sample.used, + ContextWindowRemaining: remaining, + ContextEfficiency: float64(sample.used) / float64(sample.size) * 100, + } + sent := a.sendUpdateLocked(event) + if sent { + a.contextSamples[sessionID] = sample + } + a.mu.Unlock() + if !sent { + a.logger.Warn("updates channel full, dropping event", zap.String("type", event.Type)) + return nil + } + return &event } // convertNotification converts an ACP SessionNotification to an AgentEvent. @@ -324,6 +372,7 @@ func (a *Adapter) resetContextWindowMaxSize(sessionID string) { if tr := a.usageBySession[sessionID]; tr != nil { tr.maxSize = 0 } + delete(a.contextSamples, sessionID) } // consumeUsageDelta returns the delta of cumulative `used` since the diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/dialect.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/dialect.go new file mode 100644 index 0000000000..6a1d4042df --- /dev/null +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/dialect.go @@ -0,0 +1,118 @@ +package acp + +import ( + "github.com/coder/acp-go-sdk" + "github.com/kandev/kandev/internal/agentctl/types/streams" +) + +// acpDialect contains pure compatibility decisions for an agent's observed +// ACP implementation. The Adapter remains responsible for RPC execution, +// session state, serialization, and event delivery. +type acpDialect struct { + normalizeSessionConfig func( + []streams.ConfigOption, + []modelInfo, + string, + ) []streams.ConfigOption + configAfterModelChange func( + []streams.ConfigOption, + []modelInfo, + string, + ) []streams.ConfigOption + modelChangeRequest func(dialectConfigChange) (*dialectRPC, error) + configChangeRequest func(dialectConfigChange) (*dialectRPC, error) + suppressNotification func(acp.SessionNotification) bool + contextWindow func(map[string]any, []modelInfo, []streams.ConfigOption) (contextWindowSample, bool) + normalizePromptUsage func(*streams.PromptUsage, map[string]any) *streams.PromptUsage +} + +type dialectConfigChange struct { + sessionID string + configID string + value string + models []modelInfo + config []streams.ConfigOption +} + +type dialectRPC struct { + request acp.UnstableSetSessionModelRequest + formatError func(error) error +} + +type contextWindowSample struct { + used int64 + size int64 +} + +func newACPDialect(agentID string) acpDialect { + if agentID == grokAgentID { + return newGrokACPDialect() + } + return acpDialect{} +} + +func (d acpDialect) sessionConfigOptions( + meta map[string]any, + options []acp.SessionConfigOption, + models []modelInfo, + currentModelID string, +) []streams.ConfigOption { + base := convertACPConfigOptions(options) + if len(base) == 0 { + base = extractConfigOptions(meta) + } + if d.normalizeSessionConfig == nil { + return base + } + return d.normalizeSessionConfig(base, models, currentModelID) +} + +func (d acpDialect) modelConfigAfterChange( + config []streams.ConfigOption, + models []modelInfo, + modelID string, +) []streams.ConfigOption { + if d.configAfterModelChange == nil { + return config + } + return d.configAfterModelChange(config, models, modelID) +} + +func (d acpDialect) modelRequest(change dialectConfigChange) (*dialectRPC, error) { + if d.modelChangeRequest == nil { + return nil, nil + } + return d.modelChangeRequest(change) +} + +func (d acpDialect) configRequest(change dialectConfigChange) (*dialectRPC, error) { + if d.configChangeRequest == nil { + return nil, nil + } + return d.configChangeRequest(change) +} + +func (d acpDialect) suppresses(notification acp.SessionNotification) bool { + return d.suppressNotification != nil && d.suppressNotification(notification) +} + +func (d acpDialect) contextSample( + meta map[string]any, + models []modelInfo, + config []streams.ConfigOption, +) (contextWindowSample, bool) { + if d.contextWindow == nil { + return contextWindowSample{}, false + } + return d.contextWindow(meta, models, config) +} + +func (d acpDialect) promptUsage( + usage *streams.PromptUsage, + meta map[string]any, +) *streams.PromptUsage { + if d.normalizePromptUsage == nil { + return usage + } + return d.normalizePromptUsage(usage, meta) +} diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/dialect_grok.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/dialect_grok.go new file mode 100644 index 0000000000..1c2a07ae36 --- /dev/null +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/dialect_grok.go @@ -0,0 +1,202 @@ +package acp + +import ( + "errors" + "fmt" + "strings" + + "github.com/coder/acp-go-sdk" + "github.com/kandev/kandev/internal/agentctl/acpcompat" + "github.com/kandev/kandev/internal/agentctl/types/streams" +) + +const ( + grokAgentID = acpcompat.GrokAgentID + grokReasoningEffortMetaKey = acpcompat.GrokReasoningEffortMetaKey + configOptionIDReasoningEffort = acpcompat.GrokReasoningEffortOptionID + configOptionCategoryThoughtLevel = "thought_level" +) + +func newGrokACPDialect() acpDialect { + return acpDialect{ + normalizeSessionConfig: grokNormalizeSessionConfig, + configAfterModelChange: grokConfigAfterModelChange, + modelChangeRequest: grokModelChangeRequest, + configChangeRequest: grokConfigChangeRequest, + suppressNotification: func(notification acp.SessionNotification) bool { + return notification.Update.UserMessageChunk != nil + }, + contextWindow: grokContextWindow, + normalizePromptUsage: normalizeGrokPromptUsage, + } +} + +func grokNormalizeSessionConfig( + base []streams.ConfigOption, + models []modelInfo, + currentModelID string, +) []streams.ConfigOption { + return acpcompat.NormalizeSessionConfig( + grokAgentID, + base, + toCompatModels(models), + currentModelID, + ) +} + +// grokSessionConfigOptions is retained as a focused parser entry point for +// captured-wire tests. Runtime selection goes through acpDialect. +func grokSessionConfigOptions( + typed []streams.ConfigOption, + meta map[string]any, + models []modelInfo, + currentModelID string, +) []streams.ConfigOption { + base := typed + if len(base) == 0 { + base = extractConfigOptions(meta) + } + return grokNormalizeSessionConfig(base, models, currentModelID) +} + +func grokConfigAfterModelChange( + config []streams.ConfigOption, + models []modelInfo, + modelID string, +) []streams.ConfigOption { + return acpcompat.ConfigAfterModelChange(grokAgentID, config, toCompatModels(models), modelID) +} + +func grokModelChangeRequest(change dialectConfigChange) (*dialectRPC, error) { + if len(change.models) > 0 && findModelInfo(change.models, change.value) == nil { + return nil, fmt.Errorf("model %q is not in the agent's available models", change.value) + } + req := acp.UnstableSetSessionModelRequest{ + SessionId: acp.SessionId(change.sessionID), + ModelId: change.value, + } + if previous := currentGrokEffortFromConfig(change.config); previous != "" { + if effort, err := acpcompat.ValidateGrokReasoningEffort( + toCompatModels(change.models), change.value, previous, + ); err == nil { + req.Meta = map[string]any{grokReasoningEffortMetaKey: effort} + } + } + return &dialectRPC{request: req, formatError: formatGrokSetModelError}, nil +} + +func grokConfigChangeRequest(change dialectConfigChange) (*dialectRPC, error) { + if isModelConfigID(change.configID, change.config) { + return grokModelChangeRequest(change) + } + if change.configID != configOptionIDReasoningEffort { + return nil, nil + } + modelID := currentModelFromConfig(change.config) + effort, err := acpcompat.ValidateGrokReasoningEffort( + toCompatModels(change.models), modelID, change.value, + ) + if err != nil { + return nil, err + } + return &dialectRPC{ + request: acp.UnstableSetSessionModelRequest{ + SessionId: acp.SessionId(change.sessionID), + ModelId: modelID, + Meta: map[string]any{grokReasoningEffortMetaKey: effort}, + }, + formatError: formatGrokReasoningError, + }, nil +} + +func grokContextWindow( + meta map[string]any, + models []modelInfo, + config []streams.ConfigOption, +) (contextWindowSample, bool) { + sample, ok := acpcompat.GrokContextWindow(meta, toCompatModels(models), config) + return contextWindowSample{used: sample.Used, size: sample.Size}, ok +} + +func normalizeGrokPromptUsage( + usage *streams.PromptUsage, + meta map[string]any, +) *streams.PromptUsage { + if usage != nil && usage.ThoughtTokens == 0 { + usage.ThoughtTokens = acpcompat.GrokReasoningTokens(meta) + } + return usage +} + +func toCompatModels(models []modelInfo) []acpcompat.Model { + out := make([]acpcompat.Model, 0, len(models)) + for _, model := range models { + out = append(out, acpcompat.Model{ID: model.ModelId, Name: model.Name, Meta: toStringMap(model.Meta)}) + } + return out +} + +func findModelInfo(models []modelInfo, modelID string) *modelInfo { + for i := range models { + if models[i].ModelId == modelID { + return &models[i] + } + } + return nil +} + +func currentGrokEffortFromConfig(config []streams.ConfigOption) string { + for _, option := range config { + if option.ID == configOptionIDReasoningEffort { + return option.CurrentValue + } + } + return "" +} + +func formatGrokReasoningError(err error) error { + return fmt.Errorf("set grok reasoning effort via session/set_model failed: %w", err) +} + +func formatGrokSetModelError(err error) error { + if err == nil { + return nil + } + var reqErr *acp.RequestError + if errors.As(err, &reqErr) && reqErr.Message != "" { + if isGrokIncompatibleAgentSwitchError(err) { + message := reqErr.Message + if !strings.Contains(message, "Start a new session") { + message += ". Start a new session to use this model." + } + return fmt.Errorf("%s [MODEL_SWITCH_INCOMPATIBLE_AGENT]: %w", message, err) + } + return fmt.Errorf("set grok model via session/set_model failed: %s: %w", reqErr.Message, err) + } + return fmt.Errorf("set grok model via session/set_model failed: %w", err) +} + +func isGrokIncompatibleAgentSwitchError(err error) bool { + if err == nil { + return false + } + var reqErr *acp.RequestError + if errors.As(err, &reqErr) && grokErrorDataCode(reqErr.Data) == "MODEL_SWITCH_INCOMPATIBLE_AGENT" { + return true + } + message := err.Error() + return strings.Contains(message, "MODEL_SWITCH_INCOMPATIBLE_AGENT") || + (strings.Contains(message, "requires agent") && strings.Contains(message, "Start a new session")) +} + +func grokErrorDataCode(data any) string { + switch value := data.(type) { + case map[string]any: + code, _ := value["code"].(string) + return code + case map[string]string: + return value["code"] + default: + return "" + } +} diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/dialect_grok_test.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/dialect_grok_test.go new file mode 100644 index 0000000000..38db936e84 --- /dev/null +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/dialect_grok_test.go @@ -0,0 +1,585 @@ +package acp + +import ( + "context" + "errors" + "strings" + "sync" + "testing" + "time" + + "github.com/coder/acp-go-sdk" + "github.com/kandev/kandev/internal/agentctl/types/streams" +) + +func TestGrokSessionConfig_BecomesModelAndReasoningSelect(t *testing.T) { + models := []modelInfo{{ + ModelId: "grok-4.5", + Name: "Grok 4.5", + Meta: map[string]any{ + "supportsReasoningEffort": true, + "totalContextTokens": float64(500_000), + "reasoningEffort": "high", + "reasoningEfforts": []any{"high", "medium", "low"}, + }, + }} + // No typed ACP configOptions / no extractable meta options → build from catalog. + opts := grokSessionConfigOptions(nil, nil, models, "grok-4.5") + if len(opts) != 2 { + t.Fatalf("got %d options, want 2 (model + reasoning_effort)", len(opts)) + } + + var model, effort *streams.ConfigOption + for i := range opts { + switch opts[i].ID { + case configOptionIDModel: + model = &opts[i] + case configOptionIDReasoningEffort: + effort = &opts[i] + } + } + if model == nil || model.CurrentValue != "grok-4.5" { + t.Fatalf("model option = %#v", model) + } + if effort == nil { + t.Fatal("missing reasoning_effort option") + } + if effort.Category != configOptionCategoryThoughtLevel { + t.Fatalf("effort category = %q, want %q (frontend filters mode)", effort.Category, configOptionCategoryThoughtLevel) + } + if effort.CurrentValue != "high" { + t.Fatalf("effort current = %q, want high", effort.CurrentValue) + } +} + +func TestGrokSessionConfig_UnsupportedModelHidesEffort(t *testing.T) { + models := []modelInfo{{ + ModelId: "grok-build", + Name: "Grok Build", + Meta: map[string]any{"totalContextTokens": float64(500_000)}, + }} + opts := grokSessionConfigOptions(nil, nil, models, "grok-build") + for _, o := range opts { + if o.ID == configOptionIDReasoningEffort { + t.Fatalf("reasoning_effort must be hidden for unsupported model; got %#v", o) + } + } +} + +func TestGrokSessionConfig_MergesDerivedReasoningWithTypedOptions(t *testing.T) { + modelCategory := acp.SessionConfigOptionCategoryModel + typed := []streams.ConfigOption{{ + Type: "select", + ID: configOptionIDModel, + Category: string(modelCategory), + Name: "Model", + CurrentValue: "grok-4.5", + }} + models := []modelInfo{{ + ModelId: "grok-4.5", + Meta: map[string]any{ + "supportsReasoningEffort": true, + "reasoningEffort": "high", + "reasoningEfforts": []any{"high", "medium", "low"}, + }, + }} + + opts := grokSessionConfigOptions(typed, nil, models, "grok-4.5") + + if len(opts) != 2 { + t.Fatalf("got %d options, want typed model + derived reasoning", len(opts)) + } + if opts[1].ID != configOptionIDReasoningEffort { + t.Fatalf("second option ID = %q, want %q", opts[1].ID, configOptionIDReasoningEffort) + } +} + +type fakeGrokSetModelConn struct { + reqs []acp.UnstableSetSessionModelRequest + configReqs []acp.SetSessionConfigOptionRequest + err error +} + +type blockingGrokSetModelConn struct { + mu sync.Mutex + started chan string + releaseFirst chan struct{} + calls int +} + +func (f *blockingGrokSetModelConn) SetSessionConfigOption( + context.Context, + acp.SetSessionConfigOptionRequest, +) (acp.SetSessionConfigOptionResponse, error) { + return acp.SetSessionConfigOptionResponse{}, nil +} + +func (f *blockingGrokSetModelConn) UnstableSetSessionModel( + _ context.Context, + req acp.UnstableSetSessionModelRequest, +) (acp.UnstableSetSessionModelResponse, error) { + f.mu.Lock() + f.calls++ + call := f.calls + f.mu.Unlock() + f.started <- req.ModelId + if call == 1 { + <-f.releaseFirst + } + return acp.UnstableSetSessionModelResponse{}, nil +} + +func (f *fakeGrokSetModelConn) SetSessionConfigOption( + _ context.Context, + req acp.SetSessionConfigOptionRequest, +) (acp.SetSessionConfigOptionResponse, error) { + f.configReqs = append(f.configReqs, req) + return acp.SetSessionConfigOptionResponse{}, f.err +} + +func (f *fakeGrokSetModelConn) UnstableSetSessionModel(_ context.Context, req acp.UnstableSetSessionModelRequest) (acp.UnstableSetSessionModelResponse, error) { + f.reqs = append(f.reqs, req) + return acp.UnstableSetSessionModelResponse{}, f.err +} + +func grokAdapterWithModels(t *testing.T, models []modelInfo, config []streams.ConfigOption) *Adapter { + t.Helper() + a := newTestAdapter() + a.agentID = grokAgentID + a.dialect = newGrokACPDialect() + a.sessionID = "sess-grok" + a.availableModels = models + a.availableConfigOptions = config + return a +} + +func TestSetGrokModel_UsesSessionSetModel(t *testing.T) { + models := []modelInfo{ + {ModelId: "grok-build", Name: "Grok Build"}, + {ModelId: "grok-4.5", Name: "Grok 4.5", Meta: map[string]any{ + "supportsReasoningEffort": true, + "reasoningEfforts": []any{"high", "medium", "low"}, + }}, + } + config := []streams.ConfigOption{ + {Type: "select", ID: "model", Category: "model", CurrentValue: "grok-build", + Options: []streams.ConfigOptionValue{ + {Value: "grok-build", Name: "Grok Build"}, + {Value: "grok-4.5", Name: "Grok 4.5"}, + }}, + {Type: "select", ID: configOptionIDReasoningEffort, Category: configOptionCategoryThoughtLevel, + CurrentValue: "high", Options: []streams.ConfigOptionValue{ + {Value: "low", Name: "Low"}, {Value: "medium", Name: "Medium"}, {Value: "high", Name: "High"}, + }}, + } + a := grokAdapterWithModels(t, models, config) + conn := &fakeGrokSetModelConn{} + + if err := a.setModelWithConn(context.Background(), conn, "sess-grok", "grok-4.5", models, config); err != nil { + t.Fatalf("setModel: %v", err) + } + if len(conn.reqs) != 1 { + t.Fatalf("RPC count = %d, want 1", len(conn.reqs)) + } + if conn.reqs[0].ModelId != "grok-4.5" { + t.Fatalf("modelId = %q, want grok-4.5", conn.reqs[0].ModelId) + } + // Prior effort still valid for target model → carried in meta. + if conn.reqs[0].Meta[grokReasoningEffortMetaKey] != "high" { + t.Fatalf("meta.reasoningEffort = %v, want high", conn.reqs[0].Meta[grokReasoningEffortMetaKey]) + } +} + +func TestSetGrokReasoningEffort_UsesNormalizedModelConfig(t *testing.T) { + models := []modelInfo{{ + ModelId: "grok-4.5", + Meta: map[string]any{ + "supportsReasoningEffort": true, + "reasoningEfforts": []any{"high", "medium", "low"}, + }, + }} + config := []streams.ConfigOption{ + {Type: "select", ID: "model", Category: "model", CurrentValue: "grok-4.5"}, + {Type: "select", ID: configOptionIDReasoningEffort, Category: configOptionCategoryThoughtLevel, + CurrentValue: "medium", Options: []streams.ConfigOptionValue{ + {Value: "low", Name: "Low"}, {Value: "medium", Name: "Medium"}, {Value: "high", Name: "High"}, + }}, + } + rpc, err := grokConfigChangeRequest(dialectConfigChange{ + sessionID: "sess-grok", + configID: configOptionIDReasoningEffort, + value: "high", + models: models, + config: config, + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if rpc == nil { + t.Fatal("expected Grok session/set_model request") + } + req := rpc.request + if req.ModelId != "grok-4.5" { + t.Fatalf("modelId = %q, want current model preserved", req.ModelId) + } + if req.Meta[grokReasoningEffortMetaKey] != "high" { + t.Fatalf("meta.reasoningEffort = %v, want high", req.Meta[grokReasoningEffortMetaKey]) + } +} + +func TestSetGrokModel_IncompatibleHarnessAsksForNewSession(t *testing.T) { + models := []modelInfo{ + {ModelId: "grok-build", Meta: map[string]any{"agentType": "grok-build-plan"}}, + {ModelId: "grok-composer-2.5-fast", Name: "Composer 2.5", Meta: map[string]any{"agentType": "cursor"}}, + } + config := []streams.ConfigOption{ + {Type: "select", ID: "model", Category: "model", CurrentValue: "grok-build"}, + } + a := grokAdapterWithModels(t, models, config) + conn := &fakeGrokSetModelConn{ + err: &acp.RequestError{ + Code: -32600, + Message: "Cannot switch model harness after first turn", + Data: map[string]any{"code": "MODEL_SWITCH_INCOMPATIBLE_AGENT"}, + }, + } + + err := a.setModelWithConn( + context.Background(), conn, "sess-grok", "grok-composer-2.5-fast", models, config, + ) + if err == nil || !strings.Contains(err.Error(), "Start a new session") { + t.Fatalf("setModel error = %v, want actionable new-session instruction", err) + } + if len(conn.reqs) != 1 { + t.Fatalf("set_model RPC count = %d, want 1", len(conn.reqs)) + } + if a.sessionID != "sess-grok" { + t.Fatalf("session ID = %q, want unchanged sess-grok", a.sessionID) + } + if currentModelFromConfig(a.availableConfigOptions) != "grok-build" { + t.Fatalf("current model config changed after failed switch: %#v", a.availableConfigOptions) + } +} + +func TestSetModelWithConn_GrokDriverAsksForNewSession(t *testing.T) { + models := []modelInfo{ + {ModelId: "grok-build", Meta: map[string]any{"agentType": "grok-build-plan"}}, + {ModelId: "grok-composer-2.5-fast", Meta: map[string]any{"agentType": "cursor"}}, + } + config := []streams.ConfigOption{{ + Type: "select", ID: "model", Category: "model", CurrentValue: "grok-build", + }} + adapter := grokAdapterWithModels(t, models, config) + conn := &fakeGrokSetModelConn{ + err: &acp.RequestError{ + Code: -32600, + Message: "Cannot switch model harness after first turn", + Data: map[string]any{"code": "MODEL_SWITCH_INCOMPATIBLE_AGENT"}, + }, + } + + err := adapter.setModelWithConn( + context.Background(), + conn, + "sess-grok", + "grok-composer-2.5-fast", + models, + config, + ) + if err == nil || !strings.Contains(err.Error(), "Start a new session") { + t.Fatalf("SetModel error = %v, want actionable new-session instruction", err) + } + if len(conn.configReqs) != 0 { + t.Fatalf("session/set_config_option calls = %d, want 0", len(conn.configReqs)) + } + if len(conn.reqs) != 1 { + t.Fatalf("session/set_model calls = %d, want 1", len(conn.reqs)) + } + if adapter.sessionID != "sess-grok" { + t.Fatalf("session ID = %q, want unchanged sess-grok", adapter.sessionID) + } + if currentModelFromConfig(adapter.availableConfigOptions) != "grok-build" { + t.Fatalf("current model config changed after failed switch: %#v", adapter.availableConfigOptions) + } +} + +func TestSetModelWithConn_SerializesConcurrentGrokChanges(t *testing.T) { + models := []modelInfo{{ModelId: "grok-old"}, {ModelId: "grok-first"}, {ModelId: "grok-latest"}} + config := []streams.ConfigOption{{ + Type: "select", ID: configOptionIDModel, Category: configOptionIDModel, CurrentValue: "grok-old", + }} + adapter := grokAdapterWithModels(t, models, config) + conn := &blockingGrokSetModelConn{ + started: make(chan string, 2), + releaseFirst: make(chan struct{}), + } + firstDone := make(chan error, 1) + go func() { + firstDone <- adapter.setModelWithConn( + context.Background(), conn, "sess-grok", "grok-first", models, config, + ) + }() + if got := <-conn.started; got != "grok-first" { + t.Fatalf("first RPC model = %q, want grok-first", got) + } + + secondDone := make(chan error, 1) + go func() { + secondDone <- adapter.setModelWithConn( + context.Background(), conn, "sess-grok", "grok-latest", models, config, + ) + }() + select { + case got := <-conn.started: + t.Fatalf("second RPC %q started before the first completed", got) + case <-time.After(50 * time.Millisecond): + } + + close(conn.releaseFirst) + if err := <-firstDone; err != nil { + t.Fatalf("first SetModel: %v", err) + } + if got := <-conn.started; got != "grok-latest" { + t.Fatalf("second RPC model = %q, want grok-latest", got) + } + if err := <-secondDone; err != nil { + t.Fatalf("second SetModel: %v", err) + } + if got := currentModelFromConfig(adapter.availableConfigOptions); got != "grok-latest" { + t.Fatalf("cached model = %q, want latest selection", got) + } +} + +func TestSetModelWithConn_StaleGrokCompletionDoesNotMutateReplacementSession(t *testing.T) { + models := []modelInfo{{ModelId: "grok-old"}, {ModelId: "grok-stale"}, {ModelId: "grok-new"}} + oldConfig := []streams.ConfigOption{{ + Type: "select", ID: configOptionIDModel, Category: configOptionIDModel, CurrentValue: "grok-old", + }} + adapter := grokAdapterWithModels(t, models, oldConfig) + conn := &blockingGrokSetModelConn{ + started: make(chan string, 1), + releaseFirst: make(chan struct{}), + } + done := make(chan error, 1) + go func() { + done <- adapter.setModelWithConn( + context.Background(), conn, "sess-grok", "grok-stale", models, oldConfig, + ) + }() + <-conn.started + + newConfig := []streams.ConfigOption{{ + Type: "select", ID: configOptionIDModel, Category: configOptionIDModel, CurrentValue: "grok-new", + }} + adapter.mu.Lock() + adapter.sessionID = "sess-new" + adapter.availableConfigOptions = newConfig + adapter.mu.Unlock() + close(conn.releaseFirst) + if err := <-done; err != nil { + t.Fatalf("stale SetModel: %v", err) + } + + adapter.mu.RLock() + gotSession := adapter.sessionID + gotModel := currentModelFromConfig(adapter.availableConfigOptions) + adapter.mu.RUnlock() + if gotSession != "sess-new" || gotModel != "grok-new" { + t.Fatalf("replacement state = session %q model %q, want sess-new/grok-new", gotSession, gotModel) + } +} + +func TestGrokContextFromNotificationMeta(t *testing.T) { + models := []modelInfo{{ + ModelId: "grok-4.5", + Meta: map[string]any{"totalContextTokens": float64(500_000)}, + }} + a := grokAdapterWithModels(t, models, []streams.ConfigOption{{ + Type: "select", ID: configOptionIDModel, Category: configOptionIDModel, CurrentValue: "grok-4.5", + }}) + + a.handleACPUpdate(acp.SessionNotification{ + SessionId: "sess-grok", + Meta: map[string]any{"totalTokens": float64(42_000)}, + Update: acp.SessionUpdate{ + AgentMessageChunk: &acp.SessionUpdateAgentMessageChunk{ + Content: acp.TextBlock("hi"), + }, + }, + }) + + events := drainEvents(a) + var ctx *AgentEvent + for i := range events { + if events[i].Type == streams.EventTypeContextWindow { + ctx = &events[i] + } + } + if ctx == nil { + t.Fatalf("expected context_window event; got %#v", events) + } + if ctx.ContextWindowSize != 500_000 || ctx.ContextWindowUsed != 42_000 { + t.Fatalf("size/used = %d/%d, want 500000/42000", ctx.ContextWindowSize, ctx.ContextWindowUsed) + } + + a.handleACPUpdate(acp.SessionNotification{ + SessionId: "sess-grok", + Meta: map[string]any{"totalTokens": float64(42_000)}, + Update: acp.SessionUpdate{ + AgentMessageChunk: &acp.SessionUpdateAgentMessageChunk{Content: acp.TextBlock("again")}, + }, + }) + for _, event := range drainEvents(a) { + if event.Type == streams.EventTypeContextWindow { + t.Fatal("unchanged totalTokens must not emit duplicate context_window") + } + } + + // Prompt completion consumes the shared ACP usage tracker. Driver-owned + // context dedupe must remain intact across that unrelated reset. + a.consumeUsageDelta("sess-grok") + a.handleACPUpdate(acp.SessionNotification{ + SessionId: "sess-grok", + Meta: map[string]any{"totalTokens": float64(42_000)}, + Update: acp.SessionUpdate{ + AgentMessageChunk: &acp.SessionUpdateAgentMessageChunk{Content: acp.TextBlock("next turn")}, + }, + }) + for _, event := range drainEvents(a) { + if event.Type == streams.EventTypeContextWindow { + t.Fatal("usage tracker reset must not duplicate unchanged dialect context") + } + } + + // Compaction legitimately lowers context usage and must still emit. + a.handleACPUpdate(acp.SessionNotification{ + SessionId: "sess-grok", + Meta: map[string]any{"totalTokens": float64(40_000)}, + Update: acp.SessionUpdate{ + AgentMessageChunk: &acp.SessionUpdateAgentMessageChunk{Content: acp.TextBlock("compacted")}, + }, + }) + compactedUsed := int64(0) + compactedFound := false + for _, event := range drainEvents(a) { + if event.Type == streams.EventTypeContextWindow { + compactedUsed = event.ContextWindowUsed + compactedFound = true + } + } + if !compactedFound || compactedUsed != 40_000 { + t.Fatalf("compaction context used = %d (found=%t), want 40000", compactedUsed, compactedFound) + } + + // A delayed notification from a replaced session cannot update current UI. + a.handleACPUpdate(acp.SessionNotification{ + SessionId: "stale-session", + Meta: map[string]any{"totalTokens": float64(99_000)}, + Update: acp.SessionUpdate{ + AgentMessageChunk: &acp.SessionUpdateAgentMessageChunk{Content: acp.TextBlock("stale")}, + }, + }) + for _, event := range drainEvents(a) { + if event.Type == streams.EventTypeContextWindow { + t.Fatal("stale session must not emit dialect context") + } + } +} + +func TestGrokUserMessageEchoIsSuppressedWithoutDroppingContext(t *testing.T) { + models := []modelInfo{{ + ModelId: "grok-current", + Meta: map[string]any{"totalContextTokens": float64(500_000)}, + }} + a := grokAdapterWithModels(t, models, []streams.ConfigOption{{ + Type: "select", ID: configOptionIDModel, Category: configOptionIDModel, CurrentValue: "grok-current", + }}) + + a.handleACPUpdate(acp.SessionNotification{ + SessionId: "sess-grok", + Meta: map[string]any{"totalTokens": float64(42_000)}, + Update: acp.SessionUpdate{ + UserMessageChunk: &acp.SessionUpdateUserMessageChunk{ + Content: acp.TextBlock("echoed prompt"), + }, + }, + }) + + contextFound := false + for _, event := range drainEvents(a) { + if event.Type == streams.EventTypeMessageChunk { + t.Fatalf("Grok user echo must not become a message event: %#v", event) + } + if event.Type == streams.EventTypeContextWindow { + contextFound = true + } + } + if !contextFound { + t.Fatal("suppressed user echo must still contribute context metadata") + } +} + +func TestGrokContextRetriesAfterFullUpdateChannel(t *testing.T) { + models := []modelInfo{{ + ModelId: "grok-current", + Meta: map[string]any{"totalContextTokens": float64(500_000)}, + }} + a := grokAdapterWithModels(t, models, []streams.ConfigOption{{ + Type: "select", ID: configOptionIDModel, Category: configOptionIDModel, CurrentValue: "grok-current", + }}) + a.updatesCh = make(chan AgentEvent, 1) + a.updatesCh <- AgentEvent{Type: streams.EventTypeMessageChunk} + meta := map[string]any{"totalTokens": float64(42_000)} + + if event := a.emitDialectContextWindow("sess-grok", meta); event != nil { + t.Fatalf("emitDialectContextWindow() = %#v with full channel, want nil", event) + } + if _, cached := a.contextSamples["sess-grok"]; cached { + t.Fatal("dropped context sample must not be cached") + } + <-a.updatesCh + + if event := a.emitDialectContextWindow("sess-grok", meta); event == nil { + t.Fatal("same context sample must be retried after channel capacity returns") + } +} + +func TestGrokDialect_NormalizesPrivateReasoningTokens(t *testing.T) { + response := &acp.PromptResponse{Meta: map[string]any{ + "usage": map[string]any{ + "inputTokens": float64(5), + "outputTokens": float64(3), + "totalTokens": float64(8), + "reasoningTokens": float64(2), + }, + }} + dialect := newGrokACPDialect() + usage := dialect.promptUsage(extractUsage(response), response.Meta) + if usage == nil { + t.Fatal("expected non-nil usage") + } + if usage.ThoughtTokens != 2 { + t.Fatalf("ThoughtTokens = %d, want 2", usage.ThoughtTokens) + } +} + +func TestIsGrokIncompatibleAgentSwitchError(t *testing.T) { + reqErr := &acp.RequestError{ + Code: -32600, + Message: "Cannot switch to model 'composer': it requires agent 'cursor'. Start a new session to use this model.", + Data: map[string]any{"code": "MODEL_SWITCH_INCOMPATIBLE_AGENT"}, + } + if !isGrokIncompatibleAgentSwitchError(reqErr) { + t.Fatal("expected structured error to match") + } + wrapped := formatGrokSetModelError(reqErr) + if !isGrokIncompatibleAgentSwitchError(wrapped) { + t.Fatal("expected wrapped error to still match") + } + if !strings.Contains(wrapped.Error(), "MODEL_SWITCH_INCOMPATIBLE_AGENT") { + t.Fatalf("wrapped should retain code tag, got %q", wrapped.Error()) + } + if isGrokIncompatibleAgentSwitchError(errors.New("method not found")) { + t.Fatal("unrelated error must not match") + } +} diff --git a/apps/backend/internal/agentctl/server/adapter/transport/acp/usage_test.go b/apps/backend/internal/agentctl/server/adapter/transport/acp/usage_test.go index 2f365e1b91..b02d38402c 100644 --- a/apps/backend/internal/agentctl/server/adapter/transport/acp/usage_test.go +++ b/apps/backend/internal/agentctl/server/adapter/transport/acp/usage_test.go @@ -156,6 +156,23 @@ func TestExtractUsage(t *testing.T) { } } +func TestExtractUsage_DoesNotInterpretGrokPrivateReasoningTokens(t *testing.T) { + usage := extractUsage(&acp.PromptResponse{Meta: map[string]any{ + "usage": map[string]any{ + "inputTokens": float64(5), + "outputTokens": float64(3), + "totalTokens": float64(8), + "reasoningTokens": float64(2), + }, + }}) + if usage == nil { + t.Fatal("expected non-nil usage") + } + if usage.ThoughtTokens != 0 { + t.Fatalf("ThoughtTokens = %d, want 0; Grok private fields belong to its ACP dialect", usage.ThoughtTokens) + } +} + // TestUsageTracker_CumulativeDelta asserts the codex-acp fallback path: // usage_update updates the cumulative used counter via tryConvertUntypedUpdate; // consumeUsageDelta returns the running total and resets to zero so the next diff --git a/apps/backend/internal/agentctl/server/utility/acp_executor.go b/apps/backend/internal/agentctl/server/utility/acp_executor.go index 68a6b6554c..d46a7dbf64 100644 --- a/apps/backend/internal/agentctl/server/utility/acp_executor.go +++ b/apps/backend/internal/agentctl/server/utility/acp_executor.go @@ -15,8 +15,10 @@ import ( "time" acp "github.com/coder/acp-go-sdk" + "github.com/kandev/kandev/internal/agentctl/acpcompat" acpclient "github.com/kandev/kandev/internal/agentctl/server/acp" "github.com/kandev/kandev/internal/agentctl/sessionmodel" + "github.com/kandev/kandev/internal/agentctl/types/streams" "go.uber.org/zap" ) @@ -344,7 +346,7 @@ func (e *ACPInferenceExecutor) Probe(ctx context.Context, req *ProbeRequest) (*P } defer cleanupACPCommand(ctx, cmd, lifecycle, e.logger) - resp, err := e.probeACPSession(ctx, stdin, stdout, workDir) + resp, err := e.probeACPSession(ctx, stdin, stdout, workDir, req.AgentID) if err != nil { return &ProbeResponse{ Success: false, @@ -460,6 +462,7 @@ func (e *ACPInferenceExecutor) probeACPSession( stdin io.Writer, stdout io.Reader, workDir string, + agentID string, ) (*ProbeResponse, error) { var mu sync.Mutex var commands []ProbeCommand @@ -518,7 +521,7 @@ func (e *ACPInferenceExecutor) probeACPSession( } out := buildInitProbeFields(initResp) - applySessionProbeFields(out, sessionResp) + applySessionProbeFields(out, sessionResp, agentID) mu.Lock() out.Commands = append([]ProbeCommand(nil), commands...) mu.Unlock() @@ -565,7 +568,7 @@ func buildInitProbeFields(initResp acp.InitializeResponse) *ProbeResponse { // top-level `models` field via acp.LegacyModels for agents that haven't // migrated yet (auggie 0.29.x). The legacy `modes` field is still present on // the SDK type and we keep populating from it for older agents. -func applySessionProbeFields(out *ProbeResponse, sessionResp acp.NewSessionResponse) { +func applySessionProbeFields(out *ProbeResponse, sessionResp acp.NewSessionResponse, agentID string) { out.ConfigOptions = probeConfigOptions(sessionResp.ConfigOptions) if sessionResp.Modes != nil { out.CurrentModeID = string(sessionResp.Modes.CurrentModeId) @@ -585,6 +588,53 @@ func applySessionProbeFields(out *ProbeResponse, sessionResp acp.NewSessionRespo if sessionResp.Modes == nil { applyConfigOptionsAsModes(out, sessionResp.ConfigOptions) } + applySessionCompatibility(out, agentID) +} + +func applySessionCompatibility(out *ProbeResponse, agentID string) { + models := make([]acpcompat.Model, 0, len(out.Models)) + for _, model := range out.Models { + models = append(models, acpcompat.Model{ID: model.ID, Name: model.Name, Meta: model.Meta}) + } + base := make([]streams.ConfigOption, 0, len(out.ConfigOptions)) + for _, option := range out.ConfigOptions { + converted := streams.ConfigOption{ + Type: option.Type, + ID: option.ID, + Name: option.Name, + Description: option.Description, + CurrentValue: option.CurrentValue, + Category: option.Category, + } + for _, choice := range option.Options { + converted.Options = append(converted.Options, streams.ConfigOptionValue{ + Value: choice.Value, + Name: choice.Name, + Description: choice.Description, + }) + } + base = append(base, converted) + } + normalized := acpcompat.NormalizeSessionConfig(agentID, base, models, out.CurrentModelID) + out.ConfigOptions = make([]ProbeConfigOption, 0, len(normalized)) + for _, option := range normalized { + converted := ProbeConfigOption{ + Type: option.Type, + ID: option.ID, + Name: option.Name, + Description: option.Description, + CurrentValue: option.CurrentValue, + Category: option.Category, + } + for _, choice := range option.Options { + converted.Options = append(converted.Options, ProbeConfigOptionChoice{ + Value: choice.Value, + Name: choice.Name, + Description: choice.Description, + }) + } + out.ConfigOptions = append(out.ConfigOptions, converted) + } } // applyLegacyModelsFallback fills out.Models / out.CurrentModelID from the diff --git a/apps/backend/internal/agentctl/server/utility/acp_executor_test.go b/apps/backend/internal/agentctl/server/utility/acp_executor_test.go index d2e3385339..bd9b7ffee4 100644 --- a/apps/backend/internal/agentctl/server/utility/acp_executor_test.go +++ b/apps/backend/internal/agentctl/server/utility/acp_executor_test.go @@ -50,6 +50,32 @@ func TestProbeConfigOptions_PreservesDescriptions(t *testing.T) { } } +func TestApplySessionCompatibility_PreservesDescriptions(t *testing.T) { + t.Parallel() + + out := &ProbeResponse{ConfigOptions: []ProbeConfigOption{{ + Type: "select", + ID: "reasoning_effort", + Name: "Reasoning effort", + Description: "Controls reasoning depth.", + CurrentValue: "high", + Options: []ProbeConfigOptionChoice{{ + Value: "high", + Name: "High", + Description: "More thorough reasoning.", + }}, + }}} + + applySessionCompatibility(out, "mock-agent") + + if got := out.ConfigOptions[0].Description; got != "Controls reasoning depth." { + t.Errorf("option description = %q, want provider description", got) + } + if got := out.ConfigOptions[0].Options[0].Description; got != "More thorough reasoning." { + t.Errorf("value description = %q, want provider description", got) + } +} + func TestResolveProbeCommand_AllowsEveryListedBinary(t *testing.T) { t.Parallel() @@ -265,7 +291,7 @@ func TestApplySessionProbeFields_FallsBackToConfigOptions(t *testing.T) { } out := &ProbeResponse{} - applySessionProbeFields(out, resp) + applySessionProbeFields(out, resp, "") if got, want := out.CurrentModelID, "opus"; got != want { t.Fatalf("CurrentModelID = %q, want %q", got, want) @@ -309,7 +335,7 @@ func TestApplySessionProbeFields_FallsBackToLegacyModels(t *testing.T) { } out := &ProbeResponse{} - applySessionProbeFields(out, resp) + applySessionProbeFields(out, resp, "") if got, want := out.CurrentModelID, "claude-opus-4-7"; got != want { t.Fatalf("CurrentModelID = %q, want %q", got, want) @@ -325,6 +351,35 @@ func TestApplySessionProbeFields_FallsBackToLegacyModels(t *testing.T) { } } +func TestApplySessionProbeFields_GrokLegacyModelsExposeReasoning(t *testing.T) { + t.Parallel() + + resp := acp.NewSessionResponse{ + LegacyModels: &acp.LegacyModels{ + CurrentModelId: "grok-4.5", + AvailableModels: []acp.LegacyModelInfo{{ + ModelId: "grok-4.5", + Name: "Grok 4.5", + Meta: map[string]any{ + "supportsReasoningEffort": true, + "reasoningEffort": "high", + "reasoningEfforts": []any{"high", "medium", "low"}, + }, + }}, + }, + } + + out := &ProbeResponse{} + applySessionProbeFields(out, resp, "grok-acp") + + if got, want := len(out.ConfigOptions), 2; got != want { + t.Fatalf("len(ConfigOptions) = %d, want model + reasoning", got) + } + if got, want := out.ConfigOptions[1].ID, "reasoning_effort"; got != want { + t.Fatalf("ConfigOptions[1].ID = %q, want %q", got, want) + } +} + // When both surfaces are present the typed configOptions list wins — the // legacy field is only a fallback for unmigrated agents. func TestApplySessionProbeFields_TypedConfigOptionsBeatLegacyModels(t *testing.T) { @@ -351,7 +406,7 @@ func TestApplySessionProbeFields_TypedConfigOptionsBeatLegacyModels(t *testing.T } out := &ProbeResponse{} - applySessionProbeFields(out, resp) + applySessionProbeFields(out, resp, "") if got, want := out.CurrentModelID, "opus"; got != want { t.Fatalf("CurrentModelID = %q, want %q (typed configOptions must win)", got, want) @@ -389,7 +444,7 @@ func TestApplySessionProbeFields_FlattensGroupedConfigOptions(t *testing.T) { } out := &ProbeResponse{} - applySessionProbeFields(out, resp) + applySessionProbeFields(out, resp, "") if got, want := len(out.Modes), 2; got != want { t.Fatalf("len(Modes) = %d, want %d", got, want) diff --git a/docs/decisions/0044-acp-agent-compatibility-dialects.md b/docs/decisions/0044-acp-agent-compatibility-dialects.md new file mode 100644 index 0000000000..be5c3b903f --- /dev/null +++ b/docs/decisions/0044-acp-agent-compatibility-dialects.md @@ -0,0 +1,43 @@ +# 0044: ACP Agent Compatibility Dialects + +**Status:** accepted +**Date:** 2026-07-16 +**Area:** backend, protocol + +## Context + +ACP defines common transport and session methods, but individual agent CLIs can expose models, configuration, usage, and recovery through private metadata or non-standard method combinations. Grok, for example, combines the legacy `models` response with model metadata for reasoning controls and uses `session/set_model` for both model and reasoning changes. The live adapter and the sessionless utility probe must interpret those capabilities consistently. + +## Decision + +The ACP adapter selects a package-private `acpDialect` function table by agent ID. + +- Dialect hooks are optional, pure compatibility functions that return normalized data or request descriptions. They never receive `*Adapter`, execute RPCs, mutate session state, or emit events. +- Shared adapter code owns JSON-RPC transport, session lifecycle, serialized configuration changes, generation checks, state convergence, and event delivery. +- Canonical ACP behavior is the zero-value dialect; there is no standard implementation containing no-op methods. +- Capability normalization shared by live sessions and utility probes lives in `internal/agentctl/acpcompat/`. +- Grok-specific model, reasoning, usage, notification, and incompatible-agent error translation lives in `dialect_grok.go` and the shared compatibility package. +- A dialect never replaces an active session to satisfy a model change. When an implementation requires another agent harness, Kandev returns its actionable error and the user starts a new session explicitly. + +## Consequences + +- New ACP CLI variants can add narrowly scoped functions without spreading agent checks through shared adapter files. +- Provider-specific parsing is reusable outside the live adapter when capability discovery needs the same interpretation. +- Concurrent model and configuration requests converge on the latest generation and stale session completions cannot overwrite replacement state. +- A small function table adds indirection to ACP session and update handling. +- No product spec covers internal ACP compatibility handling; this decision remains the durable record. +- Model changes cannot silently discard upstream agent context by replacing the ACP session. + +## Alternatives Considered + +### Use a stateful ACP driver interface + +Rejected because the hooks crossed ownership boundaries by receiving the concrete adapter, executing RPCs, mutating shared state, and emitting events. It also required a standard implementation made mostly of no-op methods. + +### Keep agent checks in shared ACP files + +Rejected because each new non-standard implementation would increase branching in common session and notification paths. Small optional dialect functions keep selection centralized without creating provider objects. + +### Create a complete adapter for every ACP CLI + +Rejected because transport, lifecycle, permissions, tools, and event delivery are already shared correctly and would be duplicated. diff --git a/docs/decisions/INDEX.md b/docs/decisions/INDEX.md index 7e36e909c7..ebdb6f7690 100644 --- a/docs/decisions/INDEX.md +++ b/docs/decisions/INDEX.md @@ -50,3 +50,4 @@ Read individual ADRs for full context. Create new ones via `/record decision` or | 0042 | [Project shell output and fetch it on demand](0042-project-shell-output-and-fetch-on-demand.md) | accepted | backend, frontend, protocol | 2026-07-16 | | 0043 | [Plugins read/write kandev data via capability-gated Host gRPC RPCs](0043-plugin-host-data-api.md) | accepted | backend, protocol | 2026-07-17 | | 2026-07-15-office-agent-execution-profile-routing | [Separate Office identity from routed execution profiles](2026-07-15-office-agent-execution-profile-routing.md) | proposed | backend, frontend | 2026-07-15 | +| 0044 | [ACP agent compatibility dialects](0044-acp-agent-compatibility-dialects.md) | accepted | backend, protocol | 2026-07-16 |