update see calss
This commit is contained in:
@@ -20,6 +20,7 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiaudit"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiprompts"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiprovider"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/catalog"
|
||||
@@ -58,12 +59,14 @@ func main() {
|
||||
log.Fatalf("db ping: %v", err)
|
||||
}
|
||||
|
||||
pipeline := newWorkerLikePipeline(ctx, pool, cfg)
|
||||
pipeline, auditedSvc := newWorkerLikePipeline(ctx, pool, cfg)
|
||||
if strings.TrimSpace(*llmBase) != "" {
|
||||
// Everything else stays production-identical; only the provider endpoint is
|
||||
// pinned so a local run does not depend on the configured remote model.
|
||||
pipeline.AI = fixedCompleter{
|
||||
client: processing.NewOpenAIClient(*llmKey, *llmBase, *llmModel, 0, 2),
|
||||
client: auditedSvc.WrapAudit(
|
||||
processing.NewOpenAIClient(*llmKey, *llmBase, *llmModel, 0, 2),
|
||||
uuid.Nil, "processing", processing.AIProviderInternal),
|
||||
}
|
||||
log.Printf("LLM override: base=%s model=%s", *llmBase, *llmModel)
|
||||
}
|
||||
@@ -93,7 +96,7 @@ func (f fixedCompleter) ResolveCompleterForRole(ctx context.Context, id uuid.UUI
|
||||
|
||||
// newWorkerLikePipeline mirrors cmd/worker/main.go so this proof exercises the same
|
||||
// code path production does.
|
||||
func newWorkerLikePipeline(ctx context.Context, pool *pgxpool.Pool, cfg config.Config) *processing.Pipeline {
|
||||
func newWorkerLikePipeline(ctx context.Context, pool *pgxpool.Pool, cfg config.Config) (*processing.Pipeline, *aiprovider.Service) {
|
||||
platSettings := platformsettings.NewService(pool, platformsettings.EnvConfig{
|
||||
AppEncryptionKey: cfg.AppEncryptionKey,
|
||||
CredentialsEncryptionKey: cfg.CredentialsEncryptionKey,
|
||||
@@ -115,6 +118,7 @@ func newWorkerLikePipeline(ctx context.Context, pool *pgxpool.Pool, cfg config.C
|
||||
ProcessingMaxRetries: cfg.ProcessingMaxRetries,
|
||||
})
|
||||
aiSvc.Platform = platSettings
|
||||
aiSvc.WithAudit(aiaudit.NewRecorder(pool))
|
||||
p := processing.NewPipeline(pool)
|
||||
p.AI = aiSvc
|
||||
p.Prompts = aiprompts.NewService(pool)
|
||||
@@ -130,7 +134,7 @@ func newWorkerLikePipeline(ctx context.Context, pool *pgxpool.Pool, cfg config.C
|
||||
} else {
|
||||
log.Printf("OpenAI: source=%s base=%s model=%s", oi.Source, oi.BaseURL, oi.Model)
|
||||
}
|
||||
return p
|
||||
return p, aiSvc
|
||||
}
|
||||
|
||||
func runExisting(ctx context.Context, pool *pgxpool.Pool, p *processing.Pipeline, companyName, gtin string, limit int, ptype string) error {
|
||||
@@ -179,9 +183,15 @@ func runExisting(ctx context.Context, pool *pgxpool.Pool, p *processing.Pipeline
|
||||
|
||||
func processAndReport(ctx context.Context, pool *pgxpool.Pool, p *processing.Pipeline, companyID uuid.UUID, rawIDs []uuid.UUID, ptype string) error {
|
||||
// Clear prior enhance hashes so the run cannot be skipped as "unchanged".
|
||||
// The hash lives twice: once on field_sources and once per language inside
|
||||
// localized_content — clearing only the first still hash-skips the LLM.
|
||||
if _, err := pool.Exec(ctx, `
|
||||
UPDATE processed_products
|
||||
SET field_sources = field_sources - 'enhance_input_hash'
|
||||
SET field_sources = field_sources - 'enhance_input_hash',
|
||||
localized_content = COALESCE((
|
||||
SELECT jsonb_object_agg(k, v - 'enhance_input_hash')
|
||||
FROM jsonb_each(COALESCE(localized_content, '{}'::jsonb)) AS t(k, v)
|
||||
), '{}'::jsonb)
|
||||
WHERE company_id = $1 AND raw_product_id = ANY($2)`, companyID, rawIDs); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiaudit"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiprompts"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiprovider"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/billing"
|
||||
@@ -95,6 +96,9 @@ func main() {
|
||||
ProcessingMaxRetries: cfg.ProcessingMaxRetries,
|
||||
})
|
||||
aiSvc.Platform = platSettings
|
||||
// Full prompt/response capture for /admin/ai-calls (pruned after
|
||||
// aiaudit.RetentionDays by the retention tick below).
|
||||
aiSvc.WithAudit(aiaudit.NewRecorder(pool))
|
||||
|
||||
pipeline := processing.NewPipeline(pool)
|
||||
pipeline.BatchSize = cfg.ProcessingBatchSize
|
||||
@@ -371,6 +375,11 @@ func main() {
|
||||
} else if res.JobsDeleted > 0 {
|
||||
log.Printf("retention cleanup jobs_deleted=%d", res.JobsDeleted)
|
||||
}
|
||||
if n, err := aiaudit.CleanupExpired(ctx, pool); err != nil {
|
||||
log.Printf("worker: ai call log cleanup: %v", err)
|
||||
} else if n > 0 {
|
||||
log.Printf("worker: ai call log cleanup removed=%d older_than_days=%d", n, aiaudit.RetentionDays)
|
||||
}
|
||||
if res, err := processing.CleanupExpiredSyncJobs(ctx, pool); err != nil {
|
||||
log.Printf("expired sync cleanup: %v", err)
|
||||
} else if res.SyncJobsDeleted > 0 {
|
||||
|
||||
@@ -0,0 +1,213 @@
|
||||
// Package aiaudit records the exact prompt and response of every LLM call so
|
||||
// platform admins can inspect them at /admin/ai-calls.
|
||||
//
|
||||
// It is deliberately a leaf package: it imports neither processing nor aiprovider,
|
||||
// so processing can attach per-call context (job, product, role) without an import
|
||||
// cycle while aiprovider does the actual Completer wrapping.
|
||||
//
|
||||
// Rows are a short-lived debugging buffer, not an audit trail — CleanupExpired
|
||||
// prunes them after RetentionDays. Never build billing or reporting on them.
|
||||
package aiaudit
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
// Roles mirror aiprovider role ids plus "probe" for admin connectivity tests.
|
||||
const (
|
||||
RoleProcessing = "processing"
|
||||
RoleCategorize = "categorize"
|
||||
RoleSEOMeta = "seo_meta"
|
||||
RoleCampaign = "campaign"
|
||||
RoleSupport = "support"
|
||||
RoleProbe = "probe"
|
||||
RoleOther = "other"
|
||||
)
|
||||
|
||||
// RetentionDays bounds how long captured prompts live. Prompts embed the tenant's
|
||||
// product copy and are large, so the window is intentionally short.
|
||||
const RetentionDays = 7
|
||||
|
||||
// maxBodyRunes caps each stored body. A category-formula prompt is a few KB; this
|
||||
// leaves room for outliers while stopping a runaway reply from bloating the table.
|
||||
const maxBodyRunes = 60000
|
||||
|
||||
// Call is one recorded LLM exchange.
|
||||
type Call struct {
|
||||
CompanyID uuid.UUID
|
||||
UserID uuid.UUID
|
||||
JobID uuid.UUID
|
||||
RawProductID uuid.UUID
|
||||
Role string
|
||||
ProviderMode string
|
||||
Model string
|
||||
System string
|
||||
User string
|
||||
Response string
|
||||
Error string
|
||||
FinishReason string
|
||||
PromptTokens int
|
||||
OutputTokens int
|
||||
TotalTokens int
|
||||
Duration time.Duration
|
||||
}
|
||||
|
||||
// CallContext is the per-call metadata processing/campaigns attach to ctx before
|
||||
// invoking a Completer. The recorder merges it into every Call it writes.
|
||||
type CallContext struct {
|
||||
CompanyID uuid.UUID
|
||||
UserID uuid.UUID
|
||||
JobID uuid.UUID
|
||||
RawProductID uuid.UUID
|
||||
Role string
|
||||
}
|
||||
|
||||
type ctxKey struct{}
|
||||
|
||||
// WithCall returns ctx carrying call metadata for the recorder. Fields left zero
|
||||
// on cc do not clear values already present, so an outer scope can set company/user
|
||||
// once per job and an inner scope add the product.
|
||||
func WithCall(ctx context.Context, cc CallContext) context.Context {
|
||||
cur := FromContext(ctx)
|
||||
if cc.CompanyID != uuid.Nil {
|
||||
cur.CompanyID = cc.CompanyID
|
||||
}
|
||||
if cc.UserID != uuid.Nil {
|
||||
cur.UserID = cc.UserID
|
||||
}
|
||||
if cc.JobID != uuid.Nil {
|
||||
cur.JobID = cc.JobID
|
||||
}
|
||||
if cc.RawProductID != uuid.Nil {
|
||||
cur.RawProductID = cc.RawProductID
|
||||
}
|
||||
if strings.TrimSpace(cc.Role) != "" {
|
||||
cur.Role = cc.Role
|
||||
}
|
||||
return context.WithValue(ctx, ctxKey{}, cur)
|
||||
}
|
||||
|
||||
// FromContext reads call metadata attached by WithCall (zero value when absent).
|
||||
func FromContext(ctx context.Context) CallContext {
|
||||
if ctx == nil {
|
||||
return CallContext{}
|
||||
}
|
||||
cc, _ := ctx.Value(ctxKey{}).(CallContext)
|
||||
return cc
|
||||
}
|
||||
|
||||
// Recorder writes captured calls. A nil Recorder is a no-op, so callers never need
|
||||
// to branch on whether capture is wired.
|
||||
type Recorder struct {
|
||||
pool *pgxpool.Pool
|
||||
|
||||
mu sync.Mutex
|
||||
failed bool // stop log-spamming once writes are known to fail
|
||||
}
|
||||
|
||||
// NewRecorder returns a Recorder writing to pool. nil pool yields a no-op recorder.
|
||||
func NewRecorder(pool *pgxpool.Pool) *Recorder {
|
||||
if pool == nil {
|
||||
return nil
|
||||
}
|
||||
return &Recorder{pool: pool}
|
||||
}
|
||||
|
||||
const insertCallSQL = `
|
||||
INSERT INTO ai_call_logs (
|
||||
company_id, user_id, job_id, raw_product_id, role, provider_mode, model,
|
||||
system_prompt, user_prompt, response_text, error, finish_reason,
|
||||
prompt_tokens, completion_tokens, total_tokens, duration_ms)
|
||||
VALUES (
|
||||
$1, $2, $3, $4, $5, $6, $7,
|
||||
$8, $9, $10, $11, $12,
|
||||
$13, $14, $15, $16)`
|
||||
|
||||
// Record persists one call, merging any CallContext on ctx. Capture must never
|
||||
// break the request it is observing: every failure is swallowed after one log line.
|
||||
//
|
||||
// The write uses context.WithoutCancel so a cancelled/timed-out job still records
|
||||
// the call that was in flight — those are exactly the ones worth inspecting.
|
||||
func (r *Recorder) Record(ctx context.Context, c Call) {
|
||||
if r == nil || r.pool == nil {
|
||||
return
|
||||
}
|
||||
cc := FromContext(ctx)
|
||||
if c.CompanyID == uuid.Nil {
|
||||
c.CompanyID = cc.CompanyID
|
||||
}
|
||||
if c.UserID == uuid.Nil {
|
||||
c.UserID = cc.UserID
|
||||
}
|
||||
if c.JobID == uuid.Nil {
|
||||
c.JobID = cc.JobID
|
||||
}
|
||||
if c.RawProductID == uuid.Nil {
|
||||
c.RawProductID = cc.RawProductID
|
||||
}
|
||||
if strings.TrimSpace(c.Role) == "" {
|
||||
c.Role = cc.Role
|
||||
}
|
||||
if strings.TrimSpace(c.Role) == "" {
|
||||
c.Role = RoleOther
|
||||
}
|
||||
|
||||
writeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
|
||||
defer cancel()
|
||||
_, err := r.pool.Exec(writeCtx, insertCallSQL,
|
||||
nullUUID(c.CompanyID), nullUUID(c.UserID), nullUUID(c.JobID), nullUUID(c.RawProductID),
|
||||
c.Role, c.ProviderMode, c.Model,
|
||||
clamp(c.System), clamp(c.User), clamp(c.Response), clamp(c.Error), c.FinishReason,
|
||||
c.PromptTokens, c.OutputTokens, c.TotalTokens, int(c.Duration.Milliseconds()),
|
||||
)
|
||||
if err != nil {
|
||||
r.noteFailure(err)
|
||||
}
|
||||
}
|
||||
|
||||
func (r *Recorder) noteFailure(err error) {
|
||||
r.mu.Lock()
|
||||
first := !r.failed
|
||||
r.failed = true
|
||||
r.mu.Unlock()
|
||||
if first {
|
||||
log.Printf("aiaudit: capture disabled after write error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func nullUUID(id uuid.UUID) any {
|
||||
if id == uuid.Nil {
|
||||
return nil
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
func clamp(s string) string {
|
||||
rs := []rune(s)
|
||||
if len(rs) <= maxBodyRunes {
|
||||
return s
|
||||
}
|
||||
return string(rs[:maxBodyRunes]) + "\n…[truncated by aiaudit]"
|
||||
}
|
||||
|
||||
// CleanupExpired deletes captured calls older than RetentionDays and returns how
|
||||
// many rows went. Called from the worker's retention tick.
|
||||
func CleanupExpired(ctx context.Context, pool *pgxpool.Pool) (int64, error) {
|
||||
if pool == nil {
|
||||
return 0, nil
|
||||
}
|
||||
tag, err := pool.Exec(ctx, `
|
||||
DELETE FROM ai_call_logs
|
||||
WHERE created_at < now() - ($1 || ' days')::interval`, RetentionDays)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return tag.RowsAffected(), nil
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
package aiaudit
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
// Call metadata is attached in layers: the job sets company/user/job once, then
|
||||
// each product narrows it. A later WithCall must not wipe what an outer scope set.
|
||||
func TestWithCall_mergesWithoutClearing(t *testing.T) {
|
||||
t.Parallel()
|
||||
company, user, job, product := uuid.New(), uuid.New(), uuid.New(), uuid.New()
|
||||
|
||||
ctx := WithCall(context.Background(), CallContext{
|
||||
CompanyID: company, UserID: user, JobID: job, Role: RoleProcessing,
|
||||
})
|
||||
ctx = WithCall(ctx, CallContext{RawProductID: product})
|
||||
|
||||
got := FromContext(ctx)
|
||||
if got.CompanyID != company || got.UserID != user || got.JobID != job {
|
||||
t.Fatalf("outer scope lost: %+v", got)
|
||||
}
|
||||
if got.RawProductID != product {
|
||||
t.Fatalf("product not attached: %+v", got)
|
||||
}
|
||||
if got.Role != RoleProcessing {
|
||||
t.Fatalf("role lost: %q", got.Role)
|
||||
}
|
||||
|
||||
// A narrower scope may override the role (categorize inside a processing job).
|
||||
ctx = WithCall(ctx, CallContext{Role: RoleCategorize})
|
||||
if got := FromContext(ctx); got.Role != RoleCategorize || got.JobID != job {
|
||||
t.Fatalf("role override broke context: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFromContext_zeroValueWhenAbsent(t *testing.T) {
|
||||
t.Parallel()
|
||||
if got := FromContext(context.Background()); got != (CallContext{}) {
|
||||
t.Fatalf("expected zero CallContext, got %+v", got)
|
||||
}
|
||||
//nolint:staticcheck // explicitly asserting the nil-ctx guard
|
||||
if got := FromContext(nil); got != (CallContext{}) {
|
||||
t.Fatalf("nil ctx must be safe, got %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Capture must never be able to take down the call it observes.
|
||||
func TestRecorder_nilIsNoOp(t *testing.T) {
|
||||
t.Parallel()
|
||||
var r *Recorder
|
||||
r.Record(context.Background(), Call{System: "x"}) // must not panic
|
||||
if got := NewRecorder(nil); got != nil {
|
||||
t.Fatal("nil pool must yield a nil (no-op) recorder")
|
||||
}
|
||||
}
|
||||
|
||||
// A runaway reply must not write an unbounded row.
|
||||
func TestClamp_boundsBody(t *testing.T) {
|
||||
t.Parallel()
|
||||
short := strings.Repeat("a", 10)
|
||||
if clamp(short) != short {
|
||||
t.Fatal("short bodies must pass through unchanged")
|
||||
}
|
||||
long := strings.Repeat("b", maxBodyRunes+500)
|
||||
got := clamp(long)
|
||||
if len([]rune(got)) <= maxBodyRunes {
|
||||
t.Fatalf("clamped body should keep the cap plus a marker, got %d runes", len([]rune(got)))
|
||||
}
|
||||
if !strings.HasSuffix(got, "[truncated by aiaudit]") {
|
||||
t.Fatal("clamped body must say it was truncated")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,163 @@
|
||||
package aiprovider
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiaudit"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/processing"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
// Every Completer this service hands out is wrapped here, so the admin AI
|
||||
// inspector sees processing, categorize, SEO meta, campaign and support calls from
|
||||
// one capture point instead of each caller remembering to log.
|
||||
//
|
||||
// Callers attach per-call detail (job, product, role) with aiaudit.WithCall; the
|
||||
// wrapper fills in what the resolver already knows (company, provider mode, model).
|
||||
|
||||
// Audit enables prompt/response capture. Nil disables it and the wrapper is skipped
|
||||
// entirely, so a service built without a recorder behaves exactly as before.
|
||||
func (s *Service) WithAudit(rec *aiaudit.Recorder) *Service {
|
||||
if s == nil {
|
||||
return s
|
||||
}
|
||||
s.Audit = rec
|
||||
return s
|
||||
}
|
||||
|
||||
type auditingCompleter struct {
|
||||
inner processing.Completer
|
||||
rec *aiaudit.Recorder
|
||||
companyID uuid.UUID
|
||||
role string
|
||||
providerMode string
|
||||
}
|
||||
|
||||
// Enabled forwards the wrapped client's gate so CompleterEnabled keeps working
|
||||
// (processing checks for the EnableChecker interface, not a concrete type).
|
||||
func (a *auditingCompleter) Enabled() bool {
|
||||
if c, ok := a.inner.(processing.EnableChecker); ok {
|
||||
return c.Enabled()
|
||||
}
|
||||
return a.inner != nil
|
||||
}
|
||||
|
||||
// ProviderModeLabel forwards the wrapped client's analytics label.
|
||||
func (a *auditingCompleter) ProviderModeLabel() string {
|
||||
type labeler interface{ ProviderModeLabel() string }
|
||||
if c, ok := a.inner.(labeler); ok {
|
||||
return c.ProviderModeLabel()
|
||||
}
|
||||
return a.providerMode
|
||||
}
|
||||
|
||||
func (a *auditingCompleter) Complete(ctx context.Context, system, user string) (processing.Completion, error) {
|
||||
return a.record(ctx, system, user, func() (processing.Completion, error) {
|
||||
return a.inner.Complete(ctx, system, user)
|
||||
})
|
||||
}
|
||||
|
||||
// auditingCompleterWithOptions is used when the wrapped client honours
|
||||
// CompleteOptions. processing.CompleteOnce prefers that interface and only falls
|
||||
// back to Complete, so a wrapper that did not forward it would silently drop
|
||||
// MaxTokens / Temperature / ReasoningEffort from every enhance call.
|
||||
type auditingCompleterWithOptions struct {
|
||||
*auditingCompleter
|
||||
inner processing.CompleterWithOptions
|
||||
}
|
||||
|
||||
func (a *auditingCompleterWithOptions) CompleteWithOptions(ctx context.Context, system, user string, opts processing.CompleteOptions) (processing.Completion, error) {
|
||||
return a.record(ctx, system, user, func() (processing.Completion, error) {
|
||||
return a.inner.CompleteWithOptions(ctx, system, user, opts)
|
||||
})
|
||||
}
|
||||
|
||||
func (a *auditingCompleter) record(ctx context.Context, system, user string, call func() (processing.Completion, error)) (processing.Completion, error) {
|
||||
started := time.Now()
|
||||
comp, err := call()
|
||||
// The call site knows the role better than the resolver does: one processing
|
||||
// Completer serves both the enhance and categorize steps.
|
||||
role := a.role
|
||||
if cc := aiaudit.FromContext(ctx); strings.TrimSpace(cc.Role) != "" {
|
||||
role = cc.Role
|
||||
}
|
||||
rec := aiaudit.Call{
|
||||
CompanyID: a.companyID,
|
||||
Role: role,
|
||||
ProviderMode: a.providerMode,
|
||||
Model: comp.Model,
|
||||
System: system,
|
||||
User: user,
|
||||
Response: comp.Text,
|
||||
FinishReason: finishReason(comp),
|
||||
PromptTokens: comp.PromptTokens,
|
||||
OutputTokens: comp.OutputTokens,
|
||||
TotalTokens: comp.TotalTokens,
|
||||
Duration: time.Since(started),
|
||||
}
|
||||
if err != nil {
|
||||
// The provider error text is already redacted by processing's OpenAI client.
|
||||
rec.Error = err.Error()
|
||||
}
|
||||
a.rec.Record(ctx, rec)
|
||||
return comp, err
|
||||
}
|
||||
|
||||
func finishReason(comp processing.Completion) string {
|
||||
m, ok := comp.Raw.(map[string]any)
|
||||
if !ok || m == nil {
|
||||
return ""
|
||||
}
|
||||
fr, _ := m["finish_reason"].(string)
|
||||
return strings.TrimSpace(fr)
|
||||
}
|
||||
|
||||
// WrapAudit decorates an externally supplied Completer with the same capture the
|
||||
// resolvers apply. Local tools that pin a provider endpoint (cmd/formula-e2e) use
|
||||
// it so their runs show up in the admin inspector too.
|
||||
func (s *Service) WrapAudit(c processing.Completer, companyID uuid.UUID, role, providerMode string) processing.Completer {
|
||||
return s.wrapAudit(c, companyID, role, providerMode)
|
||||
}
|
||||
|
||||
// wrapAudit decorates a resolved Completer with capture. It is idempotent (a
|
||||
// re-wrap keeps the original) and a no-op when capture is off or c is nil, so the
|
||||
// public resolvers can all call it on their way out.
|
||||
func (s *Service) wrapAudit(c processing.Completer, companyID uuid.UUID, role, providerMode string) processing.Completer {
|
||||
if s == nil || s.Audit == nil || c == nil {
|
||||
return c
|
||||
}
|
||||
switch c.(type) {
|
||||
case *auditingCompleter, *auditingCompleterWithOptions:
|
||||
return c
|
||||
}
|
||||
base := &auditingCompleter{
|
||||
inner: c,
|
||||
rec: s.Audit,
|
||||
companyID: companyID,
|
||||
role: auditRole(role),
|
||||
providerMode: providerMode,
|
||||
}
|
||||
// Only advertise the options interface when the wrapped client actually has it,
|
||||
// so a plain Completer is not handed options it cannot honour.
|
||||
if withOpts, ok := c.(processing.CompleterWithOptions); ok {
|
||||
return &auditingCompleterWithOptions{auditingCompleter: base, inner: withOpts}
|
||||
}
|
||||
return base
|
||||
}
|
||||
|
||||
// auditRole maps an aiprovider role onto the aiaudit vocabulary shown in the admin
|
||||
// filter. Unknown roles are recorded rather than dropped.
|
||||
func auditRole(role string) string {
|
||||
switch strings.TrimSpace(role) {
|
||||
case "", RoleProcessing:
|
||||
return aiaudit.RoleProcessing
|
||||
case RoleSupport:
|
||||
return aiaudit.RoleSupport
|
||||
case RoleDocsAPI:
|
||||
return aiaudit.RoleOther
|
||||
default:
|
||||
return strings.TrimSpace(role)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
package aiprovider
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiaudit"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/processing"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
type optionsStub struct{ withOptions, plain int }
|
||||
|
||||
func (s *optionsStub) Complete(context.Context, string, string) (processing.Completion, error) {
|
||||
s.plain++
|
||||
return processing.Completion{Text: "{}"}, nil
|
||||
}
|
||||
|
||||
func (s *optionsStub) CompleteWithOptions(context.Context, string, string, processing.CompleteOptions) (processing.Completion, error) {
|
||||
s.withOptions++
|
||||
return processing.Completion{Text: "{}"}, nil
|
||||
}
|
||||
|
||||
// Regression: processing.CompleteOnce prefers CompleterWithOptions and only falls
|
||||
// back to Complete. A wrapper that does not forward the options interface silently
|
||||
// drops MaxTokens / Temperature / ReasoningEffort for every enhance call.
|
||||
func TestAuditingCompleter_forwardsCompleteWithOptions(t *testing.T) {
|
||||
t.Parallel()
|
||||
inner := &optionsStub{}
|
||||
svc := &Service{Audit: &aiaudit.Recorder{}}
|
||||
wrapped := svc.WrapAudit(inner, uuid.New(), RoleProcessing, "internal")
|
||||
|
||||
if _, ok := wrapped.(processing.CompleterWithOptions); !ok {
|
||||
t.Fatal("wrapped completer must still satisfy CompleterWithOptions")
|
||||
}
|
||||
if _, err := processing.CompleteOnce(context.Background(), wrapped, "sys", "user",
|
||||
processing.CompleteOptions{MaxTokens: 1234}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if inner.withOptions != 1 || inner.plain != 0 {
|
||||
t.Fatalf("options path not forwarded: withOptions=%d plain=%d", inner.withOptions, inner.plain)
|
||||
}
|
||||
}
|
||||
|
||||
// A plain Completer must not gain the options interface just by being wrapped —
|
||||
// that would send options to a client that cannot honour them.
|
||||
func TestAuditingCompleter_plainCompleterStaysPlain(t *testing.T) {
|
||||
t.Parallel()
|
||||
inner := plainStub{}
|
||||
svc := &Service{Audit: &aiaudit.Recorder{}}
|
||||
wrapped := svc.WrapAudit(inner, uuid.New(), RoleProcessing, "internal")
|
||||
if _, ok := wrapped.(processing.CompleterWithOptions); ok {
|
||||
t.Fatal("plain completer must not advertise CompleterWithOptions")
|
||||
}
|
||||
}
|
||||
|
||||
type plainStub struct{}
|
||||
|
||||
func (plainStub) Complete(context.Context, string, string) (processing.Completion, error) {
|
||||
return processing.Completion{Text: "{}"}, nil
|
||||
}
|
||||
|
||||
func TestWrapAudit_noopWhenCaptureOff(t *testing.T) {
|
||||
t.Parallel()
|
||||
inner := plainStub{}
|
||||
var svc *Service
|
||||
if got := svc.WrapAudit(inner, uuid.Nil, RoleProcessing, ""); got != processing.Completer(inner) {
|
||||
t.Fatal("nil service must return the completer untouched")
|
||||
}
|
||||
svc = &Service{}
|
||||
if got := svc.WrapAudit(inner, uuid.Nil, RoleProcessing, ""); got != processing.Completer(inner) {
|
||||
t.Fatal("capture off must return the completer untouched")
|
||||
}
|
||||
}
|
||||
@@ -54,7 +54,8 @@ func (s *Service) ResolveCompleterForRole(ctx context.Context, companyID uuid.UU
|
||||
return nil, ModeInternalLabel, false, err
|
||||
}
|
||||
if ok && strings.TrimSpace(ep.APIKey) != "" && strings.TrimSpace(ep.Model) != "" {
|
||||
return s.completerFromEndpoint(ep)
|
||||
c, mode, byok, err := s.completerFromEndpoint(ep)
|
||||
return s.wrapAudit(c, companyID, role, mode), mode, byok, err
|
||||
}
|
||||
}
|
||||
|
||||
@@ -64,9 +65,11 @@ func (s *Service) ResolveCompleterForRole(ctx context.Context, companyID uuid.UU
|
||||
if s != nil && s.Pool != nil {
|
||||
return s.ResolveCompleter(ctx, companyID)
|
||||
}
|
||||
return s.resolvePlatformRoleCompleter(ctx, RoleProcessing)
|
||||
c, mode, byok, err := s.resolvePlatformRoleCompleter(ctx, RoleProcessing)
|
||||
return s.wrapAudit(c, companyID, RoleProcessing, mode), mode, byok, err
|
||||
case RoleDocsAPI, RoleSupport:
|
||||
return s.resolvePlatformRoleCompleter(ctx, role)
|
||||
c, mode, byok, err := s.resolvePlatformRoleCompleter(ctx, role)
|
||||
return s.wrapAudit(c, companyID, role, mode), mode, byok, err
|
||||
default:
|
||||
// Vectorization uses embeddings clients — not chat Completer.
|
||||
return nil, ModeInternalLabel, false, nil
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiaudit"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/platformsettings"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/processing"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/security"
|
||||
@@ -40,6 +41,9 @@ type Service struct {
|
||||
// When nil or a role is unset, ResolveCompleterForRole falls back to Resolve.
|
||||
Roles RoleEndpointSource
|
||||
HTTPClient *http.Client
|
||||
// Audit optionally captures every resolved Completer's prompts/responses for
|
||||
// the admin AI inspector. Nil disables capture (see audit.go).
|
||||
Audit *aiaudit.Recorder
|
||||
}
|
||||
|
||||
func NewService(pool *pgxpool.Pool, env EnvConfig) *Service {
|
||||
@@ -339,7 +343,7 @@ func (s *Service) ResolveCompleter(ctx context.Context, companyID uuid.UUID) (pr
|
||||
if err != nil {
|
||||
return nil, ModeInternalLabel, false, err
|
||||
}
|
||||
return r.Completer, r.ModeLabel, r.UsingBYOK, nil
|
||||
return s.wrapAudit(r.Completer, companyID, RoleProcessing, r.ModeLabel), r.ModeLabel, r.UsingBYOK, nil
|
||||
}
|
||||
|
||||
// TestPlatformRole probes admin platform AI role credentials (not company BYOK).
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiaudit"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiprompts"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/company"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/email"
|
||||
@@ -66,6 +67,8 @@ func (s *Service) Generate(ctx context.Context, companyID, id uuid.UUID, in Gene
|
||||
return Campaign{}, ErrInsufficientCredits
|
||||
}
|
||||
}
|
||||
// Tag the campaign role so admin AI inspector rows are not filed as processing.
|
||||
ctx := aiaudit.WithCall(ctx, aiaudit.CallContext{CompanyID: companyID, Role: aiaudit.RoleCampaign})
|
||||
var completer processing.Completer
|
||||
if s.AI != nil {
|
||||
cplt, _, _, rerr := s.AI.ResolveCompleter(ctx, companyID)
|
||||
|
||||
@@ -0,0 +1,270 @@
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiaudit"
|
||||
"github.com/go-chi/chi/v5"
|
||||
"github.com/google/uuid"
|
||||
"github.com/jackc/pgx/v5"
|
||||
)
|
||||
|
||||
// Admin AI inspector: the exact prompt and response of every LLM call.
|
||||
//
|
||||
// Platform-admin only. Prompts embed tenant product copy, so the list endpoint
|
||||
// returns previews and the full bodies are fetched one row at a time.
|
||||
|
||||
const (
|
||||
adminAICallsTimeout = 20 * time.Second
|
||||
adminAICallsDefaultLimit = 50
|
||||
adminAICallsMaxLimit = 200
|
||||
// adminAICallPreviewRunes bounds the list-view snippet of each body.
|
||||
adminAICallPreviewRunes = 240
|
||||
)
|
||||
|
||||
// handleAdminListAICalls lists captured calls, newest first.
|
||||
// GET /api/admin/ai-calls?company_id=&user_id=&job_id=&role=&outcome=&q=&limit=&before=
|
||||
func (s *Server) handleAdminListAICalls(w http.ResponseWriter, r *http.Request) {
|
||||
ctx, cancel := context.WithTimeout(r.Context(), adminAICallsTimeout)
|
||||
defer cancel()
|
||||
|
||||
q := r.URL.Query()
|
||||
where := []string{"TRUE"}
|
||||
args := []any{}
|
||||
add := func(clause string, val any) {
|
||||
args = append(args, val)
|
||||
where = append(where, strings.ReplaceAll(clause, "?", "$"+strconv.Itoa(len(args))))
|
||||
}
|
||||
|
||||
for _, f := range []struct {
|
||||
param string
|
||||
clause string
|
||||
}{
|
||||
{"company_id", "l.company_id = ?"},
|
||||
{"user_id", "l.user_id = ?"},
|
||||
{"job_id", "l.job_id = ?"},
|
||||
{"raw_product_id", "l.raw_product_id = ?"},
|
||||
} {
|
||||
raw := strings.TrimSpace(q.Get(f.param))
|
||||
if raw == "" {
|
||||
continue
|
||||
}
|
||||
id, err := uuid.Parse(raw)
|
||||
if err != nil {
|
||||
Error(w, http.StatusBadRequest, "invalid "+f.param)
|
||||
return
|
||||
}
|
||||
add(f.clause, id)
|
||||
}
|
||||
|
||||
if role := strings.TrimSpace(q.Get("role")); role != "" && role != "all" {
|
||||
add("l.role = ?", role)
|
||||
}
|
||||
switch strings.TrimSpace(q.Get("outcome")) {
|
||||
case "", "all":
|
||||
case "failed":
|
||||
where = append(where, "l.error <> ''")
|
||||
case "ok":
|
||||
where = append(where, "l.error = ''")
|
||||
default:
|
||||
Error(w, http.StatusBadRequest, "invalid outcome filter (all|ok|failed)")
|
||||
return
|
||||
}
|
||||
// Free-text search across the captured bodies — the point of the inspector is
|
||||
// answering "which call mentioned this", so it must cover prompt and response.
|
||||
if needle := strings.TrimSpace(q.Get("q")); needle != "" {
|
||||
args = append(args, "%"+needle+"%")
|
||||
i := strconv.Itoa(len(args))
|
||||
where = append(where, "(l.system_prompt ILIKE $"+i+" OR l.user_prompt ILIKE $"+i+" OR l.response_text ILIKE $"+i+")")
|
||||
}
|
||||
if before := strings.TrimSpace(q.Get("before")); before != "" {
|
||||
ts, err := time.Parse(time.RFC3339Nano, before)
|
||||
if err != nil {
|
||||
Error(w, http.StatusBadRequest, "invalid before cursor (RFC3339)")
|
||||
return
|
||||
}
|
||||
add("l.created_at < ?", ts)
|
||||
}
|
||||
|
||||
limit := adminAICallsDefaultLimit
|
||||
if raw := strings.TrimSpace(q.Get("limit")); raw != "" {
|
||||
if n, err := strconv.Atoi(raw); err == nil && n > 0 {
|
||||
limit = n
|
||||
}
|
||||
}
|
||||
if limit > adminAICallsMaxLimit {
|
||||
limit = adminAICallsMaxLimit
|
||||
}
|
||||
args = append(args, limit)
|
||||
|
||||
rows, err := s.Pool.Query(ctx, adminAICallsListSQL(strings.Join(where, " AND "), len(args)), args...)
|
||||
if err != nil {
|
||||
Error(w, http.StatusInternalServerError, "list ai calls failed")
|
||||
return
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
items := make([]map[string]any, 0, limit)
|
||||
var oldest time.Time
|
||||
for rows.Next() {
|
||||
var id uuid.UUID
|
||||
var companyID, companyName, userID, userEmail, jobID, rawProductID string
|
||||
var role, providerMode, model, callErr, finishReason string
|
||||
var promptTokens, completionTokens, totalTokens, durationMS int
|
||||
var createdAt time.Time
|
||||
var sysLen, userLen, respLen int
|
||||
var userPreview, respPreview string
|
||||
if err := rows.Scan(&id, &companyID, &companyName, &userID, &userEmail,
|
||||
&jobID, &rawProductID, &role, &providerMode, &model, &callErr, &finishReason,
|
||||
&promptTokens, &completionTokens, &totalTokens, &durationMS, &createdAt,
|
||||
&sysLen, &userLen, &respLen, &userPreview, &respPreview); err != nil {
|
||||
Error(w, http.StatusInternalServerError, "scan ai calls failed")
|
||||
return
|
||||
}
|
||||
oldest = createdAt
|
||||
outcome := "ok"
|
||||
if callErr != "" {
|
||||
outcome = "failed"
|
||||
}
|
||||
items = append(items, map[string]any{
|
||||
"id": id,
|
||||
"company_id": companyID,
|
||||
"company_name": companyName,
|
||||
"user_id": userID,
|
||||
"user_email": userEmail,
|
||||
"job_id": jobID,
|
||||
"raw_product_id": rawProductID,
|
||||
"role": role,
|
||||
"provider_mode": providerMode,
|
||||
"model": model,
|
||||
"outcome": outcome,
|
||||
"error": callErr,
|
||||
"finish_reason": finishReason,
|
||||
"prompt_tokens": promptTokens,
|
||||
"completion_tokens": completionTokens,
|
||||
"total_tokens": totalTokens,
|
||||
"duration_ms": durationMS,
|
||||
"created_at": createdAt,
|
||||
"system_length": sysLen,
|
||||
"user_length": userLen,
|
||||
"response_length": respLen,
|
||||
"user_preview": userPreview,
|
||||
"response_preview": respPreview,
|
||||
})
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
Error(w, http.StatusInternalServerError, "read ai calls failed")
|
||||
return
|
||||
}
|
||||
|
||||
out := map[string]any{
|
||||
"items": items,
|
||||
"limit": limit,
|
||||
"retention_days": aiaudit.RetentionDays,
|
||||
}
|
||||
// Cursor for the next page: callers pass it back as ?before=.
|
||||
if len(items) == limit && !oldest.IsZero() {
|
||||
out["next_before"] = oldest.Format(time.RFC3339Nano)
|
||||
}
|
||||
JSON(w, http.StatusOK, out)
|
||||
}
|
||||
|
||||
// adminAICallsListSQL builds the list query. Bodies are returned as lengths plus
|
||||
// left(...) previews only: one unfiltered request must never stream a tenant's
|
||||
// whole prompt corpus. Full bodies come from handleAdminGetAICall, one row at a time.
|
||||
func adminAICallsListSQL(where string, limitArg int) string {
|
||||
preview := strconv.Itoa(adminAICallPreviewRunes)
|
||||
return `
|
||||
SELECT l.id, COALESCE(l.company_id::text, ''), COALESCE(c.name, ''),
|
||||
COALESCE(l.user_id::text, ''), COALESCE(u.email, ''),
|
||||
COALESCE(l.job_id::text, ''), COALESCE(l.raw_product_id::text, ''),
|
||||
l.role, l.provider_mode, l.model, l.error, l.finish_reason,
|
||||
l.prompt_tokens, l.completion_tokens, l.total_tokens, l.duration_ms, l.created_at,
|
||||
length(l.system_prompt), length(l.user_prompt), length(l.response_text),
|
||||
left(l.user_prompt, ` + preview + `),
|
||||
left(l.response_text, ` + preview + `)
|
||||
FROM ai_call_logs l
|
||||
LEFT JOIN companies c ON c.id = l.company_id
|
||||
LEFT JOIN users u ON u.id = l.user_id
|
||||
WHERE ` + where + `
|
||||
ORDER BY l.created_at DESC
|
||||
LIMIT $` + strconv.Itoa(limitArg)
|
||||
}
|
||||
|
||||
// handleAdminGetAICall returns one captured call including the full prompt bodies.
|
||||
// GET /api/admin/ai-calls/{id}
|
||||
func (s *Server) handleAdminGetAICall(w http.ResponseWriter, r *http.Request) {
|
||||
ctx, cancel := context.WithTimeout(r.Context(), adminAICallsTimeout)
|
||||
defer cancel()
|
||||
|
||||
id, err := uuid.Parse(strings.TrimSpace(chi.URLParam(r, "id")))
|
||||
if err != nil {
|
||||
Error(w, http.StatusBadRequest, "invalid id")
|
||||
return
|
||||
}
|
||||
|
||||
var companyID, companyName, userID, userEmail, jobID, rawProductID string
|
||||
var role, providerMode, model, callErr, finishReason string
|
||||
var systemPrompt, userPrompt, responseText string
|
||||
var promptTokens, completionTokens, totalTokens, durationMS int
|
||||
var createdAt time.Time
|
||||
err = s.Pool.QueryRow(ctx, `
|
||||
SELECT COALESCE(l.company_id::text, ''), COALESCE(c.name, ''),
|
||||
COALESCE(l.user_id::text, ''), COALESCE(u.email, ''),
|
||||
COALESCE(l.job_id::text, ''), COALESCE(l.raw_product_id::text, ''),
|
||||
l.role, l.provider_mode, l.model, l.error, l.finish_reason,
|
||||
l.system_prompt, l.user_prompt, l.response_text,
|
||||
l.prompt_tokens, l.completion_tokens, l.total_tokens, l.duration_ms, l.created_at
|
||||
FROM ai_call_logs l
|
||||
LEFT JOIN companies c ON c.id = l.company_id
|
||||
LEFT JOIN users u ON u.id = l.user_id
|
||||
WHERE l.id = $1`, id).
|
||||
Scan(&companyID, &companyName, &userID, &userEmail, &jobID, &rawProductID,
|
||||
&role, &providerMode, &model, &callErr, &finishReason,
|
||||
&systemPrompt, &userPrompt, &responseText,
|
||||
&promptTokens, &completionTokens, &totalTokens, &durationMS, &createdAt)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
// Expired rows are the common case here, so say so rather than a bare 404.
|
||||
Error(w, http.StatusNotFound, "call not found (captured calls are kept for "+
|
||||
strconv.Itoa(aiaudit.RetentionDays)+" days)")
|
||||
return
|
||||
}
|
||||
if err != nil {
|
||||
Error(w, http.StatusInternalServerError, "load ai call failed")
|
||||
return
|
||||
}
|
||||
|
||||
outcome := "ok"
|
||||
if callErr != "" {
|
||||
outcome = "failed"
|
||||
}
|
||||
JSON(w, http.StatusOK, map[string]any{
|
||||
"id": id,
|
||||
"company_id": companyID,
|
||||
"company_name": companyName,
|
||||
"user_id": userID,
|
||||
"user_email": userEmail,
|
||||
"job_id": jobID,
|
||||
"raw_product_id": rawProductID,
|
||||
"role": role,
|
||||
"provider_mode": providerMode,
|
||||
"model": model,
|
||||
"outcome": outcome,
|
||||
"error": callErr,
|
||||
"finish_reason": finishReason,
|
||||
"system_prompt": systemPrompt,
|
||||
"user_prompt": userPrompt,
|
||||
"response_text": responseText,
|
||||
"prompt_tokens": promptTokens,
|
||||
"completion_tokens": completionTokens,
|
||||
"total_tokens": totalTokens,
|
||||
"duration_ms": durationMS,
|
||||
"created_at": createdAt,
|
||||
"retention_days": aiaudit.RetentionDays,
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,92 @@
|
||||
package httpapi
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func readSourceFile(t *testing.T, name string) string {
|
||||
t.Helper()
|
||||
b, err := os.ReadFile(name)
|
||||
if err != nil {
|
||||
t.Fatalf("read %s: %v", name, err)
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
|
||||
// Filter parsing must reject bad input before touching the database, so a typo in
|
||||
// the admin UI cannot turn into a full-table scan or a 500.
|
||||
func TestHandleAdminListAICalls_rejectsBadFilters(t *testing.T) {
|
||||
t.Parallel()
|
||||
s := &Server{}
|
||||
cases := map[string]string{
|
||||
"invalid company_id": "?company_id=not-a-uuid",
|
||||
"invalid user_id": "?user_id=123",
|
||||
"invalid job_id": "?job_id=abc",
|
||||
"invalid raw_product_id": "?raw_product_id=xyz",
|
||||
"invalid outcome": "?outcome=maybe",
|
||||
"invalid before cursor": "?before=yesterday",
|
||||
}
|
||||
for name, query := range cases {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/admin/ai-calls"+query, nil)
|
||||
rec := httptest.NewRecorder()
|
||||
s.handleAdminListAICalls(rec, req)
|
||||
if rec.Code != http.StatusBadRequest {
|
||||
t.Fatalf("status=%d want 400 body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandleAdminGetAICall_rejectsBadID(t *testing.T) {
|
||||
t.Parallel()
|
||||
s := &Server{}
|
||||
req := httptest.NewRequest(http.MethodGet, "/api/admin/ai-calls/nope", nil)
|
||||
rec := httptest.NewRecorder()
|
||||
s.handleAdminGetAICall(rec, req)
|
||||
if rec.Code != http.StatusBadRequest {
|
||||
t.Fatalf("status=%d want 400 body=%s", rec.Code, rec.Body.String())
|
||||
}
|
||||
}
|
||||
|
||||
// The list endpoint returns previews, never the full bodies — one unfiltered
|
||||
// request must not stream a tenant's whole prompt corpus out of the admin API.
|
||||
func TestAdminAICallsListSQL_selectsPreviewsNotFullBodies(t *testing.T) {
|
||||
t.Parallel()
|
||||
sql := adminAICallsListSQL("TRUE", 1)
|
||||
selectList := sql[strings.Index(sql, "SELECT"):strings.Index(sql, "FROM ai_call_logs")]
|
||||
|
||||
// A bare column followed by a comma is the whole body; the length()/left()
|
||||
// wrappers put a ")" in between.
|
||||
for _, bare := range []string{" l.system_prompt,", " l.user_prompt,", " l.response_text,"} {
|
||||
if strings.Contains(selectList, bare) {
|
||||
t.Fatalf("list selects a full body (%q):\n%s", strings.TrimSpace(bare), selectList)
|
||||
}
|
||||
}
|
||||
for _, want := range []string{
|
||||
"left(l.user_prompt, 240)",
|
||||
"left(l.response_text, 240)",
|
||||
"length(l.system_prompt)",
|
||||
} {
|
||||
if !strings.Contains(selectList, want) {
|
||||
t.Fatalf("list should select %s:\n%s", want, selectList)
|
||||
}
|
||||
}
|
||||
if !strings.Contains(sql, "ORDER BY l.created_at DESC") || !strings.Contains(sql, "LIMIT $1") {
|
||||
t.Fatalf("list must page newest-first with a bound limit:\n%s", sql)
|
||||
}
|
||||
}
|
||||
|
||||
// The detail endpoint is the only place full bodies come from.
|
||||
func TestAdminAICallsDetail_returnsFullBodies(t *testing.T) {
|
||||
t.Parallel()
|
||||
src := readSourceFile(t, "admin_ai_calls_handlers.go")
|
||||
detail := src[strings.Index(src, "func (s *Server) handleAdminGetAICall"):]
|
||||
if !strings.Contains(detail, "l.system_prompt, l.user_prompt, l.response_text") {
|
||||
t.Fatal("detail handler must select the full bodies")
|
||||
}
|
||||
}
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/alexedwards/scs/v2"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiaudit"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiprompts"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiprovider"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/auth"
|
||||
@@ -123,6 +124,9 @@ func NewServer(
|
||||
ProcessingRPM: cfg.ProcessingRPM,
|
||||
ProcessingMaxRetries: cfg.ProcessingMaxRetries,
|
||||
})
|
||||
// Campaign / SEO / support AI calls run in this process; capture them for
|
||||
// /admin/ai-calls the same way the worker captures processing.
|
||||
aiSvc.WithAudit(aiaudit.NewRecorder(pool))
|
||||
promptSvc := aiprompts.NewService(pool)
|
||||
platformSettings := platformsettings.NewService(pool, platformsettings.EnvConfig{
|
||||
AppEncryptionKey: cfg.AppEncryptionKey,
|
||||
@@ -400,6 +404,8 @@ func (s *Server) Router() http.Handler {
|
||||
r.Post("/companies/{id}/fix-catalog", s.handleAdminFixCompanyCatalog) // compat alias → Sync A1
|
||||
r.Get("/readiness", s.handleAdminReadiness)
|
||||
r.Get("/diagnostics", s.handleAdminDiagnostics)
|
||||
r.Get("/ai-calls", s.handleAdminListAICalls)
|
||||
r.Get("/ai-calls/{id}", s.handleAdminGetAICall)
|
||||
r.Get("/analytics", s.handleAdminAnalytics)
|
||||
r.Get("/jobs", s.handleAdminListJobs)
|
||||
r.Post("/jobs/stuck-cleanup", s.handleAdminStuckCleanup)
|
||||
|
||||
@@ -5,6 +5,8 @@ import (
|
||||
"log"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiaudit"
|
||||
)
|
||||
|
||||
// MaxCategorizeOptions caps the taxonomy list injected into the categorize LLM
|
||||
@@ -255,7 +257,9 @@ func runCategorizeStep(ctx context.Context, e *Engine, companyID string, out *St
|
||||
if strings.TrimSpace(out.Category) != "" {
|
||||
return
|
||||
}
|
||||
tryAICategorize(ctx, e, out, in, policy)
|
||||
// Taxonomy picks and copy generation share one Completer; tag the role so the
|
||||
// admin inspector can tell them apart.
|
||||
tryAICategorize(aiaudit.WithCall(ctx, aiaudit.CallContext{Role: aiaudit.RoleCategorize}), e, out, in, policy)
|
||||
if strings.TrimSpace(out.Category) == "" {
|
||||
appendStepLog(out.GPTResponse, StepCategorize, map[string]any{
|
||||
"status": "unset",
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiaudit"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiprompts"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/billing"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/catalog"
|
||||
@@ -588,15 +589,23 @@ func (p *Pipeline) RetryJob(ctx context.Context, companyID, id uuid.UUID) (Job,
|
||||
// Terminal job statuses (completed/cancelled/failed) are no-ops — RetryJob requeues work.
|
||||
func (p *Pipeline) ProcessJob(ctx context.Context, jobID uuid.UUID) error {
|
||||
var companyID uuid.UUID
|
||||
var jobUserID *uuid.UUID
|
||||
var status, processingType string
|
||||
var alreadyProcessed int
|
||||
err := p.Pool.QueryRow(ctx, `
|
||||
SELECT company_id, status, processing_type, processed_products
|
||||
SELECT company_id, user_id, status, processing_type, processed_products
|
||||
FROM processing_jobs WHERE id = $1`, jobID).
|
||||
Scan(&companyID, &status, &processingType, &alreadyProcessed)
|
||||
Scan(&companyID, &jobUserID, &status, &processingType, &alreadyProcessed)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Tag every LLM call this job makes so the admin AI inspector can group them
|
||||
// by tenant, job and the user who started it (internal/aiaudit).
|
||||
auditCall := aiaudit.CallContext{CompanyID: companyID, JobID: jobID, Role: aiaudit.RoleProcessing}
|
||||
if jobUserID != nil {
|
||||
auditCall.UserID = *jobUserID
|
||||
}
|
||||
ctx = aiaudit.WithCall(ctx, auditCall)
|
||||
if !IsProcessableJobStatus(status) {
|
||||
log.Printf("processing: skip job=%s status=%s reason=not_processable", jobID, status)
|
||||
return nil
|
||||
@@ -1439,6 +1448,9 @@ func (p *Pipeline) processOne(ctx context.Context, companyID, jobID uuid.UUID, i
|
||||
if it == nil {
|
||||
return false, 0, StepResult{}, fmt.Errorf("processing: nil job item")
|
||||
}
|
||||
// Narrow the job-level audit tag to this product so captured prompts are
|
||||
// addressable per item, not just per job.
|
||||
ctx = aiaudit.WithCall(ctx, aiaudit.CallContext{RawProductID: it.RawID})
|
||||
gtin := it.gtin
|
||||
mappedBytes := it.mappedBytes
|
||||
rawBytes := it.rawBytes
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"errors"
|
||||
"strings"
|
||||
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiaudit"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiprompts"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiprovider"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/billing"
|
||||
@@ -114,6 +115,7 @@ func (s *Service) Apply(ctx context.Context, companyID uuid.UUID, productID uuid
|
||||
}
|
||||
var completer processing.Completer
|
||||
if s.AI != nil {
|
||||
ctx := aiaudit.WithCall(ctx, aiaudit.CallContext{CompanyID: companyID, Role: aiaudit.RoleSEOMeta})
|
||||
c, _, _, rerr := s.AI.ResolveCompleter(ctx, companyID)
|
||||
if rerr != nil {
|
||||
return ApplyResult{}, rerr
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiaudit"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/aiprovider"
|
||||
"github.com/descrybe/descrybe-v2/apps/api/internal/processing"
|
||||
"github.com/google/uuid"
|
||||
@@ -150,6 +151,7 @@ func (r *CompleterSupportAI) resolveCompleter(ctx context.Context, companyID uui
|
||||
if resolver == nil {
|
||||
return nil, nil
|
||||
}
|
||||
ctx = aiaudit.WithCall(ctx, aiaudit.CallContext{CompanyID: companyID, Role: aiaudit.RoleSupport})
|
||||
c, _, _, err := resolver.ResolveCompleterForRole(ctx, companyID, aiprovider.RoleSupport)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
-- +goose Up
|
||||
-- Full prompt/response capture for every LLM call, for the platform-admin AI
|
||||
-- inspector (/admin/ai-calls).
|
||||
--
|
||||
-- Worker logs only ever carried a 500-rune head/tail preview and
|
||||
-- processed_products.gpt_response keeps response metadata without the prompt, so
|
||||
-- there was no way to answer "what exactly did we send for this job".
|
||||
--
|
||||
-- Rows are large (a category-formula prompt runs several KB) and high volume (one
|
||||
-- per product per language), so they are strictly short-lived: CleanupExpiredAICalls
|
||||
-- prunes them on the worker's retention tick. This table is a debugging buffer, not
|
||||
-- an audit trail — never build billing or reporting on it.
|
||||
|
||||
CREATE TABLE IF NOT EXISTS ai_call_logs (
|
||||
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
|
||||
-- Nullable: platform-level probes (admin AI role test) belong to no tenant.
|
||||
company_id UUID REFERENCES companies(id) ON DELETE CASCADE,
|
||||
-- Who triggered the work (job starter / campaign author). Kept on user delete
|
||||
-- so a tenant's recent calls stay readable; the row expires on its own.
|
||||
user_id UUID REFERENCES users(id) ON DELETE SET NULL,
|
||||
job_id UUID,
|
||||
raw_product_id UUID,
|
||||
-- aiprovider role: processing | categorize | seo_meta | campaign | support | probe.
|
||||
role TEXT NOT NULL DEFAULT 'other',
|
||||
provider_mode TEXT NOT NULL DEFAULT '',
|
||||
model TEXT NOT NULL DEFAULT '',
|
||||
system_prompt TEXT NOT NULL DEFAULT '',
|
||||
user_prompt TEXT NOT NULL DEFAULT '',
|
||||
response_text TEXT NOT NULL DEFAULT '',
|
||||
error TEXT NOT NULL DEFAULT '',
|
||||
finish_reason TEXT NOT NULL DEFAULT '',
|
||||
prompt_tokens INTEGER NOT NULL DEFAULT 0,
|
||||
completion_tokens INTEGER NOT NULL DEFAULT 0,
|
||||
total_tokens INTEGER NOT NULL DEFAULT 0,
|
||||
duration_ms INTEGER NOT NULL DEFAULT 0,
|
||||
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
-- Primary browse: one company's most recent calls.
|
||||
CREATE INDEX IF NOT EXISTS ai_call_logs_company_created_idx
|
||||
ON ai_call_logs (company_id, created_at DESC);
|
||||
|
||||
-- Drill into a single job from the Processing page.
|
||||
CREATE INDEX IF NOT EXISTS ai_call_logs_job_idx
|
||||
ON ai_call_logs (job_id, created_at DESC)
|
||||
WHERE job_id IS NOT NULL;
|
||||
|
||||
-- Retention sweep scans by age across all tenants.
|
||||
CREATE INDEX IF NOT EXISTS ai_call_logs_created_idx
|
||||
ON ai_call_logs (created_at);
|
||||
|
||||
-- Failures are the rare rows worth finding fast across a whole tenant.
|
||||
CREATE INDEX IF NOT EXISTS ai_call_logs_failed_idx
|
||||
ON ai_call_logs (company_id, created_at DESC)
|
||||
WHERE error <> '';
|
||||
|
||||
-- +goose Down
|
||||
DROP TABLE IF EXISTS ai_call_logs;
|
||||
Reference in New Issue
Block a user