From bf80efa25d0659ae1adb682382512f18c666f2c8 Mon Sep 17 00:00:00 2001 From: Sam Rolfe Date: Sat, 10 Oct 2026 17:38:24 +1100 Subject: [PATCH] AUD-49: build the Hub skeleton (hear-route-answer) - Three inbound routes from docs/hub-api.md: POST /ask, POST /approve, GET /health. - Five-step pipeline in src/hub, with packages hear, understand, check, route, answer. - Routing through OmniRoute with the capability list and routing.md. The never-route control check runs before any LLM call. - One real destination: Jervis house TTS, started out of band. - Fake /ask project (src/tests/fakeproject) for the round-trip test. - Append-only, size-capped request log. Bearer auth, source match, rate limit. - Tests: all packages pass, including the fake /ask round trip. Race clean. - Dockerfile and compose updated to mount routing.md and secrets, log on tmpfs. Co-Authored-By: Paperclip --- Dockerfile | 7 +- README.md | 37 ++- docker-compose.yml | 23 +- docs/outline/02-config-and-env.md | 32 ++ docs/outline/03-code-map.md | 72 +++-- docs/outline/04-agent-runbook.md | 78 +++++ scripts/hub-test.sh | 25 ++ src/answer/answer.go | 171 +++++++++++ src/answer/answer_test.go | 80 +++++ src/apierr/apierr.go | 41 +++ src/check/check.go | 42 +++ src/config/config.go | 285 ++++++++++++++++++ src/destinations/paperclip.go | 85 ++++++ src/destinations/project.go | 122 ++++++++ src/destinations/speakers.go | 65 ++++ src/hear/hear.go | 109 +++++++ src/hear/hear_test.go | 65 ++++ src/hub/approval.go | 80 +++++ src/hub/health.go | 48 +++ src/hub/hub.go | 282 ++++++++++++++++++ src/hub/hub_test.go | 228 ++++++++++++++ src/ids/ids.go | 87 ++++++ src/logbook/logbook.go | 151 ++++++++++ src/main.go | 429 ++++++++++++++++++++++----- src/main_test.go | 258 ++++++++++++++++ src/route/route.go | 317 ++++++++++++++++++++ src/route/route_test.go | 168 +++++++++++ src/tests/fakeproject/cmd/main.go | 28 ++ src/tests/fakeproject/fakeproject.go | 125 ++++++++ src/understand/control.go | 90 ++++++ src/understand/router.go | 213 +++++++++++++ src/understand/rules.go | 203 +++++++++++++ src/understand/understand_test.go | 89 ++++++ 33 files changed, 4025 insertions(+), 110 deletions(-) create mode 100755 scripts/hub-test.sh create mode 100644 src/answer/answer.go create mode 100644 src/answer/answer_test.go create mode 100644 src/apierr/apierr.go create mode 100644 src/check/check.go create mode 100644 src/config/config.go create mode 100644 src/destinations/paperclip.go create mode 100644 src/destinations/project.go create mode 100644 src/destinations/speakers.go create mode 100644 src/hear/hear.go create mode 100644 src/hear/hear_test.go create mode 100644 src/hub/approval.go create mode 100644 src/hub/health.go create mode 100644 src/hub/hub.go create mode 100644 src/hub/hub_test.go create mode 100644 src/ids/ids.go create mode 100644 src/logbook/logbook.go create mode 100644 src/main_test.go create mode 100644 src/route/route.go create mode 100644 src/route/route_test.go create mode 100644 src/tests/fakeproject/cmd/main.go create mode 100644 src/tests/fakeproject/fakeproject.go create mode 100644 src/understand/control.go create mode 100644 src/understand/router.go create mode 100644 src/understand/rules.go create mode 100644 src/understand/understand_test.go diff --git a/Dockerfile b/Dockerfile index c22c84e..db1bf9e 100644 --- a/Dockerfile +++ b/Dockerfile @@ -6,8 +6,11 @@ COPY src/ ./ RUN CGO_ENABLED=0 GOOS=linux go build -trimpath -ldflags='-s -w' -o /out/hub . FROM gcr.io/distroless/static-debian12:nonroot -COPY --from=build /out/hub /hub +WORKDIR /app +COPY --from=build /out/hub /app/hub ENV HUB_PORT=8788 +ENV HUB_ROUTING_FILE=/app/routing.md +ENV HUB_LOG_PATH=/tmp/hub.log EXPOSE 8788 USER nonroot:nonroot -ENTRYPOINT ["/hub"] +ENTRYPOINT ["/app/hub"] diff --git a/README.md b/README.md index 09a8c71..af9d435 100644 --- a/README.md +++ b/README.md @@ -7,6 +7,8 @@ shared by the phone app, the house voice, chat, and the admin app. - **Core Technologies**: Go 1.25 service · HTTP JSON API · MQTT · LLM intent routing · OmniRoute gateway · Docker +- **Entry routes**: `POST /ask` · `POST /approve` · `GET /health` + (frozen in `docs/hub-api.md`) - **Key Tools**: Jervis (voice) · Home Assistant · Paperclip · Artemis · internet use control · Apprise / NTFY · OmniRoute - **Engineering Capabilities**: service design · intent routing · authorization @@ -15,16 +17,41 @@ shared by the phone app, the house voice, chat, and the admin app. ## Quick Start +### Run the Hub + +```bash +# Build and run the container (Docker host 192.168.20.13). +docker compose up -d --build + +# Liveness (no token). +curl -s http://192.168.20.13:8788/health + +# Ask a question (source esp32 speaks the reply on the house speakers). +curl -s -X POST http://192.168.20.13:8788/ask \ + -H "Authorization: Bearer $HUB_TOKEN_ESP32" \ + -H "Content-Type: application/json" \ + -d '{"text":"what does Harry have on today?","asker":"sam","source":"esp32"}' +``` + +The Hub reads `HUB_TOKEN_*`, `OMNIROUTE_API_KEY`, and the project tokens from +`~/.config/environment.d/10-secrets.conf` (mounted read-only, never committed). + +### Test without a network + +```bash +scripts/hub-test.sh +``` + +The fake `/ask` project in `src/tests/fakeproject/` proves the round trip. + +### Phase 1 — prove the Paperclip round trip + ```bash -# Phase 1 — prove the Paperclip round trip # 1. Create an agent API key in Paperclip. # 2. Export it for the test only. export PAPERCLIP_API_KEY= # 3. File one job. -curl -s -X POST http://192.168.20.13:3100/api/issues \ - -H "Authorization: Bearer $PAPERCLIP_API_KEY" \ - -H "Content-Type: application/json" \ - -d '{"title":"round-trip test","description":"prove the return path"}' +scripts/paperclip-roundtrip.sh --title "round-trip test" ``` Full steps: `docs/outline/04-agent-runbook.md`. diff --git a/docker-compose.yml b/docker-compose.yml index 7135afa..e7949bf 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -10,11 +10,26 @@ services: - "8788:8788" environment: HUB_PORT: "8788" - # Secrets are never hard-coded. The Hub keys (HUB_TOKEN_*, OMNIROUTE_API_KEY) - # come from ~/.config/environment.d/10-secrets.conf. Add only the keys the - # running routes need, when they land (AUD-49+). + HUB_ROUTING_FILE: "/app/routing.md" + HUB_LOG_PATH: "/tmp/hub.log" + # Data projects that speak /ask (comma-separated name=url). Empty until a + # project is ready. The Hub reads each project's /capabilities. + HUB_PROJECTS: "${HUB_PROJECTS:-}" + # House TTS. The Hub starts speech and does not wait. + JERVIS_URL: "${JERVIS_URL:-http://192.168.20.13:8501}" + # Secrets are never hard-coded. The Hub keys (HUB_TOKEN_*, OMNIROUTE_API_KEY, + # PAPERCLIP_API_KEY, HUB_PROJECT__TOKEN) come from environment.d. The + # file is mounted read-only; the Hub reads it when the environment is empty. + HUB_SECRETS_FILE: "/run/secrets/hub.env" + volumes: + # routing.md is human-edited and stays outside the image. + - "./routing.md:/app/routing.md:ro" + - "${HOME}/.config/environment.d/10-secrets.conf:/run/secrets/hub.env:ro" + # The distroless image has no writable path except this tmpfs. The log lives here. + tmpfs: + - /tmp healthcheck: - test: ["CMD", "/hub", "-healthcheck"] + test: ["CMD", "/app/hub", "-healthcheck"] interval: 30s timeout: 3s retries: 3 diff --git a/docs/outline/02-config-and-env.md b/docs/outline/02-config-and-env.md index 3eb5ab5..5df638b 100644 --- a/docs/outline/02-config-and-env.md +++ b/docs/outline/02-config-and-env.md @@ -23,6 +23,38 @@ Relevant keys: | `HUB_TOKEN_CONSOLE` | Hub token for the console portal. | | `HUB_TOKEN_ADMIN` | Hub token for the admin app. The only trusted door. | +## Hub Environment Keys + +The Hub reads these keys from the environment, then from the secrets file. The +secrets file path is `HUB_SECRETS_FILE`, or `~/.config/environment.d/10-secrets.conf`. + +| Key | Default | Use | +|---|---|---| +| `HUB_PORT` | `8788` | Listen port. | +| `HUB_ROUTING_FILE` | `routing.md` | Human-edited routing rules. | +| `HUB_SECRETS_FILE` | `~/.config/environment.d/10-secrets.conf` | Read secrets when the environment is empty. | +| `HUB_PROJECTS` | empty | Data projects: `name=url,name2=url2`. | +| `HUB_PROJECT__TOKEN` | empty | Per-project bearer token. | +| `OMNIROUTE_URL` | `http://192.168.20.13:20129` | Model gateway base. | +| `OMNIROUTE_MODEL` | `gpt-4o-mini` | Routing model. The gateway is OpenAI-compatible. | +| `JERVIS_URL` | `http://192.168.20.13:8501` | House TTS. | +| `PAPERCLIP_BASE_URL` | `http://192.168.20.13:3100` | Paperclip board. | +| `PAPERCLIP_COMPANY_ID` | from the repo convention | Company for filed jobs. | +| `HUB_LOG_PATH` | `hub.log` | Append-only request log. | +| `HUB_LOG_MAX_BYTES` | `1048576` | Rotate size. | +| `HUB_LOG_KEEP_DAYS` | `7` | Rotated files kept. | +| `HUB_LLM_TIMEOUT` | `3s` | Routing LLM cap (`docs/hub-api.md` section 8). | +| `HUB_PROJECT_TIMEOUT` | `3s` | One project `/ask` cap. | +| `HUB_DEST_TIMEOUT` | `2s` | One destination command cap. | +| `HUB_TTS_TIMEOUT` | `15s` | House TTS completes out of band. | +| `HUB_ASK_TIMEOUT` | `8s` | `POST /ask` total. | +| `HUB_APPROVE_TIMEOUT` | `5s` | `POST /approve` total. | +| `HUB_RATE_PER_MIN` | `30` | Requests per client token per minute. | + +> Observed on 2026-10-10: the routing LLM answer through OmniRoute took 0.8 s to +> 1.9 s with `gpt-4o-mini`, inside the 3 s cap. The `auto/*` combos were slower +> than 3 s and are not the default. + ## Services and Ports | Service | Host | Port | Use | diff --git a/docs/outline/03-code-map.md b/docs/outline/03-code-map.md index a59d761..7d2a55c 100644 --- a/docs/outline/03-code-map.md +++ b/docs/outline/03-code-map.md @@ -1,15 +1,16 @@ # Central AI Hub — Code Structure Map > **Writing Style**: ASD-STE100 Simplified Technical English (`ste-writing`). -> **Status**: Skeleton built (AUD-48). Routing (AUD-49) is not built yet. +> **Status**: Hub skeleton built (AUD-49). The three inbound routes answer. -## Planned Folder Layout +## Folder Layout ```text central_ai_llm_master/ AGENTS.md project rules for agents hub_brief.md agreed design (source of truth) README.md technical readme + routing.md human-edited routing rules Dockerfile multi-stage build (static binary, distroless) docker-compose.yml one container, port 8788 .dockerignore build context filter @@ -22,40 +23,69 @@ central_ai_llm_master/ ask-contract.md Hub to project interface hub-api.md frozen inbound API (v1) vikunja-tasks.json human task list - src/ Hub service code - go.mod module hub - main.go HTTP entry point (GET /health) - hear/ takes voice or text - understand/ LLM intent routing - check/ authorization and approval - route/ dispatch to a destination - answer/ reply shaping (text, TTS, notify) - destinations/ one client per destination - tests/ unit and round-trip tests + scripts/ + paperclip-roundtrip.sh Phase 1 outbound job proof + src/ Hub service code (module hub) + go.mod module hub (Go 1.25, standard library only) + main.go HTTP layer: POST /ask, POST /approve, GET /health + apierr/ the frozen error envelope + ids/ ULID request and approval ids + config/ environment and secrets loading + hear/ step 1: decode and validate the request + understand/ step 2: never-route check and the routing LLM + check/ step 3: identity and source checks + route/ step 4: dispatch to one destination + answer/ step 5: reply shaping (text, speech, notify) + destinations/ one client per destination: + project.go (a data project through /ask) + speakers.go (Jervis house TTS) + paperclip.go (file a job after approval) + logbook/ append-only, size-capped request log + hub/ the pipeline that wires the five steps + tests/fakeproject/ fake /ask project for tests and manual runs ``` ## Entry Points -| Entry point | Use | -|---|---| -| `POST /ask` | Send a voice transcript or text. Returns a reply. | -| `POST /approve` | Admin app confirms a restricted action. | -| `GET /health` | Liveness check. | +| Entry point | Use | Handler | +|---|---|---| +| `POST /ask` | Send a voice transcript or text. Returns a reply. | `main.go`, `hub.Service.Ask` | +| `POST /approve` | Admin app confirms a Hub-owned action. | `main.go`, `hub.Service.Approve` | +| `GET /health` | Liveness check. No authentication. | `main.go`, `hub.Service.Health` | The inbound API is frozen. See `docs/hub-api.md`. +## The Five Steps in Code + +| Step | Package | Notes | +|---|---|---| +| Hear | `hear` | Decode JSON, validate the frozen shape, make a request id. | +| Understand | `understand` | The never-route check runs first, with no LLM. Then the OmniRoute LLM picks route, fanout, clarify, or no_match. | +| Check | `check` | One identity (Sam) and a source that matches the token. | +| Route | `route` | Dispatch. The action call holds no LLM. | +| Answer | `answer` | Choose the reply shape from `source` and `want`. House TTS is out of band. | + ## Primary Dependencies -- HTTP server (Go 1.25, standard library `net/http`). -- HTTP client for OmniRoute, Home Assistant, Paperclip. -- MQTT client for voice topics. -- Tavily API client for web search. +- HTTP server and client (Go 1.25, standard library `net/http`). +- No third-party module. The binary is static and small. +- OmniRoute, Home Assistant, MQTT, and Paperclip are called over HTTP. ## Destinations Each destination gets one client under `src/destinations/`. The client calls the destination API. The client does not hold business logic. +## Tests + +```bash +# Unit and round-trip tests. No network, no secrets. +docker run --rm -v "$PWD/src":/src -w /src golang:1.25-alpine go test ./... +``` + +The fake `/ask` project (`src/tests/fakeproject/`) proves the Hub-to-project +round trip. It has no real data and changes no state. + ## Source Control - Gitea: `sam/central_ai_llm_master` diff --git a/docs/outline/04-agent-runbook.md b/docs/outline/04-agent-runbook.md index 96cedcd..cf88c78 100644 --- a/docs/outline/04-agent-runbook.md +++ b/docs/outline/04-agent-runbook.md @@ -83,6 +83,84 @@ announce ran one time. - The house speaks the result. **DONE.** - This file records the result. **DONE.** +## Phase 2 — Hub Skeleton (AUD-49) + +This phase builds the Hub itself. The shape is frozen in `docs/hub-api.md`. The +code layout is in `docs/outline/03-code-map.md`. + +### What was built + +- `POST /ask`, `POST /approve`, and `GET /health`. +- The five steps: `hear` → `understand` → `check` → `route` → `answer`. +- Routing through the LLM at OmniRoute, with the capability topics and + `routing.md`. +- The never-route check runs before the LLM. A control action never reaches a + model. +- One real destination: the house speakers (Jervis `/announce`). +- A fake `/ask` project for tests and manual runs. + +### Test command + +```bash +docker run --rm -v "$PWD/src":/src -w /src golang:1.25-alpine go test ./... +``` + +Result on 2026-10-10: all packages pass, including the fake `/ask` round trip. + +### Live round trip + +The Hub ran on port 18788 with `HUB_PROJECTS="compass_nhs=http://127.0.0.1:9199"` +and the fake project on 9199. The real OmniRoute LLM chose the project. Jervis +spoke the reply on the house speakers. + +Request: + +```json +{ + "text": "what does Harry have on today?", + "asker": "sam", + "source": "esp32", + "device": "kitchen-esp32", + "request_id": "01AUD49PROOF" +} +``` + +Response: + +```json +{ + "status": "ok", + "request_id": "01AUD49PROOF", + "answer": "Harry has maths and a music lesson today.", + "shape": "speech", + "speech": {"text": "Harry has maths and a music lesson today.", "target": "house"}, + "route": {"kind": "project", "project": "compass_nhs"} +} +``` + +Hub stdout (evidence): + +```text +router: model=gpt-4o-mini status=200 elapsed=1.554598547s +speakers: announced 41 chars +``` + +The routing LLM answered in 1.5 s (inside the 3 s cap). The house announcement +ran out of band and Jervis returned HTTP 200. Jervis has no playback telemetry, +so the HTTP 200 and the spoken reply are the strongest available proof. + +### Control action proof + +`"cut the kids internet"` returned `status: "refused"`, `route.kind: "refused"`, +`route.reason: "high_consequence_control"`. No LLM call ran for that request. +The unit test `TestControlActionNeverReachesLLM` asserts the zero call count. + +### Deploy note + +The `docker-compose.yml` mounts `routing.md` and the secrets file read-only, and +uses a `/tmp` tmpfs for the log. To deploy: build the image, then restart the +container. A `HUB_TOKEN_*` key per client must exist in the secrets file first. + ## After Phase 1 Follow the build order in `AGENTS.md`. Do not start a destination before the diff --git a/scripts/hub-test.sh b/scripts/hub-test.sh new file mode 100755 index 0000000..b320697 --- /dev/null +++ b/scripts/hub-test.sh @@ -0,0 +1,25 @@ +#!/usr/bin/env bash +# +# hub-test.sh - run the Hub unit and round-trip tests. +# +# The tests use the fake /ask project. They need no network and no secrets. +# They run in the golang container so the host does not need a Go toolchain. +# +# Usage: +# scripts/hub-test.sh # run all tests +# scripts/hub-test.sh -race # extra args go to "go test" +# +set -euo pipefail + +readonly SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" +readonly PROJECT_DIR="$(cd -- "$SCRIPT_DIR/.." && pwd)" +readonly GO_IMAGE="${HUB_GO_IMAGE:-golang:1.25-alpine}" + +command -v docker >/dev/null 2>&1 || { echo "error: docker is required" >&2; exit 1; } + +# The mount is read-only so a test run can never change the source tree. +docker run --rm \ + -v "$PROJECT_DIR/src":/src:ro \ + -w /src \ + "$GO_IMAGE" \ + go test "$@" ./... diff --git a/src/answer/answer.go b/src/answer/answer.go new file mode 100644 index 0000000..470c875 --- /dev/null +++ b/src/answer/answer.go @@ -0,0 +1,171 @@ +// Package answer is step 5 of the Hub: return a short reply in the right shape +// for the device. It changes the shape only. It never changes the meaning of an +// answer (docs/hub-api.md section 9). +package answer + +import ( + "time" + + "hub/hear" + "hub/route" +) + +// Shape is the reply shape. +type Shape string + +const ( + ShapeText Shape = "text" + ShapeSpeech Shape = "speech" + ShapeNotify Shape = "notification" + ShapeNone Shape = "none" +) + +// Speech is the TTS detail. +type Speech struct { + Text string `json:"text"` + Target string `json:"target"` +} + +// RouteInfo records what the Hub did. +type RouteInfo struct { + Kind string `json:"kind"` + Project string `json:"project,omitempty"` + Reason string `json:"reason,omitempty"` +} + +// ApprovalInfo is the approval block in the reply. +type ApprovalInfo struct { + ApprovalID string `json:"approval_id"` + Action string `json:"action"` + Summary string `json:"summary"` + Reason string `json:"reason"` + Expires string `json:"expires"` +} + +// ClarifyInfo is the clarify block in the reply. +type ClarifyInfo struct { + Question string `json:"question"` + Options []string `json:"options,omitempty"` +} + +// Response is one POST /ask reply (docs/hub-api.md section 5). +type Response struct { + Status string `json:"status"` + RequestID string `json:"request_id"` + Answer string `json:"answer"` + Shape Shape `json:"shape"` + Speech *Speech `json:"speech,omitempty"` + Route *RouteInfo `json:"route,omitempty"` + Approval *ApprovalInfo `json:"approval,omitempty"` + Clarify *ClarifyInfo `json:"clarify,omitempty"` + Expires string `json:"expires,omitempty"` +} + +// Build turns a route outcome into a wire reply. +func Build(req hear.Request, outcome route.Outcome) Response { + shape, target := shapeFor(req) + + resp := Response{ + Status: string(outcome.Status), + RequestID: req.RequestID, + Answer: outcome.Answer, + Shape: shape, + Route: &RouteInfo{ + Kind: string(outcome.Kind), + Project: outcome.Project, + Reason: outcome.Reason, + }, + } + if len(outcome.Projects) > 1 { + resp.Route.Project = "" + } + if outcome.Expires != "" { + resp.Expires = outcome.Expires + } + if shape == ShapeSpeech && target != "none" && outcome.Answer != "" { + resp.Speech = &Speech{Text: speakable(outcome.Answer), Target: target} + } + if outcome.Approval != nil { + resp.Approval = &ApprovalInfo{ + ApprovalID: outcome.Approval.ID, + Action: outcome.Approval.Action, + Summary: outcome.Approval.Summary, + Reason: outcome.Approval.Reason, + Expires: outcome.Approval.Expires.UTC().Format(time.RFC3339), + } + } + if outcome.Clarify != nil { + resp.Clarify = &ClarifyInfo{Question: outcome.Clarify.Question, Options: outcome.Clarify.Options} + } + return resp +} + +// shapeFor picks the final shape and TTS target from source and want. +// +// Defaults (docs/hub-api.md section 9): +// +// esp32 -> speech / house +// phone -> speech / phone +// chat -> text / none +// console -> text / none +// admin -> text / none +func shapeFor(req hear.Request) (Shape, string) { + defShape, defTarget := ShapeText, "none" + switch req.Source { + case hear.SourceESP32: + defShape, defTarget = ShapeSpeech, "house" + case hear.SourcePhone: + defShape, defTarget = ShapeSpeech, "phone" + } + + switch req.Want { + case hear.WantText: + return ShapeText, "none" + case hear.WantSpeech: + target := defTarget + if target == "none" { + target = "house" + } + return ShapeSpeech, target + case hear.WantBoth: + target := defTarget + if target == "none" { + target = "house" + } + return ShapeSpeech, target + } + return defShape, defTarget +} + +// speakable trims a reply to one short, spoken sentence. It drops a trailing +// newline so TTS does not add a pause. +func speakable(s string) string { + out := make([]rune, 0, len(s)) + for _, r := range s { + if r == '\n' || r == '\r' { + r = ' ' + } + out = append(out, r) + } + // Collapse repeated spaces. + prevSpace := false + clean := out[:0] + for _, r := range out { + if r == ' ' { + if prevSpace { + continue + } + prevSpace = true + } else { + prevSpace = false + } + clean = append(clean, r) + } + return string(clean) +} + +// ShouldSpeak reports whether the Hub must call the house speakers for this +// reply. A phone target is part of the reply, not a house announcement. +func ShouldSpeak(resp Response) bool { + return resp.Shape == ShapeSpeech && resp.Speech != nil && resp.Speech.Target == "house" +} diff --git a/src/answer/answer_test.go b/src/answer/answer_test.go new file mode 100644 index 0000000..0304481 --- /dev/null +++ b/src/answer/answer_test.go @@ -0,0 +1,80 @@ +package answer + +import ( + "testing" + "time" + + "hub/hear" + "hub/route" +) + +func TestShapeFromSource(t *testing.T) { + cases := []struct { + source hear.Source + want hear.Want + shape Shape + target string + }{ + {hear.SourceESP32, "", ShapeSpeech, "house"}, + {hear.SourcePhone, "", ShapeSpeech, "phone"}, + {hear.SourceChat, "", ShapeText, "none"}, + {hear.SourceConsole, "", ShapeText, "none"}, + {hear.SourceAdmin, "", ShapeText, "none"}, + {hear.SourceChat, hear.WantSpeech, ShapeSpeech, "house"}, + {hear.SourceESP32, hear.WantText, ShapeText, "none"}, + {hear.SourceChat, hear.WantBoth, ShapeSpeech, "house"}, + } + for _, c := range cases { + req := hear.Request{Source: c.source, Want: c.want} + shape, target := shapeFor(req) + if shape != c.shape || target != c.target { + t.Errorf("source=%s want=%s => %s/%s, want %s/%s", c.source, c.want, shape, target, c.shape, c.target) + } + } +} + +func TestBuildSpeechReply(t *testing.T) { + req := hear.Request{Source: hear.SourceESP32, RequestID: "r1"} + out := route.Outcome{Kind: route.KindProject, Status: route.StatusOK, Answer: "Harry has maths today.", Project: "compass_nhs", Expires: "2026-10-10T09:00:00Z"} + resp := Build(req, out) + if resp.Status != "ok" || resp.Shape != ShapeSpeech { + t.Fatalf("resp = %#v", resp) + } + if resp.Speech == nil || resp.Speech.Target != "house" { + t.Fatalf("speech = %#v", resp.Speech) + } + if !ShouldSpeak(resp) { + t.Errorf("ShouldSpeak = false") + } + if resp.Route == nil || resp.Route.Project != "compass_nhs" { + t.Errorf("route = %#v", resp.Route) + } +} + +func TestBuildApproval(t *testing.T) { + req := hear.Request{Source: hear.SourceESP32, RequestID: "r1"} + out := route.Outcome{ + Kind: route.KindPaperclip, + Status: route.StatusNeedsApproval, + Answer: "I need a yes on the admin app to file that job.", + Approval: &route.Approval{ + ID: "ap_1", Action: "paperclip_job", Summary: "File a job: plan.", + Reason: "slow", Expires: time.Date(2026, 10, 10, 9, 10, 0, 0, time.UTC), + }, + } + resp := Build(req, out) + if resp.Approval == nil || resp.Approval.ApprovalID != "ap_1" { + t.Fatalf("approval = %#v", resp.Approval) + } + if resp.Approval.Expires != "2026-10-10T09:10:00Z" { + t.Errorf("expires = %q", resp.Approval.Expires) + } +} + +func TestBuildTextDoesNotSpeak(t *testing.T) { + req := hear.Request{Source: hear.SourceChat, RequestID: "r1"} + resp := Build(req, route.Outcome{Kind: route.KindProject, Status: route.StatusOK, Answer: "hi"}) + if resp.Shape != ShapeText || resp.Speech != nil || ShouldSpeak(resp) { + t.Fatalf("resp = %#v", resp) + } +} diff --git a/src/apierr/apierr.go b/src/apierr/apierr.go new file mode 100644 index 0000000..c46e971 --- /dev/null +++ b/src/apierr/apierr.go @@ -0,0 +1,41 @@ +// Package apierr holds the frozen error envelope from docs/hub-api.md +// section 4. Every package returns *apierr.Error so the HTTP layer can write +// the same JSON shape for every fault. +package apierr + +import "fmt" + +// Error is one API fault. The fields match the frozen envelope: +// +// {"status":"error","error":{"code":"bad_request","message":"...","request_id":"..."}} +// +// Message is one short sentence. It is safe to speak aloud. It holds no secret, +// no internal path, and no stack trace (docs/hub-api.md section 4). +type Error struct { + HTTP int // HTTP status code. + Code string // Machine code, for example "bad_request". + Message string // One short, speakable sentence. +} + +func (e *Error) Error() string { + return fmt.Sprintf("%s (%d): %s", e.Code, e.HTTP, e.Message) +} + +// New makes an Error. Use this instead of building the struct by hand so the +// code and the HTTP status stay paired. +func New(httpCode int, code, message string) *Error { + return &Error{HTTP: httpCode, Code: code, Message: message} +} + +// From returns err as *Error. When err is nil or is another type, it returns a +// generic internal error. The HTTP layer uses this so it never leaks a raw Go +// error to a client. +func From(err error) *Error { + if err == nil { + return nil + } + if e, ok := err.(*Error); ok { + return e + } + return New(500, "internal", "The Hub had an internal problem.") +} diff --git a/src/check/check.go b/src/check/check.go new file mode 100644 index 0000000..5a01cce --- /dev/null +++ b/src/check/check.go @@ -0,0 +1,42 @@ +// Package check is step 3 of the Hub: test whether this person may ask for +// this. It does two things. It checks the one identity (Sam), and it checks +// that the request source matches the token that opened the door. +// +// It holds no routing logic. +package check + +import ( + "hub/apierr" + "hub/hear" +) + +// Asker is the one identity in v1 (AGENTS.md rule 14). +const Asker = "sam" + +// Ask validates a POST /ask request after the bearer token is read. +// +// tokenSource is the source that owns the token. reqSource is what the body +// claims. They must match (docs/hub-api.md section 5). +func Ask(req hear.Request, tokenSource string) *apierr.Error { + if req.Asker != Asker { + return apierr.New(422, "unsupported_asker", "The Hub supports one asker, Sam.") + } + if tokenSource != string(req.Source) { + return apierr.New(403, "source_mismatch", "The token does not match the source field.") + } + return nil +} + +// Admin validates that a trusted door called POST /approve. Only the admin +// token may decide an approval (docs/hub-api.md section 6). +func Admin(tokenSource string) *apierr.Error { + if tokenSource != string(hear.SourceAdmin) { + return apierr.New(403, "forbidden", "Only the admin app may decide an approval.") + } + return nil +} + +// IsAdmin reports whether the token source is the admin app. +func IsAdmin(tokenSource string) bool { + return tokenSource == string(hear.SourceAdmin) +} diff --git a/src/config/config.go b/src/config/config.go new file mode 100644 index 0000000..63158bf --- /dev/null +++ b/src/config/config.go @@ -0,0 +1,285 @@ +// Package config loads the Hub configuration. +// +// Every secret comes from the environment or from the environment.d secrets +// file (AGENTS.md rule 2). Nothing is hard-coded. The config keeps the Hub +// testable: tests build a Config by hand and point it at fake servers. +package config + +import ( + "bufio" + "os" + "path/filepath" + "strconv" + "strings" + "time" +) + +// Defaults. A value in the environment replaces each one. +const ( + DefaultPort = "8788" + DefaultRoutingFile = "routing.md" + DefaultLogPath = "hub.log" + DefaultOmniURL = "http://192.168.20.13:20129" + DefaultOmniModel = "gpt-4o-mini" + DefaultJervisURL = "http://192.168.20.13:8501" + DefaultPaperclip = "http://192.168.20.13:3100" + + DefaultLLMTimeout = 3 * time.Second + DefaultProjectTimeout = 3 * time.Second + DefaultDestinationTime = 2 * time.Second + DefaultTTSTimeout = 15 * time.Second + DefaultAskTimeout = 8 * time.Second + DefaultApproveTimeout = 5 * time.Second + DefaultHealthTimeout = 1 * time.Second + + DefaultLogMaxBytes = 1 << 20 // 1 MiB + DefaultLogKeepDays = 7 + DefaultRatePerMin = 30 +) + +// Project is one data project that speaks the /ask contract. +type Project struct { + Name string + URL string + Token string + Topics []string // Fallback topics when /capabilities is not reachable. +} + +// Config is the full Hub configuration. +type Config struct { + Port string + RoutingFile string + + OmniURL string + OmniKey string + OmniModel string + + JervisURL string + + PaperclipURL string + PaperclipKey string + PaperclipCompanyID string + + // Tokens maps a client source to its bearer token. The token names the + // device, not the person (docs/hub-api.md section 3). + Tokens map[string]string + + // Projects is the data-project registry, keyed by project name. + Projects map[string]Project + + LLMTimeout time.Duration + ProjectTimeout time.Duration + DestTimeout time.Duration + TTSTimeout time.Duration + AskTimeout time.Duration + ApproveTimeout time.Duration + HealthTimeout time.Duration + HealthCheckTime time.Duration + + LogPath string + LogMaxBytes int64 + LogKeepDays int + RatePerMin int +} + +// Load reads the configuration from the environment, then fills gaps from the +// secrets file. It never fails on a missing secret: a missing OMNIROUTE_API_KEY +// only makes the LLM route unavailable. +func Load() *Config { + secrets := readSecrets(secretsFilePath()) + + get := func(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + if v := secrets[key]; v != "" { + return v + } + return def + } + + c := &Config{ + Port: get("HUB_PORT", DefaultPort), + RoutingFile: get("HUB_ROUTING_FILE", DefaultRoutingFile), + + OmniURL: strings.TrimRight(get("OMNIROUTE_URL", DefaultOmniURL), "/"), + OmniKey: get("OMNIROUTE_API_KEY", ""), + OmniModel: get("OMNIROUTE_MODEL", DefaultOmniModel), + + JervisURL: strings.TrimRight(get("JERVIS_URL", DefaultJervisURL), "/"), + + PaperclipURL: strings.TrimRight(get("PAPERCLIP_BASE_URL", DefaultPaperclip), "/"), + PaperclipKey: get("PAPERCLIP_API_KEY", ""), + PaperclipCompanyID: get("PAPERCLIP_COMPANY_ID", ""), + + Tokens: map[string]string{ + "phone": get("HUB_TOKEN_PHONE", ""), + "esp32": get("HUB_TOKEN_ESP32", ""), + "chat": get("HUB_TOKEN_DSH", ""), + "console": get("HUB_TOKEN_CONSOLE", ""), + "admin": get("HUB_TOKEN_ADMIN", ""), + }, + + Projects: parseProjects(get("HUB_PROJECTS", ""), get), + + LLMTimeout: duration(get("HUB_LLM_TIMEOUT", ""), DefaultLLMTimeout), + ProjectTimeout: duration(get("HUB_PROJECT_TIMEOUT", ""), DefaultProjectTimeout), + DestTimeout: duration(get("HUB_DEST_TIMEOUT", ""), DefaultDestinationTime), + TTSTimeout: duration(get("HUB_TTS_TIMEOUT", ""), DefaultTTSTimeout), + AskTimeout: duration(get("HUB_ASK_TIMEOUT", ""), DefaultAskTimeout), + ApproveTimeout: duration(get("HUB_APPROVE_TIMEOUT", ""), DefaultApproveTimeout), + HealthTimeout: duration(get("HUB_HEALTH_TIMEOUT", ""), DefaultHealthTimeout), + + LogPath: get("HUB_LOG_PATH", DefaultLogPath), + LogMaxBytes: int64(intSetting(get("HUB_LOG_MAX_BYTES", ""), DefaultLogMaxBytes)), + LogKeepDays: intSetting(get("HUB_LOG_KEEP_DAYS", ""), DefaultLogKeepDays), + RatePerMin: intSetting(get("HUB_RATE_PER_MIN", ""), DefaultRatePerMin), + } + c.HealthCheckTime = c.HealthTimeout + return c +} + +// TokenFor returns the configured token for a source, or "". +func (c *Config) TokenFor(source string) string { + return c.Tokens[source] +} + +// SourceForToken returns the source that owns a token. It compares in constant +// time (docs/hub-api.md section 3). An empty token never matches. +func (c *Config) SourceForToken(token string) (string, bool) { + if token == "" { + return "", false + } + found := "" + for source, want := range c.Tokens { + if want == "" { + continue + } + if constantTimeEqual(token, want) { + found = source + } + } + if found == "" { + return "", false + } + return found, true +} + +// secretsFilePath returns HUB_SECRETS_FILE, then the standard path. +func secretsFilePath() string { + if p := os.Getenv("HUB_SECRETS_FILE"); p != "" { + return p + } + home, err := os.UserHomeDir() + if err != nil { + return "" + } + return filepath.Join(home, ".config", "environment.d", "10-secrets.conf") +} + +// readSecrets parses a KEY=value file. It ignores comments and blank lines. +// A missing file returns an empty map, not an error. +func readSecrets(path string) map[string]string { + out := map[string]string{} + if path == "" { + return out + } + f, err := os.Open(path) + if err != nil { + return out + } + defer f.Close() + + sc := bufio.NewScanner(f) + for sc.Scan() { + line := strings.TrimSpace(sc.Text()) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + key, value, ok := strings.Cut(line, "=") + if !ok { + continue + } + key = strings.TrimSpace(key) + value = strings.TrimSpace(value) + value = strings.Trim(value, `"'`) + if key != "" { + out[key] = value + } + } + return out +} + +// parseProjects reads "name=url,name2=url2" and adds each project's token. +func parseProjects(spec string, get func(string, string) string) map[string]Project { + out := map[string]Project{} + for _, entry := range strings.Split(spec, ",") { + entry = strings.TrimSpace(entry) + if entry == "" { + continue + } + name, url, ok := strings.Cut(entry, "=") + if !ok { + continue + } + name = strings.TrimSpace(name) + url = strings.TrimRight(strings.TrimSpace(url), "/") + if name == "" || url == "" { + continue + } + envKey := "HUB_PROJECT_" + strings.ToUpper(name) + "_TOKEN" + out[name] = Project{Name: name, URL: url, Token: get(envKey, "")} + } + return out +} + +func duration(s string, def time.Duration) time.Duration { + if s == "" { + return def + } + d, err := time.ParseDuration(s) + if err != nil || d <= 0 { + return def + } + return d +} + +func intSetting(s string, def int) int { + if s == "" { + return def + } + n, err := strconv.Atoi(s) + if err != nil || n <= 0 { + return def + } + return n +} + +// constantTimeEqual compares two strings without an early exit. It is a small +// local copy so the config package has no dependency. +func constantTimeEqual(a, b string) bool { + if len(a) != len(b) { + // Still walk the longer string to keep the cost steady. + max := len(a) + if len(b) > max { + max = len(b) + } + var v byte + for i := 0; i < max; i++ { + var ca, cb byte + if i < len(a) { + ca = a[i] + } + if i < len(b) { + cb = b[i] + } + v |= ca ^ cb + } + return false + } + var v byte + for i := 0; i < len(a); i++ { + v |= a[i] ^ b[i] + } + return v == 0 +} diff --git a/src/destinations/paperclip.go b/src/destinations/paperclip.go new file mode 100644 index 0000000..0e3b590 --- /dev/null +++ b/src/destinations/paperclip.go @@ -0,0 +1,85 @@ +package destinations + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "time" +) + +// Paperclip files one slow, recorded job after a tap-to-approve +// (docs/hub-api.md section 6). The Hub files the job once. +type Paperclip struct { + URL string + APIKey string + CompanyID string + Timeout time.Duration + HTTP *http.Client +} + +// NewPaperclip builds the Paperclip client. +func NewPaperclip(url, key, companyID string, timeout time.Duration) *Paperclip { + if timeout <= 0 { + timeout = 5 * time.Second + } + return &Paperclip{ + URL: strings.TrimRight(url, "/"), + APIKey: key, + CompanyID: companyID, + Timeout: timeout, + HTTP: &http.Client{Timeout: timeout}, + } +} + +// FileJob creates one Paperclip issue. It returns the issue id. +func (p *Paperclip) FileJob(ctx context.Context, title, description string) (string, error) { + if p == nil || p.URL == "" || p.APIKey == "" || p.CompanyID == "" { + return "", fmt.Errorf("paperclip: not configured") + } + body, err := json.Marshal(map[string]string{ + "title": title, + "description": description, + "priority": "medium", + }) + if err != nil { + return "", err + } + ctx, cancel := context.WithTimeout(ctx, p.Timeout) + defer cancel() + + url := fmt.Sprintf("%s/api/companies/%s/issues", p.URL, p.CompanyID) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) + if err != nil { + return "", err + } + req.Header.Set("Authorization", "Bearer "+p.APIKey) + req.Header.Set("Content-Type", "application/json") + resp, err := p.HTTP.Do(req) + if err != nil { + return "", fmt.Errorf("paperclip: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated { + return "", fmt.Errorf("paperclip: status %d", resp.StatusCode) + } + var out struct { + Issue struct { + ID string `json:"id"` + } `json:"issue"` + ID string `json:"id"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return "", err + } + id := out.Issue.ID + if id == "" { + id = out.ID + } + if id == "" { + return "", fmt.Errorf("paperclip: reply has no issue id") + } + return id, nil +} diff --git a/src/destinations/project.go b/src/destinations/project.go new file mode 100644 index 0000000..7408ae5 --- /dev/null +++ b/src/destinations/project.go @@ -0,0 +1,122 @@ +// Package destinations holds one client per destination. A client calls a +// destination and changes nothing else. It holds no business logic and no +// routing. +package destinations + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + "time" +) + +// ProjectClient talks to one data project through the /ask contract +// (docs/ask-contract.md). The Hub never parses the project's data. +type ProjectClient struct { + Name string + URL string + Token string + Timeout time.Duration + HTTP *http.Client +} + +// NewProjectClient builds a client for one project. +func NewProjectClient(name, url, token string, timeout time.Duration) *ProjectClient { + if timeout <= 0 { + timeout = 3 * time.Second + } + return &ProjectClient{ + Name: name, + URL: strings.TrimRight(url, "/"), + Token: token, + Timeout: timeout, + HTTP: &http.Client{Timeout: timeout}, + } +} + +// AskRequest is the Hub-to-project body. +type AskRequest struct { + Question string `json:"question"` + Asker string `json:"asker"` + Person string `json:"person,omitempty"` + Source string `json:"source,omitempty"` +} + +// AskReply is the project's answer. A nil Answer means the project cannot +// answer. The Hub says so and stops (docs/ask-contract.md). +type AskReply struct { + Answer *string `json:"answer"` + Data json.RawMessage `json:"data,omitempty"` + Expires string `json:"expires,omitempty"` + Reason string `json:"reason,omitempty"` +} + +// Ask forwards one question to the project. It never parses the project's data. +func (p *ProjectClient) Ask(ctx context.Context, body AskRequest) (AskReply, error) { + var reply AskReply + raw, err := json.Marshal(body) + if err != nil { + return reply, err + } + ctx, cancel := context.WithTimeout(ctx, p.Timeout) + defer cancel() + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, p.URL+"/ask", bytes.NewReader(raw)) + if err != nil { + return reply, err + } + req.Header.Set("Content-Type", "application/json; charset=utf-8") + if p.Token != "" { + req.Header.Set("Authorization", "Bearer "+p.Token) + } + + resp, err := p.HTTP.Do(req) + if err != nil { + if ctx.Err() == context.DeadlineExceeded { + return reply, fmt.Errorf("project %s: timeout", p.Name) + } + return reply, fmt.Errorf("project %s: %w", p.Name, err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return reply, fmt.Errorf("project %s: status %d", p.Name, resp.StatusCode) + } + if err := json.NewDecoder(resp.Body).Decode(&reply); err != nil { + return reply, fmt.Errorf("project %s: bad reply: %w", p.Name, err) + } + return reply, nil +} + +// Capabilities reads the project's topic list. It is best effort: a project +// that is down does not stop the Hub. +func (p *ProjectClient) Capabilities(ctx context.Context) ([]string, error) { + ctx, cancel := context.WithTimeout(ctx, p.Timeout) + defer cancel() + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, p.URL+"/capabilities", nil) + if err != nil { + return nil, err + } + if p.Token != "" { + req.Header.Set("Authorization", "Bearer "+p.Token) + } + resp, err := p.HTTP.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("project %s: capabilities status %d", p.Name, resp.StatusCode) + } + var out struct { + Project string `json:"project"` + Topics []string `json:"topics"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + return nil, err + } + return out.Topics, nil +} diff --git a/src/destinations/speakers.go b/src/destinations/speakers.go new file mode 100644 index 0000000..cf759a8 --- /dev/null +++ b/src/destinations/speakers.go @@ -0,0 +1,65 @@ +package destinations + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "log" + "net/http" + "strings" + "time" +) + +// Speaker starts house speech. The Hub starts it and does not wait for the +// speakers (docs/hub-api.md section 8). Phone TTS is part of the reply, so the +// Hub does not call a speaker for target "phone". +type Speaker struct { + URL string + Timeout time.Duration + HTTP *http.Client +} + +// NewSpeaker builds the Jervis TTS client. +func NewSpeaker(url string, timeout time.Duration) *Speaker { + if timeout <= 0 { + timeout = 2 * time.Second + } + return &Speaker{ + URL: strings.TrimRight(url, "/"), + Timeout: timeout, + HTTP: &http.Client{Timeout: timeout}, + } +} + +// Announce starts TTS on the house speakers. It returns an error when Jervis +// does not accept the request. The Hub does not wait for playback. +func (s *Speaker) Announce(ctx context.Context, text string) error { + if s == nil || s.URL == "" { + return nil + } + raw, err := json.Marshal(map[string]string{"text": text}) + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(ctx, s.Timeout) + defer cancel() + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.URL+"/announce", bytes.NewReader(raw)) + if err != nil { + return err + } + req.Header.Set("Content-Type", "application/json") + resp, err := s.HTTP.Do(req) + if err != nil { + log.Printf("speakers: announce failed: %v", err) + return fmt.Errorf("speakers: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + log.Printf("speakers: announce status %d", resp.StatusCode) + return fmt.Errorf("speakers: status %d", resp.StatusCode) + } + log.Printf("speakers: announced %d chars", len(text)) + return nil +} diff --git a/src/hear/hear.go b/src/hear/hear.go new file mode 100644 index 0000000..7d5880c --- /dev/null +++ b/src/hear/hear.go @@ -0,0 +1,109 @@ +// Package hear is step 1 of the Hub: take voice or text from any way in. +// +// It decodes the request body of POST /ask and validates it against the frozen +// shape in docs/hub-api.md section 5. It holds no routing logic. +package hear + +import ( + "bytes" + "encoding/json" + "strings" + "unicode/utf8" + + "hub/apierr" + "hub/ids" +) + +// Source is where a request came from. The value is also the token name. +type Source string + +const ( + SourceESP32 Source = "esp32" + SourcePhone Source = "phone" + SourceChat Source = "chat" + SourceConsole Source = "console" + SourceAdmin Source = "admin" +) + +// Want is the reply shape the caller prefers. +type Want string + +const ( + WantText Want = "text" + WantSpeech Want = "speech" + WantBoth Want = "both" +) + +// Request is one inbound question or command. +type Request struct { + Text string `json:"text"` + Asker string `json:"asker"` + Source Source `json:"source"` + Person string `json:"person"` + Want Want `json:"want"` + Device string `json:"device"` + RequestID string `json:"request_id"` +} + +// Source is valid when it is one of the frozen enum values. +func (s Source) Valid() bool { + switch s { + case SourceESP32, SourcePhone, SourceChat, SourceConsole, SourceAdmin: + return true + } + return false +} + +// PersonOrDefault returns the person, or the asker when the field is absent. +func (r Request) PersonOrDefault() string { + if r.Person != "" { + return r.Person + } + return r.Asker +} + +// Parse decodes and validates one POST /ask body. A fault is an *apierr.Error +// with the HTTP status and code from docs/hub-api.md section 4. +func Parse(body []byte) (Request, *apierr.Error) { + var r Request + dec := json.NewDecoder(bytes.NewReader(body)) + // Clients may add fields. Ignore unknown fields (docs/hub-api.md section 4). + if err := dec.Decode(&r); err != nil { + return Request{}, apierr.New(400, "bad_request", "The request body is not valid JSON.") + } + + r.Text = strings.TrimSpace(r.Text) + r.Asker = strings.TrimSpace(r.Asker) + r.Person = strings.TrimSpace(r.Person) + r.Device = strings.TrimSpace(r.Device) + r.RequestID = strings.TrimSpace(r.RequestID) + + if r.Text == "" { + return Request{}, apierr.New(422, "unprocessable", "Field 'text' is required.") + } + if utf8.RuneCountInString(r.Text) > 1000 { + return Request{}, apierr.New(422, "unprocessable", "Field 'text' is over 1000 characters.") + } + if r.Asker == "" { + return Request{}, apierr.New(422, "unprocessable", "Field 'asker' is required.") + } + if !r.Source.Valid() { + return Request{}, apierr.New(422, "unprocessable", "Field 'source' has an unknown value.") + } + switch r.Want { + case "", WantText, WantSpeech, WantBoth: + default: + return Request{}, apierr.New(422, "unprocessable", "Field 'want' has an unknown value.") + } + if r.RequestID == "" { + r.RequestID = ids.New() + } + return r, nil +} + +// Unmarshal decodes a /ask body without validating. Tests use it. +func Unmarshal(body []byte) (Request, error) { + var r Request + err := json.Unmarshal(body, &r) + return r, err +} diff --git a/src/hear/hear_test.go b/src/hear/hear_test.go new file mode 100644 index 0000000..814bb26 --- /dev/null +++ b/src/hear/hear_test.go @@ -0,0 +1,65 @@ +package hear + +import ( + "testing" +) + +func TestParseValid(t *testing.T) { + body := []byte(`{ + "text": " what does Harry have on today? ", + "asker": "sam", + "source": "esp32", + "person": "", + "want": "speech", + "device": "kitchen-esp32" + }`) + req, err := Parse(body) + if err != nil { + t.Fatalf("Parse returned error: %v", err) + } + if req.Text != "what does Harry have on today?" { + t.Errorf("text = %q", req.Text) + } + if req.Person != "" { + t.Errorf("person should stay empty before default") + } + if req.PersonOrDefault() != "sam" { + t.Errorf("person default = %q, want sam", req.PersonOrDefault()) + } + if req.RequestID == "" { + t.Errorf("RequestID should be generated") + } +} + +func TestParseMissingText(t *testing.T) { + body := []byte(`{"text":"","asker":"sam","source":"esp32"}`) + _, err := Parse(body) + if err == nil { + t.Fatal("expected an error") + } + if err.Code != "unprocessable" || err.HTTP != 422 { + t.Errorf("code = %q http = %d", err.Code, err.HTTP) + } +} + +func TestParseBadJSON(t *testing.T) { + _, err := Parse([]byte(`{not json`)) + if err == nil || err.Code != "bad_request" || err.HTTP != 400 { + t.Fatalf("got %#v", err) + } +} + +func TestParseBadSource(t *testing.T) { + body := []byte(`{"text":"hi","asker":"sam","source":"smoke-signal"}`) + _, err := Parse(body) + if err == nil || err.HTTP != 422 { + t.Fatalf("got %#v", err) + } +} + +func TestParseUnknownFieldsAreIgnored(t *testing.T) { + body := []byte(`{"text":"hi","asker":"sam","source":"chat","future":"ok"}`) + if _, err := Parse(body); err != nil { + t.Fatalf("unknown field should be ignored: %v", err) + } +} diff --git a/src/hub/approval.go b/src/hub/approval.go new file mode 100644 index 0000000..eef52d0 --- /dev/null +++ b/src/hub/approval.go @@ -0,0 +1,80 @@ +package hub + +import ( + "sync" + "time" + + "hub/ids" + "hub/route" +) + +// Approvals holds Hub-owned approvals in memory. An approval is single use. It +// expires (docs/hub-api.md section 6). A restart loses pending approvals, which +// is safe: the user asks again. +type Approvals struct { + mu sync.Mutex + ttl time.Duration + items map[string]*entry +} + +type entry struct { + approval *route.Approval + decided bool +} + +// NewApprovals builds an approval store. A ttl of zero uses 10 minutes. +func NewApprovals(ttl time.Duration) *Approvals { + if ttl <= 0 { + ttl = 10 * time.Minute + } + return &Approvals{ttl: ttl, items: map[string]*entry{}} +} + +// Create stores one pending approval and returns it. It satisfies +// route.ApprovalStore. +func (a *Approvals) Create(action, summary, reason string) *route.Approval { + ap := &route.Approval{ + ID: ids.Approval(), + Action: action, + Summary: summary, + Reason: reason, + Expires: time.Now().UTC().Add(a.ttl), + } + a.mu.Lock() + defer a.mu.Unlock() + a.items[ap.ID] = &entry{approval: ap} + return ap +} + +// ErrNoApproval means the id is unknown or was lost on a restart. +var ErrNoApproval = errApproval("unknown approval") + +// ErrApprovalExpired means the approval passed its expires time. +var ErrApprovalExpired = errApproval("approval expired") + +// ErrAlreadyDecided means the approval was decided before. +var ErrAlreadyDecided = errApproval("approval already decided") + +type errApproval string + +func (e errApproval) Error() string { return string(e) } + +// Decide marks an approval approved or denied. It returns the stored approval +// on the first decision. +func (a *Approvals) Decide(id string) (*route.Approval, error) { + a.mu.Lock() + defer a.mu.Unlock() + + e, ok := a.items[id] + if !ok { + return nil, ErrNoApproval + } + if e.decided { + return nil, ErrAlreadyDecided + } + if time.Now().UTC().After(e.approval.Expires) { + return nil, ErrApprovalExpired + } + e.decided = true + return e.approval, nil +} diff --git a/src/hub/health.go b/src/hub/health.go new file mode 100644 index 0000000..e30b253 --- /dev/null +++ b/src/hub/health.go @@ -0,0 +1,48 @@ +package hub + +import ( + "context" + "time" +) + +// HealthResponse is the GET /health reply (docs/hub-api.md section 7). +type HealthResponse struct { + OK bool `json:"ok"` + Service string `json:"service"` + Version string `json:"version"` + UptimeS int64 `json:"uptime_s"` + Checks map[string]string `json:"checks"` +} + +// Health runs the liveness checks. A check never blocks longer than 1 second. +func (s *Service) Health(ctx context.Context) HealthResponse { + checks := map[string]string{ + "omniroute": "unknown", + "mqtt": "unknown", + "log": "ok", + } + if s.HealthProbe != nil { + ctx, cancel := context.WithTimeout(ctx, 1*time.Second) + defer cancel() + for k, v := range s.HealthProbe(ctx) { + checks[k] = v + } + } + ok := true + for _, v := range checks { + if v == "down" { + ok = false + } + } + uptime := int64(0) + if !s.Started.IsZero() { + uptime = int64(time.Since(s.Started).Seconds()) + } + return HealthResponse{ + OK: ok, + Service: "hub", + Version: "1", + UptimeS: uptime, + Checks: checks, + } +} diff --git a/src/hub/hub.go b/src/hub/hub.go new file mode 100644 index 0000000..758eaab --- /dev/null +++ b/src/hub/hub.go @@ -0,0 +1,282 @@ +// Package hub wires the five steps: hear, understand, check, route, answer. +// +// The HTTP layer in main.go stays thin. This package holds the pipeline so it +// can be tested against a fake project without a network. +package hub + +import ( + "bytes" + "context" + "encoding/json" + "strings" + "sync" + "time" + + "hub/answer" + "hub/apierr" + "hub/config" + "hub/hear" + "hub/logbook" + "hub/route" + "hub/understand" +) + +// Service is the wired Hub. +type Service struct { + Config *config.Config + Rules *understand.Rules + Router understand.Decider + Dispatcher *route.Dispatcher + Approvals *Approvals + Log *logbook.Log + Speaker route.Announcer + Paperclip route.JobFiler + Started time.Time + HealthProbe func(ctx context.Context) map[string]string + + capMu sync.Mutex + capTopics []string + capFetched time.Time +} + +// CapabilityTTL is how long the Hub keeps a project capability list. +const CapabilityTTL = 30 * time.Second + +// Ask runs the five steps for one POST /ask request. It never returns an error +// for a routing fault: a fault becomes a status in the reply. +func (s *Service) Ask(ctx context.Context, req hear.Request) answer.Response { + var outcome route.Outcome + + // Step 2a: the never-route check. No LLM runs on a control action. + if understand.IsControlAction(req.Text) { + outcome = route.Refused() + } else { + decision, err := s.Router.Decide(ctx, understand.Input{ + Question: req.Text, + Person: req.PersonOrDefault(), + Topics: s.capabilityTopics(ctx), + Rules: s.Rules, + }) + if err != nil { + outcome = route.Outcome{ + Kind: route.KindNoMatch, + Status: route.StatusUnavailable, + Answer: "I cannot work out where to send that.", + Reason: err.Error(), + } + } else { + outcome = s.Dispatcher.Dispatch(ctx, decision, req) + } + } + + resp := answer.Build(req, outcome) + + // Answer step: speak a house reply. House TTS is out of band; the Hub + // starts it and does not wait for the speakers (docs/hub-api.md section 8). + if !outcome.Spoken && answer.ShouldSpeak(resp) && s.Speaker != nil { + text := resp.Speech.Text + speak := s.Speaker + go func() { + _ = speak.Announce(context.Background(), text) + }() + } + return resp +} + +// capabilityTopics reads the live capability topics from every project, with a +// short cache. A project that is down contributes nothing and does not block. +func (s *Service) capabilityTopics(ctx context.Context) []string { + if s.Dispatcher == nil || len(s.Dispatcher.Projects) == 0 { + return nil + } + s.capMu.Lock() + if time.Since(s.capFetched) < CapabilityTTL && s.capTopics != nil { + topics := s.capTopics + s.capMu.Unlock() + return topics + } + s.capMu.Unlock() + + type result struct { + topics []string + } + results := make([]result, 0, len(s.Dispatcher.Projects)) + var mu sync.Mutex + var wg sync.WaitGroup + + for _, client := range s.Dispatcher.Projects { + wg.Add(1) + go func(c route.ProjectQuerier) { + defer wg.Done() + cctx, cancel := context.WithTimeout(ctx, 1*time.Second) + defer cancel() + topics, err := c.Capabilities(cctx) + if err != nil { + return + } + mu.Lock() + results = append(results, result{topics: topics}) + mu.Unlock() + }(client) + } + wg.Wait() + + var topics []string + seen := map[string]bool{} + for _, r := range results { + for _, t := range r.topics { + t = strings.ToLower(strings.TrimSpace(t)) + if t == "" || seen[t] { + continue + } + seen[t] = true + topics = append(topics, t) + } + } + + s.capMu.Lock() + s.capTopics = topics + s.capFetched = time.Now() + s.capMu.Unlock() + return topics +} + +// LogAsk writes one log line for a completed request (docs/hub-api.md section +// 10). It holds no secret. +func (s *Service) LogAsk(req hear.Request, resp answer.Response, latency time.Duration) { + if s.Log == nil { + return + } + project := "" + if resp.Route != nil { + project = resp.Route.Project + } + routeText := string(resp.Shape) + if resp.Route != nil { + routeText = resp.Route.Kind + if project != "" { + routeText += ":" + project + } + } + _ = s.Log.Write(logbook.Entry{ + Time: time.Now().UTC().Format(time.RFC3339), + RequestID: req.RequestID, + Source: string(req.Source), + Asker: req.Asker, + Text: req.Text, + Route: routeText, + Status: resp.Status, + LatencyMS: latency.Milliseconds(), + Shape: string(resp.Shape), + }) +} + +// ApproveRequest is the POST /approve body. +type ApproveRequest struct { + ApprovalID string `json:"approval_id"` + Decision string `json:"decision"` + Asker string `json:"asker"` + Device string `json:"device"` +} + +// ApproveResult is the result block. +type ApproveResult struct { + Kind string `json:"kind"` + Status string `json:"status"` + Detail string `json:"detail,omitempty"` +} + +// ApproveResponse is the POST /approve reply. +type ApproveResponse struct { + Status string `json:"status"` + ApprovalID string `json:"approval_id"` + Decision string `json:"decision"` + Result ApproveResult `json:"result"` + Answer string `json:"answer"` +} + +// ErrBadApprove is returned for a malformed /approve body. +func ParseApprove(body []byte) (ApproveRequest, *apierr.Error) { + var r ApproveRequest + if err := decodeJSON(body, &r); err != nil { + return ApproveRequest{}, apierr.New(400, "bad_request", "The request body is not valid JSON.") + } + r.ApprovalID = strings.TrimSpace(r.ApprovalID) + r.Decision = strings.TrimSpace(r.Decision) + r.Asker = strings.TrimSpace(r.Asker) + if r.ApprovalID == "" { + return ApproveRequest{}, apierr.New(422, "unprocessable", "Field 'approval_id' is required.") + } + if r.Asker != "sam" { + return ApproveRequest{}, apierr.New(422, "unsupported_asker", "The Hub supports one asker, Sam.") + } + switch r.Decision { + case "approve", "deny": + default: + return ApproveRequest{}, apierr.New(400, "bad_request", "Field 'decision' must be approve or deny.") + } + return r, nil +} + +// Approve records one decision and runs the approved action. In v1 the only +// Hub-owned approval is a Paperclip job (docs/hub-api.md section 6). +func (s *Service) Approve(ctx context.Context, req ApproveRequest) (ApproveResponse, *apierr.Error) { + ap, err := s.Approvals.Decide(req.ApprovalID) + if err != nil { + switch err { + case ErrNoApproval: + return ApproveResponse{}, apierr.New(404, "not_found", "That approval does not exist.") + case ErrApprovalExpired: + return ApproveResponse{}, apierr.New(409, "approval_expired", "That approval expired. Ask again.") + case ErrAlreadyDecided: + return ApproveResponse{}, apierr.New(409, "already_decided", "That approval was already decided.") + default: + return ApproveResponse{}, apierr.New(500, "internal", "The Hub had an internal problem.") + } + } + + resp := ApproveResponse{ + Status: "ok", + ApprovalID: ap.ID, + Decision: req.Decision, + Result: ApproveResult{Kind: ap.Action}, + } + + if req.Decision == "deny" { + resp.Result.Status = "denied" + resp.Answer = "Fine, I did not file it." + return resp, nil + } + + switch ap.Action { + case "paperclip_job": + if s.Paperclip == nil { + resp.Result.Status = "failed" + resp.Result.Detail = "Paperclip is not connected." + resp.Answer = "I could not file the job." + return resp, nil + } + _, err := s.Paperclip.FileJob(ctx, ap.Summary, "Filed by the Hub after an admin approval.") + if err != nil { + resp.Result.Status = "failed" + resp.Result.Detail = "The job did not file." + resp.Answer = "I could not file the job." + return resp, nil + } + resp.Result.Status = "filed" + resp.Result.Detail = "Issue filed." + resp.Answer = "Filed the job." + return resp, nil + default: + resp.Result.Status = "failed" + resp.Result.Detail = "Unknown action." + resp.Answer = "I do not know how to run that." + return resp, nil + } +} + +// decodeJSON decodes one JSON object. Unknown fields are ignored so clients +// may add fields (docs/hub-api.md section 4). +func decodeJSON(body []byte, v any) error { + return json.NewDecoder(bytes.NewReader(body)).Decode(v) +} diff --git a/src/hub/hub_test.go b/src/hub/hub_test.go new file mode 100644 index 0000000..0e0c292 --- /dev/null +++ b/src/hub/hub_test.go @@ -0,0 +1,228 @@ +package hub + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "hub/config" + "hub/destinations" + "hub/hear" + "hub/route" + "hub/tests/fakeproject" + "hub/understand" +) + +type recordingDecider struct { + mu sync.Mutex + calls int + decide understand.Decision + err error + lastIn understand.Input +} + +func (d *recordingDecider) Decide(ctx context.Context, in understand.Input) (understand.Decision, error) { + d.mu.Lock() + defer d.mu.Unlock() + d.calls++ + d.lastIn = in + return d.decide, d.err +} + +type recordingSpeaker struct { + mu sync.Mutex + texts []string +} + +func (s *recordingSpeaker) Announce(ctx context.Context, text string) error { + s.mu.Lock() + defer s.mu.Unlock() + s.texts = append(s.texts, text) + return nil +} + +func (s *recordingSpeaker) all() []string { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]string, len(s.texts)) + copy(out, s.texts) + return out +} + +// waitFor polls until fn is true or the deadline passes. +func waitFor(t *testing.T, fn func() bool) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if fn() { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal("condition not met before the deadline") +} + +type fakePaperclip struct { + mu sync.Mutex + titles []string + err error +} + +func (p *fakePaperclip) FileJob(ctx context.Context, title, description string) (string, error) { + p.mu.Lock() + defer p.mu.Unlock() + p.titles = append(p.titles, title) + if p.err != nil { + return "", p.err + } + return "issue-1", nil +} + +func newTestService(t *testing.T, decider understand.Decider) (*Service, *fakeproject.Project, *recordingSpeaker, *fakePaperclip) { + t.Helper() + fp := &fakeproject.Project{Name: "compass_nhs", Topics: []string{"school", "homework", "timetable"}} + srv := fp.Server() + t.Cleanup(srv.Close) + + speaker := &recordingSpeaker{} + paperclip := &fakePaperclip{} + approvals := NewApprovals(time.Minute) + client := destinations.NewProjectClient("compass_nhs", srv.URL, "", time.Second) + + svc := &Service{ + Config: &config.Config{}, + Rules: understand.DefaultRules(), + Router: decider, + Dispatcher: &route.Dispatcher{ + Rules: understand.DefaultRules(), + Projects: map[string]route.ProjectQuerier{"compass_nhs": client}, + Speaker: speaker, + Paperclip: paperclip, + Approvals: approvals, + }, + Approvals: approvals, + Speaker: speaker, + Paperclip: paperclip, + Started: time.Now(), + } + return svc, fp, speaker, paperclip +} + +func TestAskRoutesThroughLLMToFakeProject(t *testing.T) { + dec := &recordingDecider{decide: understand.Decision{Action: understand.ActionRoute, Project: "compass_nhs"}} + svc, fp, speaker, _ := newTestService(t, dec) + + req := hear.Request{Text: "what does Harry have on today?", Asker: "sam", Source: hear.SourceESP32, RequestID: "r1"} + resp := svc.Ask(context.Background(), req) + + if resp.Status != "ok" { + t.Fatalf("status = %q", resp.Status) + } + if dec.calls != 1 { + t.Fatalf("LLM calls = %d, want 1", dec.calls) + } + if got := fp.Questions(); len(got) != 1 || got[0] != req.Text { + t.Fatalf("fake project questions = %v", got) + } + if resp.Shape != "speech" { + t.Errorf("shape = %q", resp.Shape) + } + waitFor(t, func() bool { return len(speaker.all()) == 1 }) + if texts := speaker.all(); len(texts) != 1 { + t.Errorf("speaker got %v, want one announcement", texts) + } +} + +func TestControlActionNeverReachesLLM(t *testing.T) { + dec := &recordingDecider{decide: understand.Decision{Action: understand.ActionNoMatch}} + svc, _, _, _ := newTestService(t, dec) + + req := hear.Request{Text: "cut the kids' internet", Asker: "sam", Source: hear.SourceESP32, RequestID: "r2"} + resp := svc.Ask(context.Background(), req) + + if resp.Status != "refused" { + t.Fatalf("status = %q, want refused", resp.Status) + } + if dec.calls != 0 { + t.Fatalf("LLM was called %d times for a control action, want 0", dec.calls) + } + if resp.Route == nil || resp.Route.Kind != "refused" || resp.Route.Reason != "high_consequence_control" { + t.Errorf("route = %#v", resp.Route) + } +} + +func TestAskProjectDownIsUnavailable(t *testing.T) { + dec := &recordingDecider{decide: understand.Decision{Action: understand.ActionRoute, Project: "compass_nhs"}} + svc, _, _, _ := newTestService(t, dec) + // Point the project client at a dead port. + svc.Dispatcher.Projects["compass_nhs"] = destinations.NewProjectClient("compass_nhs", "http://127.0.0.1:1", "", 300*time.Millisecond) + + req := hear.Request{Text: "hello", Asker: "sam", Source: hear.SourceChat, RequestID: "r3"} + resp := svc.Ask(context.Background(), req) + if resp.Status != "unavailable" { + t.Fatalf("status = %q, want unavailable", resp.Status) + } +} + +func TestAskLLMFailureIsUnavailable(t *testing.T) { + dec := &recordingDecider{err: errors.New("gateway down")} + svc, _, _, _ := newTestService(t, dec) + req := hear.Request{Text: "hello", Asker: "sam", Source: hear.SourceChat, RequestID: "r4"} + resp := svc.Ask(context.Background(), req) + if resp.Status != "unavailable" { + t.Fatalf("status = %q, want unavailable", resp.Status) + } +} + +func TestApproveFilesPaperclipJob(t *testing.T) { + dec := &recordingDecider{decide: understand.Decision{Action: understand.ActionRoute, Project: "paperclip"}} + svc, _, _, paperclip := newTestService(t, dec) + + // Ask first to create the approval. + ask := svc.Ask(context.Background(), hear.Request{Text: "plan the kitchen lights", Asker: "sam", Source: hear.SourceESP32, RequestID: "r5"}) + if ask.Status != "needs_approval" || ask.Approval == nil { + t.Fatalf("ask = %#v", ask) + } + + resp, apiErr := svc.Approve(context.Background(), ApproveRequest{ + ApprovalID: ask.Approval.ApprovalID, + Decision: "approve", + Asker: "sam", + }) + if apiErr != nil { + t.Fatalf("approve: %v", apiErr) + } + if resp.Result.Status != "filed" { + t.Fatalf("result = %#v", resp.Result) + } + if len(paperclip.titles) != 1 { + t.Fatalf("paperclip titles = %v", paperclip.titles) + } + + // A second decision must fail. + if _, apiErr := svc.Approve(context.Background(), ApproveRequest{ApprovalID: ask.Approval.ApprovalID, Decision: "approve", Asker: "sam"}); apiErr == nil || apiErr.Code != "already_decided" { + t.Fatalf("second approve = %#v", apiErr) + } +} + +func TestApproveUnknownID(t *testing.T) { + svc, _, _, _ := newTestService(t, &recordingDecider{}) + _, apiErr := svc.Approve(context.Background(), ApproveRequest{ApprovalID: "ap_missing", Decision: "approve", Asker: "sam"}) + if apiErr == nil || apiErr.Code != "not_found" { + t.Fatalf("apiErr = %#v", apiErr) + } +} + +func TestApproveDenyDoesNotFile(t *testing.T) { + svc, _, _, paperclip := newTestService(t, &recordingDecider{decide: understand.Decision{Action: understand.ActionRoute, Project: "paperclip"}}) + ask := svc.Ask(context.Background(), hear.Request{Text: "plan this", Asker: "sam", Source: hear.SourceESP32, RequestID: "r6"}) + resp, apiErr := svc.Approve(context.Background(), ApproveRequest{ApprovalID: ask.Approval.ApprovalID, Decision: "deny", Asker: "sam"}) + if apiErr != nil { + t.Fatalf("approve: %v", apiErr) + } + if resp.Result.Status != "denied" || len(paperclip.titles) != 0 { + t.Fatalf("resp=%#v titles=%v", resp.Result, paperclip.titles) + } +} diff --git a/src/ids/ids.go b/src/ids/ids.go new file mode 100644 index 0000000..af7eb57 --- /dev/null +++ b/src/ids/ids.go @@ -0,0 +1,87 @@ +// Package ids makes the request and approval identifiers from docs/hub-api.md. +// The shape is a ULID: 26 Crockford base32 characters, time-ordered. It needs +// no dependency, so the Hub stays a small static binary. +package ids + +import ( + "crypto/rand" + "sync" + "time" +) + +// Crockford base32 alphabet. ULID uses these 32 symbols. +const alphabet = "0123456789ABCDEFGHJKMNPQRSTVWXYZ" + +var ( + mu sync.Mutex + lastMS uint64 + lastRnd [10]byte +) + +// New returns a 26-character ULID string. The first 10 characters hold the +// time. The last 16 hold randomness. Two calls in the same millisecond do not +// repeat. +func New() string { + mu.Lock() + defer mu.Unlock() + + ms := uint64(time.Now().UnixMilli()) + if ms == lastMS { + // Same millisecond: increment the random part. + for i := len(lastRnd) - 1; i >= 0; i-- { + lastRnd[i]++ + if lastRnd[i] != 0 { + break + } + } + } else { + lastMS = ms + if _, err := rand.Read(lastRnd[:]); err != nil { + // crypto/rand does not fail in practice. Fall back to the clock. + for i := range lastRnd { + lastRnd[i] = byte(ms >> (8 * (i % 8))) + } + } + } + + return encodeTime(ms) + encodeRandom(lastRnd) +} + +// Approval returns an approval id, for example "ap_01J8Z...". +func Approval() string { + return "ap_" + New() +} + +// encodeTime writes 10 base32 characters from the low 50 bits of ms. +func encodeTime(ms uint64) string { + var out [10]byte + for i := 9; i >= 0; i-- { + out[i] = alphabet[ms&0x1f] + ms >>= 5 + } + return string(out[:]) +} + +// encodeRandom writes 16 base32 characters from 10 random bytes (80 bits). +func encodeRandom(rnd [10]byte) string { + var out [16]byte + var hi, lo uint64 + hi = uint64(rnd[0])<<8 | uint64(rnd[1]) + hi = hi<<8 | uint64(rnd[2]) + hi = hi<<8 | uint64(rnd[3]) + hi = hi<<8 | uint64(rnd[4]) + lo = uint64(rnd[5])<<8 | uint64(rnd[6]) + lo = lo<<8 | uint64(rnd[7]) + lo = lo<<8 | uint64(rnd[8]) + lo = lo<<8 | uint64(rnd[9]) + + for i := 15; i >= 8; i-- { + out[i] = alphabet[lo&0x1f] + lo >>= 5 + } + for i := 7; i >= 0; i-- { + out[i] = alphabet[hi&0x1f] + hi >>= 5 + } + return string(out[:]) +} diff --git a/src/logbook/logbook.go b/src/logbook/logbook.go new file mode 100644 index 0000000..03256ab --- /dev/null +++ b/src/logbook/logbook.go @@ -0,0 +1,151 @@ +// Package logbook writes the Hub request log (docs/hub-api.md section 10). +// +// The log is append only. It rotates at a fixed size and keeps a fixed count of +// files. It holds no token and no project data. +package logbook + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "sync" +) + +// Entry is one log line. The field order matches the table in docs/hub-api.md +// section 10. +type Entry struct { + Time string `json:"time"` + RequestID string `json:"request_id"` + Source string `json:"source"` + Asker string `json:"asker"` + Text string `json:"text"` + Route string `json:"route"` + Status string `json:"status"` + LatencyMS int64 `json:"latency_ms"` + Shape string `json:"shape"` +} + +// Log is an append-only, size-capped log file. +type Log struct { + path string + maxBytes int64 + keep int + + mu sync.Mutex + file *os.File + bytes int64 +} + +// New opens the log. A blank path makes a no-op log, which tests use. +func New(path string, maxBytes int64, keep int) (*Log, error) { + l := &Log{path: path, maxBytes: maxBytes, keep: keep} + if keep <= 0 { + l.keep = 7 + } + if path == "" { + return l, nil + } + if maxBytes <= 0 { + l.maxBytes = 1 << 20 + } + if err := l.open(); err != nil { + return nil, err + } + return l, nil +} + +// open opens the file for append and records its current size. +func (l *Log) open() error { + f, err := os.OpenFile(l.path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o640) + if err != nil { + return err + } + info, err := f.Stat() + if err != nil { + f.Close() + return err + } + l.file = f + l.bytes = info.Size() + return nil +} + +// Write appends one entry as a JSON line. +func (l *Log) Write(e Entry) error { + if l == nil || l.path == "" { + return nil + } + line, err := json.Marshal(e) + if err != nil { + return err + } + line = append(line, '\n') + + l.mu.Lock() + defer l.mu.Unlock() + + if l.file == nil { + if err := l.open(); err != nil { + return err + } + } + if l.maxBytes > 0 && l.bytes+int64(len(line)) > l.maxBytes { + if err := l.rotate(); err != nil { + return err + } + } + n, err := l.file.Write(line) + l.bytes += int64(n) + return err +} + +// rotate closes the current file and shifts the backups. It keeps at most +// "keep" old files. +func (l *Log) rotate() error { + if l.file != nil { + l.file.Close() + l.file = nil + } + // Drop the oldest backup. + oldest := fmt.Sprintf("%s.%d", l.path, l.keep) + _ = os.Remove(oldest) + for i := l.keep - 1; i >= 1; i-- { + from := fmt.Sprintf("%s.%d", l.path, i) + to := fmt.Sprintf("%s.%d", l.path, i+1) + if _, err := os.Stat(from); err == nil { + _ = os.Rename(from, to) + } + } + _ = os.Rename(l.path, l.path+".1") + l.bytes = 0 + return l.open() +} + +// Close closes the log. +func (l *Log) Close() error { + if l == nil || l.file == nil { + return nil + } + l.mu.Lock() + defer l.mu.Unlock() + err := l.file.Close() + l.file = nil + return err +} + +// Path returns the log file path. +func (l *Log) Path() string { + if l == nil { + return "" + } + return l.path +} + +// Dir returns the directory that holds the log. +func (l *Log) Dir() string { + if l == nil || l.path == "" { + return "" + } + return filepath.Dir(l.path) +} diff --git a/src/main.go b/src/main.go index 7dbf630..27605ff 100644 --- a/src/main.go +++ b/src/main.go @@ -1,51 +1,294 @@ // Command hub is the Central AI Hub service. // -// Skeleton (AUD-48): it serves GET /health only. POST /ask and POST /approve -// arrive in AUD-49. The wire format is frozen in docs/hub-api.md. +// It serves the three frozen inbound routes from docs/hub-api.md: +// +// POST /ask — hear, understand, check, route, answer +// POST /approve — decide one Hub-owned approval +// GET /health — liveness only +// +// The pipeline lives in package hub. This file is the HTTP layer only. package main import ( + "context" "encoding/json" + "errors" "flag" "fmt" + "io" "log" "net/http" "os" + "strings" + "sync" "time" + + "hub/apierr" + "hub/check" + "hub/config" + "hub/destinations" + "hub/hear" + "hub/hub" + "hub/logbook" + "hub/route" + "hub/understand" ) const ( serviceName = "hub" serviceVersion = "1" - defaultPort = "8788" + + maxBodyBytes = 64 * 1024 ) -var started = time.Now() - -type healthResponse struct { - OK bool `json:"ok"` - Service string `json:"service"` - Version string `json:"version"` - UptimeS int64 `json:"uptime_s"` - Checks map[string]string `json:"checks"` +// server is the HTTP surface. +type server struct { + svc *hub.Service + cfg *config.Config + rate *rateLimiter } -type errorBody struct { - Status string `json:"status"` - Error errorInfo `json:"error"` -} +func main() { + hc := flag.Bool("healthcheck", false, "Call this process's /health and exit.") + flag.Parse() -type errorInfo struct { - Code string `json:"code"` - Message string `json:"message"` -} - -// port reads HUB_PORT. It falls back to 8788. -func port() string { - if p := os.Getenv("HUB_PORT"); p != "" { - return p + cfg := config.Load() + if *hc { + healthcheck(cfg) + return } - return defaultPort + + rules, err := understand.LoadRules(cfg.RoutingFile) + if err != nil { + log.Printf("routing: using built-in rules (%v)", err) + } + + lb, err := logbook.New(cfg.LogPath, cfg.LogMaxBytes, cfg.LogKeepDays) + if err != nil { + log.Fatalf("cannot open log %s: %v", cfg.LogPath, err) + } + defer lb.Close() + + projects := map[string]route.ProjectQuerier{} + for name, p := range cfg.Projects { + projects[name] = destinations.NewProjectClient(name, p.URL, p.Token, cfg.ProjectTimeout) + } + speaker := destinations.NewSpeaker(cfg.JervisURL, cfg.TTSTimeout) + paperclip := destinations.NewPaperclip(cfg.PaperclipURL, cfg.PaperclipKey, cfg.PaperclipCompanyID, cfg.ApproveTimeout) + approvals := hub.NewApprovals(10 * time.Minute) + + dispatcher := &route.Dispatcher{ + Rules: rules, + Projects: projects, + Speaker: speaker, + Paperclip: paperclip, + Approvals: approvals, + } + svc := &hub.Service{ + Config: cfg, + Rules: rules, + Router: understand.NewOmniRouter(cfg.OmniURL, cfg.OmniKey, cfg.OmniModel, cfg.LLMTimeout), + Dispatcher: dispatcher, + Approvals: approvals, + Log: lb, + Speaker: speaker, + Paperclip: paperclip, + Started: time.Now(), + HealthProbe: omniProbe(cfg), + } + + s := &server{svc: svc, cfg: cfg, rate: newRateLimiter(cfg.RatePerMin)} + + srv := &http.Server{ + Addr: ":" + cfg.Port, + Handler: s.routes(), + ReadHeaderTimeout: 1 * time.Second, + ReadTimeout: cfg.AskTimeout, + // WriteTimeout must be longer than the /ask deadline, or the server can + // cut the reply off when the handler finishes at the deadline. + WriteTimeout: cfg.AskTimeout + 3*time.Second, + IdleTimeout: 60 * time.Second, + } + + log.Printf("%s %s listening on :%s", serviceName, serviceVersion, cfg.Port) + if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + log.Fatalf("%s stopped: %v", serviceName, err) + } +} + +// routes builds the HTTP multiplexer. +func (s *server) routes() *http.ServeMux { + mux := http.NewServeMux() + mux.HandleFunc("/ask", s.handleAsk) + mux.HandleFunc("/approve", s.handleApprove) + mux.HandleFunc("/health", s.handleHealth) + mux.HandleFunc("/", notFound) + return mux +} + +// handleAsk serves POST /ask. +func (s *server) handleAsk(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeError(w, http.StatusMethodNotAllowed, "", "bad_request", "Use POST for this route.") + return + } + if !requireJSON(w, r) { + return + } + source, ok := s.auth(w, r) + if !ok { + return + } + if ok, retry := s.rate.allow(source); !ok { + w.Header().Set("Retry-After", fmt.Sprintf("%d", retry)) + writeError(w, http.StatusTooManyRequests, "", "rate_limited", "Too many requests. Try again shortly.") + return + } + + body, ok := readBody(w, r) + if !ok { + return + } + req, apiErr := hear.Parse(body) + if apiErr != nil { + writeAPIError(w, apiErr, "") + return + } + if apiErr := check.Ask(req, source); apiErr != nil { + writeAPIError(w, apiErr, req.RequestID) + return + } + + start := time.Now() + ctx, cancel := context.WithTimeout(r.Context(), s.cfg.AskTimeout) + defer cancel() + + resp := s.svc.Ask(ctx, req) + s.svc.LogAsk(req, resp, time.Since(start)) + writeJSON(w, http.StatusOK, resp) +} + +// handleApprove serves POST /approve. Only HUB_TOKEN_ADMIN works. +func (s *server) handleApprove(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeError(w, http.StatusMethodNotAllowed, "", "bad_request", "Use POST for this route.") + return + } + if !requireJSON(w, r) { + return + } + source, ok := s.auth(w, r) + if !ok { + return + } + if ok, retry := s.rate.allow(source); !ok { + w.Header().Set("Retry-After", fmt.Sprintf("%d", retry)) + writeError(w, http.StatusTooManyRequests, "", "rate_limited", "Too many requests. Try again shortly.") + return + } + if apiErr := check.Admin(source); apiErr != nil { + writeAPIError(w, apiErr, "") + return + } + + body, ok := readBody(w, r) + if !ok { + return + } + req, apiErr := hub.ParseApprove(body) + if apiErr != nil { + writeAPIError(w, apiErr, "") + return + } + + ctx, cancel := context.WithTimeout(r.Context(), s.cfg.ApproveTimeout) + defer cancel() + + resp, apiErr := s.svc.Approve(ctx, req) + if apiErr != nil { + writeAPIError(w, apiErr, "") + return + } + writeJSON(w, http.StatusOK, resp) +} + +// handleHealth serves GET /health. No authentication. +func (s *server) handleHealth(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + writeError(w, http.StatusMethodNotAllowed, "", "bad_request", "Use GET for this route.") + return + } + ctx, cancel := context.WithTimeout(r.Context(), s.cfg.HealthTimeout) + defer cancel() + + resp := s.svc.Health(ctx) + code := http.StatusOK + if !resp.OK { + code = http.StatusServiceUnavailable + } + writeJSON(w, code, resp) +} + +// notFound answers an unknown path with the frozen error envelope. +func notFound(w http.ResponseWriter, r *http.Request) { + writeError(w, http.StatusNotFound, "", "not_found", "Unknown path.") +} + +// auth reads the bearer token and returns the client source. It writes the 401 +// reply itself when the token is missing or unknown. +func (s *server) auth(w http.ResponseWriter, r *http.Request) (string, bool) { + token := bearer(r) + if token == "" { + writeError(w, http.StatusUnauthorized, "", "unauthorized", "A bearer token is required.") + return "", false + } + source, ok := s.cfg.SourceForToken(token) + if !ok { + writeError(w, http.StatusUnauthorized, "", "unauthorized", "The bearer token is not valid.") + return "", false + } + return source, true +} + +// bearer reads the token from the Authorization header. +func bearer(r *http.Request) string { + h := r.Header.Get("Authorization") + const prefix = "Bearer " + if len(h) <= len(prefix) || !strings.EqualFold(h[:len(prefix)], prefix) { + return "" + } + return strings.TrimSpace(h[len(prefix):]) +} + +// requireJSON checks the content type and returns false after writing 415. +func requireJSON(w http.ResponseWriter, r *http.Request) bool { + ct := r.Header.Get("Content-Type") + if ct == "" { + writeError(w, http.StatusUnsupportedMediaType, "", "unsupported_media_type", "Send the body as application/json.") + return false + } + mediaType := strings.ToLower(strings.TrimSpace(strings.Split(ct, ";")[0])) + if mediaType != "application/json" { + writeError(w, http.StatusUnsupportedMediaType, "", "unsupported_media_type", "Send the body as application/json.") + return false + } + return true +} + +// readBody reads at most 64 KiB. A larger body gets 413. +func readBody(w http.ResponseWriter, r *http.Request) ([]byte, bool) { + r.Body = http.MaxBytesReader(w, r.Body, maxBodyBytes) + body, err := io.ReadAll(r.Body) + if err != nil { + var tooBig *http.MaxBytesError + if errors.As(err, &tooBig) { + writeError(w, http.StatusRequestEntityTooLarge, "", "payload_too_large", "The request body is over 64 KiB.") + return nil, false + } + writeError(w, http.StatusBadRequest, "", "bad_request", "The request body is not valid JSON.") + return nil, false + } + return body, true } func writeJSON(w http.ResponseWriter, code int, v any) { @@ -54,42 +297,99 @@ func writeJSON(w http.ResponseWriter, code int, v any) { _ = json.NewEncoder(w).Encode(v) } -func writeError(w http.ResponseWriter, code int, errCode, message string) { +type errorBody struct { + Status string `json:"status"` + Error errorInfo `json:"error"` +} + +type errorInfo struct { + Code string `json:"code"` + Message string `json:"message"` + RequestID string `json:"request_id,omitempty"` +} + +func writeError(w http.ResponseWriter, code int, requestID, errCode, message string) { writeJSON(w, code, errorBody{ Status: "error", - Error: errorInfo{Code: errCode, Message: message}, + Error: errorInfo{Code: errCode, Message: message, RequestID: requestID}, }) } -// handleHealth is GET /health. Liveness only. No authentication. -func handleHealth(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - writeError(w, http.StatusMethodNotAllowed, "bad_request", "Use GET for this route.") - return +// writeAPIError writes an *apierr.Error with the frozen envelope. +func writeAPIError(w http.ResponseWriter, e *apierr.Error, requestID string) { + writeError(w, e.HTTP, requestID, e.Code, e.Message) +} + +// rateLimiter is a fixed-window cap per client token (docs/hub-api.md section +// 4). It protects the Hub and is not a security control. +type rateLimiter struct { + limit int + mu sync.Mutex + win map[string]*rateWindow +} + +type rateWindow struct { + start time.Time + count int +} + +func newRateLimiter(limit int) *rateLimiter { + if limit <= 0 { + limit = 30 } - // OmniRoute and MQTT are wired in AUD-49, so they read "unknown" here. - writeJSON(w, http.StatusOK, healthResponse{ - OK: true, - Service: serviceName, - Version: serviceVersion, - UptimeS: int64(time.Since(started).Seconds()), - Checks: map[string]string{ - "omniroute": "unknown", - "mqtt": "unknown", - "log": "ok", - }, - }) + return &rateLimiter{limit: limit, win: map[string]*rateWindow{}} } -// handleNotFound answers an unknown path with the frozen error envelope. -func handleNotFound(w http.ResponseWriter, r *http.Request) { - writeError(w, http.StatusNotFound, "not_found", "Unknown path.") +// allow reports whether the key may proceed. It returns the seconds to wait +// when it may not. +func (rl *rateLimiter) allow(key string) (bool, int) { + rl.mu.Lock() + defer rl.mu.Unlock() + + now := time.Now() + w, ok := rl.win[key] + if !ok || now.Sub(w.start) >= time.Minute { + rl.win[key] = &rateWindow{start: now, count: 1} + return true, 0 + } + if w.count >= rl.limit { + retry := int(time.Until(w.start.Add(time.Minute)).Seconds()) + if retry < 1 { + retry = 1 + } + return false, retry + } + w.count++ + return true, 0 } -// healthcheck calls this process's own /health. It is the container health probe, -// because the runtime image has no shell and no curl. -func healthcheck() { - url := fmt.Sprintf("http://127.0.0.1:%s/health", port()) +// omniProbe checks OmniRoute for GET /health. It never blocks over 1 second. +func omniProbe(cfg *config.Config) func(ctx context.Context) map[string]string { + return func(ctx context.Context) map[string]string { + if cfg.OmniKey == "" { + return map[string]string{"omniroute": "unknown"} + } + req, err := http.NewRequestWithContext(ctx, http.MethodGet, cfg.OmniURL+"/v1/models", nil) + if err != nil { + return map[string]string{"omniroute": "down"} + } + req.Header.Set("Authorization", "Bearer "+cfg.OmniKey) + resp, err := http.DefaultClient.Do(req) + if err != nil { + return map[string]string{"omniroute": "down"} + } + defer resp.Body.Close() + if resp.StatusCode == http.StatusOK { + return map[string]string{"omniroute": "ok"} + } + return map[string]string{"omniroute": "down"} + } +} + +// healthcheck calls this process's own /health. It is the container health +// probe, because the runtime image has no shell and no curl. +func healthcheck(cfg *config.Config) { + url := fmt.Sprintf("http://127.0.0.1:%s/health", cfg.Port) client := &http.Client{Timeout: 1 * time.Second} resp, err := client.Get(url) if err != nil { @@ -103,30 +403,3 @@ func healthcheck() { } os.Exit(0) } - -func main() { - hc := flag.Bool("healthcheck", false, "Call this process's /health and exit.") - flag.Parse() - if *hc { - healthcheck() - return - } - - mux := http.NewServeMux() - mux.HandleFunc("/health", handleHealth) - mux.HandleFunc("/", handleNotFound) - - srv := &http.Server{ - Addr: ":" + port(), - Handler: mux, - ReadHeaderTimeout: 1 * time.Second, - ReadTimeout: 1 * time.Second, - WriteTimeout: 1 * time.Second, - IdleTimeout: 60 * time.Second, - } - - log.Printf("%s %s listening on :%s", serviceName, serviceVersion, port()) - if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { - log.Fatalf("%s stopped: %v", serviceName, err) - } -} diff --git a/src/main_test.go b/src/main_test.go new file mode 100644 index 0000000..e87de10 --- /dev/null +++ b/src/main_test.go @@ -0,0 +1,258 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "hub/config" + "hub/destinations" + "hub/hub" + "hub/route" + "hub/tests/fakeproject" + "hub/understand" +) + +type testSpeaker struct { + mu sync.Mutex + texts []string +} + +func (s *testSpeaker) Announce(ctx context.Context, text string) error { + s.mu.Lock() + defer s.mu.Unlock() + s.texts = append(s.texts, text) + return nil +} + +func (s *testSpeaker) all() []string { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]string, len(s.texts)) + copy(out, s.texts) + return out +} + +// waitFor polls until fn is true or the deadline passes. +func waitFor(t *testing.T, fn func() bool) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if fn() { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal("condition not met before the deadline") +} + +type testPaperclip struct{ titles []string } + +func (p *testPaperclip) FileJob(ctx context.Context, title, description string) (string, error) { + p.titles = append(p.titles, title) + return "issue-1", nil +} + +// newTestServer builds the full HTTP surface against a fake project. +func newTestServer(t *testing.T, decider understand.Decider) (*httptest.Server, *fakeproject.Project, *testSpeaker) { + t.Helper() + + fp := &fakeproject.Project{Name: "compass_nhs", Topics: []string{"school"}} + projectSrv := fp.Server() + t.Cleanup(projectSrv.Close) + + speaker := &testSpeaker{} + paperclip := &testPaperclip{} + approvals := hub.NewApprovals(time.Minute) + + cfg := &config.Config{ + Tokens: map[string]string{ + "phone": "phone-token", + "esp32": "esp32-token", + "chat": "chat-token", + "admin": "admin-token", + }, + AskTimeout: 5 * time.Second, + ApproveTimeout: 5 * time.Second, + HealthTimeout: time.Second, + RatePerMin: 1000, + } + + dispatcher := &route.Dispatcher{ + Rules: understand.DefaultRules(), + Projects: map[string]route.ProjectQuerier{"compass_nhs": destinations.NewProjectClient("compass_nhs", projectSrv.URL, "", time.Second)}, + Speaker: speaker, + Paperclip: paperclip, + Approvals: approvals, + } + svc := &hub.Service{ + Config: cfg, + Rules: understand.DefaultRules(), + Router: decider, + Dispatcher: dispatcher, + Approvals: approvals, + Speaker: speaker, + Paperclip: paperclip, + Started: time.Now(), + } + s := &server{svc: svc, cfg: cfg, rate: newRateLimiter(cfg.RatePerMin)} + ts := httptest.NewServer(s.routes()) + t.Cleanup(ts.Close) + return ts, fp, speaker +} + +func doJSON(t *testing.T, method, url, token, contentType, body string) (int, map[string]any) { + t.Helper() + req, err := http.NewRequest(method, url, bytes.NewBufferString(body)) + if err != nil { + t.Fatal(err) + } + if token != "" { + req.Header.Set("Authorization", "Bearer "+token) + } + if contentType != "" { + req.Header.Set("Content-Type", contentType) + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + raw, _ := io.ReadAll(resp.Body) + var out map[string]any + _ = json.Unmarshal(raw, &out) + return resp.StatusCode, out +} + +func TestHealthRoute(t *testing.T) { + dec := understand.DeciderFunc(func(ctx context.Context, in understand.Input) (understand.Decision, error) { + return understand.Decision{Action: understand.ActionNoMatch}, nil + }) + ts, _, _ := newTestServer(t, dec) + + code, body := doJSON(t, http.MethodGet, ts.URL+"/health", "", "", "") + if code != http.StatusOK { + t.Fatalf("health code = %d", code) + } + if body["ok"] != true || body["service"] != "hub" { + t.Fatalf("health body = %#v", body) + } +} + +func TestAskRequiresToken(t *testing.T) { + dec := understand.DeciderFunc(func(ctx context.Context, in understand.Input) (understand.Decision, error) { + return understand.Decision{Action: understand.ActionNoMatch}, nil + }) + ts, _, _ := newTestServer(t, dec) + + code, body := doJSON(t, http.MethodPost, ts.URL+"/ask", "", "application/json", `{"text":"hi","asker":"sam","source":"esp32"}`) + if code != http.StatusUnauthorized { + t.Fatalf("code = %d body=%#v", code, body) + } +} + +func TestAskRoundTripToFakeProject(t *testing.T) { + dec := understand.DeciderFunc(func(ctx context.Context, in understand.Input) (understand.Decision, error) { + return understand.Decision{Action: understand.ActionRoute, Project: "compass_nhs"}, nil + }) + ts, fp, speaker := newTestServer(t, dec) + + code, body := doJSON(t, http.MethodPost, ts.URL+"/ask", "esp32-token", "application/json", + `{"text":"what does Harry have on today?","asker":"sam","source":"esp32","request_id":"01TEST"}`) + if code != http.StatusOK { + t.Fatalf("code = %d body = %#v", code, body) + } + if body["status"] != "ok" || body["shape"] != "speech" { + t.Fatalf("body = %#v", body) + } + if body["request_id"] != "01TEST" { + t.Errorf("request_id = %v", body["request_id"]) + } + if got := fp.Questions(); len(got) != 1 { + t.Fatalf("project questions = %v", got) + } + waitFor(t, func() bool { return len(speaker.all()) == 1 }) +} + +func TestAskSourceMismatch(t *testing.T) { + dec := understand.DeciderFunc(func(ctx context.Context, in understand.Input) (understand.Decision, error) { + return understand.Decision{Action: understand.ActionNoMatch}, nil + }) + ts, _, _ := newTestServer(t, dec) + + // The esp32 token claims source admin. + code, body := doJSON(t, http.MethodPost, ts.URL+"/ask", "esp32-token", "application/json", + `{"text":"hi","asker":"sam","source":"admin"}`) + if code != http.StatusForbidden { + t.Fatalf("code = %d body=%#v", code, body) + } +} + +func TestAskBadMediaType(t *testing.T) { + dec := understand.DeciderFunc(func(ctx context.Context, in understand.Input) (understand.Decision, error) { + return understand.Decision{Action: understand.ActionNoMatch}, nil + }) + ts, _, _ := newTestServer(t, dec) + code, _ := doJSON(t, http.MethodPost, ts.URL+"/ask", "esp32-token", "text/plain", `hi`) + if code != http.StatusUnsupportedMediaType { + t.Fatalf("code = %d", code) + } +} + +func TestApproveRoute(t *testing.T) { + dec := understand.DeciderFunc(func(ctx context.Context, in understand.Input) (understand.Decision, error) { + return understand.Decision{Action: understand.ActionRoute, Project: "paperclip"}, nil + }) + ts, _, _ := newTestServer(t, dec) + + // Ask to create an approval. + _, askBody := doJSON(t, http.MethodPost, ts.URL+"/ask", "esp32-token", "application/json", + `{"text":"plan the kitchen lights","asker":"sam","source":"esp32"}`) + approval, _ := askBody["approval"].(map[string]any) + if askBody["status"] != "needs_approval" || approval == nil { + t.Fatalf("ask body = %#v", askBody) + } + id, _ := approval["approval_id"].(string) + + // A non-admin token is refused. + code, _ := doJSON(t, http.MethodPost, ts.URL+"/approve", "chat-token", "application/json", + `{"approval_id":"`+id+`","decision":"approve","asker":"sam"}`) + if code != http.StatusForbidden { + t.Fatalf("non-admin approve code = %d", code) + } + + // The admin token files the job. + code, body := doJSON(t, http.MethodPost, ts.URL+"/approve", "admin-token", "application/json", + `{"approval_id":"`+id+`","decision":"approve","asker":"sam"}`) + if code != http.StatusOK { + t.Fatalf("admin approve code = %d body=%#v", code, body) + } + result, _ := body["result"].(map[string]any) + if result == nil || result["status"] != "filed" { + t.Fatalf("approve body = %#v", body) + } +} + +func TestControlActionRefused(t *testing.T) { + called := false + dec := understand.DeciderFunc(func(ctx context.Context, in understand.Input) (understand.Decision, error) { + called = true + return understand.Decision{Action: understand.ActionRoute, Project: "compass_nhs"}, nil + }) + ts, _, _ := newTestServer(t, dec) + + code, body := doJSON(t, http.MethodPost, ts.URL+"/ask", "esp32-token", "application/json", + `{"text":"cut the kids' internet","asker":"sam","source":"esp32"}`) + if code != http.StatusOK || body["status"] != "refused" { + t.Fatalf("code=%d body=%#v", code, body) + } + if called { + t.Fatal("the LLM decider was called for a control action") + } +} diff --git a/src/route/route.go b/src/route/route.go new file mode 100644 index 0000000..02347c0 --- /dev/null +++ b/src/route/route.go @@ -0,0 +1,317 @@ +// Package route is step 4 of the Hub: send the request to a destination. +// +// The action call holds no LLM (AGENTS.md rule 4). The routing decision is +// already made. This package only dispatches. +package route + +import ( + "context" + "strings" + "sync" + "time" + + "hub/destinations" + "hub/hear" + "hub/ids" + "hub/understand" +) + +// Kind is what the Hub did. It matches route.kind in docs/hub-api.md section 5. +type Kind string + +const ( + KindDirect Kind = "direct" + KindProject Kind = "project" + KindFanout Kind = "fanout" + KindPaperclip Kind = "paperclip" + KindClarify Kind = "clarify" + KindNoMatch Kind = "no_match" + KindRefused Kind = "refused" +) + +// Status is the top-level status in the reply. +type Status string + +const ( + StatusOK Status = "ok" + StatusNeedsClarification Status = "needs_clarification" + StatusNeedsApproval Status = "needs_approval" + StatusNoMatch Status = "no_match" + StatusRefused Status = "refused" + StatusUnavailable Status = "unavailable" +) + +// Approval is a pending Hub-owned approval. +type Approval struct { + ID string `json:"approval_id"` + Action string `json:"action"` + Summary string `json:"summary"` + Reason string `json:"reason"` + Expires time.Time `json:"expires"` +} + +// Clarify is one short question the Hub asks the user. +type Clarify struct { + Question string `json:"question"` + Options []string `json:"options,omitempty"` +} + +// Outcome is the result of step 4. +type Outcome struct { + Kind Kind + Status Status + Answer string + Project string + Projects []string + Expires string + Approval *Approval + Clarify *Clarify + Reason string + + // Spoken records that a direct destination already used the speakers, so + // the answer step must not announce again. + Spoken bool +} + +// ProjectQuerier is one data project. *destinations.ProjectClient implements it. +type ProjectQuerier interface { + Ask(ctx context.Context, body destinations.AskRequest) (destinations.AskReply, error) + Capabilities(ctx context.Context) ([]string, error) +} + +// Announcer starts house TTS. *destinations.Speaker implements it. +type Announcer interface { + Announce(ctx context.Context, text string) error +} + +// JobFiler files one Paperclip job. *destinations.Paperclip implements it. +type JobFiler interface { + FileJob(ctx context.Context, title, description string) (string, error) +} + +// ApprovalStore holds Hub-owned approvals. The hub package implements it. +type ApprovalStore interface { + Create(action, summary, reason string) *Approval +} + +// Dispatcher sends a decision to its destination. +type Dispatcher struct { + Rules *understand.Rules + Projects map[string]ProjectQuerier + Speaker Announcer + Paperclip JobFiler + Approvals ApprovalStore +} + +// Dispatch runs one decision. It never returns an error: a destination fault +// becomes status "unavailable", because one project down is not an outage +// (docs/hub-api.md section 5). +func (d *Dispatcher) Dispatch(ctx context.Context, decision understand.Decision, req hear.Request) Outcome { + switch decision.Action { + case understand.ActionRefused: + return Refused() + case understand.ActionClarify: + return Outcome{ + Kind: KindClarify, + Status: StatusNeedsClarification, + Answer: decision.Question, + Clarify: &Clarify{ + Question: decision.Question, + Options: decision.Options, + }, + } + case understand.ActionNoMatch: + return Outcome{Kind: KindNoMatch, Status: StatusNoMatch, Answer: "I do not know where to send that."} + case understand.ActionRoute: + return d.dispatchOne(ctx, decision.Project, req) + case understand.ActionFanout: + return d.dispatchFanout(ctx, decision.Projects, req) + default: + return Outcome{Kind: KindNoMatch, Status: StatusNoMatch, Answer: "I do not know where to send that."} + } +} + +// Refused is the fixed high-consequence control reply. +func Refused() Outcome { + return Outcome{ + Kind: KindRefused, + Status: StatusRefused, + Answer: "Use the admin app to change internet limits.", + Reason: "high_consequence_control", + } +} + +// dispatchOne sends to one destination. +func (d *Dispatcher) dispatchOne(ctx context.Context, project string, req hear.Request) Outcome { + kind := "project" + if d.Rules != nil { + kind = d.Rules.KindFor(project) + } + + switch kind { + case "direct": + return d.dispatchDirect(ctx, project, req) + case "paperclip": + return d.dispatchPaperclip(ctx, req) + case "peer", "way_in": + // A way in or a peer is not a destination the Hub routes to. + return Outcome{ + Kind: KindNoMatch, + Status: StatusNoMatch, + Answer: "That is a way in, not a place I send work.", + Reason: "not_a_destination", + } + default: + return d.askProject(ctx, project, req) + } +} + +// dispatchDirect calls a direct destination. Only the speakers exist in v1. +func (d *Dispatcher) dispatchDirect(ctx context.Context, project string, req hear.Request) Outcome { + switch project { + case "speakers": + if d.Speaker == nil { + return Outcome{Kind: KindDirect, Status: StatusUnavailable, Answer: "The speakers are not connected.", Project: project} + } + // House TTS is out of band. Start it and do not wait. + speak := d.Speaker + text := req.Text + go func() { _ = speak.Announce(context.Background(), text) }() + return Outcome{Kind: KindDirect, Status: StatusOK, Answer: req.Text, Project: project, Spoken: true} + default: + return Outcome{ + Kind: KindDirect, + Status: StatusUnavailable, + Answer: "That destination is not built yet.", + Project: project, + Reason: "not_built", + } + } +} + +// dispatchPaperclip creates an approval. It files nothing yet. +func (d *Dispatcher) dispatchPaperclip(ctx context.Context, req hear.Request) Outcome { + summary := "File a job: " + req.Text + if len(summary) > 200 { + summary = summary[:200] + } + var ap *Approval + if d.Approvals != nil { + ap = d.Approvals.Create("paperclip_job", summary, "A slow, multi-step job needs a yes.") + } else { + ap = &Approval{ + ID: ids.Approval(), + Action: "paperclip_job", + Summary: summary, + Reason: "A slow, multi-step job needs a yes.", + Expires: time.Now().Add(10 * time.Minute), + } + } + return Outcome{ + Kind: KindPaperclip, + Status: StatusNeedsApproval, + Answer: "I need a yes on the admin app to file that job.", + Approval: ap, + } +} + +// askProject forwards to one data project. +func (d *Dispatcher) askProject(ctx context.Context, project string, req hear.Request) Outcome { + client, ok := d.Projects[project] + if !ok || client == nil { + return Outcome{ + Kind: KindProject, + Status: StatusUnavailable, + Answer: "That service is not connected.", + Project: project, + Reason: "no_client", + } + } + reply, err := client.Ask(ctx, destinations.AskRequest{ + Question: req.Text, + Asker: req.Asker, + Person: req.PersonOrDefault(), + Source: string(req.Source), + }) + if err != nil { + return Outcome{ + Kind: KindProject, + Status: StatusUnavailable, + Answer: "That service is not answering.", + Project: project, + Reason: err.Error(), + } + } + if reply.Answer == nil || strings.TrimSpace(*reply.Answer) == "" { + reason := reply.Reason + if reason == "" { + reason = "no answer" + } + return Outcome{ + Kind: KindProject, + Status: StatusUnavailable, + Answer: "That service cannot answer that.", + Project: project, + Reason: reason, + } + } + return Outcome{ + Kind: KindProject, + Status: StatusOK, + Answer: strings.TrimSpace(*reply.Answer), + Project: project, + Expires: reply.Expires, + } +} + +// dispatchFanout asks up to three projects in parallel, then merges the answers +// into one short reply (routing.md section 2). +func (d *Dispatcher) dispatchFanout(ctx context.Context, projects []string, req hear.Request) Outcome { + if len(projects) == 0 { + return Outcome{Kind: KindFanout, Status: StatusNoMatch, Answer: "I could not work out where to ask."} + } + if len(projects) > 3 { + projects = projects[:3] + } + + type result struct { + project string + answer string + ok bool + } + results := make([]result, len(projects)) + var wg sync.WaitGroup + + for i, p := range projects { + wg.Add(1) + go func(i int, p string) { + defer wg.Done() + o := d.askProject(ctx, p, req) + results[i] = result{project: p, answer: o.Answer, ok: o.Status == StatusOK} + }(i, p) + } + wg.Wait() + + var answers []string + var used []string + for _, r := range results { + if r.ok { + answers = append(answers, r.answer) + used = append(used, r.project) + } + } + if len(answers) == 0 { + return Outcome{ + Kind: KindFanout, + Status: StatusUnavailable, + Answer: "Those services are not answering.", + Projects: projects, + } + } + return Outcome{ + Kind: KindFanout, + Status: StatusOK, + Answer: strings.Join(answers, " "), + Projects: used, + } +} diff --git a/src/route/route_test.go b/src/route/route_test.go new file mode 100644 index 0000000..dfca3b4 --- /dev/null +++ b/src/route/route_test.go @@ -0,0 +1,168 @@ +package route + +import ( + "context" + "sync" + "testing" + "time" + + "hub/destinations" + "hub/hear" + "hub/tests/fakeproject" + "hub/understand" +) + +// fakeSpeaker records announcements. +type fakeSpeaker struct { + mu sync.Mutex + texts []string + err error +} + +func (f *fakeSpeaker) Announce(ctx context.Context, text string) error { + if f.err != nil { + return f.err + } + f.mu.Lock() + defer f.mu.Unlock() + f.texts = append(f.texts, text) + return nil +} + +func (f *fakeSpeaker) all() []string { + f.mu.Lock() + defer f.mu.Unlock() + out := make([]string, len(f.texts)) + copy(out, f.texts) + return out +} + +// waitFor polls until fn is true or the deadline passes. +func waitFor(t *testing.T, fn func() bool) { + t.Helper() + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if fn() { + return + } + time.Sleep(5 * time.Millisecond) + } + t.Fatal("condition not met before the deadline") +} + +// memApprovals is a tiny approval store for tests. +type memApprovals struct{ last *Approval } + +func (m *memApprovals) Create(action, summary, reason string) *Approval { + m.last = &Approval{ID: "ap_test", Action: action, Summary: summary, Reason: reason, Expires: time.Now().Add(time.Minute)} + return m.last +} + +func newTestDispatcher(t *testing.T, project string) (*Dispatcher, *fakeproject.Project, *fakeSpeaker) { + t.Helper() + fp := &fakeproject.Project{Name: project, Topics: []string{"school"}} + srv := fp.Server() + t.Cleanup(srv.Close) + + client := destinations.NewProjectClient(project, srv.URL, "", 2*time.Second) + speaker := &fakeSpeaker{} + d := &Dispatcher{ + Rules: understand.DefaultRules(), + Projects: map[string]ProjectQuerier{project: client}, + Speaker: speaker, + Approvals: &memApprovals{}, + } + return d, fp, speaker +} + +func TestDispatchProjectRoundTrip(t *testing.T) { + d, fp, _ := newTestDispatcher(t, "compass_nhs") + req := hear.Request{Text: "what does Harry have on today?", Asker: "sam", Source: hear.SourceESP32, RequestID: "r1"} + + out := d.Dispatch(context.Background(), understand.Decision{Action: understand.ActionRoute, Project: "compass_nhs"}, req) + if out.Status != StatusOK { + t.Fatalf("status = %q (%s)", out.Status, out.Reason) + } + if out.Kind != KindProject || out.Project != "compass_nhs" { + t.Errorf("kind/project = %q/%q", out.Kind, out.Project) + } + if out.Answer == "" { + t.Errorf("empty answer") + } + if got := fp.Questions(); len(got) != 1 || got[0] != req.Text { + t.Errorf("project questions = %v", got) + } +} + +func TestDispatchProjectDown(t *testing.T) { + fp := &fakeproject.Project{Name: "compass_nhs"} + srv := fp.Server() + url := srv.URL + srv.Close() // project is down + client := destinations.NewProjectClient("compass_nhs", url, "", 500*time.Millisecond) + d := &Dispatcher{Rules: understand.DefaultRules(), Projects: map[string]ProjectQuerier{"compass_nhs": client}, Speaker: &fakeSpeaker{}} + + out := d.Dispatch(context.Background(), understand.Decision{Action: understand.ActionRoute, Project: "compass_nhs"}, hear.Request{Text: "hi"}) + if out.Status != StatusUnavailable { + t.Fatalf("status = %q, want unavailable", out.Status) + } +} + +func TestDispatchDirectSpeakers(t *testing.T) { + d, _, sp := newTestDispatcher(t, "compass_nhs") + out := d.Dispatch(context.Background(), understand.Decision{Action: understand.ActionRoute, Project: "speakers"}, hear.Request{Text: "goodnight"}) + if out.Status != StatusOK || !out.Spoken { + t.Fatalf("outcome = %#v", out) + } + waitFor(t, func() bool { return len(sp.all()) == 1 }) + if got := sp.all(); got[0] != "goodnight" { + t.Errorf("speaker texts = %v", got) + } +} + +func TestDispatchFanout(t *testing.T) { + fp := &fakeproject.Project{Name: "compass_nhs", Answers: map[string]string{"harry": "Harry has maths today."}} + srv := fp.Server() + t.Cleanup(srv.Close) + fp2 := &fakeproject.Project{Name: "email_agent", Answers: map[string]string{"harry": "You have three emails."}} + srv2 := fp2.Server() + t.Cleanup(srv2.Close) + + d := &Dispatcher{ + Rules: understand.DefaultRules(), + Projects: map[string]ProjectQuerier{ + "compass_nhs": destinations.NewProjectClient("compass_nhs", srv.URL, "", 2*time.Second), + "email_agent": destinations.NewProjectClient("email_agent", srv2.URL, "", 2*time.Second), + }, + Speaker: &fakeSpeaker{}, + } + out := d.Dispatch(context.Background(), understand.Decision{Action: understand.ActionFanout, Projects: []string{"compass_nhs", "email_agent"}}, hear.Request{Text: "what does Harry have on today?"}) + if out.Status != StatusOK { + t.Fatalf("status = %q (%s)", out.Status, out.Reason) + } + if len(out.Projects) != 2 { + t.Errorf("projects = %v", out.Projects) + } +} + +func TestDispatchPaperclipNeedsApproval(t *testing.T) { + ap := &memApprovals{} + d := &Dispatcher{Rules: understand.DefaultRules(), Projects: map[string]ProjectQuerier{}, Speaker: &fakeSpeaker{}, Approvals: ap} + out := d.Dispatch(context.Background(), understand.Decision{Action: understand.ActionRoute, Project: "paperclip"}, hear.Request{Text: "plan the kitchen lights"}) + if out.Status != StatusNeedsApproval || out.Approval == nil { + t.Fatalf("outcome = %#v", out) + } + if out.Approval.Action != "paperclip_job" { + t.Errorf("action = %q", out.Approval.Action) + } +} + +func TestRefused(t *testing.T) { + out := Refused() + if out.Status != StatusRefused || out.Kind != KindRefused { + t.Fatalf("outcome = %#v", out) + } + if out.Reason != "high_consequence_control" { + t.Errorf("reason = %q", out.Reason) + } +} diff --git a/src/tests/fakeproject/cmd/main.go b/src/tests/fakeproject/cmd/main.go new file mode 100644 index 0000000..6c2c0b4 --- /dev/null +++ b/src/tests/fakeproject/cmd/main.go @@ -0,0 +1,28 @@ +// Command fakeproject runs the fake /ask project on its own port. It is for a +// manual round-trip check with a running Hub. It has no real data. +// +// HUB_FAKE_PORT=9099 go run ./tests/fakeproject/cmd +package main + +import ( + "log" + "net/http" + "os" + + "hub/tests/fakeproject" +) + +func main() { + port := os.Getenv("HUB_FAKE_PORT") + if port == "" { + port = "9099" + } + p := &fakeproject.Project{ + Name: "compass_nhs", + Topics: []string{"school", "homework", "timetable"}, + } + log.Printf("fake /ask project listening on :%s", port) + if err := http.ListenAndServe(":"+port, p.Handler()); err != nil { + log.Fatal(err) + } +} diff --git a/src/tests/fakeproject/fakeproject.go b/src/tests/fakeproject/fakeproject.go new file mode 100644 index 0000000..7a7d31c --- /dev/null +++ b/src/tests/fakeproject/fakeproject.go @@ -0,0 +1,125 @@ +// Package fakeproject is a fake data project that speaks the /ask contract +// (docs/ask-contract.md). Tests use it to prove the Hub-to-project round trip +// without a real project. It has no real data and no state changes. +package fakeproject + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync" + "time" +) + +// DefaultAnswers maps a keyword in the question to a short answer. The first +// keyword found wins. +var DefaultAnswers = map[string]string{ + "harry": "Harry has maths and a music lesson today.", + "school": "School ends at 3pm today.", + "email": "You have three new emails.", + "inbox": "You have three new emails.", +} + +// Project is one fake /ask server. +type Project struct { + Name string + Topics []string + + // Answers maps a question keyword to a short answer. Nil uses + // DefaultAnswers. + Answers map[string]string + + // Fail makes /ask return HTTP 500. + Fail bool + // Delay waits before answering. + Delay time.Duration + + mu sync.Mutex + questions []string +} + +// Ask records that the Hub reached the project. +func (p *Project) Ask(question string) { + p.mu.Lock() + defer p.mu.Unlock() + p.questions = append(p.questions, question) +} + +// Questions returns the questions the Hub sent, in order. +func (p *Project) Questions() []string { + p.mu.Lock() + defer p.mu.Unlock() + out := make([]string, len(p.questions)) + copy(out, p.questions) + return out +} + +// Handler returns the three /ask-contract endpoints. +func (p *Project) Handler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) { + writeJSON(w, http.StatusOK, map[string]bool{"ok": true}) + }) + mux.HandleFunc("/capabilities", func(w http.ResponseWriter, r *http.Request) { + name := p.Name + if name == "" { + name = "fake_project" + } + writeJSON(w, http.StatusOK, map[string]any{ + "project": name, + "topics": p.Topics, + }) + }) + mux.HandleFunc("/ask", p.handleAsk) + return mux +} + +// Server starts the fake project on a random local port. Callers must Close it. +func (p *Project) Server() *httptest.Server { + return httptest.NewServer(p.Handler()) +} + +func (p *Project) handleAsk(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeJSON(w, http.StatusMethodNotAllowed, map[string]string{"answer": ""}) + return + } + var body struct { + Question string `json:"question"` + Asker string `json:"asker"` + Person string `json:"person"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + http.Error(w, "bad json", http.StatusBadRequest) + return + } + p.Ask(body.Question) + + if p.Delay > 0 { + time.Sleep(p.Delay) + } + if p.Fail { + http.Error(w, "fake failure", http.StatusInternalServerError) + return + } + + answers := p.Answers + if answers == nil { + answers = DefaultAnswers + } + q := strings.ToLower(body.Question) + for keyword, text := range answers { + if strings.Contains(q, strings.ToLower(keyword)) { + writeJSON(w, http.StatusOK, map[string]any{"answer": text}) + return + } + } + writeJSON(w, http.StatusOK, map[string]any{"answer": nil, "reason": "no data for that question"}) +} + +func writeJSON(w http.ResponseWriter, code int, v any) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(code) + _ = json.NewEncoder(w).Encode(v) +} diff --git a/src/understand/control.go b/src/understand/control.go new file mode 100644 index 0000000..12a071d --- /dev/null +++ b/src/understand/control.go @@ -0,0 +1,90 @@ +package understand + +import "strings" + +// ControlPhrases are the never-route actions from routing.md section 3. A match +// means the Hub refuses and points at the admin app. The check is deterministic +// and runs before any LLM call, so no model is ever in the path of a +// high-consequence control action (AGENTS.md rule 4). +var ControlPhrases = []string{ + "parental control", + "parental controls", + "cut the internet", + "cut internet", + "cut the kids", + "restore the internet", + "restore internet", + "turn off the internet", + "turn on the internet", + "disable the internet", + "disable internet", + "enable the internet", + "enable internet", + "block the internet", + "block internet", + "unblock the internet", + "unblock internet", + "internet limit", + "internet limits", + "time limit", + "time limits", + "daily allowance", + "screen time limit", + "screen time limits", + "screen time schedule", + "blocklist", + "block list", + "content filter", + "content filters", + "internet schedule", + "internet schedules", + "internet access", + "kids internet", + "kids' internet", + "child profile", + "child profiles", + "kids profile", + "device rule", + "device rules", + "bedtime internet", + "pause the internet", + "lock the device", + "unlock the device", +} + +// ControlVerbs are combined with the word "internet" to catch a control action +// the phrase list misses. +var ControlVerbs = []string{ + "cut", "restore", "block", "unblock", "pause", "limit", "allowance", + "schedule", "filter", "ban", "disable", "enable", "parental", +} + +// IsControlAction reports whether the text means a high-consequence control +// action. It is deliberately generous: a false positive only tells the user to +// use the admin app. A false negative would route a control action, which the +// brief forbids. +func IsControlAction(text string) bool { + t := " " + strings.ToLower(strings.Join(strings.Fields(text), " ")) + " " + for _, p := range ControlPhrases { + if strings.Contains(t, " "+p+" ") || strings.Contains(t, p) { + return true + } + } + if strings.Contains(t, "internet") { + for _, v := range ControlVerbs { + if strings.Contains(t, v) { + return true + } + } + } + // A direct change to a child's access is a control action. + if (strings.Contains(t, " kid") || strings.Contains(t, " child")) && + (strings.Contains(t, "internet") || strings.Contains(t, "online") || + strings.Contains(t, "wifi") || strings.Contains(t, "screen")) && + (strings.Contains(t, "cut") || strings.Contains(t, "block") || + strings.Contains(t, "limit") || strings.Contains(t, "allowance") || + strings.Contains(t, "schedule")) { + return true + } + return false +} diff --git a/src/understand/router.go b/src/understand/router.go new file mode 100644 index 0000000..ad0277c --- /dev/null +++ b/src/understand/router.go @@ -0,0 +1,213 @@ +package understand + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "log" + "net/http" + "strings" + "time" + + "hub/apierr" +) + +// Action is what the router decided. +type Action string + +const ( + // ActionRoute sends the question to one project or destination. + ActionRoute Action = "route" + // ActionFanout asks several projects, then merges the answers. + ActionFanout Action = "fanout" + // ActionClarify asks the user one short question. + ActionClarify Action = "clarify" + // ActionNoMatch says the Hub cannot route the request. + ActionNoMatch Action = "no_match" + // ActionRefused is a high-consequence control action. + ActionRefused Action = "refused" +) + +// Decision is the result of step 2. +type Decision struct { + Action Action `json:"action"` + Project string `json:"project,omitempty"` + Projects []string `json:"projects,omitempty"` + Question string `json:"question,omitempty"` + Options []string `json:"options,omitempty"` + Reason string `json:"reason,omitempty"` +} + +// Input is what the router reads. +type Input struct { + Question string + Person string + // Topics are the live capability topics from every project. + Topics []string + Rules *Rules +} + +// Decider turns an Input into a Decision. OmniRouter is the live one. Tests use +// a fake. +type Decider interface { + Decide(ctx context.Context, in Input) (Decision, error) +} + +// DeciderFunc adapts a function to the Decider interface. +type DeciderFunc func(ctx context.Context, in Input) (Decision, error) + +// Decide calls f. +func (f DeciderFunc) Decide(ctx context.Context, in Input) (Decision, error) { + return f(ctx, in) +} + +// OmniRouter routes through OmniRoute (the model gateway). No provider is ever +// called directly (AGENTS.md rule 1). +type OmniRouter struct { + URL string + APIKey string + Model string + Timeout Duration + HTTP *http.Client +} + +// NewOmniRouter builds the live router. +func NewOmniRouter(url, key, model string, timeout Duration) *OmniRouter { + if timeout <= 0 { + timeout = 3 * time.Second + } + return &OmniRouter{ + URL: strings.TrimRight(url, "/"), + APIKey: key, + Model: model, + Timeout: timeout, + HTTP: &http.Client{Timeout: timeout}, + } +} + +// Decide sends the question, the capability topics, and the routing rules to +// the model. It asks for one small JSON object back. +func (o *OmniRouter) Decide(ctx context.Context, in Input) (Decision, error) { + if o == nil || o.URL == "" || o.APIKey == "" { + return Decision{}, apierr.New(503, "unavailable", "The model gateway is not configured.") + } + if in.Rules == nil { + in.Rules = DefaultRules() + } + + ctx, cancel := context.WithTimeout(ctx, o.Timeout) + defer cancel() + start := time.Now() + + body := map[string]any{ + "model": o.Model, + "stream": false, + "max_tokens": 200, + "response_format": map[string]string{"type": "json_object"}, + "messages": []map[string]string{ + {"role": "system", "content": routingSystemPrompt(in)}, + {"role": "user", "content": in.Question}, + }, + } + raw, err := json.Marshal(body) + if err != nil { + return Decision{}, err + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, o.URL+"/v1/chat/completions", bytes.NewReader(raw)) + if err != nil { + return Decision{}, err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer "+o.APIKey) + + resp, err := o.HTTP.Do(req) + if err != nil { + if ctx.Err() == context.DeadlineExceeded { + return Decision{}, apierr.New(504, "timeout", "The model gateway did not answer in time.") + } + return Decision{}, apierr.New(503, "unavailable", "The model gateway is not answering.") + } + defer resp.Body.Close() + log.Printf("router: model=%s status=%d elapsed=%s", o.Model, resp.StatusCode, time.Since(start)) + + if resp.StatusCode != http.StatusOK { + return Decision{}, apierr.New(503, "unavailable", "The model gateway is not answering.") + } + + var out struct { + Choices []struct { + Message struct { + Content string `json:"content"` + } `json:"message"` + } `json:"choices"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil || len(out.Choices) == 0 { + return Decision{}, fmt.Errorf("router: bad model reply") + } + return parseDecision(out.Choices[0].Message.Content) +} + +// routingSystemPrompt builds the model instruction. It lists the topics and the +// route kinds, and asks for JSON only. +func routingSystemPrompt(in Input) string { + var b strings.Builder + b.WriteString("You route a home request for the Central AI Hub.\n") + b.WriteString("Answer with one JSON object only. No prose. No code fence.\n\n") + b.WriteString("Rules:\n") + b.WriteString("- A domain question goes to the project that owns the data. Do not fanout for one topic.\n") + b.WriteString("- A day summary may fanout to at most three projects.\n") + b.WriteString("- A high-consequence control action is never routed.\n") + b.WriteString("- When unsure, return clarify and ask one short question.\n\n") + b.WriteString("Reply shapes:\n") + b.WriteString("{\"action\":\"route\",\"project\":\"\",\"reason\":\"\"}\n") + b.WriteString("{\"action\":\"fanout\",\"projects\":[\"\"],\"reason\":\"\"}\n") + b.WriteString("{\"action\":\"clarify\",\"question\":\"\",\"options\":[\"\",\"\"]}\n") + b.WriteString("{\"action\":\"no_match\",\"reason\":\"\"}\n\n") + b.WriteString("Topic to project map (topic=project):\n") + b.WriteString(in.Rules.PromptTopics()) + b.WriteString("\n") + if len(in.Topics) > 0 { + b.WriteString("\nLive project capability topics:\n") + b.WriteString(strings.Join(in.Topics, ", ")) + b.WriteString("\n") + } + return b.String() +} + +// parseDecision reads the model's JSON. It tolerates a code fence and a leading +// or trailing line of prose. +func parseDecision(s string) (Decision, error) { + s = strings.TrimSpace(s) + if i := strings.Index(s, "{"); i >= 0 { + if j := strings.LastIndex(s, "}"); j > i { + s = s[i : j+1] + } + } + var d Decision + if err := json.Unmarshal([]byte(s), &d); err != nil { + return Decision{}, fmt.Errorf("router: cannot read model JSON: %w", err) + } + switch d.Action { + case ActionRoute: + if strings.TrimSpace(d.Project) == "" { + return Decision{}, fmt.Errorf("router: route with no project") + } + case ActionFanout: + if len(d.Projects) == 0 { + return Decision{}, fmt.Errorf("router: fanout with no projects") + } + if len(d.Projects) > 3 { + d.Projects = d.Projects[:3] + } + case ActionClarify: + if strings.TrimSpace(d.Question) == "" { + return Decision{}, fmt.Errorf("router: clarify with no question") + } + case ActionNoMatch: + default: + return Decision{}, fmt.Errorf("router: unknown action %q", d.Action) + } + return d, nil +} diff --git a/src/understand/rules.go b/src/understand/rules.go new file mode 100644 index 0000000..697016a --- /dev/null +++ b/src/understand/rules.go @@ -0,0 +1,203 @@ +// Package understand is step 2 of the Hub: work out what the user asked. +// +// Two parts: +// +// 1. The never-route check. It is deterministic. No LLM is in this path +// (AGENTS.md rule 4, routing.md section 3). +// 2. The routing LLM through OmniRoute. It gets the capability topics and the +// routing rules, and returns route, fanout, clarify, or no_match. +package understand + +import ( + "bufio" + "fmt" + "io" + "os" + "strings" + "time" +) + +// Duration is re-exported so callers do not import time just for the config. +type Duration = time.Duration + +// Topic is one row of the routing.md topic map. +type Topic struct { + Project string + Kind string // "project", "direct", "paperclip" +} + +// Rules is the machine form of routing.md. +type Rules struct { + // TopicToProject maps one plain-language topic (lowercase) to a project. + TopicToProject map[string]string + // KindOf maps a project name to its route kind. + KindOf map[string]string + // Topics holds the topic keys in a stable order for the LLM prompt. + Topics []string +} + +// DefaultRules mirrors routing.md v1. It is the fallback when routing.md is +// missing or unreadable, so the Hub always has a map. +func DefaultRules() *Rules { + r := &Rules{TopicToProject: map[string]string{}, KindOf: map[string]string{}} + add := func(project, kind string, topics ...string) { + r.KindOf[project] = kind + for _, t := range topics { + r.TopicToProject[t] = project + r.Topics = append(r.Topics, t) + } + } + // routing.md section 1, "Topic to project map". + add("email_agent", "project", "email", "inbox", "mail", "message", "unread") + add("compass_nhs", "project", "school", "compass", "homework", "timetable", "class", "nhs appointment") + add("internet_control", "project", "screen time", "internet use", "time limits", "what did the kids do online") + add("home_assistant", "direct", "lights", "lamp", "switch", "plug", "timer", "reminder", "calendar event", "device", "thermostat", "calendar", "today") + add("speakers", "direct", "speak", "say it out loud", "house speakers", "announce") + add("web_search", "direct", "search", "look up", "what is", "who is", "weather", "news", "facts") + add("apprise", "direct", "notify", "notification", "alert me", "remind my phone") + add("paperclip", "paperclip", "file a job", "plan this", "research this", "slow task", "work bench") + add("artemis", "direct", "phone screen", "tap", "open an app", "type on the phone", "scroll") + add("front_door_cam", "peer", "visitor", "front door", "who is at the door") + add("jervis", "way_in", "talk to the house voice", "hey jervis") + add("dsh", "peer", "chat", "per-person chat", "dsh") + return r +} + +// LoadRules reads routing.md and parses the topic map. On error it returns the +// default rules and the error, so the caller can log and keep running. +func LoadRules(path string) (*Rules, error) { + if path == "" { + return DefaultRules(), nil + } + f, err := os.Open(path) + if err != nil { + return DefaultRules(), err + } + defer f.Close() + + r, err := ParseRules(f) + if err != nil { + return DefaultRules(), err + } + return r, nil +} + +// ParseRules reads the "Topic to project map" table from routing.md. +// +// It reads rows of the form: +// +// | email, inbox, mail | `email_agent` | query `/ask` | placeholder | +// +// The Kind column decides the route kind: +// +// contains "query" -> project +// contains "command" -> direct +// contains "decision" -> paperclip +// anything else -> peer +func ParseRules(r io.Reader) (*Rules, error) { + out := &Rules{TopicToProject: map[string]string{}, KindOf: map[string]string{}} + sc := bufio.NewScanner(r) + sc.Buffer(make([]byte, 0, 64*1024), 1024*1024) + + inTable := false + for sc.Scan() { + line := strings.TrimSpace(sc.Text()) + if strings.HasPrefix(line, "## ") { + // A new section. Stop after the topic map section. + if inTable { + break + } + inTable = strings.Contains(strings.ToLower(line), "topic to project map") + continue + } + if !inTable || !strings.HasPrefix(line, "|") { + continue + } + cells := splitRow(line) + if len(cells) < 3 { + continue + } + topics := cells[0] + project := strings.Trim(cells[1], "`* ") + kindText := strings.ToLower(cells[2]) + // Skip the header and separator rows. + if project == "" || strings.EqualFold(project, "project") || + strings.HasPrefix(project, "---") || strings.HasPrefix(project, ":--") { + continue + } + kind := "peer" + switch { + case strings.Contains(kindText, "query"): + kind = "project" + case strings.Contains(kindText, "command"): + kind = "direct" + case strings.Contains(kindText, "decision"): + kind = "paperclip" + } + out.KindOf[project] = kind + for _, t := range strings.Split(topics, ",") { + t = strings.ToLower(strings.Trim(strings.TrimSpace(t), "`* ")) + if t == "" || t == "topic (what the user says)" { + continue + } + out.TopicToProject[t] = project + out.Topics = append(out.Topics, t) + } + } + if err := sc.Err(); err != nil { + return nil, err + } + if len(out.TopicToProject) == 0 { + return nil, fmt.Errorf("routing rules: no topic map rows found") + } + return out, nil +} + +// splitRow splits a markdown table row into its cells. +func splitRow(line string) []string { + line = strings.TrimPrefix(strings.TrimSpace(line), "|") + line = strings.TrimSuffix(strings.TrimSpace(line), "|") + parts := strings.Split(line, "|") + for i := range parts { + parts[i] = strings.TrimSpace(parts[i]) + } + return parts +} + +// KindFor returns the route kind for a project. An unknown project is a +// project-style query by default. +func (r *Rules) KindFor(project string) string { + if r == nil { + return "project" + } + if k, ok := r.KindOf[project]; ok { + return k + } + return "project" +} + +// ProjectForTopic returns the project that owns a topic. +func (r *Rules) ProjectForTopic(topic string) (string, bool) { + if r == nil { + return "", false + } + p, ok := r.TopicToProject[strings.ToLower(strings.TrimSpace(topic))] + return p, ok +} + +// PromptTopics returns the topic list as one short, stable string. +func (r *Rules) PromptTopics() string { + if r == nil { + return "" + } + seen := map[string]bool{} + var parts []string + for _, t := range r.Topics { + if seen[t] { + continue + } + seen[t] = true + parts = append(parts, fmt.Sprintf("%s=%s", t, r.TopicToProject[t])) + } + return strings.Join(parts, ", ") +} diff --git a/src/understand/understand_test.go b/src/understand/understand_test.go new file mode 100644 index 0000000..a331a24 --- /dev/null +++ b/src/understand/understand_test.go @@ -0,0 +1,89 @@ +package understand + +import ( + "strings" + "testing" +) + +func TestIsControlAction(t *testing.T) { + control := []string{ + "cut the kids' internet", + "turn off the internet", + "change the internet time limit", + "add that to the blocklist", + "change a parental control", + "restore internet access", + } + for _, text := range control { + if !IsControlAction(text) { + t.Errorf("IsControlAction(%q) = false, want true", text) + } + } + + safe := []string{ + "what does Harry have on today?", + "how long have the kids been on screens?", + "turn on the kitchen lamp", + "set a timer for ten minutes", + "speak the weather", + } + for _, text := range safe { + if IsControlAction(text) { + t.Errorf("IsControlAction(%q) = true, want false", text) + } + } +} + +func TestParseRules(t *testing.T) { + md := ` +## 1. Topic to project map + +| Topic (what the user says) | Project | Kind | Status | +|---|---|---|---| +| email, inbox, mail | ` + "`email_agent`" + ` | query ` + "`/ask`" + ` | placeholder | +| lights, lamp, switch | ` + "`home_assistant`" + ` | command (direct) | active | +| file a job, plan this | ` + "`paperclip`" + ` | decision (approval) | active | + +## 2. Preferred project when two overlap + +| Overlap | Preferred | +|---|---| +| today | home_assistant | +` + rules, err := ParseRules(strings.NewReader(md)) + if err != nil { + t.Fatalf("ParseRules: %v", err) + } + if got := rules.KindFor("email_agent"); got != "project" { + t.Errorf("email_agent kind = %q", got) + } + if got := rules.KindFor("home_assistant"); got != "direct" { + t.Errorf("home_assistant kind = %q", got) + } + if got := rules.KindFor("paperclip"); got != "paperclip" { + t.Errorf("paperclip kind = %q", got) + } + if p, ok := rules.ProjectForTopic("lamp"); !ok || p != "home_assistant" { + t.Errorf("lamp -> %q %v", p, ok) + } + if strings.Contains(rules.PromptTopics(), "home_assistant") == false { + t.Errorf("prompt topics missing home_assistant: %s", rules.PromptTopics()) + } +} + +func TestParseDecision(t *testing.T) { + d, err := parseDecision("```json\n{\"action\":\"route\",\"project\":\"compass_nhs\"}\n```") + if err != nil { + t.Fatalf("parseDecision: %v", err) + } + if d.Action != ActionRoute || d.Project != "compass_nhs" { + t.Errorf("got %#v", d) + } + + if _, err := parseDecision(`{"action":"route"}`); err == nil { + t.Error("route with no project should fail") + } + if _, err := parseDecision("not json"); err == nil { + t.Error("garbage should fail") + } +}