diff --git a/Makefile b/Makefile index f52b6bc..4daf58d 100644 --- a/Makefile +++ b/Makefile @@ -4,6 +4,7 @@ GO ?= go build: $(GO) build ./... + cd services/atomizer && $(GO) build ./... run: $(GO) run ./cmd/app @@ -16,6 +17,13 @@ test: test-integration: $(GO) test -tags=integration ./... +# Requires the mock services running: +# docker compose --profile mocks up -d --build atomizer-mock work-performer +test-contract: + ATOMIZER_TOKEN=$$(grep '^ATOMIZER_TOKEN=' .env | cut -d= -f2-) \ + WORK_PERFORMER_TOKEN=$$(grep '^WORK_PERFORMER_TOKEN=' .env | cut -d= -f2-) \ + $(GO) test -tags=contract -count=1 -timeout=8m ./internal/contract/ + lint: @unformatted=$$(gofmt -l .); if [ -n "$$unformatted" ]; then \ echo "gofmt needed on:"; echo "$$unformatted"; exit 1; fi diff --git a/PROGRESS.md b/PROGRESS.md index f35a237..71ef961 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -11,3 +11,4 @@ - Phase 8 (bounty board + lifecycle): developer board (pool-scoped, filters/search/sort, stale age badges, hide-competing-claims setting, saved filters), claim/withdraw/decline/approve, assign-to-AI via §5.2 client + HMAC-verified idempotent callback w/ artifact ingestion into GridFS, start/submit/abandon/unassign, comments (sanitized, @mentions→notifications), time log, review queue + per-AC checklist on timeline, immutable bountyAwards on approve (humans only), notifications center + bell + WS toasts, my-tasks kanban, task detail page, pool management UI, server-side HTML sanitizer w/ XSS vector tests — full lifecycle integration tests green. - Phase 9 (messaging): conversations (dm dedupe/group/project), sanitized rich-text messages, file/image attachments (generic POST /api/v1/files w/ rate limit), per-conversation unread counts + mark-read, chat-file access limited to participants, mentions→notifications, typing indicator over WS inbound relay, two-pane messages UI (contenteditable composer, formatting toolbar, drag&drop upload, inline previews + lightbox, live delivery, polling fallback), user directory search — integration tests green. - Phase 10 (metrics): aggregation pipelines over bountyAwards + timelines (totals, weekly $dateTrunc buckets, per-developer/per-customer, approval rate, time logged, assign→approve lead time, atomization lead time, open board depth), CSV export, leaderboard w/ opt-out (profile setting), dependency-free SVG line+bar charts, developer/consultant/admin dashboards — integration tests green. +- Phase 11 (mock services): services/atomizer (own Go module, ports 8090, bearer auth, OpenAI-compatible chat completions + native /v1/messages via LLM_API_STYLE, strict-JSON prompt, fence-stripping defensive parse, one retry, deterministic equal-split fallback, exact sum normalization); services/work-performer (Node 20, zero deps, single-concurrency queue, TASK.md + attachment download + optional git clone, claude -p --output-format json --dangerously-skip-permissions, simulated result when CLI unavailable, artifact serving, HMAC-signed callbacks with retry); compose profile mocks w/ healthchecks + ${HOME}/.claude mounts; contract tests pass live (WP ran real Claude Code via mounted host creds). UFW rule added for container→host callbacks. diff --git a/docker-compose.yml b/docker-compose.yml index 5d9e882..c5825e7 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -39,5 +39,52 @@ services: retries: 3 start_period: 10s + # --- placeholder external services (docker compose --profile mocks up) --- + # Two fully independent services: separate builds, ports, tokens (§5). + atomizer-mock: + build: ./services/atomizer + profiles: ["mocks"] + restart: unless-stopped + environment: + PORT: "8090" + ATOMIZER_TOKEN: ${ATOMIZER_TOKEN} + ANTHROPIC_API_KEY: ${ANTHROPIC_API_KEY:-} + ANTHROPIC_OPENAI_BASE_URL: ${ANTHROPIC_OPENAI_BASE_URL:-https://api.anthropic.com/v1} + ANTHROPIC_MODEL: ${ANTHROPIC_MODEL:-claude-sonnet-4-6} + LLM_API_STYLE: ${LLM_API_STYLE:-openai} + ports: + - "127.0.0.1:8090:8090" + healthcheck: + test: ["CMD", "wget", "-qO-", "http://127.0.0.1:8090/healthz"] + interval: 10s + timeout: 5s + retries: 3 + start_period: 5s + + work-performer: + build: ./services/work-performer + profiles: ["mocks"] + restart: unless-stopped + environment: + PORT: "8091" + WORK_PERFORMER_TOKEN: ${WORK_PERFORMER_TOKEN} + PUBLIC_BASE_URL: http://work-performer:8091 + # host Claude Code config/secrets, read-only (§9.2) + volumes: + - ${HOME}/.claude:/root/.claude + - ${HOME}/.claude.json:/root/.claude.json + - work-data:/work + ports: + - "127.0.0.1:8091:8091" + extra_hosts: + - "host.docker.internal:host-gateway" + healthcheck: + test: ["CMD-SHELL", "node -e \"fetch('http://127.0.0.1:8091/healthz').then(r=>process.exit(r.ok?0:1)).catch(()=>process.exit(1))\""] + interval: 10s + timeout: 5s + retries: 3 + start_period: 5s + volumes: mongo-data: + work-data: diff --git a/internal/contract/contract_test.go b/internal/contract/contract_test.go new file mode 100644 index 0000000..02c436e --- /dev/null +++ b/internal/contract/contract_test.go @@ -0,0 +1,258 @@ +//go:build contract + +// Contract tests for the two mock services (§13). They run against live +// containers: +// +// docker compose --profile mocks -f docker-compose.yml -f docker-compose.test.yml up -d --build +// ATOMIZER_TOKEN=… WORK_PERFORMER_TOKEN=… go test -tags=contract ./internal/contract/ +// +// URLs default to the loopback ports published by the mocks profile. +package contract + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "math" + "net" + "net/http" + "os" + "strings" + "testing" + "time" + + "bountyboard/internal/workperform" +) + +func env(k, def string) string { + if v := os.Getenv(k); v != "" { + return v + } + return def +} + +var ( + atomizerURL = env("CONTRACT_ATOMIZER_URL", "http://127.0.0.1:8090") + performerURL = env("CONTRACT_PERFORMER_URL", "http://127.0.0.1:8091") + atomizerToken = os.Getenv("ATOMIZER_TOKEN") + performerToken = os.Getenv("WORK_PERFORMER_TOKEN") +) + +func postJSON(t *testing.T, url, token string, body any) (*http.Response, []byte) { + t.Helper() + b, _ := json.Marshal(body) + req, _ := http.NewRequest(http.MethodPost, url, bytes.NewReader(b)) + req.Header.Set("Content-Type", "application/json") + if token != "" { + req.Header.Set("Authorization", "Bearer "+token) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("POST %s: %v", url, err) + } + data, _ := io.ReadAll(resp.Body) + resp.Body.Close() + return resp, data +} + +func TestAtomizerContract(t *testing.T) { + // healthz + resp, err := http.Get(atomizerURL + "/healthz") + if err != nil { + t.Skipf("atomizer not reachable at %s (start the mocks profile): %v", atomizerURL, err) + } + resp.Body.Close() + if resp.StatusCode != 200 { + t.Fatalf("healthz: %d", resp.StatusCode) + } + + // wrong bearer token rejected (when a token is configured) + if atomizerToken != "" { + r, _ := postJSON(t, atomizerURL+"/v1/atomize", "wrong-token", map[string]any{"taskId": "x", "title": "x"}) + if r.StatusCode != http.StatusUnauthorized { + t.Fatalf("wrong token: %d, want 401", r.StatusCode) + } + } + + // atomize honors the §5.1 response contract + r, data := postJSON(t, atomizerURL+"/v1/atomize", atomizerToken, map[string]any{ + "taskId": "01JCONTRACT", + "title": "Implement user import", + "description": "Import users from CSV with mapping and validation.", + "acceptanceCriteria": []string{"valid rows imported", "errors reported per row"}, + "constraints": map[string]int{"minTasks": 2, "maxTasks": 5}, + }) + if r.StatusCode != http.StatusOK { + t.Fatalf("atomize: %d %s", r.StatusCode, data) + } + var out struct { + Tasks []struct { + Title string `json:"title"` + Description string `json:"description"` + EffortCoefficient float64 `json:"effortCoefficient"` + } `json:"tasks"` + Model string `json:"model"` + } + if err := json.Unmarshal(data, &out); err != nil { + t.Fatalf("decode: %v: %s", err, data) + } + if len(out.Tasks) < 2 || len(out.Tasks) > 5 { + t.Fatalf("task count %d outside constraints", len(out.Tasks)) + } + sum := 0.0 + for _, task := range out.Tasks { + if task.Title == "" || task.EffortCoefficient <= 0 || task.EffortCoefficient > 1 { + t.Fatalf("bad task: %+v", task) + } + sum += task.EffortCoefficient + } + if math.Abs(sum-1) > 0.001 { + t.Fatalf("coefficient sum %v, want 1±0.001", sum) + } + + // extend returns exactly one task with coefficient in (0, 2] + r, data = postJSON(t, atomizerURL+"/v1/extend", atomizerToken, map[string]any{ + "taskId": "01JCONTRACT", "title": "Implement user import", + "description": "Import users from CSV.", + "extensionNote": "add reusable column-mapping presets", + }) + if r.StatusCode != http.StatusOK { + t.Fatalf("extend: %d %s", r.StatusCode, data) + } + var ext struct { + Task struct { + Title string `json:"title"` + EffortCoefficient float64 `json:"effortCoefficient"` + } `json:"task"` + } + if err := json.Unmarshal(data, &ext); err != nil { + t.Fatalf("decode extend: %v", err) + } + if ext.Task.Title == "" || ext.Task.EffortCoefficient <= 0 || ext.Task.EffortCoefficient > 2 { + t.Fatalf("bad extension: %+v", ext.Task) + } + + // extend without a note is rejected + r, _ = postJSON(t, atomizerURL+"/v1/extend", atomizerToken, map[string]any{ + "taskId": "x", "title": "x", + }) + if r.StatusCode != http.StatusBadRequest { + t.Fatalf("extend without note: %d, want 400", r.StatusCode) + } +} + +func TestWorkPerformerContract(t *testing.T) { + resp, err := http.Get(performerURL + "/healthz") + if err != nil { + t.Skipf("work performer not reachable at %s (start the mocks profile): %v", performerURL, err) + } + resp.Body.Close() + if resp.StatusCode != 200 { + t.Fatalf("healthz: %d", resp.StatusCode) + } + + // callback receiver on the host, reachable from the container via + // host.docker.internal (extra_hosts: host-gateway) + callbackCh := make(chan struct { + body []byte + sig string + }, 1) + ln, err := net.Listen("tcp", "0.0.0.0:0") + if err != nil { + t.Fatal(err) + } + defer ln.Close() + go http.Serve(ln, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + select { + case callbackCh <- struct { + body []byte + sig string + }{body, r.Header.Get("X-Signature")}: + default: + } + w.WriteHeader(200) + })) + port := ln.Addr().(*net.TCPAddr).Port + callbackURL := fmt.Sprintf("http://host.docker.internal:%d/cb", port) + + r, data := postJSON(t, performerURL+"/v1/jobs", performerToken, map[string]any{ + "taskId": "01JCONTRACTWP", "title": "Write hello world", + "description": "Create hello.txt containing hello world.", + "acceptanceCriteria": []string{"hello.txt exists"}, + "callbackUrl": callbackURL, + }) + if r.StatusCode != http.StatusAccepted { + t.Fatalf("submit: %d %s", r.StatusCode, data) + } + var accepted struct { + JobID string `json:"jobId"` + Status string `json:"status"` + } + if err := json.Unmarshal(data, &accepted); err != nil || accepted.JobID == "" { + t.Fatalf("bad 202 body: %s", data) + } + + // status endpoint + req, _ := http.NewRequest(http.MethodGet, performerURL+"/v1/jobs/"+accepted.JobID, nil) + req.Header.Set("Authorization", "Bearer "+performerToken) + sResp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + sBody, _ := io.ReadAll(sResp.Body) + sResp.Body.Close() + var status struct { + Status string `json:"status"` + } + if err := json.Unmarshal(sBody, &status); err != nil { + t.Fatalf("status decode: %s", sBody) + } + if !strings.Contains("queued running succeeded failed", status.Status) { + t.Fatalf("status %q", status.Status) + } + + // HMAC-signed callback arrives (the simulated path is fast; the real + // claude path may take minutes — allow a generous window) + select { + case cb := <-callbackCh: + if !workperform.VerifySignature(performerToken, cb.body, cb.sig) { + t.Fatalf("callback signature invalid") + } + var payload workperform.Callback + if err := json.Unmarshal(cb.body, &payload); err != nil { + t.Fatalf("callback decode: %v", err) + } + if payload.JobID != accepted.JobID || payload.TaskID != "01JCONTRACTWP" { + t.Fatalf("callback ids: %+v", payload) + } + if payload.Status != "succeeded" && payload.Status != "failed" { + t.Fatalf("callback status %q", payload.Status) + } + // artifacts must be fetchable (rewrite in-network host for the host-side test) + for _, a := range payload.Artifacts { + url := strings.Replace(a.URL, "http://work-performer:8091", performerURL, 1) + aResp, err := http.Get(url) + if err != nil || aResp.StatusCode != 200 { + t.Fatalf("artifact %s not fetchable: %v %v", a.URL, err, aResp) + } + aResp.Body.Close() + } + case <-time.After(5 * time.Minute): + t.Fatal("no callback within 5 minutes") + } + + // cancel is accepted for unknown-but-wellformed ids → 404, and DELETE on + // a finished job still answers + req, _ = http.NewRequest(http.MethodDelete, performerURL+"/v1/jobs/"+accepted.JobID, nil) + req.Header.Set("Authorization", "Bearer "+performerToken) + dResp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + dResp.Body.Close() + if dResp.StatusCode != 200 { + t.Fatalf("cancel: %d", dResp.StatusCode) + } +} diff --git a/services/atomizer/Dockerfile b/services/atomizer/Dockerfile new file mode 100644 index 0000000..a1922dd --- /dev/null +++ b/services/atomizer/Dockerfile @@ -0,0 +1,14 @@ +# --- build --- +FROM golang:1.26-alpine AS build +WORKDIR /src +COPY go.mod ./ +COPY *.go ./ +RUN CGO_ENABLED=0 go build -trimpath -ldflags="-s -w" -o /out/atomizer . + +# --- runtime --- +FROM alpine:3.21 +RUN apk add --no-cache ca-certificates && adduser -S -G nogroup app +USER app +COPY --from=build /out/atomizer /usr/local/bin/atomizer +EXPOSE 8090 +ENTRYPOINT ["/usr/local/bin/atomizer"] diff --git a/services/atomizer/atomizer-mock b/services/atomizer/atomizer-mock new file mode 100755 index 0000000..bcff00e Binary files /dev/null and b/services/atomizer/atomizer-mock differ diff --git a/services/atomizer/go.mod b/services/atomizer/go.mod new file mode 100644 index 0000000..62f0021 --- /dev/null +++ b/services/atomizer/go.mod @@ -0,0 +1,3 @@ +module atomizer-mock + +go 1.26 diff --git a/services/atomizer/llm.go b/services/atomizer/llm.go new file mode 100644 index 0000000..552cceb --- /dev/null +++ b/services/atomizer/llm.go @@ -0,0 +1,252 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log/slog" + "net/http" + "strings" + "time" +) + +// llmClient calls Anthropic via the OpenAI-compatible chat-completions +// endpoint (default) or the native Messages API (LLM_API_STYLE=anthropic). +// Plain net/http, no SDK (§9.1). +type llmClient struct { + cfg config + log *slog.Logger + http *http.Client +} + +func newLLMClient(cfg config, log *slog.Logger) *llmClient { + return &llmClient{cfg: cfg, log: log, http: &http.Client{Timeout: 110 * time.Second}} +} + +const atomizeSystem = `You are an expert software project planner. You split one ticket into small, well-scoped, independently deliverable developer tasks. +Respond with ONLY a JSON object, no prose and no markdown fences, exactly matching: +{"tasks":[{"title":"...","description":"...","acceptanceCriteria":["..."],"effortCoefficient":0.25}],"notes":"short advice for the consultant"} +Rules: between MIN and MAX tasks; every effortCoefficient is a number in (0,1]; all effortCoefficients MUST sum to exactly 1.0; titles are concise; descriptions are self-contained instructions; each task has 1-4 testable acceptance criteria.` + +const extendSystem = `You are an expert software project planner. You design exactly ONE additional sibling task that extends the given task's functionality per the extension note. +Respond with ONLY a JSON object, no prose and no markdown fences, exactly matching: +{"task":{"title":"...","description":"...","acceptanceCriteria":["..."],"effortCoefficient":0.4},"notes":"short advice"} +Rules: effortCoefficient is the effort RELATIVE to the source task, a number in (0,2].` + +func ticketPrompt(req atomizeRequest) string { + var sb strings.Builder + fmt.Fprintf(&sb, "TICKET: %s\n\nDESCRIPTION:\n%s\n", req.Title, req.Description) + if len(req.AcceptanceCriteria) > 0 { + sb.WriteString("\nACCEPTANCE CRITERIA:\n") + for _, c := range req.AcceptanceCriteria { + fmt.Fprintf(&sb, "- %s\n", c) + } + } + if len(req.Links) > 0 { + fmt.Fprintf(&sb, "\nLINKS: %s\n", strings.Join(req.Links, ", ")) + } + for _, a := range req.Attachments { + fmt.Fprintf(&sb, "ATTACHMENT: %s (%s) %s\n", a.Name, a.MimeType, a.URL) + } + return sb.String() +} + +func (l *llmClient) atomize(ctx context.Context, req atomizeRequest, minT, maxT int) ([]subTask, string, string, error) { + if l.cfg.apiKey == "" { + return fallbackAtomize(req, minT, maxT), "offline-fallback", "Deterministic split (no ANTHROPIC_API_KEY configured).", nil + } + system := strings.NewReplacer("MIN", fmt.Sprint(minT), "MAX", fmt.Sprint(maxT)).Replace(atomizeSystem) + user := ticketPrompt(req) + if req.SubdivisionNote != "" { + user += "\nCONSULTANT NOTE: " + req.SubdivisionNote + } + + var parsed struct { + Tasks []subTask `json:"tasks"` + Notes string `json:"notes"` + } + raw, err := l.completeWithRetry(ctx, system, user, func(text string) error { + if err := json.Unmarshal([]byte(stripFences(text)), &parsed); err != nil { + return err + } + if len(parsed.Tasks) == 0 { + return fmt.Errorf("no tasks in response") + } + for _, t := range parsed.Tasks { + if t.Title == "" || t.EffortCoefficient <= 0 { + return fmt.Errorf("invalid task entry") + } + } + return nil + }) + if err != nil { + l.log.Warn("llm atomize failed, using fallback", "err", err) + return fallbackAtomize(req, minT, maxT), "offline-fallback", + "LLM unavailable (" + err.Error() + "); deterministic split.", nil + } + _ = raw + return parsed.Tasks, l.cfg.model, parsed.Notes, nil +} + +func (l *llmClient) extend(ctx context.Context, req atomizeRequest) (subTask, string, string, error) { + if l.cfg.apiKey == "" { + return fallbackExtend(req), "offline-fallback", "Deterministic extension (no ANTHROPIC_API_KEY configured).", nil + } + user := ticketPrompt(req) + "\nEXTENSION NOTE (required scope): " + req.ExtensionNote + + var parsed struct { + Task subTask `json:"task"` + Notes string `json:"notes"` + } + _, err := l.completeWithRetry(ctx, extendSystem, user, func(text string) error { + if err := json.Unmarshal([]byte(stripFences(text)), &parsed); err != nil { + return err + } + if parsed.Task.Title == "" { + return fmt.Errorf("no task in response") + } + return nil + }) + if err != nil { + l.log.Warn("llm extend failed, using fallback", "err", err) + return fallbackExtend(req), "offline-fallback", + "LLM unavailable (" + err.Error() + "); deterministic extension.", nil + } + return parsed.Task, l.cfg.model, parsed.Notes, nil +} + +// completeWithRetry runs the chat completion and retries ONCE on parse +// failure with a corrective hint (§9.1). +func (l *llmClient) completeWithRetry(ctx context.Context, system, user string, + validate func(string) error) (string, error) { + text, err := l.complete(ctx, system, user) + if err != nil { + return "", err + } + if vErr := validate(text); vErr == nil { + return text, nil + } else { + l.log.Warn("llm response failed validation, retrying once", "err", vErr) + } + text, err = l.complete(ctx, system, + user+"\n\nIMPORTANT: your previous reply was not valid JSON matching the schema. Reply with ONLY the JSON object.") + if err != nil { + return "", err + } + if vErr := validate(text); vErr != nil { + return "", fmt.Errorf("invalid JSON after retry: %w", vErr) + } + return text, nil +} + +func (l *llmClient) complete(ctx context.Context, system, user string) (string, error) { + if l.cfg.apiStyle == "anthropic" { + return l.completeNative(ctx, system, user) + } + return l.completeOpenAI(ctx, system, user) +} + +// completeOpenAI uses Anthropic's OpenAI-compatible endpoint. +func (l *llmClient) completeOpenAI(ctx context.Context, system, user string) (string, error) { + body, _ := json.Marshal(map[string]any{ + "model": l.cfg.model, + "max_tokens": 4096, + "messages": []map[string]string{ + {"role": "system", "content": system}, + {"role": "user", "content": user}, + }, + }) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, + l.cfg.baseURL+"/chat/completions", bytes.NewReader(body)) + if err != nil { + return "", err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+l.cfg.apiKey) + data, err := l.do(req) + if err != nil { + return "", err + } + var out struct { + Choices []struct { + Message struct { + Content string `json:"content"` + } `json:"message"` + } `json:"choices"` + } + if err := json.Unmarshal(data, &out); err != nil { + return "", fmt.Errorf("decode chat completion: %w", err) + } + if len(out.Choices) == 0 { + return "", fmt.Errorf("no choices in response") + } + return out.Choices[0].Message.Content, nil +} + +// completeNative uses the native Anthropic Messages API (flip via +// LLM_API_STYLE=anthropic if the compatibility endpoint misbehaves, §9.1). +func (l *llmClient) completeNative(ctx context.Context, system, user string) (string, error) { + body, _ := json.Marshal(map[string]any{ + "model": l.cfg.model, + "max_tokens": 4096, + "system": system, + "messages": []map[string]string{{"role": "user", "content": user}}, + }) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, + l.cfg.baseURL+"/messages", bytes.NewReader(body)) + if err != nil { + return "", err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("x-api-key", l.cfg.apiKey) + req.Header.Set("anthropic-version", "2023-06-01") + data, err := l.do(req) + if err != nil { + return "", err + } + var out struct { + Content []struct { + Type string `json:"type"` + Text string `json:"text"` + } `json:"content"` + } + if err := json.Unmarshal(data, &out); err != nil { + return "", fmt.Errorf("decode messages response: %w", err) + } + for _, c := range out.Content { + if c.Type == "text" { + return c.Text, nil + } + } + return "", fmt.Errorf("no text content in response") +} + +func (l *llmClient) do(req *http.Request) ([]byte, error) { + resp, err := l.http.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + data, err := io.ReadAll(io.LimitReader(resp.Body, 10<<20)) + if err != nil { + return nil, err + } + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("llm api status %d: %.300s", resp.StatusCode, data) + } + return data, nil +} + +// stripFences defensively removes markdown code fences and surrounding prose +// (§9.1) by slicing from the first '{' to the last '}'. +func stripFences(s string) string { + s = strings.TrimSpace(s) + if i := strings.IndexByte(s, '{'); i >= 0 { + if j := strings.LastIndexByte(s, '}'); j > i { + return s[i : j+1] + } + } + return s +} diff --git a/services/atomizer/main.go b/services/atomizer/main.go new file mode 100644 index 0000000..65a531f --- /dev/null +++ b/services/atomizer/main.go @@ -0,0 +1,243 @@ +// atomizer-mock is the standalone placeholder Atomization Service (§5.1, +// §9.1). It calls Anthropic through the OpenAI-compatible chat-completions +// endpoint by default, or the native Messages API when +// LLM_API_STYLE=anthropic, and falls back to a deterministic equal split +// when no ANTHROPIC_API_KEY is configured so the whole system works offline. +package main + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "net/http" + "os" + "os/signal" + "strings" + "syscall" + "time" +) + +type config struct { + port string + token string + apiKey string + baseURL string + model string + apiStyle string // openai | anthropic +} + +func loadConfig() config { + get := func(k, def string) string { + if v := os.Getenv(k); v != "" { + return v + } + return def + } + return config{ + port: get("PORT", "8090"), + token: os.Getenv("ATOMIZER_TOKEN"), + apiKey: os.Getenv("ANTHROPIC_API_KEY"), + baseURL: strings.TrimRight(get("ANTHROPIC_OPENAI_BASE_URL", "https://api.anthropic.com/v1"), "/"), + model: get("ANTHROPIC_MODEL", "claude-sonnet-4-6"), + apiStyle: get("LLM_API_STYLE", "openai"), + } +} + +type subTask struct { + Title string `json:"title"` + Description string `json:"description"` + AcceptanceCriteria []string `json:"acceptanceCriteria"` + EffortCoefficient float64 `json:"effortCoefficient"` +} + +type atomizeRequest struct { + TaskID string `json:"taskId"` + Title string `json:"title"` + Description string `json:"description"` + AcceptanceCriteria []string `json:"acceptanceCriteria"` + Attachments []struct { + Name string `json:"name"` + URL string `json:"url"` + MimeType string `json:"mimeType"` + } `json:"attachments"` + Links []string `json:"links"` + SubdivisionNote string `json:"subdivisionNote"` + ExtensionNote string `json:"extensionNote"` + Constraints *struct { + MinTasks int `json:"minTasks"` + MaxTasks int `json:"maxTasks"` + } `json:"constraints"` +} + +type server struct { + cfg config + log *slog.Logger + llm *llmClient +} + +func main() { + cfg := loadConfig() + log := slog.New(slog.NewJSONHandler(os.Stdout, nil)) + s := &server{cfg: cfg, log: log, llm: newLLMClient(cfg, log)} + + mux := http.NewServeMux() + mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) + }) + mux.HandleFunc("POST /v1/atomize", s.auth(s.handleAtomize)) + mux.HandleFunc("POST /v1/extend", s.auth(s.handleExtend)) + + srv := &http.Server{Addr: ":" + cfg.port, Handler: mux, ReadHeaderTimeout: 10 * time.Second} + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + go func() { + log.Info("atomizer-mock listening", "port", cfg.port, + "llm", map[bool]string{true: "anthropic:" + cfg.apiStyle, false: "fallback (no api key)"}[cfg.apiKey != ""]) + if err := srv.ListenAndServe(); err != http.ErrServerClosed { + log.Error("serve", "err", err) + os.Exit(1) + } + }() + <-ctx.Done() + shutCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + srv.Shutdown(shutCtx) +} + +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + json.NewEncoder(w).Encode(v) +} + +func writeErr(w http.ResponseWriter, status int, code, msg string) { + writeJSON(w, status, map[string]any{"error": map[string]string{"code": code, "message": msg}}) +} + +func (s *server) auth(next http.HandlerFunc) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if s.cfg.token != "" && r.Header.Get("Authorization") != "Bearer "+s.cfg.token { + writeErr(w, http.StatusUnauthorized, "unauthorized", "missing or invalid bearer token") + return + } + next(w, r) + } +} + +func (s *server) handleAtomize(w http.ResponseWriter, r *http.Request) { + var req atomizeRequest + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<20)).Decode(&req); err != nil { + writeErr(w, http.StatusBadRequest, "bad_request", err.Error()) + return + } + minT, maxT := 2, 8 + if req.Constraints != nil { + if req.Constraints.MinTasks > 0 { + minT = req.Constraints.MinTasks + } + if req.Constraints.MaxTasks > 0 { + maxT = req.Constraints.MaxTasks + } + } + if minT > maxT { + minT = maxT + } + + tasks, model, notes, err := s.llm.atomize(r.Context(), req, minT, maxT) + if err != nil { + s.log.Error("atomize failed", "taskId", req.TaskID, "err", err) + writeErr(w, http.StatusBadGateway, "llm_failed", err.Error()) + return + } + normalizeSum(tasks) + writeJSON(w, http.StatusOK, map[string]any{"tasks": tasks, "model": model, "notes": notes}) +} + +func (s *server) handleExtend(w http.ResponseWriter, r *http.Request) { + var req atomizeRequest + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<20)).Decode(&req); err != nil { + writeErr(w, http.StatusBadRequest, "bad_request", err.Error()) + return + } + if strings.TrimSpace(req.ExtensionNote) == "" { + writeErr(w, http.StatusBadRequest, "note_required", "extensionNote is required") + return + } + task, model, notes, err := s.llm.extend(r.Context(), req) + if err != nil { + s.log.Error("extend failed", "taskId", req.TaskID, "err", err) + writeErr(w, http.StatusBadGateway, "llm_failed", err.Error()) + return + } + if task.EffortCoefficient <= 0 || task.EffortCoefficient > 2 { + task.EffortCoefficient = 0.3 + } + writeJSON(w, http.StatusOK, map[string]any{"task": task, "model": model, "notes": notes}) +} + +// normalizeSum forces coefficients to sum to exactly 1.0. +func normalizeSum(tasks []subTask) { + sum := 0.0 + for _, t := range tasks { + sum += t.EffortCoefficient + } + if sum <= 0 { + eq := 1.0 / float64(len(tasks)) + for i := range tasks { + tasks[i].EffortCoefficient = eq + } + sum = 1.0 + } + total := 0.0 + largest := 0 + for i := range tasks { + tasks[i].EffortCoefficient = round4(tasks[i].EffortCoefficient / sum) + total += tasks[i].EffortCoefficient + if tasks[i].EffortCoefficient > tasks[largest].EffortCoefficient { + largest = i + } + } + tasks[largest].EffortCoefficient = round4(tasks[largest].EffortCoefficient + 1 - total) +} + +func round4(v float64) float64 { + return float64(int64(v*10000+0.5)) / 10000 +} + +func fallbackAtomize(req atomizeRequest, minT, maxT int) []subTask { + n := (minT + maxT) / 2 + if n < 1 { + n = 3 + } + out := make([]subTask, n) + eq := round4(1.0 / float64(n)) + for i := range out { + out[i] = subTask{ + Title: fmt.Sprintf("%s — part %d of %d", req.Title, i+1, n), + Description: fmt.Sprintf("Deterministic offline split %d/%d of:\n\n%s", i+1, n, req.Description), + AcceptanceCriteria: append([]string{}, req.AcceptanceCriteria...), + EffortCoefficient: eq, + } + } + return out +} + +func fallbackExtend(req atomizeRequest) subTask { + return subTask{ + Title: req.Title + " — extension: " + firstLine(req.ExtensionNote, 60), + Description: "Deterministic offline extension of the source task.\n\nRequested scope: " + req.ExtensionNote, + AcceptanceCriteria: []string{"extension scope implemented: " + firstLine(req.ExtensionNote, 120)}, + EffortCoefficient: 0.3, + } +} + +func firstLine(s string, max int) string { + if i := strings.IndexByte(s, '\n'); i >= 0 { + s = s[:i] + } + if len(s) > max { + s = s[:max] + } + return s +} diff --git a/services/work-performer/Dockerfile b/services/work-performer/Dockerfile new file mode 100644 index 0000000..6ac7967 --- /dev/null +++ b/services/work-performer/Dockerfile @@ -0,0 +1,9 @@ +FROM node:20-bookworm +RUN apt-get update && apt-get install -y --no-install-recommends git ca-certificates \ + && rm -rf /var/lib/apt/lists/* \ + && npm install -g @anthropic-ai/claude-code +WORKDIR /srv +COPY server.js . +RUN mkdir -p /work +EXPOSE 8091 +CMD ["node", "server.js"] diff --git a/services/work-performer/server.js b/services/work-performer/server.js new file mode 100644 index 0000000..92ffd04 --- /dev/null +++ b/services/work-performer/server.js @@ -0,0 +1,294 @@ +#!/usr/bin/env node +// work-performer: standalone placeholder Work Performer Service (§5.2, §9.2). +// Plain Node http server, no framework. For each job it prepares +// /work/{jobId}/TASK.md and runs the Claude Code CLI; if the CLI is missing +// or fails to start (e.g. no credentials mounted), it produces a simulated +// result so the whole flow stays demonstrable offline. The HTTP contract — +// not this implementation — is the deliverable. +'use strict'; + +const http = require('http'); +const https = require('https'); +const crypto = require('crypto'); +const fs = require('fs'); +const path = require('path'); +const { execFile, execFileSync } = require('child_process'); + +const PORT = parseInt(process.env.PORT || '8091', 10); +const TOKEN = process.env.WORK_PERFORMER_TOKEN || ''; +const WORK_DIR = process.env.WORK_DIR || '/work'; +const PUBLIC_BASE = process.env.PUBLIC_BASE_URL || `http://work-performer:${PORT}`; + +const jobs = new Map(); // jobId -> {status, taskId, request, startedAt, finishedAt, cancel} +const queue = []; +let running = false; // single-job concurrency (§9.2) + +const log = (msg, extra) => + console.log(JSON.stringify({ ts: new Date().toISOString(), msg, ...extra })); + +function json(res, status, body) { + const data = JSON.stringify(body); + res.writeHead(status, { 'Content-Type': 'application/json' }); + res.end(data); +} +const errJson = (res, status, code, message) => + json(res, status, { error: { code, message } }); + +function readBody(req) { + return new Promise((resolve, reject) => { + const chunks = []; + let size = 0; + req.on('data', (c) => { + size += c.length; + if (size > 4 << 20) { reject(new Error('body too large')); req.destroy(); return; } + chunks.push(c); + }); + req.on('end', () => resolve(Buffer.concat(chunks))); + req.on('error', reject); + }); +} + +function download(url, dest) { + return new Promise((resolve, reject) => { + const mod = url.startsWith('https:') ? https : http; + const file = fs.createWriteStream(dest); + mod.get(url, (res) => { + if (res.statusCode !== 200) { + file.close(); fs.rmSync(dest, { force: true }); + reject(new Error(`download ${url}: status ${res.statusCode}`)); + return; + } + res.pipe(file); + file.on('finish', () => file.close(resolve)); + }).on('error', (e) => { file.close(); fs.rmSync(dest, { force: true }); reject(e); }); + }); +} + +function postCallback(urlStr, payload) { + return new Promise((resolve, reject) => { + const body = Buffer.from(JSON.stringify(payload)); + const sig = crypto.createHmac('sha256', TOKEN).update(body).digest('hex'); + const url = new URL(urlStr); + const mod = url.protocol === 'https:' ? https : http; + const req = mod.request(url, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + 'Content-Length': body.length, + 'X-Signature': sig, + }, + }, (res) => { res.resume(); resolve(res.statusCode); }); + req.on('error', reject); + req.end(body); + }); +} + +function claudeAvailable() { + try { + execFileSync('claude', ['--version'], { timeout: 15000, stdio: 'pipe' }); + return true; + } catch (e) { + return false; + } +} + +function buildTaskMD(r) { + const lines = [`# ${r.title}`, '', r.description || '', '']; + if ((r.acceptanceCriteria || []).length) { + lines.push('## Acceptance criteria', ''); + r.acceptanceCriteria.forEach((c) => lines.push(`- ${c}`)); + lines.push(''); + } + if ((r.links || []).length) { + lines.push('## Links', ''); + r.links.forEach((l) => lines.push(`- ${l}`)); + lines.push(''); + } + if ((r.attachments || []).length) { + lines.push('## Attachments (downloaded into ./attachments)', ''); + r.attachments.forEach((a) => lines.push(`- ${a.name}`)); + lines.push(''); + } + if (r.context && r.context.instructions) { + lines.push('## Extra instructions', '', r.context.instructions, ''); + } + lines.push('Produce your changes in this directory. Write a SUMMARY.md describing what you did.'); + return lines.join('\n'); +} + +function listProducedFiles(dir, before) { + const out = []; + const walk = (d) => { + for (const entry of fs.readdirSync(d, { withFileTypes: true })) { + if (entry.name === '.git' || entry.name === 'attachments') continue; + const full = path.join(d, entry.name); + if (entry.isDirectory()) { walk(full); continue; } + const rel = path.relative(dir, full); + if (!before.has(rel) && fs.statSync(full).size <= 20 << 20) out.push(rel); + } + }; + walk(dir); + return out.slice(0, 20); +} + +async function runJob(jobId) { + const job = jobs.get(jobId); + if (!job || job.status !== 'queued') return; + job.status = 'running'; + job.startedAt = new Date().toISOString(); + const r = job.request; + const dir = path.join(WORK_DIR, jobId); + let result; + try { + fs.mkdirSync(path.join(dir, 'attachments'), { recursive: true }); + + // optional repo clone (§9.2) + if (r.context && r.context.repositoryUrl) { + const args = ['clone', '--depth', '1']; + if (r.context.branch) args.push('-b', r.context.branch); + args.push(r.context.repositoryUrl, path.join(dir, 'repo')); + await new Promise((resolve) => { + execFile('git', args, { timeout: 120000 }, (err, _o, stderr) => { + if (err) log('git clone failed', { jobId, err: String(stderr || err) }); + resolve(); + }); + }); + } + for (const a of r.attachments || []) { + try { + await download(a.url, path.join(dir, 'attachments', path.basename(a.name))); + } catch (e) { log('attachment download failed', { jobId, name: a.name, err: e.message }); } + } + fs.writeFileSync(path.join(dir, 'TASK.md'), buildTaskMD(r)); + const before = new Set(['TASK.md']); + + if (claudeAvailable()) { + result = await new Promise((resolve) => { + const child = execFile('claude', + ['-p', fs.readFileSync(path.join(dir, 'TASK.md'), 'utf8'), + '--output-format', 'json', '--dangerously-skip-permissions'], + { cwd: dir, timeout: 30 * 60 * 1000, maxBuffer: 32 << 20 }, + (err, stdout, stderr) => { + if (job.cancel) { resolve({ status: 'failed', summary: 'job canceled', log: '' }); return; } + if (err) { + resolve({ status: 'failed', summary: `claude execution failed: ${err.message}`, + log: String(stderr || '').slice(-4000) }); + return; + } + let summary = 'Claude Code completed the task.'; + try { + const parsed = JSON.parse(stdout); + summary = parsed.result || parsed.summary || summary; + } catch (e) { summary = String(stdout).slice(0, 1000) || summary; } + resolve({ status: 'succeeded', summary: String(summary).slice(0, 4000), + log: String(stdout).slice(-4000) }); + }); + job.kill = () => child.kill('SIGTERM'); + }); + } else { + // offline placeholder result keeps the end-to-end flow demonstrable + const summary = `Simulated work performer result (Claude Code CLI not available in this container).\n` + + `Reviewed task "${r.title}" with ${(r.acceptanceCriteria || []).length} acceptance criteria.`; + fs.writeFileSync(path.join(dir, 'SUMMARY.md'), + `# Simulated result\n\n${summary}\n\nThis placeholder proves the §5.2 contract end to end.`); + result = { status: 'succeeded', summary, log: 'claude CLI unavailable; produced simulated SUMMARY.md' }; + } + + const artifacts = listProducedFiles(dir, before).map((rel) => ({ + name: rel.replace(/\//g, '_'), + url: `${PUBLIC_BASE}/artifacts/${jobId}/${encodeURIComponent(rel)}`, + })); + result.artifacts = artifacts; + } catch (e) { + result = { status: 'failed', summary: `job crashed: ${e.message}`, log: '', artifacts: [] }; + } + + job.status = result.status; + job.finishedAt = new Date().toISOString(); + const payload = { + jobId, taskId: r.taskId, status: result.status, + summary: result.summary, artifacts: result.artifacts || [], log: result.log || '', + }; + for (let attempt = 1; attempt <= 3; attempt++) { + try { + const code = await postCallback(r.callbackUrl, payload); + log('callback delivered', { jobId, code }); + break; + } catch (e) { + log('callback failed', { jobId, attempt, err: e.message }); + await new Promise((s) => setTimeout(s, attempt * 2000)); + } + } +} + +async function pump() { + if (running) return; + const next = queue.shift(); + if (!next) return; + running = true; + try { await runJob(next); } finally { + running = false; + setImmediate(pump); + } +} + +const server = http.createServer(async (req, res) => { + const url = new URL(req.url, `http://${req.headers.host}`); + + if (req.method === 'GET' && url.pathname === '/healthz') { + return json(res, 200, { status: 'ok' }); + } + + // artifact downloads are unauthenticated within the compose network + const artMatch = url.pathname.match(/^\/artifacts\/([\w]+)\/(.+)$/); + if (req.method === 'GET' && artMatch) { + const file = path.join(WORK_DIR, artMatch[1], decodeURIComponent(artMatch[2])); + if (!file.startsWith(path.join(WORK_DIR, artMatch[1]) + path.sep) || !fs.existsSync(file)) { + return errJson(res, 404, 'not_found', 'artifact not found'); + } + res.writeHead(200, { 'Content-Type': 'application/octet-stream' }); + return fs.createReadStream(file).pipe(res); + } + + if (TOKEN && req.headers.authorization !== `Bearer ${TOKEN}`) { + return errJson(res, 401, 'unauthorized', 'missing or invalid bearer token'); + } + + if (req.method === 'POST' && url.pathname === '/v1/jobs') { + let body; + try { body = JSON.parse(await readBody(req)); } + catch (e) { return errJson(res, 400, 'bad_request', e.message); } + if (!body.taskId || !body.title || !body.callbackUrl) { + return errJson(res, 400, 'bad_request', 'taskId, title and callbackUrl are required'); + } + const jobId = 'wp_' + crypto.randomBytes(8).toString('hex'); + jobs.set(jobId, { status: 'queued', taskId: body.taskId, request: body, + startedAt: null, finishedAt: null, cancel: false }); + queue.push(jobId); + setImmediate(pump); + log('job queued', { jobId, taskId: body.taskId }); + return json(res, 202, { jobId, status: 'queued' }); + } + + const jobMatch = url.pathname.match(/^\/v1\/jobs\/(wp_[\w]+)$/); + if (jobMatch) { + const job = jobs.get(jobMatch[1]); + if (!job) return errJson(res, 404, 'not_found', 'job not found'); + if (req.method === 'GET') { + return json(res, 200, { jobId: jobMatch[1], status: job.status, + startedAt: job.startedAt, finishedAt: job.finishedAt }); + } + if (req.method === 'DELETE') { // best-effort cancel (§5.2) + job.cancel = true; + const idx = queue.indexOf(jobMatch[1]); + if (idx >= 0) { queue.splice(idx, 1); job.status = 'failed'; job.finishedAt = new Date().toISOString(); } + if (job.kill) try { job.kill(); } catch (e) { /* already gone */ } + return json(res, 200, { jobId: jobMatch[1], status: 'cancel_requested' }); + } + } + + errJson(res, 404, 'not_found', 'unknown endpoint'); +}); + +server.listen(PORT, () => log('work-performer listening', { port: PORT, claude: claudeAvailable() })); +process.on('SIGTERM', () => server.close(() => process.exit(0)));