marketing-shibata50/multica
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(runtime): add local codex daemon pairing

Browse source
This commit is contained in:
Jiayuan Zhang 2026-03-24 12:03:14 +08:00
parent c6960d39b9
commit cdfa63af15
36 changed files with 2579 additions and 309 deletions
Download patch file Download diff file Expand all files Collapse all files

41
server/internal/handler/agent.go
View file

@ -12,6 +12,7 @@ import (
type AgentResponse struct {
ID string `json:"id"`
WorkspaceID string `json:"workspace_id"`
RuntimeID string `json:"runtime_id"`
Name string `json:"name"`
Description string `json:"description"`
AvatarURL *string `json:"avatar_url"`
@ -56,6 +57,7 @@ func agentToResponse(a db.Agent) AgentResponse {
return AgentResponse{
ID: uuidToString(a.ID),
WorkspaceID: uuidToString(a.WorkspaceID),
RuntimeID: uuidToString(a.RuntimeID),
Name: a.Name,
Description: a.Description,
AvatarURL: textToPtr(a.AvatarUrl),
@ -76,6 +78,7 @@ func agentToResponse(a db.Agent) AgentResponse {
type AgentTaskResponse struct {
ID string `json:"id"`
AgentID string `json:"agent_id"`
RuntimeID string `json:"runtime_id"`
IssueID string `json:"issue_id"`
Status string `json:"status"`
Priority int32 `json:"priority"`
@ -100,6 +103,7 @@ func taskToResponse(t db.AgentTaskQueue) AgentTaskResponse {
return AgentTaskResponse{
ID: uuidToString(t.ID),
AgentID: uuidToString(t.AgentID),
RuntimeID: uuidToString(t.RuntimeID),
IssueID: uuidToString(t.IssueID),
Status: t.Status,
Priority: t.Priority,
@ -146,7 +150,7 @@ type CreateAgentRequest struct {
Name string `json:"name"`
Description string `json:"description"`
AvatarURL *string `json:"avatar_url"`
RuntimeMode string `json:"runtime_mode"`
RuntimeID string `json:"runtime_id"`
RuntimeConfig any `json:"runtime_config"`
Visibility string `json:"visibility"`
MaxConcurrentTasks int32 `json:"max_concurrent_tasks"`
@ -176,8 +180,9 @@ func (h *Handler) CreateAgent(w http.ResponseWriter, r *http.Request) {
writeError(w, http.StatusBadRequest, "name is required")
return
}
if req.RuntimeMode == "" {
req.RuntimeMode = "local"
if req.RuntimeID == "" {
writeError(w, http.StatusBadRequest, "runtime_id is required")
return
}
if req.Visibility == "" {
req.Visibility = "workspace"
@ -186,6 +191,15 @@ func (h *Handler) CreateAgent(w http.ResponseWriter, r *http.Request) {
req.MaxConcurrentTasks = 1
}
runtime, err := h.Queries.GetAgentRuntimeForWorkspace(r.Context(), db.GetAgentRuntimeForWorkspaceParams{
ID: parseUUID(req.RuntimeID),
WorkspaceID: parseUUID(workspaceID),
})
if err != nil {
writeError(w, http.StatusBadRequest, "invalid runtime_id")
return
}
rc, _ := json.Marshal(req.RuntimeConfig)
if req.RuntimeConfig == nil {
rc = []byte("{}")
@ -206,8 +220,9 @@ func (h *Handler) CreateAgent(w http.ResponseWriter, r *http.Request) {
Name: req.Name,
Description: req.Description,
AvatarUrl: ptrToText(req.AvatarURL),
RuntimeMode: req.RuntimeMode,
RuntimeMode: runtime.RuntimeMode,
RuntimeConfig: rc,
RuntimeID: runtime.ID,
Visibility: req.Visibility,
MaxConcurrentTasks: req.MaxConcurrentTasks,
OwnerID: parseUUID(ownerID),
@ -220,6 +235,11 @@ func (h *Handler) CreateAgent(w http.ResponseWriter, r *http.Request) {
return
}
if runtime.Status == "online" {
h.TaskService.ReconcileAgentStatus(r.Context(), agent.ID)
agent, _ = h.Queries.GetAgent(r.Context(), agent.ID)
}
writeJSON(w, http.StatusCreated, agentToResponse(agent))
}
@ -227,6 +247,7 @@ type UpdateAgentRequest struct {
Name *string `json:"name"`
Description *string `json:"description"`
AvatarURL *string `json:"avatar_url"`
RuntimeID *string `json:"runtime_id"`
RuntimeConfig any `json:"runtime_config"`
Visibility *string `json:"visibility"`
Status *string `json:"status"`
@ -268,6 +289,18 @@ func (h *Handler) UpdateAgent(w http.ResponseWriter, r *http.Request) {
rc, _ := json.Marshal(req.RuntimeConfig)
params.RuntimeConfig = rc
}
if req.RuntimeID != nil {
runtime, err := h.Queries.GetAgentRuntimeForWorkspace(r.Context(), db.GetAgentRuntimeForWorkspaceParams{
ID: parseUUID(*req.RuntimeID),
WorkspaceID: agent.WorkspaceID,
})
if err != nil {
writeError(w, http.StatusBadRequest, "invalid runtime_id")
return
}
params.RuntimeID = runtime.ID
params.RuntimeMode = pgtype.Text{String: runtime.RuntimeMode, Valid: true}
}
if req.Visibility != nil {
params.Visibility = pgtype.Text{String: *req.Visibility, Valid: true}
}

118
server/internal/handler/daemon.go
View file

@ -2,7 +2,9 @@ package handler
import (
"encoding/json"
"fmt"
"net/http"
"strings"
"github.com/go-chi/chi/v5"
db "github.com/multica-ai/multica/server/pkg/db/generated"
@ -13,9 +15,11 @@ import (
// ---------------------------------------------------------------------------
type DaemonRegisterRequest struct {
DaemonID string `json:"daemon_id"`
AgentID string `json:"agent_id"`
Runtimes []struct {
WorkspaceID string `json:"workspace_id"`
DaemonID string `json:"daemon_id"`
DeviceName string `json:"device_name"`
Runtimes []struct {
Name string `json:"name"`
Type string `json:"type"`
Version string `json:"version"`
Status string `json:"status"`
@ -29,36 +33,68 @@ func (h *Handler) DaemonRegister(w http.ResponseWriter, r *http.Request) {
return
}
if req.DaemonID == "" || req.AgentID == "" {
writeError(w, http.StatusBadRequest, "daemon_id and agent_id are required")
if req.DaemonID == "" {
writeError(w, http.StatusBadRequest, "daemon_id is required")
return
}
if req.WorkspaceID == "" {
writeError(w, http.StatusBadRequest, "workspace_id is required")
return
}
if len(req.Runtimes) == 0 {
writeError(w, http.StatusBadRequest, "at least one runtime is required")
return
}
runtimeInfo, _ := json.Marshal(req.Runtimes)
resp := make([]AgentRuntimeResponse, 0, len(req.Runtimes))
for _, runtime := range req.Runtimes {
provider := strings.TrimSpace(runtime.Type)
if provider == "" {
provider = "unknown"
}
name := strings.TrimSpace(runtime.Name)
if name == "" {
name = provider
if req.DeviceName != "" {
name = fmt.Sprintf("%s (%s)", provider, req.DeviceName)
}
}
deviceInfo := strings.TrimSpace(req.DeviceName)
if runtime.Version != "" && deviceInfo != "" {
deviceInfo = fmt.Sprintf("%s · %s", deviceInfo, runtime.Version)
} else if runtime.Version != "" {
deviceInfo = runtime.Version
}
status := "online"
if runtime.Status == "offline" {
status = "offline"
}
metadata, _ := json.Marshal(map[string]any{
"version": runtime.Version,
})
conn, err := h.Queries.UpsertDaemonConnection(r.Context(), db.UpsertDaemonConnectionParams{
AgentID: parseUUID(req.AgentID),
DaemonID: req.DaemonID,
RuntimeInfo: runtimeInfo,
})
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to register daemon: "+err.Error())
return
registered, err := h.Queries.UpsertAgentRuntime(r.Context(), db.UpsertAgentRuntimeParams{
WorkspaceID: parseUUID(req.WorkspaceID),
DaemonID: strToText(req.DaemonID),
Name: name,
RuntimeMode: "local",
Provider: provider,
Status: status,
DeviceInfo: deviceInfo,
Metadata: metadata,
})
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to register runtime: "+err.Error())
return
}
resp = append(resp, runtimeToResponse(registered))
}
// Reconcile agent status (set to idle if no running tasks)
h.TaskService.ReconcileAgentStatus(r.Context(), parseUUID(req.AgentID))
writeJSON(w, http.StatusOK, map[string]any{
"connection_id": uuidToString(conn.ID),
"status": conn.Status,
})
writeJSON(w, http.StatusOK, map[string]any{"runtimes": resp})
}
type DaemonHeartbeatRequest struct {
DaemonID string `json:"daemon_id"`
AgentID string `json:"agent_id"`
CurrentTasks int `json:"current_tasks"`
RuntimeID string `json:"runtime_id"`
}
func (h *Handler) DaemonHeartbeat(w http.ResponseWriter, r *http.Request) {
@ -68,10 +104,12 @@ func (h *Handler) DaemonHeartbeat(w http.ResponseWriter, r *http.Request) {
return
}
err := h.Queries.UpdateDaemonHeartbeat(r.Context(), db.UpdateDaemonHeartbeatParams{
DaemonID: req.DaemonID,
AgentID: parseUUID(req.AgentID),
})
if req.RuntimeID == "" {
writeError(w, http.StatusBadRequest, "runtime_id is required")
return
}
_, err := h.Queries.UpdateAgentRuntimeHeartbeat(r.Context(), parseUUID(req.RuntimeID))
if err != nil {
writeError(w, http.StatusInternalServerError, "heartbeat failed")
return
@ -80,15 +118,11 @@ func (h *Handler) DaemonHeartbeat(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
}
// ---------------------------------------------------------------------------
// Task Lifecycle (called by daemon)
// ---------------------------------------------------------------------------
// ClaimTaskByRuntime atomically claims the next queued task for a runtime.
func (h *Handler) ClaimTaskByRuntime(w http.ResponseWriter, r *http.Request) {
runtimeID := chi.URLParam(r, "runtimeId")
// ClaimTask atomically claims the next queued task for an agent.
func (h *Handler) ClaimTask(w http.ResponseWriter, r *http.Request) {
agentID := chi.URLParam(r, "agentId")
task, err := h.TaskService.ClaimTask(r.Context(), parseUUID(agentID))
task, err := h.TaskService.ClaimTaskForRuntime(r.Context(), parseUUID(runtimeID))
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to claim task: "+err.Error())
return
@ -102,11 +136,11 @@ func (h *Handler) ClaimTask(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, map[string]any{"task": taskToResponse(*task)})
}
// ListPendingTasks returns queued/dispatched tasks for an agent.
func (h *Handler) ListPendingTasks(w http.ResponseWriter, r *http.Request) {
agentID := chi.URLParam(r, "agentId")
// ListPendingTasksByRuntime returns queued/dispatched tasks for a runtime.
func (h *Handler) ListPendingTasksByRuntime(w http.ResponseWriter, r *http.Request) {
runtimeID := chi.URLParam(r, "runtimeId")
tasks, err := h.Queries.ListPendingTasks(r.Context(), parseUUID(agentID))
tasks, err := h.Queries.ListPendingTasksByRuntime(r.Context(), parseUUID(runtimeID))
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to list pending tasks")
return
@ -120,6 +154,10 @@ func (h *Handler) ListPendingTasks(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, resp)
}
// ---------------------------------------------------------------------------
// Task Lifecycle (called by daemon)
// ---------------------------------------------------------------------------
// StartTask marks a dispatched task as running.
func (h *Handler) StartTask(w http.ResponseWriter, r *http.Request) {
taskID := chi.URLParam(r, "taskId")

386
server/internal/handler/daemon_pairing.go Normal file
View file

@ -0,0 +1,386 @@
package handler
import (
"context"
"crypto/rand"
"encoding/hex"
"encoding/json"
"fmt"
"net/http"
"net/url"
"os"
"strings"
"time"
"github.com/go-chi/chi/v5"
"github.com/jackc/pgx/v5/pgtype"
)
const daemonPairingTTL = 10 * time.Minute
type daemonPairingSessionRecord struct {
Token string
DaemonID string
DeviceName string
RuntimeName string
RuntimeType string
RuntimeVersion string
WorkspaceID pgtype.UUID
ApprovedBy pgtype.UUID
Status string
ApprovedAt pgtype.Timestamptz
ClaimedAt pgtype.Timestamptz
ExpiresAt pgtype.Timestamptz
CreatedAt pgtype.Timestamptz
UpdatedAt pgtype.Timestamptz
}
type DaemonPairingSessionResponse struct {
Token string `json:"token"`
DaemonID string `json:"daemon_id"`
DeviceName string `json:"device_name"`
RuntimeName string `json:"runtime_name"`
RuntimeType string `json:"runtime_type"`
RuntimeVersion string `json:"runtime_version"`
WorkspaceID *string `json:"workspace_id"`
Status string `json:"status"`
ApprovedAt *string `json:"approved_at"`
ClaimedAt *string `json:"claimed_at"`
ExpiresAt string `json:"expires_at"`
CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"`
LinkURL *string `json:"link_url,omitempty"`
}
type CreateDaemonPairingSessionRequest struct {
DaemonID string `json:"daemon_id"`
DeviceName string `json:"device_name"`
RuntimeName string `json:"runtime_name"`
RuntimeType string `json:"runtime_type"`
RuntimeVersion string `json:"runtime_version"`
}
type ApproveDaemonPairingSessionRequest struct {
WorkspaceID string `json:"workspace_id"`
}
func daemonAppBaseURL() string {
for _, key := range []string{"MULTICA_APP_URL", "FRONTEND_ORIGIN"} {
if value := strings.TrimSpace(os.Getenv(key)); value != "" {
return strings.TrimRight(value, "/")
}
}
return "http://localhost:3000"
}
func daemonPairingLinkURL(token string) string {
base := daemonAppBaseURL()
return base + "/pair/local?token=" + url.QueryEscape(token)
}
func daemonPairingSessionToResponse(rec daemonPairingSessionRecord, includeLink bool) DaemonPairingSessionResponse {
resp := DaemonPairingSessionResponse{
Token: rec.Token,
DaemonID: rec.DaemonID,
DeviceName: rec.DeviceName,
RuntimeName: rec.RuntimeName,
RuntimeType: rec.RuntimeType,
RuntimeVersion: rec.RuntimeVersion,
WorkspaceID: uuidToPtr(rec.WorkspaceID),
Status: rec.Status,
ApprovedAt: timestampToPtr(rec.ApprovedAt),
ClaimedAt: timestampToPtr(rec.ClaimedAt),
ExpiresAt: timestampToString(rec.ExpiresAt),
CreatedAt: timestampToString(rec.CreatedAt),
UpdatedAt: timestampToString(rec.UpdatedAt),
}
if includeLink {
link := daemonPairingLinkURL(rec.Token)
resp.LinkURL = &link
}
return resp
}
func randomDaemonPairingToken() (string, error) {
bytes := make([]byte, 16)
if _, err := rand.Read(bytes); err != nil {
return "", err
}
return hex.EncodeToString(bytes), nil
}
func (h *Handler) getDaemonPairingSession(ctx context.Context, token string) (daemonPairingSessionRecord, error) {
if h.DB == nil {
return daemonPairingSessionRecord{}, fmt.Errorf("database executor is not configured")
}
var rec daemonPairingSessionRecord
err := h.DB.QueryRow(ctx, `
SELECT
token,
daemon_id,
device_name,
runtime_name,
runtime_type,
runtime_version,
workspace_id,
approved_by,
status,
approved_at,
claimed_at,
expires_at,
created_at,
updated_at
FROM daemon_pairing_session
WHERE token = $1
`, token).Scan(
&rec.Token,
&rec.DaemonID,
&rec.DeviceName,
&rec.RuntimeName,
&rec.RuntimeType,
&rec.RuntimeVersion,
&rec.WorkspaceID,
&rec.ApprovedBy,
&rec.Status,
&rec.ApprovedAt,
&rec.ClaimedAt,
&rec.ExpiresAt,
&rec.CreatedAt,
&rec.UpdatedAt,
)
if err != nil {
return daemonPairingSessionRecord{}, err
}
if rec.Status == "pending" && rec.ExpiresAt.Valid && rec.ExpiresAt.Time.Before(time.Now()) {
if _, err := h.DB.Exec(ctx, `
UPDATE daemon_pairing_session
SET status = 'expired', updated_at = now()
WHERE token = $1 AND status = 'pending'
`, token); err == nil {
rec.Status = "expired"
rec.UpdatedAt = pgtype.Timestamptz{Time: time.Now(), Valid: true}
}
}
return rec, nil
}
func (h *Handler) CreateDaemonPairingSession(w http.ResponseWriter, r *http.Request) {
if h.DB == nil {
writeError(w, http.StatusInternalServerError, "database executor is not configured")
return
}
var req CreateDaemonPairingSessionRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "invalid request body")
return
}
req.DaemonID = strings.TrimSpace(req.DaemonID)
req.DeviceName = strings.TrimSpace(req.DeviceName)
req.RuntimeName = strings.TrimSpace(req.RuntimeName)
req.RuntimeType = strings.TrimSpace(req.RuntimeType)
req.RuntimeVersion = strings.TrimSpace(req.RuntimeVersion)
if req.DaemonID == "" {
writeError(w, http.StatusBadRequest, "daemon_id is required")
return
}
if req.DeviceName == "" {
writeError(w, http.StatusBadRequest, "device_name is required")
return
}
if req.RuntimeName == "" {
writeError(w, http.StatusBadRequest, "runtime_name is required")
return
}
if req.RuntimeType == "" {
writeError(w, http.StatusBadRequest, "runtime_type is required")
return
}
token, err := randomDaemonPairingToken()
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to create pairing token")
return
}
expiresAt := time.Now().Add(daemonPairingTTL)
var rec daemonPairingSessionRecord
err = h.DB.QueryRow(r.Context(), `
INSERT INTO daemon_pairing_session (
token,
daemon_id,
device_name,
runtime_name,
runtime_type,
runtime_version,
expires_at
) VALUES ($1, $2, $3, $4, $5, $6, $7)
RETURNING
token,
daemon_id,
device_name,
runtime_name,
runtime_type,
runtime_version,
workspace_id,
approved_by,
status,
approved_at,
claimed_at,
expires_at,
created_at,
updated_at
`,
token,
req.DaemonID,
req.DeviceName,
req.RuntimeName,
req.RuntimeType,
req.RuntimeVersion,
expiresAt,
).Scan(
&rec.Token,
&rec.DaemonID,
&rec.DeviceName,
&rec.RuntimeName,
&rec.RuntimeType,
&rec.RuntimeVersion,
&rec.WorkspaceID,
&rec.ApprovedBy,
&rec.Status,
&rec.ApprovedAt,
&rec.ClaimedAt,
&rec.ExpiresAt,
&rec.CreatedAt,
&rec.UpdatedAt,
)
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to create pairing session")
return
}
writeJSON(w, http.StatusCreated, daemonPairingSessionToResponse(rec, true))
}
func (h *Handler) GetDaemonPairingSession(w http.ResponseWriter, r *http.Request) {
token := chi.URLParam(r, "token")
rec, err := h.getDaemonPairingSession(r.Context(), token)
if err != nil {
writeError(w, http.StatusNotFound, "pairing session not found")
return
}
writeJSON(w, http.StatusOK, daemonPairingSessionToResponse(rec, true))
}
func (h *Handler) ApproveDaemonPairingSession(w http.ResponseWriter, r *http.Request) {
token := chi.URLParam(r, "token")
rec, err := h.getDaemonPairingSession(r.Context(), token)
if err != nil {
writeError(w, http.StatusNotFound, "pairing session not found")
return
}
if rec.Status == "expired" {
writeError(w, http.StatusBadRequest, "pairing session expired")
return
}
if rec.Status == "claimed" {
writeError(w, http.StatusBadRequest, "pairing session already claimed")
return
}
if rec.Status == "approved" {
writeJSON(w, http.StatusOK, daemonPairingSessionToResponse(rec, true))
return
}
var req ApproveDaemonPairingSessionRequest
if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
writeError(w, http.StatusBadRequest, "invalid request body")
return
}
if req.WorkspaceID == "" {
writeError(w, http.StatusBadRequest, "workspace_id is required")
return
}
userID, ok := requireUserID(w, r)
if !ok {
return
}
if _, ok := h.requireWorkspaceMember(w, r, req.WorkspaceID, "workspace not found"); !ok {
return
}
if h.DB == nil {
writeError(w, http.StatusInternalServerError, "database executor is not configured")
return
}
if _, err := h.DB.Exec(r.Context(), `
UPDATE daemon_pairing_session
SET
workspace_id = $2,
approved_by = $3,
status = 'approved',
approved_at = now(),
updated_at = now()
WHERE token = $1 AND status = 'pending'
`, token, parseUUID(req.WorkspaceID), parseUUID(userID)); err != nil {
writeError(w, http.StatusInternalServerError, "failed to approve pairing session")
return
}
rec, err = h.getDaemonPairingSession(r.Context(), token)
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to reload pairing session")
return
}
writeJSON(w, http.StatusOK, daemonPairingSessionToResponse(rec, true))
}
func (h *Handler) ClaimDaemonPairingSession(w http.ResponseWriter, r *http.Request) {
token := chi.URLParam(r, "token")
rec, err := h.getDaemonPairingSession(r.Context(), token)
if err != nil {
writeError(w, http.StatusNotFound, "pairing session not found")
return
}
if rec.Status == "claimed" {
writeJSON(w, http.StatusOK, daemonPairingSessionToResponse(rec, true))
return
}
if rec.Status != "approved" {
writeError(w, http.StatusBadRequest, "pairing session is not approved")
return
}
if h.DB == nil {
writeError(w, http.StatusInternalServerError, "database executor is not configured")
return
}
if _, err := h.DB.Exec(r.Context(), `
UPDATE daemon_pairing_session
SET
status = 'claimed',
claimed_at = now(),
updated_at = now()
WHERE token = $1 AND status = 'approved'
`, token); err != nil {
writeError(w, http.StatusInternalServerError, "failed to claim pairing session")
return
}
rec, err = h.getDaemonPairingSession(r.Context(), token)
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to reload pairing session")
return
}
writeJSON(w, http.StatusOK, daemonPairingSessionToResponse(rec, true))
}

13
server/internal/handler/handler.go
View file

@ -20,16 +20,29 @@ type txStarter interface {
Begin(ctx context.Context) (pgx.Tx, error)
}
type dbExecutor interface {
Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)
Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
QueryRow(ctx context.Context, sql string, args ...any) pgx.Row
}
type Handler struct {
Queries *db.Queries
DB dbExecutor
TxStarter txStarter
Hub *realtime.Hub
TaskService *service.TaskService
}
func New(queries *db.Queries, txStarter txStarter, hub *realtime.Hub) *Handler {
var executor dbExecutor
if candidate, ok := txStarter.(dbExecutor); ok {
executor = candidate
}
return &Handler{
Queries: queries,
DB: executor,
TxStarter: txStarter,
Hub: hub,
TaskService: service.NewTaskService(queries, hub),

102
server/internal/handler/issue.go
View file

@ -1,6 +1,7 @@
package handler
import (
"context"
"encoding/json"
"io"
"net/http"
@ -14,24 +15,30 @@ import (
// IssueResponse is the JSON response for an issue.
type IssueResponse struct {
ID string `json:"id"`
WorkspaceID string `json:"workspace_id"`
Title string `json:"title"`
Description *string `json:"description"`
Status string `json:"status"`
Priority string `json:"priority"`
AssigneeType *string `json:"assignee_type"`
AssigneeID *string `json:"assignee_id"`
CreatorType string `json:"creator_type"`
CreatorID string `json:"creator_id"`
ParentIssueID *string `json:"parent_issue_id"`
AcceptanceCriteria []any `json:"acceptance_criteria"`
ContextRefs []any `json:"context_refs"`
Repository any `json:"repository"`
Position float64 `json:"position"`
DueDate *string `json:"due_date"`
CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"`
ID string `json:"id"`
WorkspaceID string `json:"workspace_id"`
Title string `json:"title"`
Description *string `json:"description"`
Status string `json:"status"`
Priority string `json:"priority"`
AssigneeType *string `json:"assignee_type"`
AssigneeID *string `json:"assignee_id"`
CreatorType string `json:"creator_type"`
CreatorID string `json:"creator_id"`
ParentIssueID *string `json:"parent_issue_id"`
AcceptanceCriteria []any `json:"acceptance_criteria"`
ContextRefs []any `json:"context_refs"`
Repository any `json:"repository"`
Position float64 `json:"position"`
DueDate *string `json:"due_date"`
CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"`
}
type agentTriggerSnapshot struct {
Type string `json:"type"`
Enabled bool `json:"enabled"`
Config map[string]any `json:"config"`
}
func issueToResponse(i db.Issue) IssueResponse {
@ -258,8 +265,8 @@ func (h *Handler) CreateIssue(w http.ResponseWriter, r *http.Request) {
h.broadcast("inbox:new", map[string]any{"item": inboxToResponse(inboxItem)})
}
// If assigned to an agent, enqueue a task with context
if issue.AssigneeType.String == "agent" {
// Only ready issues in todo are enqueued for agents.
if h.shouldEnqueueAgentTask(r.Context(), issue) {
h.TaskService.EnqueueTaskForIssue(r.Context(), issue)
}
}
@ -283,12 +290,12 @@ type UpdateIssueRequest struct {
func (h *Handler) UpdateIssue(w http.ResponseWriter, r *http.Request) {
id := chi.URLParam(r, "id")
current, ok := h.loadIssueForUser(w, r, id)
prevIssue, ok := h.loadIssueForUser(w, r, id)
if !ok {
return
}
// Read body as raw bytes so we can detect which fields were explicitly sent
// Read body as raw bytes so we can detect which fields were explicitly sent.
bodyBytes, err := io.ReadAll(r.Body)
if err != nil {
writeError(w, http.StatusBadRequest, "failed to read request body")
@ -307,10 +314,10 @@ func (h *Handler) UpdateIssue(w http.ResponseWriter, r *http.Request) {
// Pre-fill nullable fields (bare sqlc.narg) with current values
params := db.UpdateIssueParams{
ID: current.ID,
AssigneeType: current.AssigneeType,
AssigneeID: current.AssigneeID,
DueDate: current.DueDate,
ID: prevIssue.ID,
AssigneeType: prevIssue.AssigneeType,
AssigneeID: prevIssue.AssigneeID,
DueDate: prevIssue.DueDate,
}
// COALESCE fields — only set when explicitly provided
@ -379,16 +386,21 @@ func (h *Handler) UpdateIssue(w http.ResponseWriter, r *http.Request) {
resp := issueToResponse(issue)
h.broadcast("issue:updated", map[string]any{"issue": resp})
// If assignee changed, handle agent task queue
if req.AssigneeType != nil || req.AssigneeID != nil {
// Cancel any existing agent tasks for this issue
assigneeChanged := (req.AssigneeType != nil || req.AssigneeID != nil) &&
(prevIssue.AssigneeType.String != issue.AssigneeType.String || uuidToString(prevIssue.AssigneeID) != uuidToString(issue.AssigneeID))
statusChanged := req.Status != nil && prevIssue.Status != issue.Status
// If assignee or readiness status changed, reconcile the task queue.
if assigneeChanged || statusChanged {
h.TaskService.CancelTasksForIssue(r.Context(), issue.ID)
// If newly assigned to an agent, enqueue a task with context
if issue.AssigneeType.Valid && issue.AssigneeType.String == "agent" && issue.AssigneeID.Valid {
if h.shouldEnqueueAgentTask(r.Context(), issue) {
h.TaskService.EnqueueTaskForIssue(r.Context(), issue)
}
}
// If assignee changed, create a notification for the new assignee.
if assigneeChanged {
// Create inbox notification for new assignee
if issue.AssigneeType.Valid && issue.AssigneeID.Valid {
inboxItem, err := h.Queries.CreateInboxItem(r.Context(), db.CreateInboxItemParams{
@ -427,6 +439,34 @@ func (h *Handler) UpdateIssue(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusOK, resp)
}
func (h *Handler) shouldEnqueueAgentTask(ctx context.Context, issue db.Issue) bool {
if issue.Status != "todo" {
return false
}
if !issue.AssigneeType.Valid || issue.AssigneeType.String != "agent" || !issue.AssigneeID.Valid {
return false
}
agent, err := h.Queries.GetAgent(ctx, issue.AssigneeID)
if err != nil || !agent.RuntimeID.Valid {
return false
}
if agent.Triggers == nil || len(agent.Triggers) == 0 {
return true
}
var triggers []agentTriggerSnapshot
if err := json.Unmarshal(agent.Triggers, &triggers); err != nil {
return false
}
for _, trigger := range triggers {
if trigger.Type == "on_assign" && trigger.Enabled {
return true
}
}
return false
}
func (h *Handler) DeleteIssue(w http.ResponseWriter, r *http.Request) {
id := chi.URLParam(r, "id")
if _, ok := h.loadIssueForUser(w, r, id); !ok {

68
server/internal/handler/runtime.go Normal file
View file

@ -0,0 +1,68 @@
package handler
import (
"encoding/json"
"net/http"
db "github.com/multica-ai/multica/server/pkg/db/generated"
)
type AgentRuntimeResponse struct {
ID string `json:"id"`
WorkspaceID string `json:"workspace_id"`
DaemonID *string `json:"daemon_id"`
Name string `json:"name"`
RuntimeMode string `json:"runtime_mode"`
Provider string `json:"provider"`
Status string `json:"status"`
DeviceInfo string `json:"device_info"`
Metadata any `json:"metadata"`
LastSeenAt *string `json:"last_seen_at"`
CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"`
}
func runtimeToResponse(rt db.AgentRuntime) AgentRuntimeResponse {
var metadata any
if rt.Metadata != nil {
json.Unmarshal(rt.Metadata, &metadata)
}
if metadata == nil {
metadata = map[string]any{}
}
return AgentRuntimeResponse{
ID: uuidToString(rt.ID),
WorkspaceID: uuidToString(rt.WorkspaceID),
DaemonID: textToPtr(rt.DaemonID),
Name: rt.Name,
RuntimeMode: rt.RuntimeMode,
Provider: rt.Provider,
Status: rt.Status,
DeviceInfo: rt.DeviceInfo,
Metadata: metadata,
LastSeenAt: timestampToPtr(rt.LastSeenAt),
CreatedAt: timestampToString(rt.CreatedAt),
UpdatedAt: timestampToString(rt.UpdatedAt),
}
}
func (h *Handler) ListAgentRuntimes(w http.ResponseWriter, r *http.Request) {
workspaceID := resolveWorkspaceID(r)
if _, ok := h.requireWorkspaceMember(w, r, workspaceID, "workspace not found"); !ok {
return
}
runtimes, err := h.Queries.ListAgentRuntimes(r.Context(), parseUUID(workspaceID))
if err != nil {
writeError(w, http.StatusInternalServerError, "failed to list runtimes")
return
}
resp := make([]AgentRuntimeResponse, len(runtimes))
for i, rt := range runtimes {
resp[i] = runtimeToResponse(rt)
}
writeJSON(w, http.StatusOK, resp)
}

237
server/internal/service/task.go
View file

@ -8,9 +8,9 @@ import (
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgtype"
db "github.com/multica-ai/multica/server/pkg/db/generated"
"github.com/multica-ai/multica/server/internal/realtime"
"github.com/multica-ai/multica/server/internal/util"
db "github.com/multica-ai/multica/server/pkg/db/generated"
"github.com/multica-ai/multica/server/pkg/protocol"
)
@ -29,14 +29,28 @@ func (s *TaskService) EnqueueTaskForIssue(ctx context.Context, issue db.Issue) (
return db.AgentTaskQueue{}, fmt.Errorf("issue has no assignee")
}
snapshot := buildContextSnapshot(issue)
agent, err := s.Queries.GetAgent(ctx, issue.AssigneeID)
if err != nil {
return db.AgentTaskQueue{}, fmt.Errorf("load agent: %w", err)
}
if !agent.RuntimeID.Valid {
return db.AgentTaskQueue{}, fmt.Errorf("agent has no runtime")
}
runtime, err := s.Queries.GetAgentRuntime(ctx, agent.RuntimeID)
if err != nil {
return db.AgentTaskQueue{}, fmt.Errorf("load runtime: %w", err)
}
snapshot := buildContextSnapshot(issue, agent, runtime)
contextJSON, _ := json.Marshal(snapshot)
task, err := s.Queries.CreateAgentTaskWithContext(ctx, db.CreateAgentTaskWithContextParams{
AgentID: issue.AssigneeID,
IssueID: issue.ID,
Priority: priorityToInt(issue.Priority),
Context: contextJSON,
AgentID: issue.AssigneeID,
RuntimeID: agent.RuntimeID,
IssueID: issue.ID,
Priority: priorityToInt(issue.Priority),
Context: contextJSON,
})
if err != nil {
return db.AgentTaskQueue{}, fmt.Errorf("create task: %w", err)
@ -83,6 +97,34 @@ func (s *TaskService) ClaimTask(ctx context.Context, agentID pgtype.UUID) (*db.A
return &task, nil
}
// ClaimTaskForRuntime claims the next runnable task for a runtime while
// still respecting each agent's max_concurrent_tasks limit.
func (s *TaskService) ClaimTaskForRuntime(ctx context.Context, runtimeID pgtype.UUID) (*db.AgentTaskQueue, error) {
tasks, err := s.Queries.ListPendingTasksByRuntime(ctx, runtimeID)
if err != nil {
return nil, fmt.Errorf("list pending tasks: %w", err)
}
triedAgents := map[string]struct{}{}
for _, candidate := range tasks {
agentKey := util.UUIDToString(candidate.AgentID)
if _, seen := triedAgents[agentKey]; seen {
continue
}
triedAgents[agentKey] = struct{}{}
task, err := s.ClaimTask(ctx, candidate.AgentID)
if err != nil {
return nil, err
}
if task != nil && task.RuntimeID == runtimeID {
return task, nil
}
}
return nil, nil
}
// StartTask transitions a dispatched task to running and syncs issue status.
func (s *TaskService) StartTask(ctx context.Context, taskID pgtype.UUID) (*db.AgentTaskQueue, error) {
task, err := s.Queries.StartAgentTask(ctx, taskID)
@ -91,10 +133,13 @@ func (s *TaskService) StartTask(ctx context.Context, taskID pgtype.UUID) (*db.Ag
}
// Sync issue → in_progress
s.Queries.UpdateIssueStatus(ctx, db.UpdateIssueStatusParams{
issue, err := s.Queries.UpdateIssueStatus(ctx, db.UpdateIssueStatusParams{
ID: task.IssueID,
Status: "in_progress",
})
if err == nil {
s.broadcastIssueUpdated(issue)
}
return &task, nil
}
@ -110,10 +155,23 @@ func (s *TaskService) CompleteTask(ctx context.Context, taskID pgtype.UUID, resu
}
// Sync issue → in_review
s.Queries.UpdateIssueStatus(ctx, db.UpdateIssueStatusParams{
issue, issueErr := s.Queries.UpdateIssueStatus(ctx, db.UpdateIssueStatusParams{
ID: task.IssueID,
Status: "in_review",
})
if issueErr == nil {
s.broadcastIssueUpdated(issue)
}
var payload protocol.TaskCompletedPayload
if err := json.Unmarshal(result, &payload); err == nil {
if payload.Output != "" {
s.createAgentComment(ctx, task.IssueID, task.AgentID, payload.Output, "comment")
}
}
if issueErr == nil {
s.createInboxForIssueCreator(ctx, issue, "review_requested", "attention", "Review requested: "+issue.Title, "")
}
// Reconcile agent status
s.ReconcileAgentStatus(ctx, task.AgentID)
@ -135,10 +193,19 @@ func (s *TaskService) FailTask(ctx context.Context, taskID pgtype.UUID, errMsg s
}
// Sync issue → blocked
s.Queries.UpdateIssueStatus(ctx, db.UpdateIssueStatusParams{
issue, issueErr := s.Queries.UpdateIssueStatus(ctx, db.UpdateIssueStatusParams{
ID: task.IssueID,
Status: "blocked",
})
if issueErr == nil {
s.broadcastIssueUpdated(issue)
}
if errMsg != "" {
s.createAgentComment(ctx, task.IssueID, task.AgentID, errMsg, "system")
}
if issueErr == nil {
s.createInboxForIssueCreator(ctx, issue, "agent_blocked", "action_required", "Agent blocked: "+issue.Title, errMsg)
}
// Reconcile agent status
s.ReconcileAgentStatus(ctx, task.AgentID)
@ -183,7 +250,7 @@ func (s *TaskService) updateAgentStatus(ctx context.Context, agentID pgtype.UUID
s.broadcast(protocol.EventAgentStatus, map[string]any{"agent": agentToMap(agent)})
}
func buildContextSnapshot(issue db.Issue) protocol.TaskDispatchPayload {
func buildContextSnapshot(issue db.Issue, agent db.Agent, runtime db.AgentRuntime) map[string]any {
var ac []string
if issue.AcceptanceCriteria != nil {
json.Unmarshal(issue.AcceptanceCriteria, &ac)
@ -198,13 +265,38 @@ func buildContextSnapshot(issue db.Issue) protocol.TaskDispatchPayload {
json.Unmarshal(issue.Repository, repo)
}
return protocol.TaskDispatchPayload{
IssueID: util.UUIDToString(issue.ID),
Title: issue.Title,
Description: issue.Description.String,
AcceptanceCriteria: ac,
ContextRefs: cr,
Repository: repo,
var tools any
if agent.Tools != nil {
json.Unmarshal(agent.Tools, &tools)
}
var metadata any
if runtime.Metadata != nil {
json.Unmarshal(runtime.Metadata, &metadata)
}
return map[string]any{
"issue": map[string]any{
"id": util.UUIDToString(issue.ID),
"title": issue.Title,
"description": issue.Description.String,
"acceptance_criteria": ac,
"context_refs": cr,
"repository": repo,
},
"agent": map[string]any{
"id": util.UUIDToString(agent.ID),
"name": agent.Name,
"skills": agent.Skills,
"tools": tools,
},
"runtime": map[string]any{
"id": util.UUIDToString(runtime.ID),
"name": runtime.Name,
"runtime_mode": runtime.RuntimeMode,
"provider": runtime.Provider,
"device_info": runtime.DeviceInfo,
"metadata": metadata,
},
}
}
@ -224,11 +316,15 @@ func priorityToInt(p string) int32 {
}
func (s *TaskService) broadcastTaskDispatch(task db.AgentTaskQueue) {
var payload protocol.TaskDispatchPayload
var payload map[string]any
if task.Context != nil {
json.Unmarshal(task.Context, &payload)
}
payload.TaskID = util.UUIDToString(task.ID)
if payload == nil {
payload = map[string]any{}
}
payload["task_id"] = util.UUIDToString(task.ID)
payload["runtime_id"] = util.UUIDToString(task.RuntimeID)
s.broadcast(protocol.EventTaskDispatch, payload)
}
@ -253,6 +349,108 @@ func (s *TaskService) broadcast(eventType string, payload any) {
s.Hub.Broadcast(data)
}
func (s *TaskService) broadcastIssueUpdated(issue db.Issue) {
s.broadcast(protocol.EventIssueUpdated, map[string]any{
"issue": issueToMap(issue),
})
}
func (s *TaskService) createAgentComment(ctx context.Context, issueID, agentID pgtype.UUID, content, commentType string) {
if content == "" {
return
}
s.Queries.CreateComment(ctx, db.CreateCommentParams{
IssueID: issueID,
AuthorType: "agent",
AuthorID: agentID,
Content: content,
Type: commentType,
})
}
func (s *TaskService) createInboxForIssueCreator(ctx context.Context, issue db.Issue, itemType, severity, title, body string) {
if issue.CreatorType != "member" {
return
}
item, err := s.Queries.CreateInboxItem(ctx, db.CreateInboxItemParams{
WorkspaceID: issue.WorkspaceID,
RecipientType: "member",
RecipientID: issue.CreatorID,
Type: itemType,
Severity: severity,
IssueID: issue.ID,
Title: title,
Body: util.PtrToText(&body),
})
if err != nil {
return
}
s.broadcast(protocol.EventInboxNew, map[string]any{
"item": inboxToMap(item),
})
}
func issueToMap(issue db.Issue) map[string]any {
var ac []any
if issue.AcceptanceCriteria != nil {
json.Unmarshal(issue.AcceptanceCriteria, &ac)
}
if ac == nil {
ac = []any{}
}
var cr []any
if issue.ContextRefs != nil {
json.Unmarshal(issue.ContextRefs, &cr)
}
if cr == nil {
cr = []any{}
}
var repo any
if issue.Repository != nil {
json.Unmarshal(issue.Repository, &repo)
}
return map[string]any{
"id": util.UUIDToString(issue.ID),
"workspace_id": util.UUIDToString(issue.WorkspaceID),
"title": issue.Title,
"description": util.TextToPtr(issue.Description),
"status": issue.Status,
"priority": issue.Priority,
"assignee_type": util.TextToPtr(issue.AssigneeType),
"assignee_id": util.UUIDToPtr(issue.AssigneeID),
"creator_type": issue.CreatorType,
"creator_id": util.UUIDToString(issue.CreatorID),
"parent_issue_id": util.UUIDToPtr(issue.ParentIssueID),
"acceptance_criteria": ac,
"context_refs": cr,
"repository": repo,
"position": issue.Position,
"due_date": util.TimestampToPtr(issue.DueDate),
"created_at": util.TimestampToString(issue.CreatedAt),
"updated_at": util.TimestampToString(issue.UpdatedAt),
}
}
func inboxToMap(item db.InboxItem) map[string]any {
return map[string]any{
"id": util.UUIDToString(item.ID),
"workspace_id": util.UUIDToString(item.WorkspaceID),
"recipient_type": item.RecipientType,
"recipient_id": util.UUIDToString(item.RecipientID),
"type": item.Type,
"severity": item.Severity,
"issue_id": util.UUIDToPtr(item.IssueID),
"title": item.Title,
"body": util.TextToPtr(item.Body),
"read": item.Read,
"archived": item.Archived,
"created_at": util.TimestampToString(item.CreatedAt),
}
}
// agentToMap builds a simple map for broadcasting agent status updates.
func agentToMap(a db.Agent) map[string]any {
var rc any
@ -270,6 +468,7 @@ func agentToMap(a db.Agent) map[string]any {
return map[string]any{
"id": util.UUIDToString(a.ID),
"workspace_id": util.UUIDToString(a.WorkspaceID),
"runtime_id": util.UUIDToString(a.RuntimeID),
"name": a.Name,
"description": a.Description,
"avatar_url": util.TextToPtr(a.AvatarUrl),