toki/nomadcode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

update: add a2a/openai adapters, agent module, model, config tests, and scheduler updates

Browse source
This commit is contained in:
toki 2026-05-20 21:09:23 +09:00
parent fa65adcb39
commit b585b5d134
18 changed files with 1515 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
.antigravityignore Normal file
View file

@ -0,0 +1 @@
agent-task/archive/**

3
.gitignore vendored
View file

@ -4,3 +4,6 @@ coverage.out
# Agent-Ops Private Rules
agent-ops/rules/private/
# AI/IDE Cache
/.antigravitycli/

8
Makefile
View file

@ -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

26
README.md
View file

@ -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

11
bin/run
View file

@ -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 "$@"

22
cmd/server/main.go
View file

@ -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
}

9
docker-compose.yml
View file

@ -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: ""

299
internal/adapters/a2a/client.go Normal file
View file

@ -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())
}

162
internal/adapters/a2a/client_test.go Normal file
View file

@ -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")
}
}

266
internal/adapters/openai/client.go Normal file
View file

@ -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 ""
}

140
internal/adapters/openai/client_test.go Normal file
View file

@ -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)
}
}
}

104
internal/agent/agent.go Normal file
View file

@ -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"`
}

44
internal/config/config.go
View file

@ -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
}

53
internal/config/config_test.go Normal file
View file

@ -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)
}
}

34
internal/model/model.go Normal file
View file

@ -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
}

233
internal/scheduler/jobs.go
View file

@ -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

108
internal/scheduler/jobs_test.go Normal file
View file

@ -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
}

6
internal/scheduler/river.go
View file

@ -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,
})