Add AI agent conversations landed by the AI Sessionizer

.github/workflows/skywalking.yaml

@@ -488,6 +488,15 @@ jobs:
             env: ES_VERSION=8.18.8
           - name: Log BanyanDB
             config: test/e2e-v2/cases/log/banyandb/e2e.yaml
+          - name: AI Agent Conversations BanyanDB
+            config: test/e2e-v2/cases/ai-agent/banyandb/e2e.yaml
+          - name: AI Agent Conversations ES 8.18.8
+            config: test/e2e-v2/cases/ai-agent/es/e2e.yaml
+            env: ES_VERSION=8.18.8
+          - name: AI Agent Conversations MySQL
+            config: test/e2e-v2/cases/ai-agent/mysql/e2e.yaml
+          - name: AI Agent Conversations Postgres
+            config: test/e2e-v2/cases/ai-agent/postgres/e2e.yaml
 
           - name: Log FluentBit ES 8.18.8
             config: test/e2e-v2/cases/log/fluent-bit/e2e.yaml

.licenserc.yaml

@@ -25,6 +25,7 @@ header:
     - '.github/ISSUE_TEMPLATE'
     - '.github/PULL_REQUEST_TEMPLATE'
     - '**/.gitignore'
+    - '**/.gitattributes'
     - '.gitmodules'
     - '.lift'
     - '.mvn'
@@ -48,6 +49,8 @@ header:
     - '**/*.pem'
     - '**/*.key'
     - '**/*.txt'
+    # AI Sessionizer data files and its rendered asz.view document, used as test fixtures as-is
+    - 'oap-server/analyzer/ai-agent-conversation/src/test/resources/fixtures/**'
     - 'LICENSE'
     - 'NOTICE'
     - '**/src/main/fbs/istio/**'

docs/en/api/query-protocol.md

@@ -323,6 +323,26 @@ extend type Query {
 }
 ```
 
+### AI Agent Conversations
+Provide [AI agent conversation](../setup/backend/ai-agent-conversation.md) query APIs since 11.1.0: the list page and the raw-file export
+are GraphQL queries; the conversation itself is an HTTP route on the same server, because its `asz.view` document is as
+large as the conversation and is streamed.
+```graphql
+extend type Query {
+    # The conversations of a service active in the duration, newest first.
+    listConversations(condition: ConversationListCondition!, duration: Duration!, debug: Boolean): ConversationList
+    # Every file of a conversation, as stored. Select `body` to export them.
+    getConversationRawFiles(condition: ConversationCondition!, files: [ID!], debug: Boolean): ConversationRawFiles
+}
+```
+```
+GET /ai-agent/conversations/{conversation}/v1/view?service={serviceName}[&instance={instanceName}]
+```
+The body is one `asz.view` 1.0 document, streamed, and its `Content-Type` names the format and the version:
+`application/vnd.skywalking.asz.view+json; version=1.0`, or the `+yaml` twin when `Accept` asks for YAML; compressed on
+`Accept-Encoding`. `v1` in the path is the document version. 400 without a service, 404 when no round of the conversation
+is stored, 500 on a storage failure, each as `application/problem+json`.
+
 ## Condition
 ### Duration
 Duration is a widely used parameter type as the APM data is time-related. See the following for more details. 

docs/en/changes/changes.md

@@ -11,6 +11,7 @@
 * Add BanyanDB trace tail sampling metrics to the BanyanDB self-observability layer, in a new `otel-rules/banyandb/banyandb-trace-sampling.yaml` rule file. It covers the whole `banyandb_trace_pipeline_*` / `banyandb_trace_tst_pipeline_*` catalog a sampler plugin chain emits — pipeline reconciliation, per-plugin `Decide` execution rate and latency, chain batching, the trace-level evaluated / retained / dropped / immature outcomes, every fail-open guard and bounded-retention counter, drop-set capacity and finalization state, the plugin telemetry-host safety bounds, and the first-party `sw-trace-sampler` / `zipkin-trace-sampler` decision and row metrics. The plugin chain is optional, and the metrics follow it: on a cluster with no sampler configured the wire families are never registered, so every metric here stays absent rather than reading zero. Modeled at Service scope with `group` kept as a metric label rather than at Endpoint scope, so one cluster-wide page can render per-group series and cluster totals alike — OAP does no cross-scope rollup, so an Endpoint-scope metric could not have been aggregated back up to the cluster.
 * Fix a second `CounterWindow` key collision in the v2 MAL engine, this time ACROSS rules. `rate()` / `increase()` / `irate()` resolve their lower bound from a process-wide window keyed on the counter's own name plus its post-`.sum(...)` label set, with nothing identifying the rule doing the evaluation. Two rules that read one wire family, tell their streams apart with `tagEqual(...)`, and then `.sum(...)` away the label they filtered on therefore collapse onto one window slot and difference against each other's values. The queue is ordered by (timestamp, value), so the smaller counter wins the lower-bound lookup and still reads correctly while its partner is inflated by the gap between them — which is why this went unnoticed. A collision needs the discriminating label to be DROPPED by the `.sum(...)`: where it survives, the rules' label values differ and the window keeps them apart. Auditing the shipped rules on that basis gives 10 colliding keys over ~25 rules — `meter_activemq_cluster_gc_parallel_young_collection_count` reported ~9000/min of young-gen collections from a completely idle broker (differencing against the old-gen counter); MySQL `commands_*` / `tps` rate against each other; so do the GenAI gateway input/output token rates, four Envoy `cluster_*` counters, APISIX matched/unmatched instance bandwidth, and BanyanDB's own `network_recv` / `network_sent`, which drop the `kind` label that separates bytes-received from bytes-sent on one interface. Measured against two live scrapes of the demo cluster's FODC proxy, that last pair was wrong on every interface: `network_sent` read a flat 0 B/s and `network_recv` read large negative values (down to -778 MB/s) from differencing against the sent counter, where both now match the byte delta exactly. No rule changes were needed for any of these -- each rule already reduces to the labels it should; only the window key was wrong. The window is now keyed by (owning rule, counter name, labels). This is the complement of the within-rule collision fixed earlier by keying on the counter's own name: neither name alone is sufficient, because the two collisions are independent. `RunningContext.metricName` — written on every rule evaluation and read by nobody since that earlier fix — is what supplies the rule identity, so no code generation or MAL syntax changes. Note the whole-rule-set comparison suite could not have caught this: it resets the shared window before every rule, the one condition under which the collision cannot appear.
 * Fix `meter_rabbitmq_node_outgoing_messages_total` double-counting one of its terms. The rule summed six delivery-rate terms but `rabbitmq_global_messages_delivered_get_auto_ack_total` appeared twice, so auto-ack `basic.get` deliveries were counted once more than the other four delivery paths and the reported outgoing rate ran high whenever polling consumers were in use. The duplicate term is removed, leaving the five distinct families (redelivered, consume auto/manual ack, get auto/manual ack).
+* Add AI agent conversations landed by the AI Sessionizer: the `AI_AGENT` layer, the bundled `lal/ai-agent.yaml` rule with the `ConversationFile` output builder that verifies and stores Session Data and Session Flow files, the `ai_agent_session_data` and `ai_agent_session_flow` models in a new BanyanDB group `recordsAIAgent`, the `ai-agent-conversation` module that folds a conversation into one `asz.view` document, and the `listConversations` / `getConversationRawFiles` GraphQL queries and the streamed `GET /ai-agent/conversations/{conversation}/v1/view` route that serves the document.
 
 #### UI
 * Add a Virtual GenAI evaluation-record page and evaluation-score chart in Horizon UI, so operators can inspect evaluation result, level, reason, judge model, timestamp, trace linkage, and the `gen_ai_model_evaluation_score_ppm` trend for evaluated records.

docs/en/setup/backend/ai-agent-conversation.md

@@ -0,0 +1,155 @@
+# AI Agent Conversations
+
+Since 11.1.0, SkyWalking stores and serves the conversations of long-lived AI agents. The feature requires the
+[SkyWalking AI Sessionizer](https://github.com/apache/skywalking-ai-sessionizer) as the sender: it is the
+producer of the files described here, and a record under this layer without their attributes is rejected. The
+Sessionizer collects an agent
+runtime's transcripts into two file formats, Session Data (`.sd`, the records as collected) and Session Flow
+(`.sf`, an append-only chain of rounds that describe the conversation's structure), and pushes every file as one
+OTLP log record. The OAP verifies each file on arrival, stores it verbatim, and answers a conversation query with
+one `asz.view` document that a viewer renders without opening any file.
+
+In the Sessionizer's model a **conversation** is the unit of storage, analysis and export. A **session** is the
+source-runtime context a record came from, carried as provenance: one conversation may contain several sessions,
+and a session belongs to exactly one conversation.
+
+## How a file reaches the OAP
+
+The sender puts these resource attributes on every request:
+
+| Attribute             | Value                                                                          |
+|-----------------------|--------------------------------------------------------------------------------|
+| `service.name`        | the name the sender is configured with, or else the runtime that produced the session, such as `Claude Code` |
+| `service.instance.id` | who is pushing, in words the people reading the OAP recognise: a mailbox, a name or a machine, `user@host` of the pushing machine by default |
+| `service.layer`       | `AI_AGENT`                                                                     |
+
+Each log record is one file. The body is the file's text. The record attributes name the file (`asz.format`,
+`asz.file`, `asz.file.digest`, `asz.lines`, `asz.session`, `asz.seq` for a Session Data file; `asz.conversation`,
+`asz.round`, the conversation's time range and its title and counts for a round). The two file formats are
+documented by the Sessionizer under
+[Session Data](https://skywalking.apache.org/docs/skywalking-ai-sessionizer/next/en/formats/session-data/) and
+[Session Flow](https://skywalking.apache.org/docs/skywalking-ai-sessionizer/next/en/formats/session-flow/), and
+the wire attributes under
+[Export over OpenTelemetry](https://skywalking.apache.org/docs/skywalking-ai-sessionizer/next/en/setup/export-otlp/).
+
+The OAP routes these records like every other OTLP log: by layer, to the bundled LAL rule
+`lal/ai-agent.yaml`. The rule's output type, `ConversationFile`, checks the body's sha256 against
+`asz.file.digest` and its line count against `asz.lines`, and stores the file in the table its format names. A
+file that fails either check is dropped and counted in the `ai_agent_conversation_files_rejected` self-observability
+metric with the reason as a label; a stored file is a verified file. The service and its instance appear on the
+service list under the `AI_AGENT` layer as for any other log sender.
+
+Nothing is folded or decoded at ingest, so an OAP cluster needs no shared state for this feature.
+
+## Storage
+
+Two record models, both super datasets:
+
+| Model                   | One row per        | Keys                                                           | Stored only                                                                                                        |
+|-------------------------|--------------------|----------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------|
+| `ai_agent_session_data` | Session Data file  | `service_id`, `service_instance_id`, `session`, indexed `seq`  | `digest`, `body`                                                                                                   |
+| `ai_agent_session_flow` | Session Flow round | `service_id`, `service_instance_id`, indexed `conversation`, `round` | `session_from_time`, `title`, `talks`, `steps`, `streams`, `segments`, `unresolved`, `digest`, `body`     |
+
+A Session Data row carries nothing but its keys and the file: the file's kind, stream or run, time range and name
+are on its first line and are read from there. A round's stored-only columns exist for the list page, which reads
+them without opening a body; its `round` number is queryable so a long chain is read window by window. The row's
+timestamp is the file's latest record time, or the conversation's last activity for a round, so a conversation's
+files are found by its own time range. A row belongs to its sender: its id is the service, the instance and the
+file's digest, so the same file pushed again by the same sender lands on the same row, and pushed by another
+service or sender makes another.
+
+- **BanyanDB**: both models live in their own group, `recordsAIAgent`, configured like the log group with hot,
+  warm and cold stages under `SW_STORAGE_BANYANDB_AI_AGENT_*`, 30 days hot by default. Both tables expire
+  together, because a round whose files are gone is a broken chain. See the
+  [BanyanDB storage document](storages/banyandb.md).
+- **Elasticsearch**: two super-dataset index families, `sw_ai_agent_session_data-*` and
+  `sw_ai_agent_session_flow-*`, sharded by `superDatasetIndexShardsFactor`; retention is the single
+  `recordDataTTL`. The columns the reads sort and range on, `seq`, `round` and `timestamp`, keep doc values.
+- **JDBC** (MySQL, PostgreSQL, H2): two tables of their own; the body is `MEDIUMTEXT` on MySQL and H2 and `TEXT`
+  on PostgreSQL; retention is `recordDataTTL`.
+
+## Query
+
+The list and the export are GraphQL queries in `ai-agent-conversation.graphqls`; the conversation itself is an
+HTTP route on the same server, because its document is as large as the conversation.
+
+- `listConversations(condition, duration)` lists one row per conversation of a service, optionally of one sender,
+  from the newest round's attributes. The rounds are read newest first, at most `limit` (default 1000), then
+  folded to one row per conversation.
+- `getConversationRawFiles(condition, files)` lists every landed file and round of a conversation with its id,
+  digest and size; selecting `body` returns the files verbatim, which is the export path. The optional `files`
+  argument narrows the read to named files.
+
+### The conversation view route
+
+```
+GET /ai-agent/conversations/{conversation}/v1/view?service={serviceName}[&instance={instanceName}]
+```
+
+It answers with the whole conversation, once, as one `asz.view` version 1.0 document, the document the
+Sessionizer defines under
+[The asz.view document](https://skywalking.apache.org/docs/skywalking-ai-sessionizer/next/en/formats/asz-view/)
+and serves from its own viewer; the OAP's document equals it, key for key, for the same files. `v1` in the path
+is the document version. The OAP reads the conversation's rounds over the whole retention window, then the
+files of each session the head round names over the time range the head round carries, checks the chain, folds
+the rounds, resolves every reference into the landed records, and renders the document. Verification is
+content, not an error: a missing round or file, or a failed digest, is written into the document's
+`summary.state` and `summary.problems`, and the rest of the document holds whatever could still be folded. The
+fold stops before a round that is missing, that does not read, or that belongs to another conversation,
+session, parser or policy; `head` names the last round it reached, and the rounds after it are listed and not
+verified, exactly as the Sessionizer's own viewer does. The document is built on every call and nothing is cached.
+
+| Parameter or header | Meaning |
+|---|---|
+| `service` / `serviceId` | the service by name, or by id; one of them is required |
+| `instance` | optional, the sender's instance name from the list row; with it, every storage read is a full series lookup |
+| `Accept` | `application/vnd.skywalking.asz.view+yaml`, or any type naming `yaml`, for YAML; anything else, JSON, as `asz conversation -json` prints it |
+| `Content-Type` | names the document and its version, the HTTP way: `application/vnd.skywalking.asz.view+json; version=1.0` or `application/vnd.skywalking.asz.view+yaml; version=1.0`. The document's own first two keys, `format` and `version`, say the same |
+| `Accept-Encoding` | the body is compressed when the client allows; a document is repetitive text and shrinks several times over |
+| status | 200 with the document; 400 when no service is named; 404 when the service stores no round of the conversation; 500 on a storage failure. An error is `application/problem+json` ([RFC 9457](https://www.rfc-editor.org/rfc/rfc9457)): `{"type": "about:blank", "title": "Not Found", "status": 404, "detail": "..."}` |
+
+The route is on the core HTTP server beside `/graphql`, so it has the same host, port, context path and TLS
+settings, and serves HTTP/1.1 and HTTP/2 alike. The body is streamed: it is written to the response as it is
+rendered, never held whole in memory, and a slow client holds back the render. The route runs under its own
+timeout, `viewRequestTimeout`, in place of the server's default of ten seconds, because the floor for a large
+conversation is seconds of storage reads plus seconds of fold and render.
+
+The conversation page of the UI makes one call, this route, and nothing else.
+
+## Configuration
+
+```yaml
+ai-agent-conversation:
+  selector: ${SW_AI_AGENT_CONVERSATION:default}
+  default:
+    fileReadWindow: ${SW_AI_AGENT_CONVERSATION_FILE_READ_WINDOW:16}
+    roundReadWindow: ${SW_AI_AGENT_CONVERSATION_ROUND_READ_WINDOW:16}
+    maxListLimit: ${SW_AI_AGENT_CONVERSATION_MAX_LIST_LIMIT:10000}
+    viewRequestTimeout: ${SW_AI_AGENT_CONVERSATION_VIEW_REQUEST_TIMEOUT:120}
+```
+
+| Key              | Meaning                                                                                                                                     |
+|------------------|---------------------------------------------------------------------------------------------------------------------------------------------|
+| `fileReadWindow` | how many Session Data files one storage query fetches, a batch size and not a limit: the view and the raw-file export read every file of the conversation, this many per query. Files are cut at 2 MiB, and the BanyanDB client caps one response at 50 MB. |
+| `roundReadWindow` | how many Session Flow rounds one storage query fetches, the same way: the head round is fixed first, then the chain is read from round 1 to the head, this many per query. A round is cut at 2 MiB by the Sessionizer. |
+| `maxListLimit`   | the most rounds one list query reads before folding, and the ceiling of the query's `limit` argument.                                       |
+| `viewRequestTimeout` | how long one conversation view request may take, in seconds. |
+
+The GraphQL query module requires this module, so it cannot be disabled while the GraphQL query module is active.
+
+## Limits on the path
+
+- The OAP's OTLP/HTTP endpoint accepts requests of up to 10 MiB, the HTTP server's default. The Sessionizer's
+  request budget defaults to 8 MiB for that reason; a single file is cut at 2 MiB, so it always fits.
+- The BanyanDB client caps one query response at 50 MB, and Elasticsearch answers at most 10,000 hits to one search.
+  The files of a conversation are read in windows of `fileReadWindow` files, and its rounds in windows of
+  `roundReadWindow` rounds, per storage query, inside one view request.
+- A read that is not bound to a duration, the view and the export, covers every retained stage: on BanyanDB the
+  default stages and, when the group keeps one, the cold stage. A conversation the list found in cold storage
+  is served, and one that spans stages is served whole.
+- When the caller names no sender, the view and the export read across every sender of the service and keep one
+  copy of a file or round two senders both pushed, so a Sessionizer renamed between pushes still yields the
+  whole conversation.
+- The `asz.view` document grows with the conversation. A session of 136 MB of landed files renders to a 70 MB
+  document in about five seconds after about six seconds of storage reads, which is why the view is a streamed
+  route with its own timeout and not a GraphQL query.