MonkeyCode
从源码学会它怎么工作

  554  // ==================== CreateTask 类型 ====================
  555  
  556  // CodingAgent 编码代理类型
  557  type CodingAgent int
  558  
  559  const (
  560  	CodingAgentCodex CodingAgent = iota + 1
  561  	CodingAgentClaude
  562  	CodingAgentMCAIReview
  563  	CodingAgentOpenCode
  564  )
  565  
  566  // LLM 模型配置
  567  type LLM struct {
  568  	ApiKey      string   `json:"api_key"`
  569  	BaseURL     string   `json:"base_url"`
  570  	Model       string   `json:"model"`
  571  	ApiType     string   `json:"api_type,omitempty"` // 接口类型 anthropic | openai
  572  	Temperature *float32 `json:"temperature,omitempty"`
  573  }
  574  
  575  // ConfigFile 配置文件
  576  type ConfigFile struct {
  577  	Path    string  `json:"path"`
  578  	Content string  `json:"content"`
  579  	Mode    *uint32 `json:"mode,omitempty"`
  580  }
  581  
  582  // TaskExecutionConfig 任务运行配置
  583  type TaskExecutionConfig struct {
  584  	Envs           map[string]string `json:"envs,omitempty"`
  585  	ConfigFiles    []ConfigFile      `json:"config_files,omitempty"`
  586  	McpServers     []McpServerConfig `json:"mcp_servers,omitempty"`
  587  	AgentResources *AgentResources   `json:"agent_resources,omitempty"`

  556  	limit, err := a.resolveTaskConcurrencyLimit(ctx, user.ID)
  557  	if err != nil {
  558  		return nil, err
  559  	}
  560  	ctx = entx.WithTaskConcurrencyLimit(ctx, limit)
  561  
  562  	vmID := fmt.Sprintf("agent_%s", uuid.NewString())
  563  	prepared, err := a.repo.PrepareCreate(ctx, user, req, token, vmID)
  564  	if err != nil {
  565  		a.logger.With("error", err, "req", req).ErrorContext(ctx, "failed to create task")
  566  		return nil, err
  567  	}
  568  	if prepared == nil || prepared.ProjectTask == nil || prepared.Model == nil || prepared.Image == nil {
  569  		return nil, fmt.Errorf("failed to prepare task")
  570  	}
  571  	pt := prepared.ProjectTask
  572  	m := prepared.Model
  573  	i := prepared.Image
  574  	t := pt.Edges.Task
  575  	if t == nil {
  576  		return nil, fmt.Errorf("task edge is nil")
  577  	}
  578  	if git.URL == "" {
  579  		git.URL = pt.RepoURL
  580  	}
      … 26 lines omitted; exact range 556–617 …
  607  		LLM: taskflow.LLMProviderReq{
  608  			Provider: taskflow.LlmProviderOpenAI,
  609  			ApiKey:   m.APIKey,
  610  			BaseURL:  m.BaseURL,
  611  			Model:    m.Model,
  612  		},
  613  		Cores:    "2",
  614  		Memory:   8 << 30,
  615  		Envs:     env,
  616  		LogStore: normalizeTaskLogStore(t.LogStore),
  617  	})

  104  		reqKey := fmt.Sprintf("task:create_req:%s", id.String())
  105  		val, err := h.redis.Get(ctx, reqKey).Result()
  106  		if err != nil {
  107  			h.logger.With("task_id", id, "error", err).ErrorContext(ctx, "failed to get CreateTaskReq from redis")
  108  			return fmt.Errorf("failed to get CreateTaskReq from Redis: %w", err)
  109  		}
  110  
  111  		defer h.redis.Del(ctx, reqKey)
  112  
  113  		if err := h.repo.Update(ctx, &domain.User{}, id, func(up *db.TaskUpdateOne) error {
  114  			up.SetStatus(consts.TaskStatusProcessing)
  115  			return nil
  116  		}); err != nil {
  117  			return fmt.Errorf("failed to update task status: %w", err)
  118  		}
  119  
  120  		var createReq taskflow.CreateTaskReq
  121  		if err := json.Unmarshal([]byte(val), &createReq); err != nil {
  122  			h.logger.With("task_id", id, "error", err).ErrorContext(ctx, "failed to unmarshal CreateTaskReq")
  123  			return fmt.Errorf("failed to unmarshal CreateTaskReq: %w", err)

  585  	var runtimeToken string
  586  	if keys := m.Edges.Apikeys; len(keys) > 0 {
  587  		m.APIKey = keys[0].APIKey
  588  		m.BaseURL = a.cfg.LLMProxy.BaseURL + "/v1"
  589  		runtimeToken = keys[0].APIKey
  590  	}

   28  const upstreamFailureMessage = "连接上游模型失败，请检查模型配置，或重试"
   29  
   30  var allowPaths = map[string]string{
   31  	"/v1/chat/completions": "/chat/completions",
   32  	"/v1/responses":        "/responses",
   33  	"/v1/messages":         "/messages",
   34  }

  165  // SwitchModel 切换运行中任务使用的模型
  166  func (a *TaskUsecase) SwitchModel(ctx context.Context, user *domain.User, taskID uuid.UUID, req domain.SwitchTaskModelReq) (*domain.SwitchTaskModelResp, error) {
  167  	t, owner, err := a.Info(ctx, user, taskID)
  168  	if err != nil {
  169  		return nil, err
  170  	}
  171  	if !owner && !a.isPrivileged(ctx, user.ID) {
  172  		return nil, errcode.ErrForbidden
  173  	}
  174  	if t.Status != consts.TaskStatusProcessing {
  175  		return nil, fmt.Errorf("task is not processing")
  176  	}
  177  	if t.VirtualMachine == nil {
  178  		return nil, fmt.Errorf("task virtual machine is nil")
  179  	}
  180  
  181  	taskOwnerID := t.UserID
  182  	if a.modelHook != nil {
  183  		if err := a.modelHook.ValidateAccess(ctx, taskOwnerID, req.ModelID.String()); err != nil {
  184  			return nil, err
  185  		}
  186  	}
  187  	model, err := a.modelRepo.Get(ctx, taskOwnerID, req.ModelID)
  188  	if err != nil {
  189  		return nil, err
      … 18 lines omitted; exact range 165–218 …
  208  	var skillIDs, pluginIDs []string
  209  	if t.Extra != nil {
  210  		skillIDs = t.Extra.SkillIDs
  211  		pluginIDs = t.Extra.PluginIDs
  212  	}
  213  	coding, configs, agentRes, err := a.getCodingConfigs(ctx, t.CliName, model, skillIDs, pluginIDs, a.userScope(ctx, user), false)
  214  	if err != nil {
  215  		return nil, err
  216  	}
  217  	if coding != taskflow.CodingAgentOpenCode {
  218  		return nil, fmt.Errorf("switch model only supports opencode runtime")

   99  // ScopeFilter constrains a listing or dispatch query to a subset of the
  100  // three scope tiers. Used by the *Scoped Repo methods.
  101  //
  102  // Semantic: rows whose scope_type is "global" are included when
  103  // IncludeGlobal is true; rows whose scope is "team" are included when
  104  // TeamID != nil and matches; same for user. Multiple flags may be set —
  105  // the result is the union, with name-based override (user > team > global)
  106  // applied by the call sites that need overrides (Listing / dispatch).
  107  type ScopeFilter struct {
  108  	IncludeGlobal bool
  109  	TeamID        *uuid.UUID
  110  	UserID        *uuid.UUID
  111  }
  112  
  113  // GlobalOnlyScope is the default scope for back-compat call sites that
  114  // haven't migrated to ScopeFilter yet. Matches the historical "global only"
  115  // behavior of the unsuffixed Repo methods.
  116  func GlobalOnlyScope() ScopeFilter {
  117  	return ScopeFilter{IncludeGlobal: true}
  118  }
  119  
  120  // SkillSelection bundles a ScopeFilter with the user-picked skill IDs for
  121  // the dispatch path. ListActiveSkillsScoped takes this so the same struct
  122  // flows from the task usecase down to the SQL.
  123  type SkillSelection struct {
  124  	Scope           ScopeFilter
  125  	UserSelectedIDs []uuid.UUID
  126  }

  741  func (a *TaskUsecase) buildMCPConfigs(taskID uuid.UUID, token string) []taskflow.McpServerConfig {
  742  	mcps := []taskflow.McpServerConfig{
  743  		{
  744  			Type: "http",
  745  			Name: "mcaiBuiltin",
  746  			Url:  proto.String(fmt.Sprintf("http://127.0.0.1:65510/mcp?task_id=%s", taskID.String())),
  747  		},
  748  	}
  749  
  750  	if token != "" {
  751  		mcps = append(mcps, taskflow.McpServerConfig{
  752  			Type: "http",
  753  			Name: "monkeycode-ai",
  754  			Url:  proto.String(fmt.Sprintf("%s/mcp", strings.TrimRight(a.cfg.Server.BaseURL, "/"))),
  755  			Headers: []*taskflow.McpHttpHeader{
  756  				{
  757  					Name:  "Authorization",
  758  					Value: fmt.Sprintf("Bearer %s", token),
  759  				},
  760  			},
  761  			Command: new(string),
  762  			Args:    []string{},
  763  			Env:     map[string]string{},
  764  		})
  765  	}
  766  
  767  	return mcps

   56  func (s *Service) ListEffectiveTools(ctx context.Context, userID uuid.UUID) ([]repo.ToolSnapshot, error) {
   57  	platform, err := s.listPlatformTools(ctx)
   58  	if err != nil {
   59  		return nil, err
   60  	}
   61  	userTools, err := s.repo.ListUserPublishedTools(ctx, userID)
   62  	if err != nil {
   63  		return nil, err
   64  	}
   65  	teamTools, err := s.repo.ListTeamPublishedTools(ctx, userID)
   66  	if err != nil {
   67  		return nil, err
   68  	}
   69  
   70  	settings := map[uuid.UUID]bool{}
   71  	if s.userSettings != nil {
   72  		settings, err = s.userSettings.ListEnabledMap(ctx, userID)
   73  		if err != nil {
   74  			return nil, err
   75  		}
   76  	}
   77  
   78  	tools := append(platform, userTools...)
   79  	tools = append(tools, teamTools...)
   80  	return applyUserSettingsDefaultEnabled(tools, settings), nil

  228  	if h.billing != nil {
  229  		if err := h.billing.CanConsume(ctx, call); err != nil {
  230  			_ = h.calls.MarkFailed(ctx, call.ID, err.Error(), "")
  231  			return nil, err
  232  		}
  233  	}
  234  
  235  	result, upstreamRequestID, err := h.upstream.CallTool(ctx, upstream, tool, CallToolParams{
  236  		Name:      tool.Name,
  237  		Arguments: args,
  238  	})
  239  	switch classifyUpstreamError(err) {
  240  	case repo.ToolCallStatusSuccess:
  241  		if err := h.calls.MarkSuccess(ctx, call.ID, result, upstreamRequestID); err != nil {
  242  			return nil, err
  243  		}
  244  		if h.billing != nil {
  245  			completed := *call
  246  			completed.Status = repo.ToolCallStatusSuccess
  247  			completed.ResultJSON = result
  248  			completed.UpstreamRequestID = upstreamRequestID
  249  			if err := h.billing.Consume(ctx, &completed); err != nil {
  250  				return nil, err
  251  			}

  788  func modelRuntimeDefaults(m *db.Model) (thinking bool, contextLimit int, outputLimit int) {
  789  	thinking = m.ThinkingEnabled
  790  	contextLimit = cmp.Or(m.ContextLimit, 200000)
  791  	outputLimit = cmp.Or(m.OutputLimit, 32000)
  792  	return thinking, contextLimit, outputLimit

   32  // TaskSummaryService 任务摘要生成服务
   33  type TaskSummaryService struct {
   34  	cfg                *config.Config
   35  	db                 *db.Client
   36  	llm                *llm.Client
   37  	summaryQueue       *delayqueue.TaskSummaryQueue
   38  	logger             *slog.Logger
   39  	conversationReader ConversationReader
   40  
   41  	// 生命周期管理
   42  	cancel context.CancelFunc
   43  	wg     sync.WaitGroup
   44  }
   45  
   46  type tasklogGateway interface {
   47  	QueryTurns(ctx context.Context, taskID uuid.UUID, taskCreatedAt time.Time, opts tasklog.QueryTurnsOpts, store consts.LogStore) (*tasklog.QueryTurnsResp, error)
   48  }
   49  
   50  type ConversationReader interface {
   51  	Fetch(ctx context.Context, taskID uuid.UUID, createdAt time.Time, store consts.LogStore, initialContent string, maxRounds int) ([]llm.Message, error)
   52  }
   53  
   54  type tasklogConversationReader struct {
   55  	gateway tasklogGateway
   56  	logger  *slog.Logger
      … 25 lines omitted; exact range 32–92 …
   82  		db:                 d,
   83  		llm:                llmClient,
   84  		summaryQueue:       sq,
   85  		logger:             logger,
   86  		conversationReader: newTasklogConversationReader(tlg, logger),
   87  	}
   88  
   89  	// 启动消费者
   90  	s.Start(context.Background())
   91  
   92  	return s, nil

   72  // VirtualMachine 虚拟机信息
   73  type VirtualMachine struct {
   74  	ID            string               `json:"id"`
   75  	AccessToken   string               `json:"access_token,omitempty"`
   76  	EnvironmentID string               `json:"environment_id"`
   77  	HostID        string               `json:"host_id"`
   78  	Hostname      string               `json:"hostname"`
   79  	Arch          string               `json:"arch"`
   80  	OS            string               `json:"os"`
   81  	Name          string               `json:"name"`
   82  	Repository    string               `json:"repository"`
   83  	Status        VirtualMachineStatus `json:"status"`
   84  	StatusMessage string               `json:"status_message"`
   85  	Cores         int32                `json:"cores"`
   86  	Memory        uint64               `json:"memory"`
   87  	Disk          uint64               `json:"disk"`
   88  	TTL           TTL                  `json:"ttl"`
   89  	ExternalIP    string               `json:"external_ip"`
   90  	CreatedAt     int64                `json:"created_at"`
   91  	Version       string               `json:"version"`
   92  }

  157  // AutoApprove implements domain.TaskUsecase.
  158  func (a *TaskUsecase) AutoApprove(ctx context.Context, _ *domain.User, id uuid.UUID, approve bool) error {
  159  	return a.taskflow.TaskManager().AutoApprove(ctx, taskflow.TaskApproveReq{
  160  		ID:          id,
  161  		AutoApprove: &approve,
  162  	})
  163  }

   12  // UnzipLimits guards the in-memory unzipper against zip bombs and overly
   13  // large archives. All limits are inclusive of the value (size == limit is OK).
   14  type UnzipLimits struct {
   15  	MaxFileSize  int64 // per-entry uncompressed size
   16  	MaxTotalSize int64 // sum of all entries' uncompressed sizes
   17  	MaxFiles     int   // max number of file entries
   18  }
   19  
   20  // DefaultUnzipLimits is the policy used by the Resolver.
   21  var DefaultUnzipLimits = UnzipLimits{
   22  	MaxFileSize:  32 << 20,  // 32 MiB
   23  	MaxTotalSize: 256 << 20, // 256 MiB
   24  	MaxFiles:     1000,
   25  }
   26  
   27  // unzipToMemory reads a zip archive entirely from memory and returns its
   28  // regular file entries. Directories are skipped. Paths are validated to
   29  // reject zip-slip ("../..") and absolute paths. Entries that exceed the
   30  // limits cause the entire archive to be rejected so callers can fall back
   31  // to skipping the whole asset (matching the Resolver's per-skill policy).

   53  func (g *Guard) ValidateURL(ctx context.Context, rawURL string) error {
   54  	u, err := url.Parse(strings.TrimSpace(rawURL))
   55  	if err != nil {
   56  		return fmt.Errorf("parse url: %w", err)
   57  	}
   58  	if u.Scheme != "http" && u.Scheme != "https" {
   59  		return fmt.Errorf("unsupported scheme %q", u.Scheme)
   60  	}
   61  	if u.Hostname() == "" {
   62  		return errors.New("empty host")
   63  	}
   64  	if !g.Enabled() {
   65  		return nil
   66  	}
   67  	_, err = g.resolveHost(ctx, u.Hostname())
   68  	return err
   69  }

  795  // agentRuleBaseDir / agentSkillBaseDir / agentPluginBaseDir 是 codingmatrix 在
  796  // VM 内部约定的 .ai-ready/ 投放路径。rule 走 .md 平铺；skill / plugin 解 zip
  797  // 后按目录结构展开；plugin 的 entry 字段再以 file:// 注入到 opencode.json 的
  798  // `plugin` 数组里。Claude / Codex 不消费 plugin（spec §6.3）。
  799  const (
  800  	agentRuleBaseDir   = "${HOME}/.codingmatrix/project-tpl/.ai-ready/rules/"
  801  	agentSkillBaseDir  = "${HOME}/.codingmatrix/project-tpl/.ai-ready/skills/"
  802  	agentPluginBaseDir = "${HOME}/.codingmatrix/project-tpl/.ai-ready/plugins/"
  803  )

  323  // Stream 任务数据流 WebSocket
  324  //
  325  //	@Summary		任务数据流 WebSocket
  326  //	@Description	功能定位：该接口通过 WebSocket 转发任务运行数据。任务对话继续输入使用 `type=user-input`。
  327  //	@Description	数据格式约定：当前仅支持文本帧透传。服务端将 Agent 的原始文本数据包装为如下结构返回给前端（对应 domain.TaskStream）：
  328  //	@Description	```json
  329  //	@Description	{ "type": "string", "data": "string", "kind": "string", "timestamp": 0 }
  330  //	@Description	```
  331  //	@Description	user-input 上行新格式：
  332  //	@Description	```json
  333  //	@Description	{ "type": "user-input", "data": "{\"content\":\"57un57ut5aSE55CG6L+Z5Liq6Zeu6aKY\",\"attachments\":[{\"url\":\"https://example-bucket.oss-cn-hangzhou.aliyuncs.com/temp/a.txt\",\"filename\":\"a.txt\"}]}" }
  334  //	@Description	```
  335  //	@Description	user-input 上行旧格式仍兼容：
  336  //	@Description	```json
  337  //	@Description	{ "type": "user-input", "data": "继续处理这个问题" }
  338  //	@Description	```
  339  //	@Description	user-input 下行和历史返回统一使用新 JSON payload 字符串：
  340  //	@Description	```json
  341  //	@Description	{ "type": "user-input", "data": "{\"content\":\"57un57ut5aSE55CG6L+Z5Liq6Zeu6aKY\",\"attachments\":[]}", "timestamp": 0 }
  342  //	@Description	```
  343  //	@Description	`attachments` 为可选附件列表，最多 10 个；每项包含 `url` 和 `filename`，URL 需要匹配后端配置的附件白名单前缀。
  344  //	@Description	type 字段说明：
  345  //	@Description	- task-started: 本轮任务启动
  346  //	@Description	- task-ended: 本轮任务结束
  347  //	@Description	- task-error: 本轮任务发生错误
      … 26 lines omitted; exact range 323–384 …
  374  func (h *TaskHandler) Stream(c *web.Context, req domain.TaskStreamReq) error {
  375  	user := middleware.GetUser(c)
  376  	task, owner, err := h.usecase.Info(c.Request().Context(), user, req.ID)
  377  	if err != nil {
  378  		return err
  379  	}
  380  
  381  	if req.Mode == "" {
  382  		req.Mode = "new"
  383  	}
  384  	return h.stream(c, user, task, owner, req.Mode)

  626  // CreateTaskReq 创建任务请求
  627  type CreateTaskReq struct {
  628  	ID             uuid.UUID         `json:"id"`
  629  	VMID           string            `json:"vm_id"`
  630  	SystemPrompt   string            `json:"system_prompt,omitempty"`
  631  	Text           string            `json:"text,omitempty"`
  632  	Attachments    []Attachment      `json:"attachments,omitempty"`
  633  	LLM            LLM               `json:"llm,omitzero"`
  634  	CodingAgent    CodingAgent       `json:"coding_agent,omitempty"`
  635  	Configs        []ConfigFile      `json:"configs,omitzero"`
  636  	McpConfigs     []McpServerConfig `json:"mcp_configs,omitzero"`
  637  	Env            map[string]string `json:"env,omitempty"`
  638  	LogStore       string            `json:"log_store,omitempty"`
  639  	AgentResources *AgentResources   `json:"agent_resources,omitempty"` // skill/plugin presigned URLs + rule content forwarded to codingmatrix agent
  640  }

  246  func (p *Proxy) modifyResponse(resp *http.Response) error {
  247  	if resp == nil || resp.Body == nil {
  248  		return nil
  249  	}
  250  	if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
  251  		return nil
  252  	}
  253  	ctx, ok := resp.Request.Context().Value(contextKey{}).(*proxyContext)
  254  	if !ok || ctx == nil || ctx.model == nil {
  255  		return nil
  256  	}
  257  	resp.Body = NewUsageCapture(p.logger, resp.Body, &UsageCaptureContext{
  258  		ctx:      resp.Request.Context(),
  259  		path:     normalizeUsageCapturePath(resp.Request.URL.Path),
  260  		stream:   ctx.stream,
  261  		proxyCtx: ctx,
  262  		proxy:    p,
  263  	})
  264  	return nil

   29  func Setup(ctx context.Context, cfg Config) (Shutdown, error) {
   30  	otel.SetTextMapPropagator(propagation.TraceContext{})
   31  	if !enabled(cfg) {
   32  		return func(context.Context) error { return nil }, nil
   33  	}
   34  
   35  	opts := []otlptracegrpc.Option{}
   36  	if endpoint := strings.TrimSpace(cfg.Endpoint); endpoint != "" {
   37  		if strings.Contains(endpoint, "://") {
   38  			opts = append(opts, otlptracegrpc.WithEndpointURL(endpoint))
   39  		} else {
   40  			opts = append(opts, otlptracegrpc.WithEndpoint(endpoint))
   41  		}
   42  	}
   43  	if cfg.Insecure {
   44  		opts = append(opts, otlptracegrpc.WithInsecure())
   45  	}
   46  
   47  	exporter, err := otlptracegrpc.New(ctx, opts...)
   48  	if err != nil {
   49  		return nil, err
   50  	}
   51  	tp, ok := otel.GetTracerProvider().(*sdktrace.TracerProvider)
   52  	if !ok {
   53  		_ = exporter.Shutdown(ctx)
      … 1 lines omitted; exact range 29–65 …
   55  	}
   56  
   57  	processor := sdktrace.NewBatchSpanProcessor(
   58  		&resourceExporter{SpanExporter: exporter, resource: buildResource(cfg)},
   59  		sdktrace.WithMaxQueueSize(2048),
   60  		sdktrace.WithMaxExportBatchSize(512),
   61  		sdktrace.WithBatchTimeout(5*time.Second),
   62  		sdktrace.WithExportTimeout(5*time.Second),
   63  	)
   64  	tp.RegisterSpanProcessor(processor)
   65  	return processor.Shutdown, nil