From 322608733905435f6ba9a939bb3f024071b2ff41 Mon Sep 17 00:00:00 2001 From: Wu Sheng Date: Tue, 15 Sep 2026 22:28:10 +0800 Subject: [PATCH] Read AI agent conversation files over the OAP's HTTP route The OAP drops the getConversationRawFiles GraphQL query. A conversation's stored Session Data files are read from its HTTP route beside the view, GET /ai-agent/conversations/{conversation}/v1/files, and the view and the files route now both require the service and the sender's instance. - ai-agent files requires --instance-name, --session and --seqs. A file is chosen by its session and its landed seq, which the conversation's asz.view document lists; there is no read of every file. The seqs are asked for 32 at a time, the route's limit, over one HTTP client. - The body, application/vnd.skywalking.asz.files+ndjson, is read file by file: a naming line, then exactly the bytes it names, then the newline that follows a non-empty file not ending with one. Each file is checked against its sha256, and a size no response carries is never allocated. - --export writes through os.Root, so a file's name cannot reach outside the export directory, through ".." or through a symbolic link. - ai-agent view requires --instance-name. --- CHANGES.md | 2 +- .../aiagent/ConversationRawFiles.graphql | 34 --- internal/commands/aiagent/aiagent.go | 12 +- internal/commands/aiagent/files.go | 152 +++++++------ internal/commands/aiagent/files_test.go | 55 +++++ internal/commands/aiagent/view.go | 7 +- pkg/aiagent/files/files.go | 161 ++++++++++++++ pkg/aiagent/files/files_test.go | 200 ++++++++++++++++++ pkg/aiagent/view/view.go | 17 +- pkg/aiagent/view/view_test.go | 2 +- pkg/graphql/aiagent/conversation.go | 20 +- 11 files changed, 532 insertions(+), 130 deletions(-) delete mode 100644 assets/graphqls/aiagent/ConversationRawFiles.graphql create mode 100644 internal/commands/aiagent/files_test.go create mode 100644 pkg/aiagent/files/files.go create mode 100644 pkg/aiagent/files/files_test.go diff --git a/CHANGES.md b/CHANGES.md index 5e1abc2..63d55f8 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -7,7 +7,7 @@ Release Notes. ### Features -* Add the `ai-agent` commands, `list`, `files` and `view`, for the AI agent conversations the AI Sessionizer lands in the OAP (11.1.0+); `view` reads the whole conversation as one `asz.view` document from the OAP's streamed route on the GraphQL host by @wu-sheng in https://github.com/apache/skywalking-cli/pull/234 +* Add the `ai-agent` commands, `list`, `files` and `view`, for the AI agent conversations the AI Sessionizer lands in the OAP (11.1.0+); `view` reads the whole conversation as one `asz.view` document, and `files` lists or exports its stored files by name, both from the OAP's streamed routes on the GraphQL host and both requiring the service and the sender's instance by @wu-sheng in https://github.com/apache/skywalking-cli/pull/234 * Add the sub-command `profiling async` for async-profiler query API by @zhengziyi0117 in https://github.com/apache/skywalking-cli/pull/203 * Support the owner in MQE response by using [10.2 MQE query protocol](https://github.com/apache/skywalking-query-protocol/pull/141) by @zhengziyi0117 in https://github.com/apache/skywalking-cli/pull/203 * Add the sub-command `alarm autocomplete-keys` and `alarm auto-complete-values` for alarm query API by @mrproliu in https://github.com/apache/skywalking-cli/pull/210 diff --git a/assets/graphqls/aiagent/ConversationRawFiles.graphql b/assets/graphqls/aiagent/ConversationRawFiles.graphql deleted file mode 100644 index 824e2c8..0000000 --- a/assets/graphqls/aiagent/ConversationRawFiles.graphql +++ /dev/null @@ -1,34 +0,0 @@ -# Licensed to Apache Software Foundation (ASF) under one or more contributor -# license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright -# ownership. Apache Software Foundation (ASF) licenses this file to you under -# the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. - -# The body is read from storage only when selected; $body is true on the export path. -query ($condition: ConversationCondition!, $files: [ID!], $body: Boolean!) { - result: getConversationRawFiles(condition: $condition, files: $files) { - errorReason - files { - id - format - session - seq - round - digest - bytes - timestamp - body @include(if: $body) - } - } -} diff --git a/internal/commands/aiagent/aiagent.go b/internal/commands/aiagent/aiagent.go index 6eec3df..6ad5c3c 100644 --- a/internal/commands/aiagent/aiagent.go +++ b/internal/commands/aiagent/aiagent.go @@ -17,8 +17,8 @@ // Package aiagent holds the commands for the conversations of long-lived AI agents // that the AI Sessionizer (apache/skywalking-ai-sessionizer) lands in the OAP under -// the AI_AGENT layer: the list page, the raw-file export, and the conversation itself -// as one asz.view document. +// the AI_AGENT layer: the list page, the conversation itself as one asz.view document, +// and its stored files. package aiagent import ( @@ -29,9 +29,11 @@ var Command = &cli.Command{ Name: "ai-agent", Usage: "AI agent conversations landed by the AI Sessionizer", UsageText: `The AI Sessionizer collects an agent runtime's transcripts and pushes them to the OAP -under the AI_AGENT layer. "list" and "files" are GraphQL queries on the "--base-url" -endpoint; "view" reads the whole conversation as one asz.view document from the OAP's -streamed route on the same host, GET /ai-agent/conversations/{conversation}/v1/view.`, +under the AI_AGENT layer. "list" is a GraphQL query on the "--base-url" endpoint. "view" +and "files" read the OAP's streamed routes on the same host: the whole conversation as one +asz.view document, GET /ai-agent/conversations/{conversation}/v1/view, and its stored files +by name, GET /ai-agent/conversations/{conversation}/v1/files. Both need the service and the +sender's instance, as "list" names them.`, Subcommands: []*cli.Command{ listCommand, filesCommand, diff --git a/internal/commands/aiagent/files.go b/internal/commands/aiagent/files.go index f3cb293..b0ac9b0 100644 --- a/internal/commands/aiagent/files.go +++ b/internal/commands/aiagent/files.go @@ -21,37 +21,35 @@ import ( "fmt" "os" "path/filepath" + "strconv" "strings" - api "skywalking.apache.org/repo/goapi/query" - "github.com/urfave/cli/v2" "github.com/apache/skywalking-cli/internal/commands/interceptor" "github.com/apache/skywalking-cli/internal/flags" + "github.com/apache/skywalking-cli/pkg/aiagent/files" "github.com/apache/skywalking-cli/pkg/display" "github.com/apache/skywalking-cli/pkg/display/displayable" - "github.com/apache/skywalking-cli/pkg/graphql/aiagent" ) var filesCommand = &cli.Command{ Name: "files", - Usage: "List or export the raw files of a conversation, as the OAP stores them", - UsageText: `List every landed file and round of a conversation with its digest and size, or -export them: "--export DIR" reads each body and writes it to its id path under DIR, -which gives a storage root that "asz verify" and "asz view" read like the original. + Usage: "List or export chosen stored files of a conversation, as the OAP stores them", + UsageText: `Read chosen Session Data files of a conversation's session from the OAP's route +GET /ai-agent/conversations/{conversation}/v1/files, on the "--base-url" host, and list each +with its digest and size, or export them: "--export DIR" writes each file to its name under DIR. +A file is chosen by "--session" and "--seqs", its landed seq; the asz.view document's files +list gives both. Each file's bytes are checked against the digest the OAP names. Examples: -1. The files of a conversation: -$ swctl ai-agent files --service-name "Claude Code" --conversation 7a3c882e-0dc0-46a0-b814-6613d24b7ac2 - -2. Export them all: -$ swctl ai-agent files --service-name "Claude Code" --conversation 7a3c882e-0dc0-46a0-b814-6613d24b7ac2 --export ./root +1. Two files of a session, listed: +$ swctl ai-agent files --service-name "Claude Code" --instance-name laptop --conversation 7a3c882e-0dc0-46a0-b814-6613d24b7ac2 \ + --session 7a3c882e-0dc0-46a0-b814-6613d24b7ac2 --seqs 408,409 -3. Export two named files: -$ swctl ai-agent files --service-name "Claude Code" --conversation 7a3c882e-0dc0-46a0-b814-6613d24b7ac2 \ - --files 7a3c882e-0dc0-46a0-b814-6613d24b7ac2/streams/main/transcript-20260904T152815.774957000Z-000408.sd \ - --export ./root`, +2. The same files, exported: +$ swctl ai-agent files --service-name "Claude Code" --instance-name laptop --conversation 7a3c882e-0dc0-46a0-b814-6613d24b7ac2 \ + --session 7a3c882e-0dc0-46a0-b814-6613d24b7ac2 --seqs 408,409 --export ./root`, Flags: flags.Flags( flags.ServiceFlags, flags.InstanceFlags, @@ -62,80 +60,112 @@ $ swctl ai-agent files --service-name "Claude Code" --conversation 7a3c882e-0dc0 Required: true, }, &cli.StringFlag{ - Name: "files", - Usage: "only these file `ids`, comma separated; without it, every file of the conversation", + Name: "session", + Usage: "the `session` the files belong to", + Required: true, + }, + &cli.StringFlag{ + Name: "seqs", + Usage: "the landed `seqs` of the files, comma separated", + Required: true, }, &cli.StringFlag{ Name: "export", - Usage: "write each file's body to its id path under this `directory`", + Usage: "write each file to its name under this `directory`", }, }, ), Before: interceptor.BeforeChain( interceptor.ParseService(true), - interceptor.ParseInstance(false), + interceptor.ParseInstance(true), ), Action: func(ctx *cli.Context) error { - condition := &api.ConversationCondition{ - Service: &api.ServiceCondition{ServiceName: ctx.String("service-name")}, - Conversation: ctx.String("conversation"), - Instance: instanceCondition(ctx), - } - var files []string - if arg := strings.TrimSpace(ctx.String("files")); arg != "" { - files = strings.Split(arg, ",") - } - exportDir := ctx.String("export") - - raw, err := aiagent.RawFiles(ctx.Context, condition, files, exportDir != "") + seqs, err := numbers(ctx.String("seqs")) if err != nil { return err } - if raw.ErrorReason != nil && *raw.ErrorReason != "" { - return fmt.Errorf("%s", *raw.ErrorReason) + if len(seqs) == 0 { + return fmt.Errorf("--seqs needs at least one number") } - if exportDir == "" { - return display.Display(ctx.Context, &displayable.Displayable{Data: raw, Condition: condition}) + var root *os.Root + if exportDir := ctx.String("export"); exportDir != "" { + if mkErr := os.MkdirAll(exportDir, 0o755); mkErr != nil { + return mkErr + } + if root, err = os.OpenRoot(exportDir); err != nil { + return err + } + defer root.Close() } - written, err := export(exportDir, raw.Files) + out := List{Files: []files.File{}} + var written []Exported + err = files.Read(ctx.Context, ctx.String("conversation"), ctx.String("service-name"), ctx.String("instance-name"), + ctx.String("session"), seqs, func(f files.File, content []byte) error { + if root == nil { + out.Files = append(out.Files, f) + return nil + } + path, exportErr := export(root, f.ID, content) + if exportErr != nil { + return exportErr + } + written = append(written, Exported{ID: f.ID, Path: path, Bytes: len(content)}) + return nil + }) if err != nil { return err } - return display.Display(ctx.Context, &displayable.Displayable{Data: written, Condition: condition}) + if root == nil { + return display.Display(ctx.Context, &displayable.Displayable{Data: out}) + } + return display.Display(ctx.Context, &displayable.Displayable{Data: written}) }, } -// Exported is one file written by "--export": its id path and size, the body left out. +// List is what "files" prints without "--export": each stored file's naming line. +type List struct { + Files []files.File `json:"files"` +} + +// Exported is one file written by "--export": its name, its path and its size. type Exported struct { - ID string `json:"id"` + ID string `json:"file"` Path string `json:"path"` Bytes int `json:"bytes"` } -// export writes each body to its id path under dir. An id is a relative path inside the -// Sessionizer's storage root; one that would leave dir is refused. -func export(dir string, files []*api.ConversationRawFile) ([]Exported, error) { - root, err := filepath.Abs(dir) - if err != nil { - return nil, err - } - out := make([]Exported, 0, len(files)) - for _, f := range files { - if f.Body == nil { - return nil, fmt.Errorf("the OAP returned no body for %s", f.ID) - } - path := filepath.Join(root, filepath.FromSlash(f.ID)) - if !strings.HasPrefix(path, root+string(filepath.Separator)) { - return nil, fmt.Errorf("refusing to write %s outside %s", f.ID, root) +// numbers reads a comma separated list of positive whole numbers. +func numbers(arg string) ([]int64, error) { + var out []int64 + for _, part := range strings.Split(arg, ",") { + if part = strings.TrimSpace(part); part == "" { + continue } - if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { - return nil, err + n, err := strconv.ParseInt(part, 10, 64) + if err != nil || n <= 0 { + return nil, fmt.Errorf("%q is not a positive whole number", part) } - if err := os.WriteFile(path, []byte(*f.Body), 0o644); err != nil { // #nosec G306 -- a landed file is readable by design - return nil, err - } - out = append(out, Exported{ID: f.ID, Path: path, Bytes: len(*f.Body)}) + out = append(out, n) } return out, nil } + +// export writes one file to its name under root. A name is a relative path inside the Sessionizer's +// storage root. The root refuses a name that would leave it, through ".." or through a symbolic +// link, so a file the OAP names can only land inside the export directory. +func export(root *os.Root, name string, content []byte) (string, error) { + rel := filepath.FromSlash(name) + if !filepath.IsLocal(rel) { + return "", fmt.Errorf("refusing to write %s outside %s", name, root.Name()) + } + if dir := filepath.Dir(rel); dir != "." { + if err := root.MkdirAll(dir, 0o755); err != nil { + return "", err + } + } + if err := root.WriteFile(rel, content, 0o644); err != nil { // #nosec G306 -- a landed file is readable by design + return "", err + } + return filepath.Join(root.Name(), rel), nil +} diff --git a/internal/commands/aiagent/files_test.go b/internal/commands/aiagent/files_test.go new file mode 100644 index 0000000..1ed6964 --- /dev/null +++ b/internal/commands/aiagent/files_test.go @@ -0,0 +1,55 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package aiagent + +import ( + "os" + "path/filepath" + "testing" +) + +// An export lands inside its directory whatever a file's name says: a name climbing out is refused, +// and so is one that would reach outside through a symbolic link. +func TestExportStaysInsideItsDirectory(t *testing.T) { + dir := t.TempDir() + outside := t.TempDir() + if err := os.Symlink(outside, filepath.Join(dir, "linked")); err != nil { + t.Skip("no symbolic links here:", err) + } + root, err := os.OpenRoot(dir) + if err != nil { + t.Fatal(err) + } + defer root.Close() + + if _, err := export(root, "s/streams/main/transcript-20260101T000000.000000000Z-000001.sd", []byte("x\n")); err != nil { + t.Fatal(err) + } + written := filepath.Join(dir, "s", "streams", "main", "transcript-20260101T000000.000000000Z-000001.sd") + if b, err := os.ReadFile(written); err != nil || string(b) != "x\n" { + t.Fatalf("written: %q, %v", b, err) + } + for _, name := range []string{"../escaped.sd", "linked/escaped.sd", "/abs/escaped.sd"} { + if _, err := export(root, name, []byte("x\n")); err == nil { + t.Errorf("%s: written", name) + } + } + if entries, _ := os.ReadDir(outside); len(entries) != 0 { + t.Errorf("the outside directory holds %d entries", len(entries)) + } +} diff --git a/internal/commands/aiagent/view.go b/internal/commands/aiagent/view.go index 2bbb385..1eaf882 100644 --- a/internal/commands/aiagent/view.go +++ b/internal/commands/aiagent/view.go @@ -38,10 +38,11 @@ The "--display" option does not apply; the document is printed as the OAP sends Examples: 1. A conversation as JSON, into a file: -$ swctl ai-agent view --service-name "Claude Code" --conversation 7a3c882e-0dc0-46a0-b814-6613d24b7ac2 --output conversation.json +$ swctl ai-agent view --service-name "Claude Code" --instance-name laptop \ + --conversation 7a3c882e-0dc0-46a0-b814-6613d24b7ac2 --output conversation.json 2. As YAML, on the terminal: -$ swctl ai-agent view --service-name "Claude Code" --conversation 7a3c882e-0dc0-46a0-b814-6613d24b7ac2 --yaml`, +$ swctl ai-agent view --service-name "Claude Code" --instance-name laptop --conversation 7a3c882e-0dc0-46a0-b814-6613d24b7ac2 --yaml`, Flags: flags.Flags( flags.ServiceFlags, flags.InstanceFlags, @@ -63,7 +64,7 @@ $ swctl ai-agent view --service-name "Claude Code" --conversation 7a3c882e-0dc0- ), Before: interceptor.BeforeChain( interceptor.ParseService(true), - interceptor.ParseInstance(false), + interceptor.ParseInstance(true), ), Action: func(ctx *cli.Context) error { var out io.Writer = os.Stdout diff --git a/pkg/aiagent/files/files.go b/pkg/aiagent/files/files.go new file mode 100644 index 0000000..c8a0ae2 --- /dev/null +++ b/pkg/aiagent/files/files.go @@ -0,0 +1,161 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package files reads chosen Session Data files of an AI agent conversation's session from the +// OAP's GET /ai-agent/conversations/{conversation}/v1/files route, on the query host beside the view +// route. A file is chosen by its session and its landed seq; there is no read of every file. The +// body is application/vnd.skywalking.asz.files+ndjson: for each stored file a line naming it, then +// exactly as many bytes as it says, the file. Files are read as they arrive, never the whole +// response at once. +package files + +import ( + "bufio" + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strconv" + + "github.com/apache/skywalking-cli/pkg/aiagent/view" + "github.com/apache/skywalking-cli/pkg/contextkey" + "github.com/apache/skywalking-cli/pkg/transport" +) + +const ( + // MediaType is the route's one body format. + MediaType = "application/vnd.skywalking.asz.files+ndjson" + // MaxSeqs is the most seqs one request carries, the route's own limit: the Sessionizer cuts a + // file at 2 MiB, so one response holds about 64 MiB at most. + MaxSeqs = 32 + + defaultBaseURL = "http://127.0.0.1:12800/graphql" +) + +// File is the line naming one stored file. +type File struct { + ID string `json:"file"` + Seq int64 `json:"seq"` + Lines int `json:"lines"` + Bytes int `json:"bytes"` + Digest string `json:"digest"` +} + +// Path is the files route of one conversation. +func Path(conversation string) string { + return "/ai-agent/conversations/" + url.PathEscape(conversation) + "/v1/files" +} + +// Read asks for the chosen files of session, at most MaxSeqs a request, and hands each stored file +// to fn with its bytes, checked against the digest its naming line gives. A seq no stored file +// answers is left out. serviceName, instanceName, session and at least one seq are required. +func Read(ctx context.Context, conversation, serviceName, instanceName, session string, seqs []int64, + fn func(File, []byte) error) error { + switch { + case serviceName == "" || instanceName == "": + return errors.New("the service and the instance are both required") + case session == "": + return errors.New("the session is required") + case len(seqs) == 0: + return errors.New("at least one seq is required") + } + // one client for every batch, so a long read reuses its connections instead of opening one per batch + client := transport.HTTPClient(ctx) + defer client.CloseIdleConnections() + for start := 0; start < len(seqs); start += MaxSeqs { + end := start + MaxSeqs + if end > len(seqs) { + end = len(seqs) + } + query := url.Values{"service": {serviceName}, "instance": {instanceName}, "session": {session}} + for _, n := range seqs[start:end] { + query.Add("seq", strconv.FormatInt(n, 10)) + } + if err := readBatch(ctx, client, conversation, query, fn); err != nil { + return err + } + } + return nil +} + +func readBatch(ctx context.Context, client *http.Client, conversation string, query url.Values, fn func(File, []byte) error) error { + full := view.CoreURL(transport.GetValue(ctx, contextkey.BaseURL{}, defaultBaseURL)) + Path(conversation) + "?" + query.Encode() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, full, http.NoBody) + if err != nil { + return err + } + req.Header.Set("Accept", MediaType) + if authorization := transport.AuthHeader(ctx); authorization != "" { + req.Header.Set("Authorization", authorization) + } + resp, err := client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return view.ReadError(resp, full) + } + return Parse(resp.Body, fn) +} + +// Parse reads a files body: a naming line, then exactly as many bytes as it says, the file, then +// the one newline that follows a file not ending with its own. A file can be as large as the +// largest stored, so it is read by its size, never line by line with a bounded scanner. +func Parse(body io.Reader, fn func(File, []byte) error) error { + r := bufio.NewReaderSize(body, 256*1024) + for { + line, err := r.ReadBytes('\n') + if errors.Is(err, io.EOF) && len(line) == 0 { + return nil + } + if err != nil { + return fmt.Errorf("the files response ends inside a naming line: %w", err) + } + var f File + if err := json.Unmarshal(line, &f); err != nil { + return fmt.Errorf("not a naming line: %w", err) + } + if f.Bytes < 0 { + return fmt.Errorf("%s: a naming line with %d bytes", f.ID, f.Bytes) + } + // the buffer grows with the bytes that actually arrive, so a size no response carries never allocates + var buf bytes.Buffer + if n, err := io.Copy(&buf, io.LimitReader(r, int64(f.Bytes))); err != nil || n != int64(f.Bytes) { + return fmt.Errorf("%s ends after %d of its %d bytes: %v", f.ID, n, f.Bytes, err) + } + content := buf.Bytes() + if f.Bytes > 0 && content[f.Bytes-1] != '\n' { + if b, err := r.ReadByte(); err != nil || b != '\n' { + return fmt.Errorf("%s: no newline after a file that does not end with one", f.ID) + } + } + sum := sha256.Sum256(content) + if got := hex.EncodeToString(sum[:]); got != f.Digest { + return fmt.Errorf("%s: its bytes hash to %s, the OAP names %s", f.ID, got, f.Digest) + } + if err := fn(f, content); err != nil { + return err + } + } +} diff --git a/pkg/aiagent/files/files_test.go b/pkg/aiagent/files/files_test.go new file mode 100644 index 0000000..3b00357 --- /dev/null +++ b/pkg/aiagent/files/files_test.go @@ -0,0 +1,200 @@ +// Licensed to Apache Software Foundation (ASF) under one or more contributor +// license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright +// ownership. Apache Software Foundation (ASF) licenses this file to you under +// the Apache License, Version 2.0 (the "License"); you may +// not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package files + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync" + "testing" + + "github.com/apache/skywalking-cli/pkg/aiagent/view" + "github.com/apache/skywalking-cli/pkg/contextkey" +) + +type storedFile struct { + name string + content []byte +} + +// stored are the Session Data files of session "s": a small transcript; a provider file whose one +// record line is past any scanner's default buffer; a file without a final newline; and an empty +// one. +var stored = map[int64]storedFile{ + 1: {"s/streams/main/transcript-20260101T000000.000000000Z-000001.sd", []byte("{\"h\":1}\n{\"ord\":1}\n{\"t\":\"end\"}\n")}, + 2: {"s/provider_body/provider_body-20260101T000000.000000000Z-000002.sd", []byte( + "{\"h\":1}\n{\"ord\":1,\"parts\":[{\"k\":\"data\",\"data\":\"" + strings.Repeat("x", 3<<20) + "\"}]}\n{\"t\":\"end\"}\n")}, + 3: {"s/unknown-000003.sd", []byte("{\"h\":1}\n{\"t\":\"end\"}")}, + 4: {"s/unknown-000004.sd", nil}, +} + +func naming(f storedFile, seq int64) string { + sum := sha256.Sum256(f.content) + return fmt.Sprintf(`{"file":%q,"seq":%d,"lines":%d,"bytes":%d,"digest":%q}`+"\n", + f.name, seq, bytes.Count(f.content, []byte("\n")), len(f.content), hex.EncodeToString(sum[:])) +} + +// frame writes one file the way the OAP does: its naming line, its bytes, and a newline after a +// non-empty file that does not end with one. +func frame(f storedFile, seq int64) string { + out := naming(f, seq) + string(f.content) + if len(f.content) > 0 && f.content[len(f.content)-1] != '\n' { + out += "\n" + } + return out +} + +// server answers the files route the way the OAP does. It records how many seqs each request +// carried. +func server(t *testing.T, batches *[]int, mu *sync.Mutex) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + if q.Get("service") != "agent" || q.Get("instance") != "sender" || q.Get("session") != "s" || + r.Header.Get("Accept") != MediaType { + w.WriteHeader(http.StatusTeapot) + return + } + if r.URL.Path != "/ai-agent/conversations/c/v1/files" { + w.Header().Set("Content-Type", "application/problem+json") + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte(`{"type":"about:blank","title":"Not Found","status":404,"detail":"no round"}`)) + return + } + mu.Lock() + *batches = append(*batches, len(q["seq"])) + mu.Unlock() + w.Header().Set("Content-Type", MediaType+"; charset=utf-8") + for _, text := range q["seq"] { + n, _ := strconv.ParseInt(text, 10, 64) + if f, ok := stored[n]; ok { + _, _ = w.Write([]byte(frame(f, n))) + } + } + })) +} + +func testContext(serverURL string) context.Context { + return context.WithValue(context.Background(), contextkey.BaseURL{}, serverURL+"/graphql") +} + +func TestEveryChosenFileComesBackByteForByte(t *testing.T) { + var batches []int + var mu sync.Mutex + srv := server(t, &batches, &mu) + defer srv.Close() + + got := map[string][]byte{} + take := func(f File, content []byte) error { + got[f.ID] = content + return nil + } + if err := Read(testContext(srv.URL), "c", "agent", "sender", "s", []int64{1, 2, 3, 4, 9}, take); err != nil { + t.Fatal(err) + } + if len(got) != len(stored) { + t.Fatalf("%d files, want %d", len(got), len(stored)) + } + for _, f := range stored { + if !bytes.Equal(got[f.name], f.content) { + t.Errorf("%s: %d bytes, want %d", f.name, len(got[f.name]), len(f.content)) + } + } +} + +func TestSeqsAreAskedForInBatchesUnderTheRouteLimit(t *testing.T) { + var batches []int + var mu sync.Mutex + srv := server(t, &batches, &mu) + defer srv.Close() + + many := make([]int64, 0, 2*MaxSeqs+1) + for i := int64(0); i < 2*MaxSeqs+1; i++ { + many = append(many, 1000+i) + } + if err := Read(testContext(srv.URL), "c", "agent", "sender", "s", many, func(File, []byte) error { return nil }); err != nil { + t.Fatal(err) + } + if fmt.Sprint(batches) != fmt.Sprint([]int{MaxSeqs, MaxSeqs, 1}) { + t.Fatalf("batches %v", batches) + } +} + +func TestParseRefusesWhatDoesNotMatchItsNamingLine(t *testing.T) { + f := storedFile{"a", []byte("{\"h\":1}\n{\"t\":\"end\"}\n")} + unended := storedFile{"b", []byte("{\"h\":1}")} + cases := map[string]string{ + "a wrong digest": strings.Replace(naming(f, 1), `"digest":"`, `"digest":"00`, 1) + string(f.content), + "a short file": naming(f, 1) + "{\"h\":1}\n", + "no naming line": "not json\n", + "a cut naming line": `{"file":"a"`, + "no newline after unended": naming(unended, 2) + string(unended.content), + "a negative size": `{"file":"a","bytes":-1,"lines":0,"digest":""}` + "\n", + "a size no response carries": `{"file":"a","seq":1,"lines":0,"bytes":9223372036854775807,"digest":""}` + "\n", + "another byte after unended": naming(unended, 2) + string(unended.content) + "x", + } + for what, body := range cases { + if err := Parse(strings.NewReader(body), func(File, []byte) error { return nil }); err == nil { + t.Errorf("%s: no error", what) + } + } + // an empty file is followed by nothing, and the next naming line comes straight after its own + empty := storedFile{"e", nil} + whole := frame(f, 1) + frame(empty, 3) + frame(unended, 2) + var got [][]byte + if err := Parse(strings.NewReader(whole), func(_ File, content []byte) error { + got = append(got, content) + return nil + }); err != nil || len(got) != 3 || len(got[1]) != 0 || !bytes.Equal(got[2], unended.content) { + t.Errorf("a whole stream: %v, %d files", err, len(got)) + } +} + +func TestTheSenderTheSessionAndASeqAreRequired(t *testing.T) { + noop := func(File, []byte) error { return nil } + for what, err := range map[string]error{ + "no instance": Read(context.Background(), "c", "agent", "", "s", []int64{1}, noop), + "no session": Read(context.Background(), "c", "agent", "sender", "", []int64{1}, noop), + "no seq": Read(context.Background(), "c", "agent", "sender", "s", nil, noop), + } { + if err == nil { + t.Errorf("%s: no error", what) + } + } +} + +func TestAProblemDocumentIsTheError(t *testing.T) { + var batches []int + var mu sync.Mutex + srv := server(t, &batches, &mu) + defer srv.Close() + + var problem *view.Problem + err := Read(testContext(srv.URL), "missing", "agent", "sender", "s", []int64{1}, func(File, []byte) error { return nil }) + if !errors.As(err, &problem) || problem.Status != 404 { + t.Fatalf("problem: %v", err) + } +} diff --git a/pkg/aiagent/view/view.go b/pkg/aiagent/view/view.go index 94546c6..9042ada 100644 --- a/pkg/aiagent/view/view.go +++ b/pkg/aiagent/view/view.go @@ -81,13 +81,14 @@ func Path(conversation string) string { } // Fetch streams the document of the conversation to out and returns the Content-Type -// it came with. serviceName is required; instanceName narrows the read to one sender. -// A non-2xx answer is returned as a *Problem when the OAP sent one. +// it came with. serviceName and instanceName are both required, as a list row names the +// sender of every conversation. A non-2xx answer is returned as a *Problem when the OAP +// sent one. func Fetch(ctx context.Context, conversation, serviceName, instanceName string, yaml bool, out io.Writer) (string, error) { - query := url.Values{"service": {serviceName}} - if instanceName != "" { - query.Set("instance", instanceName) + if serviceName == "" || instanceName == "" { + return "", fmt.Errorf("the service and the instance are both required") } + query := url.Values{"service": {serviceName}, "instance": {instanceName}} full := CoreURL(transport.GetValue(ctx, contextkey.BaseURL{}, defaultBaseURL)) + Path(conversation) + "?" + query.Encode() req, err := http.NewRequestWithContext(ctx, http.MethodGet, full, http.NoBody) @@ -111,15 +112,15 @@ func Fetch(ctx context.Context, conversation, serviceName, instanceName string, contentType := resp.Header.Get("Content-Type") if resp.StatusCode < 200 || resp.StatusCode >= 300 { - return contentType, readError(resp, full) + return contentType, ReadError(resp, full) } _, err = io.Copy(out, resp.Body) return contentType, err } -// readError turns a non-2xx response into an error: the problem document when the OAP +// ReadError turns a non-2xx response into an error: the problem document when the OAP // sent one, otherwise the status and whatever the body says. -func readError(resp *http.Response, full string) error { +func ReadError(resp *http.Response, full string) error { body, _ := io.ReadAll(io.LimitReader(resp.Body, 64*1024)) mediaType, _, _ := mime.ParseMediaType(resp.Header.Get("Content-Type")) if mediaType == problemType { diff --git a/pkg/aiagent/view/view_test.go b/pkg/aiagent/view/view_test.go index dadf616..fb9089c 100644 --- a/pkg/aiagent/view/view_test.go +++ b/pkg/aiagent/view/view_test.go @@ -123,7 +123,7 @@ func TestAProblemDocumentIsTheError(t *testing.T) { } defer resp.Body.Close() var problem *Problem - if err := readError(resp, req.URL.String()); !errors.As(err, &problem) || + if err := ReadError(resp, req.URL.String()); !errors.As(err, &problem) || problem.Status != 404 || problem.Detail != "no round" || problem.Title != "Not Found" { t.Fatalf("problem: %v", err) } diff --git a/pkg/graphql/aiagent/conversation.go b/pkg/graphql/aiagent/conversation.go index fce0246..b9ebd26 100644 --- a/pkg/graphql/aiagent/conversation.go +++ b/pkg/graphql/aiagent/conversation.go @@ -15,9 +15,9 @@ // specific language governing permissions and limitations // under the License. -// Package aiagent wraps the GraphQL queries of ai-agent-conversation.graphqls: the -// list page and the raw-file export of the conversations the AI Sessionizer lands. -// The conversation document itself is not a GraphQL query; see pkg/aiagent/view. +// Package aiagent wraps the GraphQL query of ai-agent-conversation.graphqls: the list +// page of the conversations the AI Sessionizer lands. The conversation document and its +// files are not GraphQL queries; see pkg/aiagent/view and pkg/aiagent/files. package aiagent import ( @@ -42,17 +42,3 @@ func ListConversations(ctx context.Context, condition *api.ConversationListCondi err := client.ExecuteQuery(ctx, request, &response) return response["result"], err } - -// RawFiles lists every landed file and round of a conversation as stored, or only the -// named ones; with body, each file comes verbatim, which is the export path. -func RawFiles(ctx context.Context, condition *api.ConversationCondition, files []string, body bool) (api.ConversationRawFiles, error) { - var response map[string]api.ConversationRawFiles - - request := graphql.NewRequest(assets.Read("graphqls/aiagent/ConversationRawFiles.graphql")) - request.Var("condition", condition) - request.Var("files", files) - request.Var("body", body) - - err := client.ExecuteQuery(ctx, request, &response) - return response["result"], err -}