Read AI agent conversation files over the OAP's HTTP route (#235)
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.diff --git a/CHANGES.md b/CHANGES.md
index 5e1abc2..63d55f8 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -7,7 +7,7 @@
### 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 @@
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 @@
"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
+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
-2. Export them all:
-$ swctl ai-agent files --service-name "Claude Code" --conversation 7a3c882e-0dc0-46a0-b814-6613d24b7ac2 --export ./root
-
-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 @@
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)
+// 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
}
- 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)
+ 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.MkdirAll(filepath.Dir(path), 0o755); err != nil {
- return nil, err
- }
- 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 @@
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 @@
),
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 @@
}
// 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 @@
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 @@
}
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 @@
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
-}
