package daemon import ( "bytes" "context" "encoding/json" "errors" "fmt" "io" "net/http" "runtime" "strings" "time" "github.com/multica-ai/multica/server/pkg/protocol" ) // requestError is returned by postJSON/getJSON when the server responds with an error status. type requestError struct { Method string Path string StatusCode int Body string } func (e *requestError) Error() string { return fmt.Sprintf("%s %s returned %d: %s", e.Method, e.Path, e.StatusCode, e.Body) } // isWorkspaceNotFoundError returns true if the error is a 404 with "workspace not found" body. func isWorkspaceNotFoundError(err error) bool { var reqErr *requestError if !errors.As(err, &reqErr) { return false } if reqErr.StatusCode != http.StatusNotFound { return false } return strings.Contains(strings.ToLower(reqErr.Body), "workspace not found") } // Client handles HTTP communication with the Multica server daemon API. type Client struct { baseURL string token string client *http.Client // Identity headers sent on every request as X-Client-*. Populated by // SetIdentity(); empty values are simply omitted. platform string version string os string } // NewClient creates a new daemon API client. func NewClient(baseURL string) *Client { return &Client{ baseURL: baseURL, client: &http.Client{Timeout: 30 * time.Second}, platform: "daemon", os: normalizeGOOS(runtime.GOOS), } } // normalizeGOOS maps Go's runtime.GOOS values to the protocol vocabulary // used by X-Client-OS / client_os ("macos" / "windows" / "linux"). func normalizeGOOS(goos string) string { switch goos { case "darwin": return "macos" case "windows": return "windows" case "linux": return "linux" default: return goos } } // SetVersion records the daemon's CLI version, sent as X-Client-Version. // Called by Daemon.Run after config is loaded. func (c *Client) SetVersion(v string) { c.version = v } // setIdentityHeaders attaches X-Client-Platform/Version/OS to req when set. func (c *Client) setIdentityHeaders(req *http.Request) { if c.platform != "" { req.Header.Set("X-Client-Platform", c.platform) } if c.version != "" { req.Header.Set("X-Client-Version", c.version) } if c.os != "" { req.Header.Set("X-Client-OS", c.os) } } // SetToken sets the auth token for authenticated requests. func (c *Client) SetToken(token string) { c.token = token } // Token returns the current auth token. func (c *Client) Token() string { return c.token } func (c *Client) ClaimTask(ctx context.Context, runtimeID string) (*Task, error) { var resp struct { Task *Task `json:"task"` } if err := c.postJSON(ctx, fmt.Sprintf("/api/daemon/runtimes/%s/tasks/claim", runtimeID), map[string]any{}, &resp); err != nil { return nil, err } return resp.Task, nil } func (c *Client) StartTask(ctx context.Context, taskID string) error { return c.postJSON(ctx, fmt.Sprintf("/api/daemon/tasks/%s/start", taskID), map[string]any{}, nil) } func (c *Client) ReportProgress(ctx context.Context, taskID, summary string, step, total int) error { return c.postJSON(ctx, fmt.Sprintf("/api/daemon/tasks/%s/progress", taskID), map[string]any{ "summary": summary, "step": step, "total": total, }, nil) } // TaskMessageData represents a single agent execution message for batch reporting. type TaskMessageData struct { Seq int `json:"seq"` Type string `json:"type"` Tool string `json:"tool,omitempty"` Content string `json:"content,omitempty"` Input map[string]any `json:"input,omitempty"` Output string `json:"output,omitempty"` } func (c *Client) ReportTaskMessages(ctx context.Context, taskID string, messages []TaskMessageData) error { return c.postJSON(ctx, fmt.Sprintf("/api/daemon/tasks/%s/messages", taskID), map[string]any{ "messages": messages, }, nil) } func (c *Client) CompleteTask(ctx context.Context, taskID, output, branchName, sessionID, workDir string) error { body := map[string]any{"output": output} if branchName != "" { body["branch_name"] = branchName } if sessionID != "" { body["session_id"] = sessionID } if workDir != "" { body["work_dir"] = workDir } return c.postJSON(ctx, fmt.Sprintf("/api/daemon/tasks/%s/complete", taskID), body, nil) } func (c *Client) ReportTaskUsage(ctx context.Context, taskID string, usage []TaskUsageEntry) error { if len(usage) == 0 { return nil } return c.postJSON(ctx, fmt.Sprintf("/api/daemon/tasks/%s/usage", taskID), map[string]any{ "usage": usage, }, nil) } func (c *Client) FailTask(ctx context.Context, taskID, errMsg, sessionID, workDir, failureReason string) error { body := map[string]any{"error": errMsg} if sessionID != "" { body["session_id"] = sessionID } if workDir != "" { body["work_dir"] = workDir } if failureReason != "" { body["failure_reason"] = failureReason } return c.postJSON(ctx, fmt.Sprintf("/api/daemon/tasks/%s/fail", taskID), body, nil) } // PinTaskSession persists the agent's session_id and work_dir on the task // row mid-flight so a daemon crash doesn't lose the resume pointer. func (c *Client) PinTaskSession(ctx context.Context, taskID, sessionID, workDir string) error { if sessionID == "" && workDir == "" { return nil } body := map[string]any{} if sessionID != "" { body["session_id"] = sessionID } if workDir != "" { body["work_dir"] = workDir } return c.postJSON(ctx, fmt.Sprintf("/api/daemon/tasks/%s/session", taskID), body, nil) } // RecoverOrphans tells the server to fail any dispatched/running tasks the // previous daemon process for this runtime left behind. The server will // auto-retry eligible tasks. func (c *Client) RecoverOrphans(ctx context.Context, runtimeID string) error { return c.postJSON(ctx, fmt.Sprintf("/api/daemon/runtimes/%s/recover-orphans", runtimeID), map[string]any{}, nil) } // GetTaskStatus returns the current status of a task. Used by the daemon to // detect if a task was cancelled while it was executing. func (c *Client) GetTaskStatus(ctx context.Context, taskID string) (string, error) { var resp struct { Status string `json:"status"` } if err := c.getJSON(ctx, fmt.Sprintf("/api/daemon/tasks/%s/status", taskID), &resp); err != nil { return "", err } return resp.Status, nil } // HeartbeatResponse, PendingUpdate, etc. alias the wire types so HTTP and WS // heartbeat paths share a single type and a single decoder shape. Aliases // (rather than wrappers) keep call sites unchanged. type ( HeartbeatResponse = protocol.DaemonHeartbeatAckPayload PendingUpdate = protocol.DaemonHeartbeatPendingUpdate PendingModelList = protocol.DaemonHeartbeatPendingModelList PendingLocalSkills = protocol.DaemonHeartbeatPendingLocalSkills PendingLocalSkillImport = protocol.DaemonHeartbeatPendingLocalSkillImport ) func (c *Client) SendHeartbeat(ctx context.Context, runtimeID string) (*HeartbeatResponse, error) { var resp HeartbeatResponse if err := c.postJSON(ctx, "/api/daemon/heartbeat", map[string]string{ "runtime_id": runtimeID, }, &resp); err != nil { return nil, err } return &resp, nil } // ReportUpdateResult sends the CLI update result back to the server. func (c *Client) ReportUpdateResult(ctx context.Context, runtimeID, updateID string, result map[string]any) error { return c.postJSON(ctx, fmt.Sprintf("/api/daemon/runtimes/%s/update/%s/result", runtimeID, updateID), result, nil) } // ReportModelListResult sends the model-discovery result back to the server. func (c *Client) ReportModelListResult(ctx context.Context, runtimeID, requestID string, result map[string]any) error { return c.postJSON(ctx, fmt.Sprintf("/api/daemon/runtimes/%s/models/%s/result", runtimeID, requestID), result, nil) } // ReportLocalSkillListResult sends the runtime-local-skill inventory back to the server. func (c *Client) ReportLocalSkillListResult(ctx context.Context, runtimeID, requestID string, result map[string]any) error { return c.postJSON(ctx, fmt.Sprintf("/api/daemon/runtimes/%s/local-skills/%s/result", runtimeID, requestID), result, nil) } // ReportLocalSkillImportResult sends a runtime-local-skill bundle back to the server. func (c *Client) ReportLocalSkillImportResult(ctx context.Context, runtimeID, requestID string, result map[string]any) error { return c.postJSON(ctx, fmt.Sprintf("/api/daemon/runtimes/%s/local-skills/import/%s/result", runtimeID, requestID), result, nil) } // WorkspaceInfo holds minimal workspace metadata returned by the API. type WorkspaceInfo struct { ID string `json:"id"` Name string `json:"name"` } // ListWorkspaces fetches all workspaces the authenticated user belongs to. func (c *Client) ListWorkspaces(ctx context.Context) ([]WorkspaceInfo, error) { var workspaces []WorkspaceInfo if err := c.getJSON(ctx, "/api/workspaces", &workspaces); err != nil { return nil, err } return workspaces, nil } // IssueGCStatus holds the minimal issue info returned by the GC check endpoint. type IssueGCStatus struct { Status string `json:"status"` UpdatedAt time.Time `json:"updated_at"` } // GetIssueGCCheck returns the status and updated_at of an issue for GC decisions. func (c *Client) GetIssueGCCheck(ctx context.Context, issueID string) (*IssueGCStatus, error) { var resp IssueGCStatus if err := c.getJSON(ctx, fmt.Sprintf("/api/daemon/issues/%s/gc-check", issueID), &resp); err != nil { return nil, err } return &resp, nil } func (c *Client) Deregister(ctx context.Context, runtimeIDs []string) error { return c.postJSON(ctx, "/api/daemon/deregister", map[string]any{ "runtime_ids": runtimeIDs, }, nil) } // RegisterResponse holds the server's response to a daemon registration. type RegisterResponse struct { Runtimes []Runtime `json:"runtimes"` Repos []RepoData `json:"repos"` ReposVersion string `json:"repos_version"` } func (c *Client) Register(ctx context.Context, req map[string]any) (*RegisterResponse, error) { var resp RegisterResponse if err := c.postJSON(ctx, "/api/daemon/register", req, &resp); err != nil { return nil, err } return &resp, nil } type WorkspaceReposResponse struct { WorkspaceID string `json:"workspace_id"` Repos []RepoData `json:"repos"` ReposVersion string `json:"repos_version"` } func (c *Client) GetWorkspaceRepos(ctx context.Context, workspaceID string) (*WorkspaceReposResponse, error) { var resp WorkspaceReposResponse if err := c.getJSON(ctx, fmt.Sprintf("/api/daemon/workspaces/%s/repos", workspaceID), &resp); err != nil { return nil, err } return &resp, nil } func (c *Client) postJSON(ctx context.Context, path string, reqBody any, respBody any) error { var body io.Reader if reqBody != nil { data, err := json.Marshal(reqBody) if err != nil { return err } body = bytes.NewReader(data) } req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+path, body) if err != nil { return err } req.Header.Set("Content-Type", "application/json") if c.token != "" { req.Header.Set("Authorization", "Bearer "+c.token) } c.setIdentityHeaders(req) resp, err := c.client.Do(req) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode >= 400 { data, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) return &requestError{Method: http.MethodPost, Path: path, StatusCode: resp.StatusCode, Body: strings.TrimSpace(string(data))} } if respBody == nil { io.Copy(io.Discard, resp.Body) return nil } return json.NewDecoder(resp.Body).Decode(respBody) } func (c *Client) getJSON(ctx context.Context, path string, respBody any) error { req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.baseURL+path, nil) if err != nil { return err } if c.token != "" { req.Header.Set("Authorization", "Bearer "+c.token) } c.setIdentityHeaders(req) resp, err := c.client.Do(req) if err != nil { return err } defer resp.Body.Close() if resp.StatusCode >= 400 { data, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) return &requestError{Method: http.MethodGet, Path: path, StatusCode: resp.StatusCode, Body: strings.TrimSpace(string(data))} } if respBody == nil { io.Copy(io.Discard, resp.Body) return nil } return json.NewDecoder(resp.Body).Decode(respBody) }