feat: initialize external DobotAI agent gateway

2026-07-20 11:47:37 +08:00
commit a8cb26203b
22 changed files with 1288 additions and 0 deletions
@@ -0,0 +1,12 @@
+# DobotAI configuration
+DOBOTAI_LISTEN_ADDR=127.0.0.1:8090
+DOBOTAI_INGRESS_TOKEN=replace-me
+DOBOTAI_GROUP_SESSION_POLICY=shared
+DOBOTAI_AGENT_BINDINGS={"dingtalk:*":"replace-with-agent-id"}
+
+# Official WeKnora API only
+WEKNORA_BASE_URL=http://127.0.0.1:8080
+WEKNORA_API_KEY=replace-me
+
+# Independent DobotAI PostgreSQL; never point this at WeKnora's database
+DATABASE_URL=postgres://dobotai:replace-me@127.0.0.1:5432/dobotai?sslmode=disable
@@ -0,0 +1,11 @@
+# Build artifacts and local secrets
+/bin/
+/dist/
+.env
+*.local
+coverage.out
+
+# Editors and OS
+.idea/
+.vscode/
+.DS_Store
@@ -0,0 +1,66 @@
+# DobotAI Platform
+
+DobotAI is the external enterprise integration layer for official WeKnora. It does not import WeKnora packages, modify WeKnora source, or access the WeKnora database.
+
+Current milestone: channel-bound Agent QA core.
+
+Implemented:
+
+- official `POST /api/v1/sessions` adapter;
+- official `POST /api/v1/agent-chat/:session_id` adapter;
+- fail-closed SSE parsing for `references`, `answer`, `error`, and terminal `done`;
+- per-channel/per-conversation Agent binding;
+- long-lived direct-user sessions;
+- shared-group sessions, with optional group+user policy;
+- external-message idempotency;
+- independent PostgreSQL session/turn storage;
+- authenticated internal `POST /v1/qa` ingress endpoint.
+
+## WeKnora baseline
+
+Target policy: latest stable official Release/Tag. The newest locally fetched stable tag is `v0.6.3`. Online GitHub verification is pending because DNS/proxy access was unavailable during initialization.
+
+## Run
+
+1. Create an independent PostgreSQL database named `dobotai`.
+2. Apply `db/migrations/000001_core.up.sql` to that database only.
+3. Copy `.env.example` to a local `.env` and supply secrets outside Git.
+4. Export the variables and run:
+
+```bash
+go run ./cmd/dobotai
+```
+
+Health check:
+
+```bash
+curl http://127.0.0.1:8090/healthz
+```
+
+Internal QA request:
+
+```bash
+curl -X POST http://127.0.0.1:8090/v1/qa \
+  -H "Authorization: Bearer $DOBOTAI_INGRESS_TOKEN" \
+  -H 'Content-Type: application/json' \
+  -d '{
+    "tenant_key":"default",
+    "channel":"dingtalk",
+    "conversation_type":"direct",
+    "external_user_id":"user-1",
+    "external_message_id":"message-1",
+    "query":"hello"
+  }'
+```
+
+## Hard boundaries
+
+- No WeKnora source modifications.
+- No custom WeKnora tables or columns.
+- No direct WeKnora database access.
+- No private WeKnora package imports.
+- All compatibility logic stays in `internal/weknora`.
+- Business data stays in DobotAI PostgreSQL.
+- Redis will be used only for queues, retries, idempotency leases, and temporary state.
+
+See `docs/architecture.md` for ownership and migration decisions.
@@ -0,0 +1,73 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"log"
+	"net/http"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+	"time"
+
+	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/agentbinding"
+	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/httpapi"
+	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/qa"
+	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/session"
+	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/store/postgres"
+	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/weknora"
+)
+
+func main() {
+	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer cancel()
+	storeCtx, storeCancel := context.WithTimeout(ctx, 10*time.Second)
+	defer storeCancel()
+	store, err := postgres.Open(storeCtx, os.Getenv("DATABASE_URL"))
+	if err != nil {
+		log.Fatal(err)
+	}
+	defer store.Close()
+	client, err := weknora.NewClient(os.Getenv("WEKNORA_BASE_URL"), os.Getenv("WEKNORA_API_KEY"), 120*time.Second)
+	if err != nil {
+		log.Fatal(err)
+	}
+	bindings, err := agentbinding.New(os.Getenv("DOBOTAI_AGENT_BINDINGS"))
+	if err != nil {
+		log.Fatal(err)
+	}
+	groupPolicy := session.GroupPolicy(strings.TrimSpace(os.Getenv("DOBOTAI_GROUP_SESSION_POLICY")))
+	if groupPolicy == "" {
+		groupPolicy = session.GroupShared
+	}
+	if groupPolicy != session.GroupShared && groupPolicy != session.GroupPerUser {
+		log.Fatal("DOBOTAI_GROUP_SESSION_POLICY must be shared or per_user")
+	}
+	qaService, err := qa.New(store, bindings, client, groupPolicy)
+	if err != nil {
+		log.Fatal(err)
+	}
+	api, err := httpapi.New(qaService, os.Getenv("DOBOTAI_INGRESS_TOKEN"))
+	if err != nil {
+		log.Fatal(err)
+	}
+	address := strings.TrimSpace(os.Getenv("DOBOTAI_LISTEN_ADDR"))
+	if address == "" {
+		address = "127.0.0.1:8090"
+	}
+	server := &http.Server{Addr: address, Handler: api.Handler(), ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 15 * time.Second, WriteTimeout: 130 * time.Second, IdleTimeout: 60 * time.Second}
+	go func() {
+		log.Printf("DobotAI ingress listening on %s", address)
+		if err := server.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+			log.Printf("HTTP server failed: %v", err)
+			cancel()
+		}
+	}()
+	<-ctx.Done()
+	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer shutdownCancel()
+	if err := server.Shutdown(shutdownCtx); err != nil {
+		log.Printf("HTTP shutdown: %v", err)
+	}
+}
@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS turns;
+DROP TABLE IF EXISTS session_mappings;
@@ -0,0 +1,35 @@
+CREATE TABLE session_mappings (
+    session_key TEXT PRIMARY KEY,
+    tenant_key TEXT NOT NULL,
+    channel TEXT NOT NULL,
+    agent_id TEXT NOT NULL,
+    conversation_type TEXT NOT NULL CHECK (conversation_type IN ('direct','group')),
+    external_subject_id TEXT NOT NULL,
+    weknora_session_id TEXT NOT NULL,
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX idx_session_mappings_tenant_channel
+    ON session_mappings (tenant_key, channel, external_subject_id);
+
+CREATE TABLE turns (
+    id TEXT PRIMARY KEY,
+    tenant_key TEXT NOT NULL,
+    channel TEXT NOT NULL,
+    external_message_id TEXT NOT NULL,
+    session_key TEXT NOT NULL,
+    weknora_session_id TEXT NOT NULL DEFAULT '',
+    weknora_request_id TEXT NOT NULL DEFAULT '',
+    question TEXT NOT NULL,
+    answer TEXT NOT NULL DEFAULT '',
+    references JSONB NOT NULL DEFAULT '[]'::JSONB,
+    status TEXT NOT NULL CHECK (status IN ('processing','succeeded','failed','cancelled')),
+    error_message TEXT NOT NULL DEFAULT '',
+    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    UNIQUE (tenant_key, channel, external_message_id)
+);
+
+CREATE INDEX idx_turns_tenant_created ON turns (tenant_key, created_at DESC);
+CREATE INDEX idx_turns_session_created ON turns (session_key, created_at DESC);
@@ -0,0 +1,21 @@
+# WeKnora API Capability Gaps
+
+These gaps must not be solved with private WeKnora source or database changes.
+
+| Capability | Current external handling |
+|---|---|
+| Like/dislike and unresolved reason | DobotAI PostgreSQL and API |
+| Formal covered/not-covered decision | DobotAI evaluation rules using references, scores, answer status, and offline calibration |
+| Support case, owner, SLA, and status | DobotAI workflow |
+| Human reply to original channel | DobotAI channel adapter; not written to WeKnora history by default |
+| Stable external turn correlation | DobotAI mapping of external message, turn, WeKnora session/request, and gateway trace |
+| Complete multi-call token usage | Model gateway/Langfuse; unknown remains null/unknown |
+| Strict image failure semantics | Adapter contract test; reject image traffic on failure |
+| Arbitrary human assistant-message append | Keep in DobotAI and submit an upstream API request only if context injection becomes necessary |
+
+Allowed responses to a gap:
+
+1. adjust the product design;
+2. implement it outside WeKnora;
+3. open an upstream issue;
+4. submit a general upstream pull request.
@@ -0,0 +1,72 @@
+# DobotAI External Architecture
+
+## Boundary
+
+```text
+DingTalk / WeCom / Web / Software
+  -> DobotAI ingress and workflows
+  -> WeKnora Adapter
+  -> official WeKnora REST API
+  -> official WeKnora RAG and Agent
+```
+
+DobotAI PostgreSQL owns channel identities, external messages, session mappings, turns, feedback, coverage decisions, support cases, human replies, FAQ workflow state, and content-source mappings. WeKnora owns only its official data model.
+
+## Session policy
+
+Default policy:
+
+- direct: one long-lived session per tenant + channel + Agent + external user;
+- group: one shared session per tenant + channel + Agent + external group;
+- group messages are admitted only when the bot is mentioned or an explicit command triggers it;
+- sender display name is prefixed to the query so a shared Agent can distinguish speakers;
+- future controls must include idle rotation and an explicit new-conversation command;
+- confidential channels may switch to `per_user`, yielding group + user sessions.
+
+Shared group sessions are the common collaboration-bot pattern because all members see the same conversation. Group+user sessions provide stronger context isolation but make replies less coherent for collaborative group discussions.
+
+## Agent binding
+
+Agents are configuration-owned. Channel callers cannot choose an arbitrary `agent_id`.
+
+Resolution order:
+
+1. exact `<channel>:<external_chat_id>` binding;
+2. channel wildcard `<channel>:*`;
+3. fail closed when no binding exists.
+
+## Image failure policy
+
+Before enabling image traffic, the WeKnora Adapter must run a contract test against the selected official release and configured vision model.
+
+If any image path fails:
+
+1. do not resend the question without the image;
+2. do not return partial model output as success;
+3. return a typed `vision_unavailable` failure to the channel adapter;
+4. DingTalk tells the user that image recognition is temporarily unavailable and asks them to retry or send text;
+5. emit an operational alert to verify Agent image upload settings, selected vision model, provider connectivity, and model capability;
+6. keep image traffic disabled until the contract test passes.
+
+No WeKnora source patch is allowed as a workaround.
+
+## Service ownership
+
+- `internal/weknora`: official REST/SSE compatibility only.
+- `internal/qa`: business orchestration and terminal-state ownership.
+- `internal/session`: external-to-WeKnora session policy.
+- `internal/store/postgres`: DobotAI business persistence only.
+- future `channels/*`: platform callbacks and outbound replies.
+- future `content-sync/*`: DingTalk KB, NAS, and CMS synchronization.
+- model gateway: token accounting, routing, quotas, circuit breaking, and provider observability.
+- Redis: queue, retry, short leases, and temporary stream state only.
+
+## Upgrade contract
+
+For every official WeKnora upgrade:
+
+1. update the pinned official release/tag;
+2. run official database migrations against WeKnora only;
+3. run DobotAI Adapter contract tests;
+4. verify text Agent Chat, references, terminal errors, Stop, images, ingestion, and FAQ;
+5. promote the official image only after the contract suite passes.
@@ -0,0 +1,13 @@
+module gitea.zheromiku.top/Weknora/dobotai-platform
+
+go 1.25.0
+
+require github.com/jackc/pgx/v5 v5.9.2
+
+require (
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
+	golang.org/x/sync v0.20.0 // indirect
+	golang.org/x/text v0.37.0 // indirect
+)
@@ -0,0 +1,19 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.9.2 h1:3ZhOzMWnR4yJ+RW1XImIPsD1aNSz4T4fyP7zlQb56hw=
+github.com/jackc/pgx/v5 v5.9.2/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
+golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
+golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc=
+golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
@@ -0,0 +1,36 @@
+package agentbinding
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+)
+
+type Resolver struct{ bindings map[string]string }
+
+func New(rawJSON string) (*Resolver, error) {
+	bindings := map[string]string{}
+	if err := json.Unmarshal([]byte(rawJSON), &bindings); err != nil {
+		return nil, fmt.Errorf("decode DOBOTAI_AGENT_BINDINGS: %w", err)
+	}
+	for key, value := range bindings {
+		if strings.TrimSpace(key) == "" || strings.TrimSpace(value) == "" {
+			return nil, errors.New("agent binding keys and values must be non-empty")
+		}
+	}
+	return &Resolver{bindings: bindings}, nil
+}
+
+func (r *Resolver) Resolve(channel, externalChatID string) (string, error) {
+	if r == nil {
+		return "", errors.New("agent resolver is not configured")
+	}
+	if agentID := r.bindings[channel+":"+externalChatID]; agentID != "" {
+		return agentID, nil
+	}
+	if agentID := r.bindings[channel+":*"]; agentID != "" {
+		return agentID, nil
+	}
+	return "", fmt.Errorf("no agent binding for %s:%s", channel, externalChatID)
+}
@@ -0,0 +1,18 @@
+package agentbinding
+
+import "testing"
+
+func TestResolverPrefersConversationBinding(t *testing.T) {
+	resolver, err := New(`{"dingtalk:*":"default-agent","dingtalk:group-1":"group-agent"}`)
+	if err != nil {
+		t.Fatalf("New: %v", err)
+	}
+	got, err := resolver.Resolve("dingtalk", "group-1")
+	if err != nil || got != "group-agent" {
+		t.Fatalf("got=%q err=%v", got, err)
+	}
+	got, err = resolver.Resolve("dingtalk", "group-2")
+	if err != nil || got != "default-agent" {
+		t.Fatalf("got=%q err=%v", got, err)
+	}
+}
@@ -0,0 +1,76 @@
+package core
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+)
+
+type ConversationType string
+
+const (
+	ConversationDirect ConversationType = "direct"
+	ConversationGroup  ConversationType = "group"
+)
+
+var ErrTurnInProgress = errors.New("turn is already processing")
+
+type QARequest struct {
+	TenantKey         string
+	Channel           string
+	ConversationType  ConversationType
+	ExternalChatID    string
+	ExternalUserID    string
+	ExternalMessageID string
+	SenderName        string
+	Query             string
+}
+
+type Reference struct {
+	ID             string          `json:"id,omitempty"`
+	KnowledgeID    string          `json:"knowledge_id,omitempty"`
+	KnowledgeTitle string          `json:"knowledge_title,omitempty"`
+	Content        string          `json:"content,omitempty"`
+	Score          float64         `json:"score,omitempty"`
+	Raw            json.RawMessage `json:"-"`
+}
+
+type QAResult struct {
+	TurnID           string      `json:"turn_id"`
+	WeKnoraSessionID string      `json:"weknora_session_id"`
+	WeKnoraRequestID string      `json:"weknora_request_id,omitempty"`
+	Answer           string      `json:"answer"`
+	References       []Reference `json:"references,omitempty"`
+	Replayed         bool        `json:"replayed"`
+}
+
+type SessionMapping struct {
+	SessionKey        string
+	TenantKey         string
+	Channel           string
+	AgentID           string
+	ConversationType  ConversationType
+	ExternalSubjectID string
+	WeKnoraSessionID  string
+}
+
+type Turn struct {
+	ID                string
+	TenantKey         string
+	Channel           string
+	ExternalMessageID string
+	SessionKey        string
+	WeKnoraSessionID  string
+	WeKnoraRequestID  string
+	Question          string
+	Answer            string
+	References        []Reference
+	Status            string
+	ErrorMessage      string
+}
+
+type Repository interface {
+	BeginTurn(ctx context.Context, turn Turn) (Turn, bool, error)
+	ResolveSession(ctx context.Context, mapping SessionMapping, create func(context.Context) (string, error)) (string, error)
+	FinishTurn(ctx context.Context, turn Turn) error
+}
@@ -0,0 +1,91 @@
+package httpapi
+
+import (
+	"context"
+	"crypto/subtle"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"strings"
+
+	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
+)
+
+type QAService interface {
+	Ask(context.Context, core.QARequest) (core.QAResult, error)
+}
+
+type Server struct {
+	service QAService
+	token   string
+	mux     *http.ServeMux
+}
+
+func New(service QAService, ingressToken string) (*Server, error) {
+	if service == nil {
+		return nil, errors.New("QA service is required")
+	}
+	if strings.TrimSpace(ingressToken) == "" {
+		return nil, errors.New("DOBOTAI_INGRESS_TOKEN is required")
+	}
+	s := &Server{service: service, token: ingressToken, mux: http.NewServeMux()}
+	s.mux.HandleFunc("GET /healthz", s.health)
+	s.mux.HandleFunc("POST /v1/qa", s.authorize(s.ask))
+	return s, nil
+}
+func (s *Server) Handler() http.Handler { return s.mux }
+
+func (s *Server) authorize(next http.HandlerFunc) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		provided := strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")
+		if len(provided) != len(s.token) || subtle.ConstantTimeCompare([]byte(provided), []byte(s.token)) != 1 {
+			writeError(w, http.StatusUnauthorized, "unauthorized")
+			return
+		}
+		next(w, r)
+	}
+}
+func (s *Server) health(w http.ResponseWriter, _ *http.Request) {
+	writeJSON(w, http.StatusOK, map[string]string{"status": "ok"})
+}
+func (s *Server) ask(w http.ResponseWriter, r *http.Request) {
+	r.Body = http.MaxBytesReader(w, r.Body, 1<<20)
+	defer r.Body.Close()
+	var request struct {
+		TenantKey         string                `json:"tenant_key"`
+		Channel           string                `json:"channel"`
+		ConversationType  core.ConversationType `json:"conversation_type"`
+		ExternalChatID    string                `json:"external_chat_id"`
+		ExternalUserID    string                `json:"external_user_id"`
+		ExternalMessageID string                `json:"external_message_id"`
+		SenderName        string                `json:"sender_name"`
+		Query             string                `json:"query"`
+	}
+	if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
+		if errors.Is(err, io.EOF) {
+			writeError(w, http.StatusBadRequest, "request body is required")
+		} else {
+			writeError(w, http.StatusBadRequest, "invalid JSON request")
+		}
+		return
+	}
+	result, err := s.service.Ask(r.Context(), core.QARequest{TenantKey: request.TenantKey, Channel: request.Channel, ConversationType: request.ConversationType, ExternalChatID: request.ExternalChatID, ExternalUserID: request.ExternalUserID, ExternalMessageID: request.ExternalMessageID, SenderName: request.SenderName, Query: request.Query})
+	if err != nil {
+		if errors.Is(err, core.ErrTurnInProgress) {
+			writeError(w, http.StatusConflict, "message is already processing")
+			return
+		}
+		writeError(w, http.StatusBadGateway, "question processing failed")
+		return
+	}
+	writeJSON(w, http.StatusOK, result)
+}
+func writeError(w http.ResponseWriter, status int, message string) {
+	writeJSON(w, status, map[string]string{"error": message})
+}
+func writeJSON(w http.ResponseWriter, status int, value any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(value)
+}
@@ -0,0 +1,38 @@
+package httpapi
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
+)
+
+type fakeQA struct{}
+
+func (fakeQA) Ask(context.Context, core.QARequest) (core.QAResult, error) {
+	return core.QAResult{TurnID: "turn-1", Answer: "answer"}, nil
+}
+
+func TestQAEndpointRequiresTokenAndReturnsResult(t *testing.T) {
+	server, err := New(fakeQA{}, "ingress-secret")
+	if err != nil {
+		t.Fatal(err)
+	}
+	body := `{"tenant_key":"tenant","channel":"dingtalk","conversation_type":"direct","external_user_id":"user","external_message_id":"message","query":"hello"}`
+	unauthorized := httptest.NewRequest(http.MethodPost, "/v1/qa", strings.NewReader(body))
+	unauthorizedResult := httptest.NewRecorder()
+	server.Handler().ServeHTTP(unauthorizedResult, unauthorized)
+	if unauthorizedResult.Code != http.StatusUnauthorized {
+		t.Fatalf("unauthorized status=%d", unauthorizedResult.Code)
+	}
+	authorized := httptest.NewRequest(http.MethodPost, "/v1/qa", strings.NewReader(body))
+	authorized.Header.Set("Authorization", "Bearer ingress-secret")
+	result := httptest.NewRecorder()
+	server.Handler().ServeHTTP(result, authorized)
+	if result.Code != http.StatusOK || !strings.Contains(result.Body.String(), `"turn_id":"turn-1"`) {
+		t.Fatalf("status=%d body=%s", result.Code, result.Body.String())
+	}
+}
@@ -0,0 +1,125 @@
+package qa
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"errors"
+	"fmt"
+	"strings"
+
+	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
+	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/session"
+	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/weknora"
+)
+
+type AgentResolver interface {
+	Resolve(channel, externalChatID string) (string, error)
+}
+
+type WeKnora interface {
+	CreateSession(ctx context.Context, title, description string) (string, error)
+	AgentChat(ctx context.Context, sessionID string, request weknora.AgentChatRequest) (weknora.ChatResult, error)
+}
+
+type Service struct {
+	repository  core.Repository
+	agents      AgentResolver
+	weknora     WeKnora
+	groupPolicy session.GroupPolicy
+}
+
+func New(repository core.Repository, agents AgentResolver, client WeKnora, groupPolicy session.GroupPolicy) (*Service, error) {
+	if repository == nil || agents == nil || client == nil {
+		return nil, errors.New("repository, agent resolver and weknora client are required")
+	}
+	if groupPolicy == "" {
+		groupPolicy = session.GroupShared
+	}
+	return &Service{repository: repository, agents: agents, weknora: client, groupPolicy: groupPolicy}, nil
+}
+
+func (s *Service) Ask(ctx context.Context, request core.QARequest) (core.QAResult, error) {
+	if err := validateRequest(request); err != nil {
+		return core.QAResult{}, err
+	}
+	agentID, err := s.agents.Resolve(request.Channel, request.ExternalChatID)
+	if err != nil {
+		return core.QAResult{}, err
+	}
+	sessionKey, subjectID, err := session.BuildKey(session.Identity{
+		TenantKey: request.TenantKey, Channel: request.Channel, AgentID: agentID,
+		ConversationType: request.ConversationType, ExternalChatID: request.ExternalChatID,
+		ExternalUserID: request.ExternalUserID, GroupPolicy: s.groupPolicy,
+	})
+	if err != nil {
+		return core.QAResult{}, err
+	}
+
+	turn := core.Turn{ID: newID(), TenantKey: request.TenantKey, Channel: request.Channel,
+		ExternalMessageID: request.ExternalMessageID, SessionKey: sessionKey,
+		Question: request.Query, Status: "processing"}
+	stored, created, err := s.repository.BeginTurn(ctx, turn)
+	if err != nil {
+		return core.QAResult{}, fmt.Errorf("begin turn: %w", err)
+	}
+	if !created {
+		if stored.Status == "succeeded" {
+			return core.QAResult{TurnID: stored.ID, WeKnoraSessionID: stored.WeKnoraSessionID,
+				WeKnoraRequestID: stored.WeKnoraRequestID, Answer: stored.Answer,
+				References: stored.References, Replayed: true}, nil
+		}
+		return core.QAResult{}, core.ErrTurnInProgress
+	}
+	turn = stored
+
+	mapping := core.SessionMapping{SessionKey: sessionKey, TenantKey: request.TenantKey,
+		Channel: request.Channel, AgentID: agentID, ConversationType: request.ConversationType,
+		ExternalSubjectID: subjectID}
+	sessionID, err := s.repository.ResolveSession(ctx, mapping, func(createCtx context.Context) (string, error) {
+		title := fmt.Sprintf("%s %s", request.Channel, subjectID)
+		return s.weknora.CreateSession(createCtx, title, "Managed by DobotAI; business state remains outside WeKnora")
+	})
+	if err != nil {
+		turn.Status, turn.ErrorMessage = "failed", err.Error()
+		_ = s.repository.FinishTurn(ctx, turn)
+		return core.QAResult{}, fmt.Errorf("resolve weknora session: %w", err)
+	}
+	turn.WeKnoraSessionID = sessionID
+
+	query := request.Query
+	if request.ConversationType == core.ConversationGroup && strings.TrimSpace(request.SenderName) != "" {
+		query = request.SenderName + ": " + request.Query
+	}
+	chat, err := s.weknora.AgentChat(ctx, sessionID, weknora.AgentChatRequest{
+		Query: query, AgentID: agentID, AgentEnabled: true, Channel: "im", EnableMemory: true,
+	})
+	if err != nil {
+		turn.Status, turn.ErrorMessage = "failed", err.Error()
+		_ = s.repository.FinishTurn(ctx, turn)
+		return core.QAResult{}, fmt.Errorf("weknora agent chat: %w", err)
+	}
+	turn.Status, turn.Answer, turn.References = "succeeded", chat.Answer, chat.References
+	turn.WeKnoraRequestID = chat.RequestID
+	if err := s.repository.FinishTurn(ctx, turn); err != nil {
+		return core.QAResult{}, fmt.Errorf("finish turn: %w", err)
+	}
+	return core.QAResult{TurnID: turn.ID, WeKnoraSessionID: sessionID,
+		WeKnoraRequestID: chat.RequestID, Answer: chat.Answer, References: chat.References}, nil
+}
+
+func validateRequest(request core.QARequest) error {
+	if strings.TrimSpace(request.TenantKey) == "" || strings.TrimSpace(request.Channel) == "" ||
+		strings.TrimSpace(request.ExternalMessageID) == "" || strings.TrimSpace(request.Query) == "" {
+		return errors.New("tenant_key, channel, external_message_id and query are required")
+	}
+	return nil
+}
+
+func newID() string {
+	var raw [16]byte
+	if _, err := rand.Read(raw[:]); err != nil {
+		panic("crypto/rand unavailable")
+	}
+	return hex.EncodeToString(raw[:])
+}
@@ -0,0 +1,105 @@
+package qa
+
+import (
+	"context"
+	"sync"
+	"testing"
+
+	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
+	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/session"
+	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/weknora"
+)
+
+type memoryRepository struct {
+	mu       sync.Mutex
+	turns    map[string]core.Turn
+	sessions map[string]string
+}
+
+func newMemoryRepository() *memoryRepository {
+	return &memoryRepository{turns: map[string]core.Turn{}, sessions: map[string]string{}}
+}
+func (r *memoryRepository) BeginTurn(_ context.Context, turn core.Turn) (core.Turn, bool, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	key := turn.TenantKey + ":" + turn.Channel + ":" + turn.ExternalMessageID
+	if old, ok := r.turns[key]; ok {
+		return old, false, nil
+	}
+	r.turns[key] = turn
+	return turn, true, nil
+}
+func (r *memoryRepository) ResolveSession(ctx context.Context, m core.SessionMapping, create func(context.Context) (string, error)) (string, error) {
+	r.mu.Lock()
+	if id := r.sessions[m.SessionKey]; id != "" {
+		r.mu.Unlock()
+		return id, nil
+	}
+	r.mu.Unlock()
+	id, err := create(ctx)
+	if err != nil {
+		return "", err
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if old := r.sessions[m.SessionKey]; old != "" {
+		return old, nil
+	}
+	r.sessions[m.SessionKey] = id
+	return id, nil
+}
+func (r *memoryRepository) FinishTurn(_ context.Context, turn core.Turn) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	r.turns[turn.TenantKey+":"+turn.Channel+":"+turn.ExternalMessageID] = turn
+	return nil
+}
+
+type fixedResolver struct{}
+
+func (fixedResolver) Resolve(_, _ string) (string, error) { return "agent-1", nil }
+
+type fakeWeKnora struct {
+	mu       sync.Mutex
+	sessions int
+	queries  []string
+}
+
+func (f *fakeWeKnora) CreateSession(context.Context, string, string) (string, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.sessions++
+	return "session-1", nil
+}
+func (f *fakeWeKnora) AgentChat(_ context.Context, sessionID string, request weknora.AgentChatRequest) (weknora.ChatResult, error) {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	f.queries = append(f.queries, request.Query)
+	return weknora.ChatResult{RequestID: "request-1", Answer: "answer", References: []core.Reference{{KnowledgeID: "kb-1"}}}, nil
+}
+
+func TestAskUsesSharedGroupSessionAndReplaysDuplicate(t *testing.T) {
+	repo := newMemoryRepository()
+	client := &fakeWeKnora{}
+	service, err := New(repo, fixedResolver{}, client, session.GroupShared)
+	if err != nil {
+		t.Fatal(err)
+	}
+	base := core.QARequest{TenantKey: "tenant", Channel: "dingtalk", ConversationType: core.ConversationGroup, ExternalChatID: "group-1", ExternalUserID: "user-1", ExternalMessageID: "message-1", SenderName: "Alice", Query: "question"}
+	first, err := service.Ask(context.Background(), base)
+	if err != nil {
+		t.Fatalf("Ask: %v", err)
+	}
+	second, err := service.Ask(context.Background(), base)
+	if err != nil {
+		t.Fatalf("replay: %v", err)
+	}
+	if first.Answer != "answer" || !second.Replayed || first.TurnID != second.TurnID {
+		t.Fatalf("first=%+v second=%+v", first, second)
+	}
+	client.mu.Lock()
+	defer client.mu.Unlock()
+	if client.sessions != 1 || len(client.queries) != 1 || client.queries[0] != "Alice: question" {
+		t.Fatalf("sessions=%d queries=%v", client.sessions, client.queries)
+	}
+}
@@ -0,0 +1,61 @@
+package session
+
+import (
+	"errors"
+	"fmt"
+	"strings"
+
+	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
+)
+
+type GroupPolicy string
+
+const (
+	GroupShared  GroupPolicy = "shared"
+	GroupPerUser GroupPolicy = "per_user"
+)
+
+type Identity struct {
+	TenantKey        string
+	Channel          string
+	AgentID          string
+	ConversationType core.ConversationType
+	ExternalChatID   string
+	ExternalUserID   string
+	GroupPolicy      GroupPolicy
+}
+
+func BuildKey(identity Identity) (key string, subjectID string, err error) {
+	parts := []string{identity.TenantKey, identity.Channel, identity.AgentID}
+	for _, part := range parts {
+		if strings.TrimSpace(part) == "" {
+			return "", "", errors.New("tenant, channel and agent are required")
+		}
+	}
+
+	switch identity.ConversationType {
+	case core.ConversationDirect:
+		subjectID = strings.TrimSpace(identity.ExternalUserID)
+		if subjectID == "" {
+			return "", "", errors.New("direct conversation requires external user id")
+		}
+		return fmt.Sprintf("v1:%s:%s:%s:user:%s", identity.TenantKey, identity.Channel, identity.AgentID, subjectID), subjectID, nil
+	case core.ConversationGroup:
+		chatID := strings.TrimSpace(identity.ExternalChatID)
+		if chatID == "" {
+			return "", "", errors.New("group conversation requires external chat id")
+		}
+		if identity.GroupPolicy == GroupPerUser {
+			userID := strings.TrimSpace(identity.ExternalUserID)
+			if userID == "" {
+				return "", "", errors.New("per-user group policy requires external user id")
+			}
+			subjectID = chatID + ":" + userID
+			return fmt.Sprintf("v1:%s:%s:%s:group-user:%s:%s", identity.TenantKey, identity.Channel, identity.AgentID, chatID, userID), subjectID, nil
+		}
+		subjectID = chatID
+		return fmt.Sprintf("v1:%s:%s:%s:group:%s", identity.TenantKey, identity.Channel, identity.AgentID, chatID), subjectID, nil
+	default:
+		return "", "", fmt.Errorf("unsupported conversation type %q", identity.ConversationType)
+	}
+}
@@ -0,0 +1,42 @@
+package session
+
+import (
+	"testing"
+
+	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
+)
+
+func TestBuildKeyPolicies(t *testing.T) {
+	tests := []struct {
+		name string
+		in   Identity
+		want string
+	}{
+		{
+			name: "direct user long lived",
+			in:   Identity{TenantKey: "tenant-a", Channel: "dingtalk", AgentID: "agent-1", ConversationType: core.ConversationDirect, ExternalUserID: "user-7"},
+			want: "v1:tenant-a:dingtalk:agent-1:user:user-7",
+		},
+		{
+			name: "group shared",
+			in:   Identity{TenantKey: "tenant-a", Channel: "dingtalk", AgentID: "agent-1", ConversationType: core.ConversationGroup, ExternalChatID: "group-9", ExternalUserID: "user-7", GroupPolicy: GroupShared},
+			want: "v1:tenant-a:dingtalk:agent-1:group:group-9",
+		},
+		{
+			name: "group per user",
+			in:   Identity{TenantKey: "tenant-a", Channel: "dingtalk", AgentID: "agent-1", ConversationType: core.ConversationGroup, ExternalChatID: "group-9", ExternalUserID: "user-7", GroupPolicy: GroupPerUser},
+			want: "v1:tenant-a:dingtalk:agent-1:group-user:group-9:user-7",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, _, err := BuildKey(tt.in)
+			if err != nil {
+				t.Fatalf("BuildKey: %v", err)
+			}
+			if got != tt.want {
+				t.Fatalf("key=%q want=%q", got, tt.want)
+			}
+		})
+	}
+}
@@ -0,0 +1,104 @@
+package postgres
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
+)
+
+type Store struct{ pool *pgxpool.Pool }
+
+func Open(ctx context.Context, databaseURL string) (*Store, error) {
+	if databaseURL == "" {
+		return nil, errors.New("DATABASE_URL is required")
+	}
+	pool, err := pgxpool.New(ctx, databaseURL)
+	if err != nil {
+		return nil, fmt.Errorf("open postgres: %w", err)
+	}
+	if err := pool.Ping(ctx); err != nil {
+		pool.Close()
+		return nil, fmt.Errorf("ping postgres: %w", err)
+	}
+	return &Store{pool: pool}, nil
+}
+func (s *Store) Close() {
+	if s != nil && s.pool != nil {
+		s.pool.Close()
+	}
+}
+
+func (s *Store) BeginTurn(ctx context.Context, turn core.Turn) (core.Turn, bool, error) {
+	command, err := s.pool.Exec(ctx, `INSERT INTO turns
+		(id, tenant_key, channel, external_message_id, session_key, question, status)
+		VALUES ($1,$2,$3,$4,$5,$6,'processing') ON CONFLICT DO NOTHING`,
+		turn.ID, turn.TenantKey, turn.Channel, turn.ExternalMessageID, turn.SessionKey, turn.Question)
+	if err != nil {
+		return core.Turn{}, false, err
+	}
+	if command.RowsAffected() == 1 {
+		return turn, true, nil
+	}
+	stored, err := s.loadTurn(ctx, turn.TenantKey, turn.Channel, turn.ExternalMessageID)
+	return stored, false, err
+}
+
+func (s *Store) ResolveSession(ctx context.Context, mapping core.SessionMapping, create func(context.Context) (string, error)) (string, error) {
+	tx, err := s.pool.BeginTx(ctx, pgx.TxOptions{})
+	if err != nil {
+		return "", err
+	}
+	defer func() { _ = tx.Rollback(ctx) }()
+	if _, err = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, mapping.SessionKey); err != nil {
+		return "", err
+	}
+	var sessionID string
+	err = tx.QueryRow(ctx, `SELECT weknora_session_id FROM session_mappings WHERE session_key=$1`, mapping.SessionKey).Scan(&sessionID)
+	if err == nil {
+		return sessionID, tx.Commit(ctx)
+	}
+	if !errors.Is(err, pgx.ErrNoRows) {
+		return "", err
+	}
+	sessionID, err = create(ctx)
+	if err != nil {
+		return "", err
+	}
+	_, err = tx.Exec(ctx, `INSERT INTO session_mappings
+		(session_key,tenant_key,channel,agent_id,conversation_type,external_subject_id,weknora_session_id)
+		VALUES($1,$2,$3,$4,$5,$6,$7)`, mapping.SessionKey, mapping.TenantKey, mapping.Channel, mapping.AgentID, string(mapping.ConversationType), mapping.ExternalSubjectID, sessionID)
+	if err != nil {
+		return "", err
+	}
+	return sessionID, tx.Commit(ctx)
+}
+
+func (s *Store) FinishTurn(ctx context.Context, turn core.Turn) error {
+	references, err := json.Marshal(turn.References)
+	if err != nil {
+		return err
+	}
+	_, err = s.pool.Exec(ctx, `UPDATE turns SET weknora_session_id=$2,weknora_request_id=$3,answer=$4,references=$5,status=$6,error_message=$7,updated_at=NOW() WHERE id=$1`, turn.ID, turn.WeKnoraSessionID, turn.WeKnoraRequestID, turn.Answer, references, turn.Status, turn.ErrorMessage)
+	return err
+}
+
+func (s *Store) loadTurn(ctx context.Context, tenantKey, channel, messageID string) (core.Turn, error) {
+	var turn core.Turn
+	var references []byte
+	err := s.pool.QueryRow(ctx, `SELECT id,tenant_key,channel,external_message_id,session_key,weknora_session_id,weknora_request_id,question,answer,references,status,error_message FROM turns WHERE tenant_key=$1 AND channel=$2 AND external_message_id=$3`, tenantKey, channel, messageID).Scan(&turn.ID, &turn.TenantKey, &turn.Channel, &turn.ExternalMessageID, &turn.SessionKey, &turn.WeKnoraSessionID, &turn.WeKnoraRequestID, &turn.Question, &turn.Answer, &references, &turn.Status, &turn.ErrorMessage)
+	if err != nil {
+		return core.Turn{}, err
+	}
+	if len(references) > 0 {
+		if err := json.Unmarshal(references, &turn.References); err != nil {
+			return core.Turn{}, err
+		}
+	}
+	return turn, nil
+}
@@ -0,0 +1,203 @@
+package weknora
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+
+	"gitea.zheromiku.top/Weknora/dobotai-platform/internal/core"
+)
+
+var ErrIncompleteStream = errors.New("weknora stream ended without a terminal event")
+
+type Client struct {
+	baseURL string
+	apiKey  string
+	http    *http.Client
+}
+
+type AgentChatRequest struct {
+	Query            string   `json:"query"`
+	AgentEnabled     bool     `json:"agent_enabled"`
+	AgentID          string   `json:"agent_id"`
+	Channel          string   `json:"channel,omitempty"`
+	EnableMemory     bool     `json:"enable_memory,omitempty"`
+	DisableTitle     bool     `json:"disable_title,omitempty"`
+	KnowledgeBaseIDs []string `json:"knowledge_base_ids,omitempty"`
+	Images           []Image  `json:"images,omitempty"`
+}
+
+type Image struct {
+	Data string `json:"data"`
+}
+
+type ChatResult struct {
+	RequestID  string
+	Answer     string
+	References []core.Reference
+}
+
+func NewClient(baseURL, apiKey string, timeout time.Duration) (*Client, error) {
+	parsed, err := url.Parse(strings.TrimRight(strings.TrimSpace(baseURL), "/"))
+	if err != nil || parsed.Scheme == "" || parsed.Host == "" {
+		return nil, errors.New("invalid WEKNORA_BASE_URL")
+	}
+	if strings.TrimSpace(apiKey) == "" {
+		return nil, errors.New("WEKNORA_API_KEY is required")
+	}
+	if timeout <= 0 {
+		timeout = 120 * time.Second
+	}
+	return &Client{baseURL: parsed.String(), apiKey: apiKey, http: &http.Client{Timeout: timeout}}, nil
+}
+
+func (c *Client) CreateSession(ctx context.Context, title, description string) (string, error) {
+	var response struct {
+		Success bool `json:"success"`
+		Data    struct {
+			ID string `json:"id"`
+		} `json:"data"`
+	}
+	if err := c.doJSON(ctx, http.MethodPost, "/api/v1/sessions", map[string]string{
+		"title": title, "description": description,
+	}, &response); err != nil {
+		return "", err
+	}
+	if !response.Success || strings.TrimSpace(response.Data.ID) == "" {
+		return "", errors.New("weknora returned an empty session id")
+	}
+	return response.Data.ID, nil
+}
+
+func (c *Client) AgentChat(ctx context.Context, sessionID string, request AgentChatRequest) (ChatResult, error) {
+	if strings.TrimSpace(sessionID) == "" || strings.TrimSpace(request.AgentID) == "" || strings.TrimSpace(request.Query) == "" {
+		return ChatResult{}, errors.New("session id, agent id and query are required")
+	}
+	request.AgentEnabled = true
+	body, err := json.Marshal(request)
+	if err != nil {
+		return ChatResult{}, fmt.Errorf("encode agent chat request: %w", err)
+	}
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/api/v1/agent-chat/"+url.PathEscape(sessionID), bytes.NewReader(body))
+	if err != nil {
+		return ChatResult{}, fmt.Errorf("create agent chat request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("Accept", "text/event-stream")
+	req.Header.Set("X-API-Key", c.apiKey)
+	resp, err := c.http.Do(req)
+	if err != nil {
+		return ChatResult{}, fmt.Errorf("call weknora agent chat: %w", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		message, _ := io.ReadAll(io.LimitReader(resp.Body, 32<<10))
+		return ChatResult{}, fmt.Errorf("weknora agent chat returned HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(message)))
+	}
+	return parseSSE(resp.Body)
+}
+
+func (c *Client) doJSON(ctx context.Context, method, path string, input, output any) error {
+	body, err := json.Marshal(input)
+	if err != nil {
+		return fmt.Errorf("encode weknora request: %w", err)
+	}
+	req, err := http.NewRequestWithContext(ctx, method, c.baseURL+path, bytes.NewReader(body))
+	if err != nil {
+		return fmt.Errorf("create weknora request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+	req.Header.Set("X-API-Key", c.apiKey)
+	resp, err := c.http.Do(req)
+	if err != nil {
+		return fmt.Errorf("call weknora: %w", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		message, _ := io.ReadAll(io.LimitReader(resp.Body, 32<<10))
+		return fmt.Errorf("weknora returned HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(message)))
+	}
+	if err := json.NewDecoder(io.LimitReader(resp.Body, 2<<20)).Decode(output); err != nil {
+		return fmt.Errorf("decode weknora response: %w", err)
+	}
+	return nil
+}
+
+func parseSSE(reader io.Reader) (ChatResult, error) {
+	scanner := bufio.NewScanner(reader)
+	scanner.Buffer(make([]byte, 64<<10), 2<<20)
+	var result ChatResult
+	var dataLines []string
+	terminal := false
+
+	flush := func() error {
+		if len(dataLines) == 0 {
+			return nil
+		}
+		raw := strings.Join(dataLines, "\n")
+		dataLines = dataLines[:0]
+		var event struct {
+			ID                  string           `json:"id"`
+			ResponseType        string           `json:"response_type"`
+			Content             string           `json:"content"`
+			Done                bool             `json:"done"`
+			KnowledgeReferences []core.Reference `json:"knowledge_references"`
+		}
+		if err := json.Unmarshal([]byte(raw), &event); err != nil {
+			return fmt.Errorf("decode weknora SSE event: %w", err)
+		}
+		if event.ID != "" {
+			result.RequestID = event.ID
+		}
+		switch event.ResponseType {
+		case "references":
+			result.References = append([]core.Reference(nil), event.KnowledgeReferences...)
+		case "answer":
+			result.Answer += event.Content
+			if event.Done {
+				terminal = true
+			}
+		case "error":
+			if event.Content == "" {
+				event.Content = "weknora returned an unspecified error"
+			}
+			return errors.New(event.Content)
+		case "complete":
+			if event.Done {
+				terminal = true
+			}
+		}
+		return nil
+	}
+
+	for scanner.Scan() {
+		line := scanner.Text()
+		if line == "" {
+			if err := flush(); err != nil {
+				return ChatResult{}, err
+			}
+			continue
+		}
+		if strings.HasPrefix(line, "data:") {
+			dataLines = append(dataLines, strings.TrimSpace(strings.TrimPrefix(line, "data:")))
+		}
+	}
+	if err := scanner.Err(); err != nil {
+		return ChatResult{}, fmt.Errorf("read weknora SSE stream: %w", err)
+	}
+	if err := flush(); err != nil {
+		return ChatResult{}, err
+	}
+	if !terminal {
+		return ChatResult{}, ErrIncompleteStream
+	}
+	return result, nil
+}
@@ -0,0 +1,65 @@
+package weknora
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestClientCreateSessionAndAgentChat(t *testing.T) {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/api/v1/sessions", func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get("X-API-Key") != "test-key" {
+			t.Errorf("missing api key")
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = w.Write([]byte(`{"success":true,"data":{"id":"session-1"}}`))
+	})
+	mux.HandleFunc("/api/v1/agent-chat/session-1", func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get("Accept") != "text/event-stream" {
+			t.Errorf("missing SSE accept header")
+		}
+		w.Header().Set("Content-Type", "text/event-stream")
+		_, _ = w.Write([]byte("data: {\"id\":\"request-1\",\"response_type\":\"references\",\"knowledge_references\":[{\"knowledge_id\":\"kb-1\",\"content\":\"reference\"}]}\n\n"))
+		_, _ = w.Write([]byte("data: {\"id\":\"request-1\",\"response_type\":\"answer\",\"content\":\"hello \",\"done\":false}\n\n"))
+		_, _ = w.Write([]byte("data: {\"id\":\"request-1\",\"response_type\":\"answer\",\"content\":\"world\",\"done\":true}\n\n"))
+	})
+	server := httptest.NewServer(mux)
+	defer server.Close()
+
+	client, err := NewClient(server.URL, "test-key", 5*time.Second)
+	if err != nil {
+		t.Fatalf("NewClient: %v", err)
+	}
+	sessionID, err := client.CreateSession(context.Background(), "title", "description")
+	if err != nil {
+		t.Fatalf("CreateSession: %v", err)
+	}
+	result, err := client.AgentChat(context.Background(), sessionID, AgentChatRequest{Query: "question", AgentID: "agent-1", Channel: "im"})
+	if err != nil {
+		t.Fatalf("AgentChat: %v", err)
+	}
+	if result.Answer != "hello world" || result.RequestID != "request-1" || len(result.References) != 1 {
+		t.Fatalf("unexpected result: %+v", result)
+	}
+}
+
+func TestParseSSEFailsClosedOnPrematureEOF(t *testing.T) {
+	_, err := parseSSE(strings.NewReader("data: {\"response_type\":\"answer\",\"content\":\"partial\",\"done\":false}\n\n"))
+	if !errors.Is(err, ErrIncompleteStream) {
+		t.Fatalf("error=%v want ErrIncompleteStream", err)
+	}
+}
+
+func TestParseSSEErrorDominatesPartialAnswer(t *testing.T) {
+	stream := "data: {\"response_type\":\"answer\",\"content\":\"partial\",\"done\":false}\n\n" +
+		"data: {\"response_type\":\"error\",\"content\":\"vision model unavailable\",\"done\":true}\n\n"
+	_, err := parseSSE(strings.NewReader(stream))
+	if err == nil || !strings.Contains(err.Error(), "vision model unavailable") {
+		t.Fatalf("expected terminal vision error, got %v", err)
+	}
+}