feat: mock atomizer and work-performer services with contract tests (phase 11)
- services/atomizer: standalone Go module implementing §5.1 — Anthropic
OpenAI-compatible chat completions by default, native /v1/messages when
LLM_API_STYLE=anthropic, strict-JSON prompting, defensive fence-stripping
parse with one retry, deterministic equal-split fallback without an API
key, exact coefficient-sum normalization, bearer auth
- services/work-performer: Node 20 http server implementing §5.2 — single
concurrency, /work/{jobId}/TASK.md preparation, attachment downloads,
optional shallow git clone, claude CLI execution with JSON output,
simulated success when the CLI is unavailable (offline demo), artifact
endpoint, idempotent HMAC-signed callbacks with retry, best-effort cancel
- compose profile 'mocks': separate builds/ports/tokens, healthchecks,
${HOME}/.claude(.json) mounted read-only into the performer
- contract tests (go test -tags=contract) for both services; Makefile
test-contract target; verified live incl. a real Claude Code job run
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
etalon
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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"]
|
||||
Executable
BIN
Binary file not shown.
@@ -0,0 +1,3 @@
|
||||
module atomizer-mock
|
||||
|
||||
go 1.26
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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"]
|
||||
@@ -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)));
|
||||
Reference in New Issue
Block a user