package agent import ( "bufio" "context" "encoding/json" "fmt" "log/slog" "os/exec" "strings" "time" ) // qwenBackend drives Qwen Code's native non-interactive JSONL protocol: // qwen -p --output-format stream-json. The event schema is based on // Qwen Code 0.20.0 captures in testdata/qwen-code-0.20.0-stream-json.jsonl. type qwenBackend struct { cfg Config } // qwenBlockedArgs are owned by Multica. Qwen accepts the task prompt and stream // protocol as flags, so custom args must not replace either. Model/session are // also selected by Multica, and safe mode disables the QWEN.md context file. // --yolo/-y, --approval-mode, and --core-tools are daemon-owned permission // flags; users may not disable bypass mode or narrow the core tool registry // from custom_args (use --exclude-tools to hard-deny specific tools instead). var qwenBlockedArgs = map[string]blockedArgMode{ "-p": blockedWithValue, "--prompt": blockedWithValue, "-i": blockedWithValue, "--prompt-interactive": blockedWithValue, "-o": blockedWithValue, "--output-format": blockedWithValue, "-m": blockedWithValue, "--model": blockedWithValue, "-r": blockedWithValue, "--resume": blockedWithValue, "-c": blockedStandalone, "--continue": blockedStandalone, "--chat-recording": blockedWithValue, "--mcp-config": blockedWithValue, "--safe-mode": blockedStandalone, "--yolo": blockedStandalone, "-y": blockedStandalone, "--approval-mode": blockedWithValue, "--core-tools": blockedWithValue, } func buildQwenArgs(prompt string, opts ExecOptions, logger *slog.Logger) []string { args := []string{"-p", prompt, "--output-format", "stream-json"} if opts.Model != "" { args = append(args, "--model", opts.Model) } if opts.ResumeSessionID != "" { args = append(args, "--resume", opts.ResumeSessionID) } // --yolo is daemon-owned: Qwen Code's non-interactive mode filters out // approval-requiring tools (run_shell_command, edit, write_file, etc.) // unless bypass mode is active. All other adapters use an equivalent // mechanism; this keeps Qwen headless runs consistent (MUL-5134). args = append(args, "--yolo") args = append(args, filterCustomArgs(opts.ExtraArgs, qwenBlockedArgs, logger)...) args = append(args, filterCustomArgs(opts.CustomArgs, qwenBlockedArgs, logger)...) return args } func (b *qwenBackend) Execute(ctx context.Context, prompt string, opts ExecOptions) (*Session, error) { execPath := b.cfg.ExecutablePath if execPath == "" { execPath = "qwen" } if _, err := exec.LookPath(execPath); err != nil { return nil, fmt.Errorf("qwen executable not found at %q: %w", execPath, err) } timeout := opts.Timeout runCtx, cancel := runContext(ctx, timeout) args := buildQwenArgs(prompt, opts, b.cfg.Logger) // Qwen Code 0.20.0 accepts a JSON string or a file path through // --mcp-config. Materialise a managed config into a 0600 temp file so it // does not appear in argv or logs, then remove it when the process exits. var mcpConfigPath string var mcpFileCleanup func() if hasManagedMcpConfig(opts.McpConfig) { path, err := writeMcpConfigToTemp(opts.McpConfig) if err != nil { cancel() return nil, fmt.Errorf("write qwen mcp_config: %w", err) } mcpConfigPath = path mcpFileCleanup = func() { cleanupMcpConfigTemp(mcpConfigPath) } args = append(args, "--mcp-config", mcpConfigPath) } // Clean up if a later setup step returns before the result goroutine owns it. defer func() { if mcpFileCleanup != nil { mcpFileCleanup() } }() cmd := exec.CommandContext(runCtx, execPath, args...) hideAgentWindow(cmd) // args contain the task prompt; never expose it in daemon logs. b.cfg.Logger.Info("agent command", "exec", execPath, "provider", "qwen") cmd.WaitDelay = 10 * time.Second if opts.Cwd != "" { cmd.Dir = opts.Cwd } cmd.Env = buildEnv(b.cfg.Env) stdout, err := cmd.StdoutPipe() if err != nil { cancel() return nil, fmt.Errorf("qwen stdout pipe: %w", err) } stderrBuf := newStderrTail(newLogWriter(b.cfg.Logger, "[qwen:stderr] "), agentStderrTailBytes) cmd.Stderr = stderrBuf if err := cmd.Start(); err != nil { cancel() return nil, fmt.Errorf("start qwen: %w", err) } // cmd.Start succeeded; result goroutine now owns cleanup. mcpFileCleanup = nil b.cfg.Logger.Info("qwen started", "pid", cmd.Process.Pid, "cwd", opts.Cwd, "model", opts.Model) msgCh := make(chan Message, 256) resCh := make(chan Result, 1) go func() { defer cancel() defer close(msgCh) defer close(resCh) if mcpConfigPath != "" { defer cleanupMcpConfigTemp(mcpConfigPath) } started := time.Now() state := qwenStreamState{model: opts.Model, usage: make(map[string]TokenUsage)} go func() { <-runCtx.Done() _ = stdout.Close() }() scanner := bufio.NewScanner(stdout) scanner.Buffer(make([]byte, 0, 1024*1024), 10*1024*1024) for scanner.Scan() { line := strings.TrimSpace(scanner.Text()) if line == "" { continue } var event qwenStreamEvent if err := json.Unmarshal([]byte(line), &event); err != nil { state.invalidEventCount++ continue } state.eventCount++ state.lastEventType = event.Type handleQwenEvent(event, msgCh, &state) } scanErr := scanner.Err() if scanErr != nil { _ = stdout.Close() } exitErr := cmd.Wait() duration := time.Since(started) status, output, errMsg := finalizeStreamResult("qwen", timeout, runCtx.Err(), nil, exitErr, state.sessionID, streamTerminalState{ lastAssistantText: state.lastAssistantText, finalResultText: state.finalResultText, sawResult: state.sawResult, resultIsError: state.resultIsError, scanErr: scanErr, }, "") if errMsg != "" { errMsg = withAgentStderr(errMsg, "qwen", stderrBuf.Tail()) } logStreamProtocolObservation(b.cfg.Logger, streamProtocolObservation{ provider: "qwen", cliVersion: b.cfg.CLIVersion, model: state.model, exitCode: streamProcessExitCode(exitErr), eventCount: state.eventCount, invalidEventCount: state.invalidEventCount, assistantEventCount: state.assistantEventCount, toolUseCount: state.toolUseCount, sawResult: state.sawResult, resultIsError: state.resultIsError, resultBytes: len(state.finalResultText), lastAssistantBytes: len(state.lastAssistantText), scannerError: scanErr != nil, lastEventType: state.lastEventType, }) b.cfg.Logger.Info("qwen finished", "pid", cmd.Process.Pid, "status", status, "duration", duration.Round(time.Millisecond).String()) resCh <- Result{ Status: status, Output: output, Error: errMsg, DurationMs: duration.Milliseconds(), SessionID: resolveSessionID(opts.ResumeSessionID, state.sessionID, status == "failed", errMsg), Usage: state.usage, ResumeRejected: resumeWasRejected(opts.ResumeSessionID, state.sessionID, status == "failed", errMsg), } }() return &Session{Messages: msgCh, Result: resCh}, nil } type qwenStreamEvent struct { Type string `json:"type"` Subtype string `json:"subtype,omitempty"` SessionID string `json:"session_id,omitempty"` Model string `json:"model,omitempty"` Message json.RawMessage `json:"message,omitempty"` Result string `json:"result,omitempty"` IsError bool `json:"is_error,omitempty"` Usage *qwenUsage `json:"usage,omitempty"` Error json.RawMessage `json:"error,omitempty"` } type qwenMessage struct { Model string `json:"model,omitempty"` Content []qwenContentBlock `json:"content"` Usage *qwenUsage `json:"usage,omitempty"` } type qwenUsage struct { InputTokens int64 `json:"input_tokens"` OutputTokens int64 `json:"output_tokens"` CacheReadInputTokens int64 `json:"cache_read_input_tokens"` } type qwenContentBlock struct { Type string `json:"type"` Thinking string `json:"thinking,omitempty"` Text string `json:"text,omitempty"` ID string `json:"id,omitempty"` Name string `json:"name,omitempty"` Input json.RawMessage `json:"input,omitempty"` ToolUseID string `json:"tool_use_id,omitempty"` Content json.RawMessage `json:"content,omitempty"` } type qwenStreamState struct { sessionID, model, lastAssistantText, finalResultText, lastEventType string sawResult, resultIsError bool usage map[string]TokenUsage eventCount, invalidEventCount, assistantEventCount, toolUseCount int } func handleQwenEvent(event qwenStreamEvent, ch chan<- Message, state *qwenStreamState) { if event.SessionID != "" { state.sessionID = event.SessionID } if event.Model != "" { state.model = event.Model } switch event.Type { case "system": trySend(ch, Message{Type: MessageStatus, Status: "running", SessionID: state.sessionID}) case "assistant": state.assistantEventCount++ text, tools, model := handleQwenAssistant(event.Message, ch, state.usage) if model != "" { state.model = model } state.toolUseCount += tools if tools == 0 && text != "" { state.lastAssistantText = text } else if tools > 0 { state.lastAssistantText = "" } case "user": handleQwenUser(event.Message, ch) case "result": state.sawResult = true state.resultIsError = event.IsError || event.Subtype == "error" || event.Subtype == "failed" if state.resultIsError { // Qwen 0.20.0 result errors omit result; their actionable detail // is in error.message. state.finalResultText = qwenErrorText(event) } else { state.finalResultText = event.Result } if usage := qwenResultUsage(event.Usage, state.model); len(usage) > 0 { state.usage = usage } case "error": // Be fail-closed if a later Qwen release emits a terminal error event. state.sawResult = true state.resultIsError = true state.finalResultText = qwenErrorText(event) } } func handleQwenAssistant(raw json.RawMessage, ch chan<- Message, usage map[string]TokenUsage) (string, int, string) { var message qwenMessage if json.Unmarshal(raw, &message) != nil { return "", 0, "" } if message.Usage != nil && message.Model != "" { usage[message.Model] = qwenTokenUsage(message.Usage) } var text strings.Builder tools := 0 for _, block := range message.Content { switch block.Type { case "thinking": if block.Thinking != "" { trySend(ch, Message{Type: MessageThinking, Content: block.Thinking}) } case "text": if block.Text != "" { text.WriteString(block.Text) trySend(ch, Message{Type: MessageText, Content: block.Text}) } case "tool_use": tools++ var input map[string]any if len(block.Input) > 0 { _ = json.Unmarshal(block.Input, &input) } trySend(ch, Message{Type: MessageToolUse, Tool: block.Name, CallID: block.ID, Input: input}) } } return text.String(), tools, message.Model } func handleQwenUser(raw json.RawMessage, ch chan<- Message) { var message qwenMessage if json.Unmarshal(raw, &message) != nil { return } for _, block := range message.Content { if block.Type == "tool_result" { trySend(ch, Message{Type: MessageToolResult, CallID: block.ToolUseID, Output: qwenToolResultOutput(block.Content)}) } } } func qwenTokenUsage(usage *qwenUsage) TokenUsage { return TokenUsage{InputTokens: usage.InputTokens, OutputTokens: usage.OutputTokens, CacheReadTokens: usage.CacheReadInputTokens} } func qwenResultUsage(usage *qwenUsage, model string) map[string]TokenUsage { if usage == nil || model == "" || (usage.InputTokens == 0 && usage.OutputTokens == 0 && usage.CacheReadInputTokens == 0) { return nil } return map[string]TokenUsage{model: qwenTokenUsage(usage)} } func qwenToolResultOutput(raw json.RawMessage) string { var text string if json.Unmarshal(raw, &text) == nil { return text } return string(raw) } func qwenErrorText(event qwenStreamEvent) string { if event.Result != "" { return event.Result } var body struct { Message string `json:"message"` } if json.Unmarshal(event.Error, &body) == nil && body.Message != "" { return body.Message } if len(event.Error) > 0 { return string(event.Error) } return "qwen returned an error event without details" }