From b585b5d13484e4edd8b4c921088bde9196234f9e Mon Sep 17 00:00:00 2001 From: toki Date: Wed, 20 May 2026 21:09:23 +0900 Subject: [PATCH] update: add a2a/openai adapters, agent module, model, config tests, and scheduler updates --- .antigravityignore | 1 + .gitignore | 3 + Makefile | 8 + README.md | 26 ++- bin/run | 11 + cmd/server/main.go | 22 +- docker-compose.yml | 9 + internal/adapters/a2a/client.go | 299 ++++++++++++++++++++++++ internal/adapters/a2a/client_test.go | 162 +++++++++++++ internal/adapters/openai/client.go | 266 +++++++++++++++++++++ internal/adapters/openai/client_test.go | 140 +++++++++++ internal/agent/agent.go | 104 +++++++++ internal/config/config.go | 44 +++- internal/config/config_test.go | 53 +++++ internal/model/model.go | 34 +++ internal/scheduler/jobs.go | 233 +++++++++++++++++- internal/scheduler/jobs_test.go | 108 +++++++++ internal/scheduler/river.go | 6 +- 18 files changed, 1515 insertions(+), 14 deletions(-) create mode 100644 .antigravityignore create mode 100644 internal/adapters/a2a/client.go create mode 100644 internal/adapters/a2a/client_test.go create mode 100644 internal/adapters/openai/client.go create mode 100644 internal/adapters/openai/client_test.go create mode 100644 internal/agent/agent.go create mode 100644 internal/config/config_test.go create mode 100644 internal/model/model.go create mode 100644 internal/scheduler/jobs_test.go diff --git a/.antigravityignore b/.antigravityignore new file mode 100644 index 0000000..477be46 --- /dev/null +++ b/.antigravityignore @@ -0,0 +1 @@ +agent-task/archive/** diff --git a/.gitignore b/.gitignore index d008d13..583fe3e 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,6 @@ coverage.out # Agent-Ops Private Rules agent-ops/rules/private/ + +# AI/IDE Cache +/.antigravitycli/ diff --git a/Makefile b/Makefile index b776e46..ae0de94 100644 --- a/Makefile +++ b/Makefile @@ -6,6 +6,14 @@ export REDIS_KEY_PREFIX ?= nomadcode-core:local export DEV_REDIS_URL ?= redis://code-server-redis:6379/4 export DEV_REDIS_KEY_PREFIX ?= nomadcode-core:dev export AUTH_USERNAME ?= nomadcode +export MODEL_BASE_URL ?= http://192.168.0.91:11434 +export MODEL_API_KEY ?= ollama +export MODEL_NAME ?= qwen3.6:35b-a3b-bf16 +export MODEL_CONTEXT_SIZE ?= 262144 +export MODEL_TIMEOUT_SEC ?= 300 +export A2A_EDGE_URL ?= +export A2A_AGENT_URL ?= +export A2A_TIMEOUT_SEC ?= 300 export GOOSE_DRIVER ?= postgres export GOOSE_DBSTRING ?= $(DATABASE_URL) export OUTPUT ?= .build/nomadcode-core diff --git a/README.md b/README.md index b7b2e5d..acdd333 100644 --- a/README.md +++ b/README.md @@ -12,7 +12,9 @@ NomadCode Core는 사용자 요청을 작업 단위로 받고, 작업 상태를 - PostgreSQL 연결 - goose migration - sqlc 기반 DB query 생성 구조 -- River dummy job +- River task job +- IOP Edge/OpenAI-compatible 모델 호출 경로(Ollama direct fallback 가능) +- IOP Edge/A2A-compatible agent 호출 인터페이스 - Plane Adapter stub - Mattermost Adapter stub - 선택적 Docker Compose 실행 환경(PostgreSQL, Redis) @@ -21,7 +23,7 @@ NomadCode Core는 사용자 요청을 작업 단위로 받고, 작업 상태를 - 실제 Plane API 호출 - 실제 Mattermost 메시지 발송 -- IOP 연동 +- IOP native protocol 연동 - Agent Integrator 연동 - Outline / Forgejo / Nextcloud 연동 - MCP 서버 @@ -39,7 +41,7 @@ NomadCode Core는 사용자 요청을 작업 단위로 받고, 작업 상태를 목표: - 서버 실행 골격 구성 - task 저장 구조 구성 -- dummy 비동기 job 실행 +- 로컬 모델 호출 기반 비동기 job 실행 - Adapter stub 생성 ### Phase 2. Workflow Core @@ -57,10 +59,12 @@ NomadCode Core는 사용자 요청을 작업 단위로 받고, 작업 상태를 - Mattermost 메시지 발송 구현 - Agent Integrator 호출 구조 추가 - IOP 호출 구조 추가 +- IOP의 OpenAI-compatible API 경로는 단순 모델/chat completion 호환 호출에 사용 +- IOP의 A2A JSON-RPC 경로는 Core가 Edge 실행 그룹이나 외부 agent에게 작업을 위임하고 task 상태/artifact/cancel 흐름을 공유할 때 사용 ## 실행 방법 -로컬 실행은 현재 개발 호스트의 Go와 `code-server` compose에 붙은 PostgreSQL/Redis가 있다는 전제로 진행합니다. local 기본 `DATABASE_URL`은 `postgres://nomadcode:nomadcode@code-server-postgres:5432/nomadcode-core-local?sslmode=disable` 이고, local 기본 `REDIS_URL`은 `redis://code-server-redis:6379/3`, `REDIS_KEY_PREFIX`는 `nomadcode-core:local` 입니다. dev 배포는 Docker Compose로 실행하며, 같은 `code-server-postgres` Postgres와 `code-server-redis` Redis를 사용합니다. dev DB명은 `nomad-core-dev` 이고, dev 기본 `REDIS_URL`은 `redis://code-server-redis:6379/4`, `REDIS_KEY_PREFIX`는 `nomadcode-core:dev` 입니다. `AUTH_PASSWORD`를 설정하면 `/readyz`와 `/api/*`에 HTTP Basic Auth가 적용됩니다. +로컬 실행은 현재 개발 호스트의 Go와 `code-server` compose에 붙은 PostgreSQL/Redis가 있다는 전제로 진행합니다. local 기본 `DATABASE_URL`은 `postgres://nomadcode:nomadcode@code-server-postgres:5432/nomadcode-core-local?sslmode=disable` 이고, local 기본 `REDIS_URL`은 `redis://code-server-redis:6379/3`, `REDIS_KEY_PREFIX`는 `nomadcode-core:local` 입니다. dev 배포는 Docker Compose로 실행하며, 같은 `code-server-postgres` Postgres와 `code-server-redis` Redis를 사용합니다. dev DB명은 `nomad-core-dev` 이고, dev 기본 `REDIS_URL`은 `redis://code-server-redis:6379/4`, `REDIS_KEY_PREFIX`는 `nomadcode-core:dev` 입니다. `AUTH_PASSWORD`를 설정하면 `/readyz`와 `/api/*`에 HTTP Basic Auth가 적용됩니다. 모델 호출의 장기 대상은 IOP Edge의 OpenAI-compatible input surface이며, 현재 core model client는 Responses API 형식을 사용합니다. fallback 기본값은 direct Ollama `MODEL_BASE_URL=http://192.168.0.91:11434`, `MODEL_NAME=qwen3.6:35b-a3b-bf16`, `MODEL_CONTEXT_SIZE=262144`, `MODEL_TIMEOUT_SEC=300` 입니다. A2A agent 호출 endpoint는 `A2A_EDGE_URL`로 설정하고, 기존 `A2A_AGENT_URL`도 fallback alias로 받습니다. bearer token은 `A2A_TOKEN`, timeout은 `A2A_TIMEOUT_SEC`로 설정합니다. `code-server` PostgreSQL 컨테이너에 DB를 생성하는 예시: @@ -95,6 +99,20 @@ DATABASE_URL="postgres://user:password@localhost:5432/dbname?sslmode=disable" ./ DATABASE_URL="postgres://user:password@localhost:5432/dbname?sslmode=disable" ./bin/run ``` +다른 모델 endpoint를 사용할 경우: + +```bash +MODEL_BASE_URL="http://localhost:11434" \ +MODEL_NAME="qwen3.6:35b-a3b-bf16" \ +MODEL_CONTEXT_SIZE="262144" \ +MODEL_TIMEOUT_SEC="300" \ +./bin/run +``` + +모델 호출은 OpenAI-compatible Responses API의 non-streaming `POST /v1/responses` 형식을 사용합니다. IOP Edge OpenAI-compatible listener를 `MODEL_BASE_URL`로 쓰려면 해당 listener가 `/v1/responses`를 제공해야 합니다. direct Ollama fallback에서는 `MODEL_CONTEXT_SIZE`를 Ollama 전용 option인 `options.num_ctx`로 전달합니다. + +A2A agent 호출 인터페이스는 JSON-RPC 2.0 `message/send`, `tasks/get`, `tasks/cancel`을 우선 지원합니다. `A2A_EDGE_URL`을 설정하면 worker는 A2A `message/send`를 blocking 호출하고, 완료된 task/message 응답만 local task completion으로 반영합니다. 기존 `A2A_AGENT_URL`도 fallback alias로 받습니다. 설정하지 않으면 기존 OpenAI-compatible 모델 호출 경로를 사용합니다. + 테스트 실행: ```bash diff --git a/bin/run b/bin/run index d7e8248..d5638de 100755 --- a/bin/run +++ b/bin/run @@ -11,6 +11,11 @@ legacy_code_server_database_url="postgres://nomadcode:nomadcode@code-server-post default_redis_url="redis://code-server-redis:6379/3" legacy_code_server_redis_url="redis://code-server-redis:6379/0" default_redis_key_prefix="nomadcode-core:local" +default_model_base_url="http://192.168.0.91:11434" +default_model_name="qwen3.6:35b-a3b-bf16" +default_model_context_size="262144" +default_model_timeout_sec="300" +default_a2a_timeout_sec="300" # code-server exports a shared DATABASE_URL; remap that default to this project's local DB. if [[ "${DATABASE_URL:-}" == "$legacy_code_server_database_url" ]]; then export DATABASE_URL="$default_database_url" @@ -25,5 +30,11 @@ else fi export REDIS_KEY_PREFIX="${REDIS_KEY_PREFIX:-$default_redis_key_prefix}" export AUTH_USERNAME="${AUTH_USERNAME:-nomadcode}" +export MODEL_BASE_URL="${MODEL_BASE_URL:-$default_model_base_url}" +export MODEL_API_KEY="${MODEL_API_KEY:-ollama}" +export MODEL_NAME="${MODEL_NAME:-$default_model_name}" +export MODEL_CONTEXT_SIZE="${MODEL_CONTEXT_SIZE:-$default_model_context_size}" +export MODEL_TIMEOUT_SEC="${MODEL_TIMEOUT_SEC:-$default_model_timeout_sec}" +export A2A_TIMEOUT_SEC="${A2A_TIMEOUT_SEC:-$default_a2a_timeout_sec}" exec go run ./cmd/server "$@" diff --git a/cmd/server/main.go b/cmd/server/main.go index 54bf5f6..bae1e1f 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -10,7 +10,10 @@ import ( "syscall" "time" + agenta2a "github.com/nomadcode/nomadcode-core/internal/adapters/a2a" "github.com/nomadcode/nomadcode-core/internal/adapters/mattermost" + modelopenai "github.com/nomadcode/nomadcode-core/internal/adapters/openai" + "github.com/nomadcode/nomadcode-core/internal/agent" "github.com/nomadcode/nomadcode-core/internal/config" "github.com/nomadcode/nomadcode-core/internal/db" apphttp "github.com/nomadcode/nomadcode-core/internal/http" @@ -46,8 +49,25 @@ func run(logger *slog.Logger) error { Token: cfg.MattermostToken, }, logger) notificationService := notification.NewService(mattermostClient, logger) + modelClient := modelopenai.NewClient(modelopenai.Config{ + BaseURL: cfg.ModelBaseURL, + APIKey: cfg.ModelAPIKey, + Model: cfg.ModelName, + ContextSize: cfg.ModelContextSize, + TimeoutSec: cfg.ModelTimeoutSec, + }, logger) - taskScheduler, err := scheduler.New(pool, store, notificationService, logger) + var agentClient agent.Client + if cfg.A2AEdgeURL != "" { + agentClient = agenta2a.NewClient(agenta2a.Config{ + URL: cfg.A2AEdgeURL, + Token: cfg.A2AToken, + TimeoutSec: cfg.A2ATimeoutSec, + }, logger) + logger.Info("a2a edge input enabled", "url", cfg.A2AEdgeURL) + } + + taskScheduler, err := scheduler.New(pool, store, notificationService, agentClient, modelClient, logger) if err != nil { return err } diff --git a/docker-compose.yml b/docker-compose.yml index 515e597..6e27682 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -11,6 +11,15 @@ services: REDIS_KEY_PREFIX: ${DEV_REDIS_KEY_PREFIX:-nomadcode-core:dev} AUTH_USERNAME: ${AUTH_USERNAME:-nomadcode} AUTH_PASSWORD: ${AUTH_PASSWORD:-} + MODEL_BASE_URL: ${MODEL_BASE_URL:-http://192.168.0.91:11434} + MODEL_API_KEY: ${MODEL_API_KEY:-ollama} + MODEL_NAME: ${MODEL_NAME:-qwen3.6:35b-a3b-bf16} + MODEL_CONTEXT_SIZE: ${MODEL_CONTEXT_SIZE:-262144} + MODEL_TIMEOUT_SEC: ${MODEL_TIMEOUT_SEC:-300} + A2A_EDGE_URL: ${A2A_EDGE_URL:-} + A2A_AGENT_URL: ${A2A_AGENT_URL:-} + A2A_TOKEN: ${A2A_TOKEN:-} + A2A_TIMEOUT_SEC: ${A2A_TIMEOUT_SEC:-300} MATTERMOST_BASE_URL: "" MATTERMOST_TOKEN: "" PLANE_BASE_URL: "" diff --git a/internal/adapters/a2a/client.go b/internal/adapters/a2a/client.go new file mode 100644 index 0000000..5749a12 --- /dev/null +++ b/internal/adapters/a2a/client.go @@ -0,0 +1,299 @@ +package a2a + +import ( + "bytes" + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "net/url" + "strings" + "sync/atomic" + "time" + + "github.com/nomadcode/nomadcode-core/internal/agent" +) + +const ( + defaultTimeoutSec = 300 + jsonRPCVersion = "2.0" +) + +type Config struct { + URL string + Token string + TimeoutSec int +} + +type Client struct { + cfg Config + httpClient *http.Client + logger *slog.Logger + nextID atomic.Uint64 +} + +func NewClient(cfg Config, logger *slog.Logger) *Client { + if cfg.TimeoutSec <= 0 { + cfg.TimeoutSec = defaultTimeoutSec + } + return &Client{ + cfg: cfg, + httpClient: &http.Client{Timeout: time.Duration(cfg.TimeoutSec) * time.Second}, + logger: logger, + } +} + +func (c *Client) SendMessage(ctx context.Context, input agent.SendMessageInput) (agent.SendMessageResult, error) { + if strings.TrimSpace(input.Text) == "" { + return agent.SendMessageResult{}, fmt.Errorf("a2a message text is required") + } + + messageID := newMessageID() + msg := agent.Message{ + Kind: "message", + Role: "user", + MessageID: messageID, + ContextID: input.ContextID, + TaskID: input.TaskID, + Parts: []agent.Part{{ + Kind: "text", + Text: input.Text, + }}, + } + + params := messageSendParams{ + Message: msg, + Configuration: &messageSendConfiguration{ + AcceptedOutputModes: input.AcceptedOutputModes, + HistoryLength: input.HistoryLength, + Blocking: input.Blocking, + }, + Metadata: input.Metadata, + } + if params.Configuration.empty() { + params.Configuration = nil + } + + raw, err := c.call(ctx, "message/send", params) + if err != nil { + return agent.SendMessageResult{}, err + } + result, err := decodeMessageSendResult(raw) + if err != nil { + return agent.SendMessageResult{}, err + } + result.Raw = raw + return result, nil +} + +func (c *Client) GetTask(ctx context.Context, input agent.GetTaskInput) (agent.Task, error) { + if strings.TrimSpace(input.TaskID) == "" { + return agent.Task{}, fmt.Errorf("a2a task id is required") + } + raw, err := c.call(ctx, "tasks/get", taskIDParams{ + ID: input.TaskID, + HistoryLength: input.HistoryLength, + Metadata: input.Metadata, + }) + if err != nil { + return agent.Task{}, err + } + var task agent.Task + if err := json.Unmarshal(raw, &task); err != nil { + return agent.Task{}, fmt.Errorf("decode a2a task: %w", err) + } + return task, nil +} + +func (c *Client) CancelTask(ctx context.Context, input agent.CancelTaskInput) (agent.Task, error) { + if strings.TrimSpace(input.TaskID) == "" { + return agent.Task{}, fmt.Errorf("a2a task id is required") + } + raw, err := c.call(ctx, "tasks/cancel", taskIDParams{ + ID: input.TaskID, + Metadata: input.Metadata, + }) + if err != nil { + return agent.Task{}, err + } + var task agent.Task + if err := json.Unmarshal(raw, &task); err != nil { + return agent.Task{}, fmt.Errorf("decode canceled a2a task: %w", err) + } + return task, nil +} + +type messageSendParams struct { + Message agent.Message `json:"message"` + Configuration *messageSendConfiguration `json:"configuration,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +type messageSendConfiguration struct { + AcceptedOutputModes []string `json:"acceptedOutputModes,omitempty"` + HistoryLength int `json:"historyLength,omitempty"` + Blocking bool `json:"blocking,omitempty"` +} + +func (c *messageSendConfiguration) empty() bool { + return c == nil || (len(c.AcceptedOutputModes) == 0 && c.HistoryLength == 0 && !c.Blocking) +} + +type taskIDParams struct { + ID string `json:"id"` + HistoryLength int `json:"historyLength,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +type jsonRPCRequest struct { + JSONRPC string `json:"jsonrpc"` + ID string `json:"id"` + Method string `json:"method"` + Params any `json:"params,omitempty"` +} + +type jsonRPCResponse struct { + JSONRPC string `json:"jsonrpc"` + ID string `json:"id"` + Result json.RawMessage `json:"result,omitempty"` + Error *jsonRPCError `json:"error,omitempty"` +} + +type jsonRPCError struct { + Code int `json:"code"` + Message string `json:"message"` + Data json.RawMessage `json:"data,omitempty"` +} + +func (c *Client) call(ctx context.Context, method string, params any) (json.RawMessage, error) { + endpoint, err := endpointURL(c.cfg.URL) + if err != nil { + return nil, err + } + + requestID := c.requestID() + payload, err := json.Marshal(jsonRPCRequest{ + JSONRPC: jsonRPCVersion, + ID: requestID, + Method: method, + Params: params, + }) + if err != nil { + return nil, err + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(payload)) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/json") + if c.cfg.Token != "" { + req.Header.Set("Authorization", "Bearer "+c.cfg.Token) + } + + if c.logger != nil { + c.logger.Info("a2a json-rpc request", "method", method, "endpoint", endpoint) + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20)) + if err != nil { + return nil, err + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return nil, fmt.Errorf("a2a request failed: status %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + } + + var rpcResp jsonRPCResponse + if err := json.Unmarshal(body, &rpcResp); err != nil { + return nil, fmt.Errorf("decode a2a json-rpc response: %w", err) + } + if rpcResp.Error != nil { + return nil, fmt.Errorf("a2a json-rpc error %d: %s", rpcResp.Error.Code, rpcResp.Error.Message) + } + if rpcResp.ID != "" && rpcResp.ID != requestID { + return nil, fmt.Errorf("a2a response id mismatch: got %q want %q", rpcResp.ID, requestID) + } + if len(rpcResp.Result) == 0 { + return nil, fmt.Errorf("a2a response result is empty") + } + return rpcResp.Result, nil +} + +func (c *Client) requestID() string { + return fmt.Sprintf("nomadcode-%d", c.nextID.Add(1)) +} + +func decodeMessageSendResult(raw json.RawMessage) (agent.SendMessageResult, error) { + var envelope struct { + Kind string `json:"kind"` + } + _ = json.Unmarshal(raw, &envelope) + + switch envelope.Kind { + case "message": + var msg agent.Message + if err := json.Unmarshal(raw, &msg); err != nil { + return agent.SendMessageResult{}, fmt.Errorf("decode a2a message: %w", err) + } + return agent.SendMessageResult{Message: &msg}, nil + case "task": + var task agent.Task + if err := json.Unmarshal(raw, &task); err != nil { + return agent.SendMessageResult{}, fmt.Errorf("decode a2a task: %w", err) + } + return agent.SendMessageResult{Task: &task}, nil + } + + var probe map[string]json.RawMessage + if err := json.Unmarshal(raw, &probe); err != nil { + return agent.SendMessageResult{}, fmt.Errorf("decode a2a result: %w", err) + } + if _, ok := probe["status"]; ok { + var task agent.Task + if err := json.Unmarshal(raw, &task); err != nil { + return agent.SendMessageResult{}, fmt.Errorf("decode a2a task: %w", err) + } + return agent.SendMessageResult{Task: &task}, nil + } + if _, ok := probe["role"]; ok { + var msg agent.Message + if err := json.Unmarshal(raw, &msg); err != nil { + return agent.SendMessageResult{}, fmt.Errorf("decode a2a message: %w", err) + } + return agent.SendMessageResult{Message: &msg}, nil + } + return agent.SendMessageResult{}, fmt.Errorf("a2a result is neither message nor task") +} + +func endpointURL(rawURL string) (string, error) { + rawURL = strings.TrimSpace(rawURL) + if rawURL == "" { + return "", fmt.Errorf("a2a endpoint URL is required") + } + parsed, err := url.Parse(rawURL) + if err != nil { + return "", err + } + if parsed.Scheme == "" || parsed.Host == "" { + return "", fmt.Errorf("a2a endpoint URL must include scheme and host") + } + return parsed.String(), nil +} + +func newMessageID() string { + var b [16]byte + if _, err := rand.Read(b[:]); err == nil { + return "msg-" + hex.EncodeToString(b[:]) + } + return fmt.Sprintf("msg-%d", time.Now().UnixNano()) +} diff --git a/internal/adapters/a2a/client_test.go b/internal/adapters/a2a/client_test.go new file mode 100644 index 0000000..c298761 --- /dev/null +++ b/internal/adapters/a2a/client_test.go @@ -0,0 +1,162 @@ +package a2a + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/nomadcode/nomadcode-core/internal/agent" +) + +func TestSendMessagePostsJSONRPCMessageSend(t *testing.T) { + var gotAuth string + var gotReq jsonRPCRequest + var gotParams map[string]any + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotAuth = r.Header.Get("Authorization") + if err := json.NewDecoder(r.Body).Decode(&gotReq); err != nil { + t.Fatalf("decode request: %v", err) + } + paramsBytes, _ := json.Marshal(gotReq.Params) + if err := json.Unmarshal(paramsBytes, &gotParams); err != nil { + t.Fatalf("decode params: %v", err) + } + _, _ = w.Write([]byte(`{ + "jsonrpc": "2.0", + "id": "nomadcode-1", + "result": { + "kind": "task", + "id": "task-1", + "contextId": "ctx-1", + "status": {"state": "working"} + } + }`)) + })) + defer server.Close() + + client := NewClient(Config{URL: server.URL, Token: "secret"}, nil) + result, err := client.SendMessage(context.Background(), agent.SendMessageInput{ + Text: "fix issue", + ContextID: "ctx-1", + AcceptedOutputModes: []string{"text/plain"}, + Blocking: true, + Metadata: map[string]any{ + "nomad_task_id": "task-local", + }, + }) + if err != nil { + t.Fatalf("SendMessage returned error: %v", err) + } + + if gotAuth != "Bearer secret" { + t.Fatalf("unexpected auth header: %q", gotAuth) + } + if gotReq.JSONRPC != "2.0" || gotReq.Method != "message/send" || gotReq.ID != "nomadcode-1" { + t.Fatalf("unexpected json-rpc request: %#v", gotReq) + } + message := gotParams["message"].(map[string]any) + if message["role"] != "user" || message["kind"] != "message" || message["contextId"] != "ctx-1" { + t.Fatalf("unexpected message: %#v", message) + } + parts := message["parts"].([]any) + part := parts[0].(map[string]any) + if part["kind"] != "text" || part["text"] != "fix issue" { + t.Fatalf("unexpected part: %#v", part) + } + config := gotParams["configuration"].(map[string]any) + if config["blocking"] != true { + t.Fatalf("expected blocking=true, got %#v", config["blocking"]) + } + metadata := gotParams["metadata"].(map[string]any) + if metadata["nomad_task_id"] != "task-local" { + t.Fatalf("unexpected metadata: %#v", metadata) + } + if result.Task == nil || result.Task.ID != "task-1" || result.Task.Status.State != agent.TaskStateWorking { + t.Fatalf("unexpected result: %#v", result) + } +} + +func TestSendMessageReturnsDirectMessage(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`{ + "jsonrpc": "2.0", + "id": "nomadcode-1", + "result": { + "kind": "message", + "role": "agent", + "messageId": "msg-2", + "parts": [{"kind": "text", "text": "done"}] + } + }`)) + })) + defer server.Close() + + client := NewClient(Config{URL: server.URL}, nil) + result, err := client.SendMessage(context.Background(), agent.SendMessageInput{Text: "status"}) + if err != nil { + t.Fatalf("SendMessage returned error: %v", err) + } + if result.Message == nil || result.Message.Role != "agent" || result.Message.Parts[0].Text != "done" { + t.Fatalf("unexpected result: %#v", result) + } +} + +func TestGetAndCancelTask(t *testing.T) { + var methods []string + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req jsonRPCRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Fatalf("decode request: %v", err) + } + methods = append(methods, req.Method) + _, _ = w.Write([]byte(`{ + "jsonrpc": "2.0", + "id": "` + req.ID + `", + "result": { + "kind": "task", + "id": "task-1", + "status": {"state": "completed"} + } + }`)) + })) + defer server.Close() + + client := NewClient(Config{URL: server.URL}, nil) + task, err := client.GetTask(context.Background(), agent.GetTaskInput{TaskID: "task-1"}) + if err != nil { + t.Fatalf("GetTask returned error: %v", err) + } + if task.Status.State != agent.TaskStateCompleted { + t.Fatalf("unexpected get task: %#v", task) + } + + task, err = client.CancelTask(context.Background(), agent.CancelTaskInput{TaskID: "task-1"}) + if err != nil { + t.Fatalf("CancelTask returned error: %v", err) + } + if task.ID != "task-1" { + t.Fatalf("unexpected cancel task: %#v", task) + } + if len(methods) != 2 || methods[0] != "tasks/get" || methods[1] != "tasks/cancel" { + t.Fatalf("unexpected methods: %#v", methods) + } +} + +func TestJSONRPCError(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`{ + "jsonrpc": "2.0", + "id": "nomadcode-1", + "error": {"code": -32602, "message": "invalid params"} + }`)) + })) + defer server.Close() + + client := NewClient(Config{URL: server.URL}, nil) + if _, err := client.SendMessage(context.Background(), agent.SendMessageInput{Text: "hello"}); err == nil { + t.Fatal("expected JSON-RPC error") + } +} diff --git a/internal/adapters/openai/client.go b/internal/adapters/openai/client.go new file mode 100644 index 0000000..e2f0d0e --- /dev/null +++ b/internal/adapters/openai/client.go @@ -0,0 +1,266 @@ +package openai + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "net/url" + "path" + "strings" + "time" + + "github.com/nomadcode/nomadcode-core/internal/model" +) + +const ( + defaultAPIKey = "ollama" + defaultTimeoutSec = 300 + responsesPath = "/v1/responses" +) + +type Config struct { + BaseURL string + APIKey string + Model string + ContextSize int + TimeoutSec int +} + +type Client struct { + cfg Config + httpClient *http.Client + logger *slog.Logger +} + +func NewClient(cfg Config, logger *slog.Logger) *Client { + if cfg.APIKey == "" { + cfg.APIKey = defaultAPIKey + } + if cfg.TimeoutSec <= 0 { + cfg.TimeoutSec = defaultTimeoutSec + } + return &Client{ + cfg: cfg, + httpClient: &http.Client{Timeout: time.Duration(cfg.TimeoutSec) * time.Second}, + logger: logger, + } +} + +func (c *Client) Generate(ctx context.Context, input model.GenerateInput) (model.GenerateResult, error) { + endpoint, err := responsesURL(c.cfg.BaseURL) + if err != nil { + return model.GenerateResult{}, err + } + + modelName := strings.TrimSpace(input.Model) + if modelName == "" { + modelName = c.cfg.Model + } + if strings.TrimSpace(modelName) == "" { + return model.GenerateResult{}, fmt.Errorf("model name is required") + } + if strings.TrimSpace(input.Input) == "" { + return model.GenerateResult{}, fmt.Errorf("model input is required") + } + + reqBody := responsesRequest{ + Model: modelName, + Input: input.Input, + Instructions: input.Instructions, + Stream: false, + Temperature: input.Temperature, + TopP: input.TopP, + MaxOutputTokens: input.MaxOutputTokens, + } + if c.cfg.ContextSize > 0 { + reqBody.Options = &responsesOptions{NumCtx: c.cfg.ContextSize} + } + body, err := json.Marshal(reqBody) + if err != nil { + return model.GenerateResult{}, err + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body)) + if err != nil { + return model.GenerateResult{}, err + } + req.Header.Set("Content-Type", "application/json") + if c.cfg.APIKey != "" { + req.Header.Set("Authorization", "Bearer "+c.cfg.APIKey) + } + + if c.logger != nil { + c.logger.Info( + "model responses request", + "endpoint", endpoint, + "model", modelName, + "num_ctx", c.cfg.ContextSize, + ) + } + + resp, err := c.httpClient.Do(req) + if err != nil { + return model.GenerateResult{}, err + } + defer resp.Body.Close() + + raw, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20)) + if err != nil { + return model.GenerateResult{}, err + } + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return model.GenerateResult{}, responseError(resp.StatusCode, raw) + } + + var parsed responsesResponse + if err := json.Unmarshal(raw, &parsed); err != nil { + return model.GenerateResult{}, err + } + + usage := model.Usage{} + if parsed.Usage != nil { + usage.InputTokens = firstNonZero(parsed.Usage.InputTokens, parsed.Usage.PromptTokens) + usage.OutputTokens = firstNonZero(parsed.Usage.OutputTokens, parsed.Usage.CompletionTokens) + usage.TotalTokens = parsed.Usage.TotalTokens + if usage.TotalTokens == 0 { + usage.TotalTokens = usage.InputTokens + usage.OutputTokens + } + } + + return model.GenerateResult{ + ID: parsed.ID, + Model: firstNonEmpty(parsed.Model, modelName), + Text: parsed.text(), + Usage: usage, + Raw: json.RawMessage(raw), + }, nil +} + +type responsesRequest struct { + Model string `json:"model"` + Input string `json:"input"` + Instructions string `json:"instructions,omitempty"` + Stream bool `json:"stream"` + Temperature *float64 `json:"temperature,omitempty"` + TopP *float64 `json:"top_p,omitempty"` + MaxOutputTokens int `json:"max_output_tokens,omitempty"` + Options *responsesOptions `json:"options,omitempty"` +} + +type responsesOptions struct { + NumCtx int `json:"num_ctx,omitempty"` +} + +type responsesResponse struct { + ID string `json:"id"` + Model string `json:"model"` + OutputText string `json:"output_text"` + Output []responsesOutput `json:"output"` + Usage *responsesUsage `json:"usage"` +} + +type responsesOutput struct { + Type string `json:"type"` + Text string `json:"text"` + Content []responsesContent `json:"content"` +} + +type responsesContent struct { + Type string `json:"type"` + Text string `json:"text"` +} + +type responsesUsage struct { + InputTokens int `json:"input_tokens"` + OutputTokens int `json:"output_tokens"` + TotalTokens int `json:"total_tokens"` + PromptTokens int `json:"prompt_tokens"` + CompletionTokens int `json:"completion_tokens"` +} + +func (r responsesResponse) text() string { + if r.OutputText != "" { + return r.OutputText + } + + var b strings.Builder + for _, output := range r.Output { + if output.Text != "" { + b.WriteString(output.Text) + } + for _, content := range output.Content { + if content.Text != "" { + b.WriteString(content.Text) + } + } + } + return b.String() +} + +type errorResponse struct { + Error struct { + Message string `json:"message"` + Type string `json:"type"` + Code string `json:"code"` + } `json:"error"` +} + +func responseError(status int, raw []byte) error { + var parsed errorResponse + if err := json.Unmarshal(raw, &parsed); err == nil && parsed.Error.Message != "" { + return fmt.Errorf("responses API request failed: status %d: %s", status, parsed.Error.Message) + } + msg := strings.TrimSpace(string(raw)) + if msg == "" { + msg = http.StatusText(status) + } + return fmt.Errorf("responses API request failed: status %d: %s", status, msg) +} + +func responsesURL(base string) (string, error) { + base = strings.TrimSpace(base) + if base == "" { + return "", fmt.Errorf("model base URL is required") + } + + parsed, err := url.Parse(base) + if err != nil { + return "", err + } + if parsed.Scheme == "" || parsed.Host == "" { + return "", fmt.Errorf("model base URL must include scheme and host") + } + + cleanPath := strings.TrimRight(parsed.Path, "/") + switch { + case strings.HasSuffix(cleanPath, "/v1/responses"): + parsed.Path = cleanPath + case strings.HasSuffix(cleanPath, "/v1"): + parsed.Path = path.Join(cleanPath, "responses") + default: + parsed.Path = path.Join(cleanPath, responsesPath) + } + return parsed.String(), nil +} + +func firstNonZero(values ...int) int { + for _, value := range values { + if value != 0 { + return value + } + } + return 0 +} + +func firstNonEmpty(values ...string) string { + for _, value := range values { + if value != "" { + return value + } + } + return "" +} diff --git a/internal/adapters/openai/client_test.go b/internal/adapters/openai/client_test.go new file mode 100644 index 0000000..2b42e7b --- /dev/null +++ b/internal/adapters/openai/client_test.go @@ -0,0 +1,140 @@ +package openai + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/nomadcode/nomadcode-core/internal/model" +) + +func TestGenerateCallsResponsesAPI(t *testing.T) { + var gotPath string + var gotAuth string + var gotBody map[string]any + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotPath = r.URL.Path + gotAuth = r.Header.Get("Authorization") + if err := json.NewDecoder(r.Body).Decode(&gotBody); err != nil { + t.Fatalf("decode request: %v", err) + } + + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{ + "id": "resp_test", + "model": "qwen3.6:35b-a3b-bf16", + "output_text": "hello from model", + "usage": { + "input_tokens": 3, + "output_tokens": 4, + "total_tokens": 7 + } + }`)) + })) + defer server.Close() + + client := NewClient(Config{ + BaseURL: server.URL, + APIKey: "test-key", + Model: "qwen3.6:35b-a3b-bf16", + ContextSize: 262144, + }, nil) + + result, err := client.Generate(context.Background(), model.GenerateInput{ + Input: "say hello", + Instructions: "be brief", + }) + if err != nil { + t.Fatalf("Generate returned error: %v", err) + } + + if gotPath != "/v1/responses" { + t.Fatalf("expected /v1/responses, got %q", gotPath) + } + if gotAuth != "Bearer test-key" { + t.Fatalf("expected auth header, got %q", gotAuth) + } + if gotBody["model"] != "qwen3.6:35b-a3b-bf16" { + t.Fatalf("unexpected model: %#v", gotBody["model"]) + } + if gotBody["input"] != "say hello" { + t.Fatalf("unexpected input: %#v", gotBody["input"]) + } + if gotBody["instructions"] != "be brief" { + t.Fatalf("unexpected instructions: %#v", gotBody["instructions"]) + } + if _, ok := gotBody["context_size"]; ok { + t.Fatal("Responses API request must not include unsupported context_size") + } + options, ok := gotBody["options"].(map[string]any) + if !ok { + t.Fatalf("expected options object, got %#v", gotBody["options"]) + } + if options["num_ctx"] != float64(262144) { + t.Fatalf("unexpected options.num_ctx: %#v", options["num_ctx"]) + } + if result.Text != "hello from model" { + t.Fatalf("unexpected result text: %q", result.Text) + } + if result.Usage.TotalTokens != 7 { + t.Fatalf("unexpected usage: %#v", result.Usage) + } +} + +func TestGenerateExtractsOutputContent(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{ + "id": "resp_test", + "output": [ + { + "type": "message", + "content": [ + {"type": "output_text", "text": "hello "}, + {"type": "output_text", "text": "again"} + ] + } + ], + "usage": { + "prompt_tokens": 2, + "completion_tokens": 5 + } + }`)) + })) + defer server.Close() + + client := NewClient(Config{BaseURL: server.URL, Model: "model-a"}, nil) + result, err := client.Generate(context.Background(), model.GenerateInput{Input: "say hello"}) + if err != nil { + t.Fatalf("Generate returned error: %v", err) + } + if result.Text != "hello again" { + t.Fatalf("unexpected result text: %q", result.Text) + } + if result.Usage.InputTokens != 2 || result.Usage.OutputTokens != 5 || result.Usage.TotalTokens != 7 { + t.Fatalf("unexpected usage: %#v", result.Usage) + } +} + +func TestResponsesURL(t *testing.T) { + tests := map[string]string{ + "http://localhost:11434": "http://localhost:11434/v1/responses", + "http://localhost:11434/": "http://localhost:11434/v1/responses", + "http://localhost:11434/v1": "http://localhost:11434/v1/responses", + "http://localhost:11434/v1/": "http://localhost:11434/v1/responses", + "http://localhost:11434/v1/responses": "http://localhost:11434/v1/responses", + } + + for input, want := range tests { + got, err := responsesURL(input) + if err != nil { + t.Fatalf("responsesURL(%q) returned error: %v", input, err) + } + if got != want { + t.Fatalf("responsesURL(%q) = %q, want %q", input, got, want) + } + } +} diff --git a/internal/agent/agent.go b/internal/agent/agent.go new file mode 100644 index 0000000..ee28be7 --- /dev/null +++ b/internal/agent/agent.go @@ -0,0 +1,104 @@ +package agent + +import ( + "context" + "encoding/json" +) + +type Client interface { + SendMessage(ctx context.Context, input SendMessageInput) (SendMessageResult, error) + GetTask(ctx context.Context, input GetTaskInput) (Task, error) + CancelTask(ctx context.Context, input CancelTaskInput) (Task, error) +} + +type SendMessageInput struct { + Text string + ContextID string + TaskID string + Metadata map[string]any + AcceptedOutputModes []string + HistoryLength int + Blocking bool +} + +type SendMessageResult struct { + Message *Message + Task *Task + Raw json.RawMessage +} + +type GetTaskInput struct { + TaskID string + HistoryLength int + Metadata map[string]any +} + +type CancelTaskInput struct { + TaskID string + Metadata map[string]any +} + +type TaskState string + +const ( + TaskStateSubmitted TaskState = "submitted" + TaskStateWorking TaskState = "working" + TaskStateInputNeeded TaskState = "input-required" + TaskStateCompleted TaskState = "completed" + TaskStateCanceled TaskState = "canceled" + TaskStateFailed TaskState = "failed" + TaskStateRejected TaskState = "rejected" + TaskStateAuthRequired TaskState = "auth-required" + TaskStateUnknown TaskState = "unknown" +) + +type Task struct { + Kind string `json:"kind,omitempty"` + ID string `json:"id"` + ContextID string `json:"contextId,omitempty"` + Status TaskStatus `json:"status"` + Artifacts []Artifact `json:"artifacts,omitempty"` + History []Message `json:"history,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +type TaskStatus struct { + State TaskState `json:"state"` + Message *Message `json:"message,omitempty"` + Timestamp string `json:"timestamp,omitempty"` +} + +type Message struct { + Kind string `json:"kind,omitempty"` + Role string `json:"role"` + Parts []Part `json:"parts"` + MessageID string `json:"messageId"` + TaskID string `json:"taskId,omitempty"` + ContextID string `json:"contextId,omitempty"` + ReferenceTaskIDs []string `json:"referenceTaskIds,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` + Extensions []string `json:"extensions,omitempty"` +} + +type Artifact struct { + ArtifactID string `json:"artifactId,omitempty"` + Name string `json:"name,omitempty"` + Description string `json:"description,omitempty"` + Parts []Part `json:"parts,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +type Part struct { + Kind string `json:"kind"` + Text string `json:"text,omitempty"` + Data map[string]any `json:"data,omitempty"` + File *FilePart `json:"file,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` +} + +type FilePart struct { + Name string `json:"name,omitempty"` + MimeType string `json:"mimeType,omitempty"` + Bytes string `json:"bytes,omitempty"` + URI string `json:"uri,omitempty"` +} diff --git a/internal/config/config.go b/internal/config/config.go index fb6e14c..626d3a2 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -1,6 +1,9 @@ package config -import "os" +import ( + "os" + "strconv" +) type Config struct { AppEnv string @@ -10,6 +13,15 @@ type Config struct { RedisKeyPrefix string AuthUsername string AuthPassword string + ModelBaseURL string + ModelAPIKey string + ModelName string + ModelContextSize int + ModelTimeoutSec int + A2AEdgeURL string + A2AAgentURL string + A2AToken string + A2ATimeoutSec int MattermostBaseURL string MattermostToken string PlaneBaseURL string @@ -25,6 +37,15 @@ func Load() Config { RedisKeyPrefix: os.Getenv("REDIS_KEY_PREFIX"), AuthUsername: getEnv("AUTH_USERNAME", "nomadcode"), AuthPassword: os.Getenv("AUTH_PASSWORD"), + ModelBaseURL: getEnv("MODEL_BASE_URL", "http://192.168.0.91:11434"), + ModelAPIKey: getEnv("MODEL_API_KEY", "ollama"), + ModelName: getEnv("MODEL_NAME", "qwen3.6:35b-a3b-bf16"), + ModelContextSize: getEnvInt("MODEL_CONTEXT_SIZE", 262144), + ModelTimeoutSec: getEnvInt("MODEL_TIMEOUT_SEC", 300), + A2AEdgeURL: firstEnv("A2A_EDGE_URL", "A2A_AGENT_URL"), + A2AAgentURL: firstEnv("A2A_AGENT_URL", "A2A_EDGE_URL"), + A2AToken: os.Getenv("A2A_TOKEN"), + A2ATimeoutSec: getEnvInt("A2A_TIMEOUT_SEC", 300), MattermostBaseURL: os.Getenv("MATTERMOST_BASE_URL"), MattermostToken: os.Getenv("MATTERMOST_TOKEN"), PlaneBaseURL: os.Getenv("PLANE_BASE_URL"), @@ -39,3 +60,24 @@ func getEnv(key, fallback string) string { } return value } + +func firstEnv(keys ...string) string { + for _, key := range keys { + if value := os.Getenv(key); value != "" { + return value + } + } + return "" +} + +func getEnvInt(key string, fallback int) int { + value := os.Getenv(key) + if value == "" { + return fallback + } + parsed, err := strconv.Atoi(value) + if err != nil { + return fallback + } + return parsed +} diff --git a/internal/config/config_test.go b/internal/config/config_test.go new file mode 100644 index 0000000..3983073 --- /dev/null +++ b/internal/config/config_test.go @@ -0,0 +1,53 @@ +package config + +import "testing" + +func TestLoadA2AEdgeURLPrefersEdgeURL(t *testing.T) { + t.Setenv("A2A_EDGE_URL", "http://edge.example/a2a") + t.Setenv("A2A_AGENT_URL", "http://legacy.example/a2a") + + cfg := Load() + + if cfg.A2AEdgeURL != "http://edge.example/a2a" { + t.Fatalf("A2AEdgeURL: got %q", cfg.A2AEdgeURL) + } + if cfg.A2AAgentURL != "http://legacy.example/a2a" { + t.Fatalf("A2AAgentURL: got %q", cfg.A2AAgentURL) + } +} + +func TestLoadA2AEdgeURLFallsBackToLegacyAgentURL(t *testing.T) { + t.Setenv("A2A_AGENT_URL", "http://legacy.example/a2a") + + cfg := Load() + + if cfg.A2AEdgeURL != "http://legacy.example/a2a" { + t.Fatalf("A2AEdgeURL fallback: got %q", cfg.A2AEdgeURL) + } + if cfg.A2AAgentURL != "http://legacy.example/a2a" { + t.Fatalf("A2AAgentURL: got %q", cfg.A2AAgentURL) + } +} + +func TestLoadModelAndA2ADefaults(t *testing.T) { + cfg := Load() + + if cfg.ModelBaseURL != "http://192.168.0.91:11434" { + t.Fatalf("ModelBaseURL: got %q", cfg.ModelBaseURL) + } + if cfg.ModelAPIKey != "ollama" { + t.Fatalf("ModelAPIKey: got %q", cfg.ModelAPIKey) + } + if cfg.ModelName != "qwen3.6:35b-a3b-bf16" { + t.Fatalf("ModelName: got %q", cfg.ModelName) + } + if cfg.ModelContextSize != 262144 { + t.Fatalf("ModelContextSize: got %d", cfg.ModelContextSize) + } + if cfg.ModelTimeoutSec != 300 { + t.Fatalf("ModelTimeoutSec: got %d", cfg.ModelTimeoutSec) + } + if cfg.A2ATimeoutSec != 300 { + t.Fatalf("A2ATimeoutSec: got %d", cfg.A2ATimeoutSec) + } +} diff --git a/internal/model/model.go b/internal/model/model.go new file mode 100644 index 0000000..4b0dc67 --- /dev/null +++ b/internal/model/model.go @@ -0,0 +1,34 @@ +package model + +import ( + "context" + "encoding/json" +) + +type Client interface { + Generate(ctx context.Context, input GenerateInput) (GenerateResult, error) +} + +type GenerateInput struct { + Model string + Instructions string + Input string + Metadata map[string]string + MaxOutputTokens int + Temperature *float64 + TopP *float64 +} + +type GenerateResult struct { + ID string + Model string + Text string + Usage Usage + Raw json.RawMessage +} + +type Usage struct { + InputTokens int + OutputTokens int + TotalTokens int +} diff --git a/internal/scheduler/jobs.go b/internal/scheduler/jobs.go index 747be9c..3e32f7b 100644 --- a/internal/scheduler/jobs.go +++ b/internal/scheduler/jobs.go @@ -3,11 +3,15 @@ package scheduler import ( "context" "encoding/json" + "fmt" "log/slog" + "strings" "time" "github.com/riverqueue/river" + "github.com/nomadcode/nomadcode-core/internal/agent" + "github.com/nomadcode/nomadcode-core/internal/model" "github.com/nomadcode/nomadcode-core/internal/notification" "github.com/nomadcode/nomadcode-core/internal/storage" "github.com/nomadcode/nomadcode-core/internal/workflow" @@ -26,6 +30,8 @@ type TaskWorker struct { Store *storage.Store Notifications *notification.Service + Agent agent.Client + Model model.Client Logger *slog.Logger } @@ -40,14 +46,12 @@ func (w *TaskWorker) Work(ctx context.Context, job *river.Job[TaskJobArgs]) erro return err } - select { - case <-time.After(500 * time.Millisecond): - case <-ctx.Done(): - w.markFailed(taskID, ctx.Err()) - return ctx.Err() + result, message, err := w.runTask(ctx, task) + if err != nil { + w.markFailed(taskID, err) + return err } - result := json.RawMessage(`{"message":"dummy task completed"}`) task, err = w.Store.CompleteTask(ctx, taskID, result) if err != nil { w.markFailed(taskID, err) @@ -59,7 +63,7 @@ func (w *TaskWorker) Work(ctx context.Context, job *river.Job[TaskJobArgs]) erro TaskID: task.ID, Title: task.Title, Status: task.Status, - Message: "dummy task completed", + Message: message, }) if err != nil && w.Logger != nil { w.Logger.Warn("task notification failed", "task_id", taskID, "error", err) @@ -72,6 +76,221 @@ func (w *TaskWorker) Work(ctx context.Context, job *river.Job[TaskJobArgs]) erro return nil } +func (w *TaskWorker) runTask(ctx context.Context, task storage.Task) (json.RawMessage, string, error) { + if w.Agent != nil { + return w.runAgentTask(ctx, task) + } + + if w.Model == nil { + select { + case <-time.After(500 * time.Millisecond): + case <-ctx.Done(): + return nil, "", ctx.Err() + } + return json.RawMessage(`{"message":"dummy task completed","mode":"dummy"}`), "dummy task completed", nil + } + + generated, err := w.Model.Generate(ctx, buildGenerateInput(task)) + if err != nil { + return nil, "", err + } + + result := map[string]any{ + "message": generated.Text, + "mode": "openai_responses", + "model": generated.Model, + "response_id": generated.ID, + "usage": map[string]int{ + "input_tokens": generated.Usage.InputTokens, + "output_tokens": generated.Usage.OutputTokens, + "total_tokens": generated.Usage.TotalTokens, + }, + } + raw, err := json.Marshal(result) + if err != nil { + return nil, "", err + } + return raw, "model task completed", nil +} + +func (w *TaskWorker) runAgentTask(ctx context.Context, task storage.Task) (json.RawMessage, string, error) { + result, err := w.Agent.SendMessage(ctx, buildAgentInput(task)) + if err != nil { + return nil, "", err + } + + output := map[string]any{ + "mode": "a2a", + } + if len(result.Raw) > 0 { + output["raw"] = json.RawMessage(result.Raw) + } + + message := "a2a task completed" + if result.Message != nil { + text := messageText(*result.Message) + if text != "" { + message = text + output["message"] = text + } + output["message_id"] = result.Message.MessageID + raw, err := json.Marshal(output) + if err != nil { + return nil, "", err + } + return raw, message, nil + } + + if result.Task == nil { + return nil, "", fmt.Errorf("a2a result does not include message or task") + } + + state := result.Task.Status.State + output["a2a_task_id"] = result.Task.ID + output["a2a_state"] = state + + text := taskOutputText(*result.Task) + if text != "" { + message = text + output["message"] = text + } + + switch state { + case agent.TaskStateCompleted: + case agent.TaskStateFailed, agent.TaskStateCanceled, agent.TaskStateRejected, agent.TaskStateInputNeeded, agent.TaskStateAuthRequired: + if text == "" { + text = fmt.Sprintf("a2a task ended with state %s", state) + } + return nil, "", fmt.Errorf("%s", text) + default: + return nil, "", fmt.Errorf("a2a blocking call returned non-terminal task state %s", state) + } + + raw, err := json.Marshal(output) + if err != nil { + return nil, "", err + } + return raw, message, nil +} + +func buildGenerateInput(task storage.Task) model.GenerateInput { + input := strings.TrimSpace(task.Title) + instructions := "" + + if len(task.Payload) > 0 && string(task.Payload) != "null" { + if payloadInput, payloadInstructions, ok := parsePayloadPrompt(task.Payload); ok { + if payloadInput != "" { + input = payloadInput + } + instructions = payloadInstructions + } else if string(task.Payload) != "{}" { + input = fmt.Sprintf("Task title: %s\nPayload:\n%s", task.Title, string(task.Payload)) + } + } + + return model.GenerateInput{ + Input: input, + Instructions: instructions, + Metadata: map[string]string{ + "task_id": task.ID, + "source": task.Source, + }, + } +} + +func buildAgentInput(task storage.Task) agent.SendMessageInput { + generateInput := buildGenerateInput(task) + text := strings.TrimSpace(generateInput.Input) + if generateInput.Instructions != "" { + text = fmt.Sprintf("Instructions:\n%s\n\nTask:\n%s", generateInput.Instructions, text) + } + if text == "" { + text = task.Title + } + + return agent.SendMessageInput{ + Text: text, + AcceptedOutputModes: []string{"text/plain", "application/json"}, + Blocking: true, + Metadata: map[string]any{ + "nomadcode_task_id": task.ID, + "source": task.Source, + "title": task.Title, + }, + } +} + +func parsePayloadPrompt(payload json.RawMessage) (input string, instructions string, ok bool) { + var fields map[string]json.RawMessage + if err := json.Unmarshal(payload, &fields); err != nil { + return "", "", false + } + ok = true + input = firstStringField(fields, "prompt", "message", "input") + instructions = firstStringField(fields, "instructions", "system") + return input, instructions, ok +} + +func firstStringField(fields map[string]json.RawMessage, names ...string) string { + for _, name := range names { + raw, exists := fields[name] + if !exists { + continue + } + var value string + if err := json.Unmarshal(raw, &value); err == nil { + if value = strings.TrimSpace(value); value != "" { + return value + } + } + } + return "" +} + +func taskOutputText(task agent.Task) string { + if task.Status.Message != nil { + if text := messageText(*task.Status.Message); text != "" { + return text + } + } + + for _, artifact := range task.Artifacts { + if text := partsText(artifact.Parts); text != "" { + return text + } + } + + for i := len(task.History) - 1; i >= 0; i-- { + if task.History[i].Role == "agent" || task.History[i].Role == "assistant" { + if text := messageText(task.History[i]); text != "" { + return text + } + } + } + for i := len(task.History) - 1; i >= 0; i-- { + if text := messageText(task.History[i]); text != "" { + return text + } + } + + return "" +} + +func messageText(message agent.Message) string { + return partsText(message.Parts) +} + +func partsText(parts []agent.Part) string { + texts := make([]string, 0, len(parts)) + for _, part := range parts { + text := strings.TrimSpace(part.Text) + if text != "" { + texts = append(texts, text) + } + } + return strings.Join(texts, "\n") +} + func (w *TaskWorker) markFailed(taskID string, err error) { if err == nil || w.Store == nil { return diff --git a/internal/scheduler/jobs_test.go b/internal/scheduler/jobs_test.go new file mode 100644 index 0000000..5842087 --- /dev/null +++ b/internal/scheduler/jobs_test.go @@ -0,0 +1,108 @@ +package scheduler + +import ( + "context" + "encoding/json" + "strings" + "testing" + + "github.com/nomadcode/nomadcode-core/internal/agent" + "github.com/nomadcode/nomadcode-core/internal/storage" +) + +func TestRunAgentTaskCompletesFromMessage(t *testing.T) { + worker := &TaskWorker{ + Agent: fakeAgentClient{ + result: agent.SendMessageResult{ + Message: &agent.Message{ + Kind: "message", + Role: "agent", + MessageID: "msg-1", + Parts: []agent.Part{{Kind: "text", Text: "done"}}, + }, + Raw: json.RawMessage(`{"kind":"message","messageId":"msg-1"}`), + }, + }, + } + + raw, message, err := worker.runAgentTask(context.Background(), storage.Task{ + ID: "task-1", + Title: "Fix issue", + Source: "manual", + Payload: json.RawMessage(`{"prompt":"please fix it"}`), + }) + if err != nil { + t.Fatalf("runAgentTask returned error: %v", err) + } + if message != "done" { + t.Fatalf("unexpected message: %q", message) + } + + var output map[string]any + if err := json.Unmarshal(raw, &output); err != nil { + t.Fatalf("decode output: %v", err) + } + if output["mode"] != "a2a" || output["message"] != "done" || output["message_id"] != "msg-1" { + t.Fatalf("unexpected output: %#v", output) + } +} + +func TestRunAgentTaskRejectsNonTerminalTask(t *testing.T) { + worker := &TaskWorker{ + Agent: fakeAgentClient{ + result: agent.SendMessageResult{ + Task: &agent.Task{ + Kind: "task", + ID: "remote-1", + Status: agent.TaskStatus{State: agent.TaskStateWorking}, + }, + }, + }, + } + + _, _, err := worker.runAgentTask(context.Background(), storage.Task{ + ID: "task-1", + Title: "Fix issue", + Source: "manual", + Payload: json.RawMessage(`{}`), + }) + if err == nil || !strings.Contains(err.Error(), "non-terminal") { + t.Fatalf("expected non-terminal state error, got %v", err) + } +} + +func TestBuildAgentInputUsesPromptPayload(t *testing.T) { + input := buildAgentInput(storage.Task{ + ID: "task-1", + Title: "Fallback title", + Source: "manual", + Payload: json.RawMessage(`{"prompt":"do the thing","instructions":"be concise"}`), + }) + + if !input.Blocking { + t.Fatal("expected blocking A2A call") + } + if !strings.Contains(input.Text, "be concise") || !strings.Contains(input.Text, "do the thing") { + t.Fatalf("unexpected text: %q", input.Text) + } + if input.Metadata["nomadcode_task_id"] != "task-1" { + t.Fatalf("unexpected metadata: %#v", input.Metadata) + } +} + +type fakeAgentClient struct { + result agent.SendMessageResult + err error +} + +func (c fakeAgentClient) SendMessage(context.Context, agent.SendMessageInput) (agent.SendMessageResult, error) { + return c.result, c.err +} + +func (c fakeAgentClient) GetTask(context.Context, agent.GetTaskInput) (agent.Task, error) { + return agent.Task{}, nil +} + +func (c fakeAgentClient) CancelTask(context.Context, agent.CancelTaskInput) (agent.Task, error) { + return agent.Task{}, nil +} diff --git a/internal/scheduler/river.go b/internal/scheduler/river.go index c2d5d83..7b0f732 100644 --- a/internal/scheduler/river.go +++ b/internal/scheduler/river.go @@ -10,6 +10,8 @@ import ( "github.com/riverqueue/river/riverdriver/riverpgxv5" "github.com/riverqueue/river/rivermigrate" + "github.com/nomadcode/nomadcode-core/internal/agent" + "github.com/nomadcode/nomadcode-core/internal/model" "github.com/nomadcode/nomadcode-core/internal/notification" "github.com/nomadcode/nomadcode-core/internal/storage" ) @@ -20,11 +22,13 @@ type Client struct { logger *slog.Logger } -func New(pool *pgxpool.Pool, store *storage.Store, notifications *notification.Service, logger *slog.Logger) (*Client, error) { +func New(pool *pgxpool.Pool, store *storage.Store, notifications *notification.Service, agentClient agent.Client, modelClient model.Client, logger *slog.Logger) (*Client, error) { workers := river.NewWorkers() river.AddWorker(workers, &TaskWorker{ Store: store, Notifications: notifications, + Agent: agentClient, + Model: modelClient, Logger: logger, })