diff --git a/.github/workflows/skywalking.yaml b/.github/workflows/skywalking.yaml
index c177cc086e3d..8d17d3124e29 100644
--- a/.github/workflows/skywalking.yaml
+++ b/.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
diff --git a/.licenserc.yaml b/.licenserc.yaml
index 556b7965154b..1f50cd5cb4a4 100644
--- a/.licenserc.yaml
+++ b/.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/**'
diff --git a/docs/en/api/query-protocol.md b/docs/en/api/query-protocol.md
index 90c9fb818844..efca61e97a7a 100644
--- a/docs/en/api/query-protocol.md
+++ b/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.
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index f094b4c0b898..cb81d5527cc8 100644
--- a/docs/en/changes/changes.md
+++ b/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.
diff --git a/docs/en/setup/backend/ai-agent-conversation.md b/docs/en/setup/backend/ai-agent-conversation.md
new file mode 100644
index 000000000000..aa6808bb18b3
--- /dev/null
+++ b/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.
diff --git a/docs/en/setup/backend/configuration-vocabulary.md b/docs/en/setup/backend/configuration-vocabulary.md
index f1235a27e1a9..4628d1af8fef 100644
--- a/docs/en/setup/backend/configuration-vocabulary.md
+++ b/docs/en/setup/backend/configuration-vocabulary.md
@@ -172,6 +172,11 @@ It divided into several modules, each of which has its own settings. The followi
| log-analyzer | default | Log Analyzer. | SW_LOG_ANALYZER | default | |
| - | - | lalFiles | The LAL configuration file names (without file extension) to be activated. Read [LAL](../../concepts-and-designs/lal.md) for more details. | SW_LOG_LAL_FILES | default |
| - | - | malFiles | The MAL configuration file names (without file extension) to be activated. Read [LAL](../../concepts-and-designs/lal.md) for more details. | SW_LOG_MAL_FILES | "" |
+| ai-agent-conversation | default | Conversations of long-lived AI agents landed by the AI Sessionizer as Session Data and Session Flow files over OTLP logs under the `AI_AGENT` layer. | SW_AI_AGENT_CONVERSATION | default | - |
+| - | - | fileReadWindow | How many Session Data files one storage read fetches; keeps one BanyanDB response under its inbound cap. | SW_AI_AGENT_CONVERSATION_FILE_READ_WINDOW | 16 |
+| - | - | roundReadWindow | How many Session Flow rounds one storage read fetches; a long conversation is read window by window up to its head. | SW_AI_AGENT_CONVERSATION_ROUND_READ_WINDOW | 16 |
+| - | - | maxListLimit | The most rounds one list query reads before folding to one row per conversation. | SW_AI_AGENT_CONVERSATION_MAX_LIST_LIMIT | 10000 |
+| - | - | viewRequestTimeout | How long one conversation view request may take, in seconds, in place of the HTTP server's default. | SW_AI_AGENT_CONVERSATION_VIEW_REQUEST_TIMEOUT | 120 |
| event-analyzer | default | Event Analyzer. | SW_EVENT_ANALYZER | default | |
| receiver-register | default | gRPC and HTTPRestful services that provide service, service instance and endpoint register. | - | - | |
| receiver-trace | default | gRPC and HTTPRestful services that accept SkyWalking format traces. | - | - | |
diff --git a/docs/en/setup/backend/storages/banyandb.md b/docs/en/setup/backend/storages/banyandb.md
index 5ddf4188181c..662ca29b535d 100644
--- a/docs/en/setup/backend/storages/banyandb.md
+++ b/docs/en/setup/backend/storages/banyandb.md
@@ -199,6 +199,29 @@ groups:
ttl: ${SW_STORAGE_BANYANDB_LOG_COLD_TTL_DAYS:30}
replicas: ${SW_STORAGE_BANYANDB_LOG_COLD_REPLICAS:0}
nodeSelector: ${SW_STORAGE_BANYANDB_LOG_COLD_NODE_SELECTOR:"type=cold"}
+ # The "recordsAIAgent" group holds AI agent conversation files landed by the AI Sessionizer: Session Data (.sd)
+ # and Session Flow (.sf) rows of up to 2 MiB each. It mirrors the log group's keys but has its own retention,
+ # because conversations are kept for weeks and a round whose files have expired is a broken chain.
+ # See the "AI Agent Conversations" document.
+ recordsAIAgent:
+ shardNum: ${SW_STORAGE_BANYANDB_AI_AGENT_SHARD_NUM:2}
+ segmentInterval: ${SW_STORAGE_BANYANDB_AI_AGENT_SI_DAYS:1}
+ ttl: ${SW_STORAGE_BANYANDB_AI_AGENT_TTL_DAYS:30}
+ replicas: ${SW_STORAGE_BANYANDB_AI_AGENT_REPLICAS:0}
+ enableWarmStage: ${SW_STORAGE_BANYANDB_AI_AGENT_ENABLE_WARM_STAGE:false}
+ enableColdStage: ${SW_STORAGE_BANYANDB_AI_AGENT_ENABLE_COLD_STAGE:false}
+ warm:
+ shardNum: ${SW_STORAGE_BANYANDB_AI_AGENT_WARM_SHARD_NUM:2}
+ segmentInterval: ${SW_STORAGE_BANYANDB_AI_AGENT_WARM_SI_DAYS:1}
+ ttl: ${SW_STORAGE_BANYANDB_AI_AGENT_WARM_TTL_DAYS:60}
+ replicas: ${SW_STORAGE_BANYANDB_AI_AGENT_WARM_REPLICAS:0}
+ nodeSelector: ${SW_STORAGE_BANYANDB_AI_AGENT_WARM_NODE_SELECTOR:"type=warm"}
+ cold:
+ shardNum: ${SW_STORAGE_BANYANDB_AI_AGENT_COLD_SHARD_NUM:2}
+ segmentInterval: ${SW_STORAGE_BANYANDB_AI_AGENT_COLD_SI_DAYS:1}
+ ttl: ${SW_STORAGE_BANYANDB_AI_AGENT_COLD_TTL_DAYS:180}
+ replicas: ${SW_STORAGE_BANYANDB_AI_AGENT_COLD_REPLICAS:0}
+ nodeSelector: ${SW_STORAGE_BANYANDB_AI_AGENT_COLD_NODE_SELECTOR:"type=cold"}
recordsBrowserErrorLog:
shardNum: ${SW_STORAGE_BANYANDB_BROWSER_ERROR_LOG_SHARD_NUM:2}
segmentInterval: ${SW_STORAGE_BANYANDB_BROWSER_ERROR_LOG_SI_DAYS:1}
diff --git a/docs/en/setup/backend/storages/elasticsearch.md b/docs/en/setup/backend/storages/elasticsearch.md
index e83892fe7d73..d78abcb3ee7f 100644
--- a/docs/en/setup/backend/storages/elasticsearch.md
+++ b/docs/en/setup/backend/storages/elasticsearch.md
@@ -188,6 +188,8 @@ And also you can [specify the settings for each index individually.](#specify-se
| sw_segment-`${day-format}` | indexShardsNumber * superDatasetIndexShardsFactor | superDatasetIndexReplicasNumber |
| sw_browser_error_log-`${day-format}` | indexShardsNumber * superDatasetIndexShardsFactor | superDatasetIndexReplicasNumber |
| sw_zipkin_span-`${day-format}` | indexShardsNumber * superDatasetIndexShardsFactor | superDatasetIndexReplicasNumber |
+| sw_ai_agent_session_data-`${day-format}` | indexShardsNumber * superDatasetIndexShardsFactor | superDatasetIndexReplicasNumber |
+| sw_ai_agent_session_flow-`${day-format}` | indexShardsNumber * superDatasetIndexShardsFactor | superDatasetIndexReplicasNumber |
| sw_records-all-`${day-format}` | indexShardsNumber | indexReplicasNumber |
#### Advanced Configurations For Elasticsearch Index
diff --git a/docs/menu.yml b/docs/menu.yml
index d0e4cc3fe57b..d9a069c9db89 100644
--- a/docs/menu.yml
+++ b/docs/menu.yml
@@ -166,6 +166,8 @@ catalog:
path: "/en/setup/service-agent/virtual-genai"
- name: "AI Evaluation"
path: "/en/setup/backend/ai-evaluation"
+ - name: "AI Agent Conversations"
+ path: "/en/setup/backend/ai-agent-conversation"
- name: "Envoy AI Gateway"
path: "/en/setup/backend/backend-envoy-ai-gateway-monitoring"
- name: "Self Observability"
diff --git a/oap-server/analyzer/ai-agent-conversation/pom.xml b/oap-server/analyzer/ai-agent-conversation/pom.xml
new file mode 100644
index 000000000000..07c2b5239560
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/pom.xml
@@ -0,0 +1,74 @@
+
+
+
+
+ 4.0.0
+
+
+ org.apache.skywalking
+ analyzer
+ ${revision}
+
+
+ org.apache.skywalking
+ ai-agent-conversation
+
+
+ AI agent conversations landed by the AI Sessionizer: the LAL output builder that verifies and stores
+ Session Data and Session Flow files, and the read side that folds a conversation's rounds into one
+ asz.view document.
+
+
+
+
+ org.apache.skywalking
+ library-module
+ ${project.version}
+
+
+ org.apache.skywalking
+ telemetry-api
+ ${project.version}
+
+
+ org.apache.skywalking
+ server-core
+ ${project.version}
+
+
+ org.apache.skywalking
+ log-analyzer
+ ${project.version}
+ test
+
+
+ com.google.code.gson
+ gson
+
+
+ com.linecorp.armeria
+ armeria-junit5
+ test
+
+
+ org.yaml
+ snakeyaml
+
+
+
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/AIAgentConversationConfig.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/AIAgentConversationConfig.java
new file mode 100644
index 000000000000..e958b9345625
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/AIAgentConversationConfig.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation;
+
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.skywalking.oap.server.library.module.ModuleConfig;
+
+@Getter
+@Setter
+public class AIAgentConversationConfig extends ModuleConfig {
+ /**
+ * How many Session Data files one storage read fetches. Files are cut at 2 MiB, and the BanyanDB client caps
+ * one response at 50 MB, so the window keeps a single response well under the cap.
+ */
+ private int fileReadWindow = 16;
+ /**
+ * How many Session Flow rounds one storage read fetches. A round is cut at 2 MiB by the Sessionizer, so the
+ * window keeps a single response well under the BanyanDB client's cap and Elasticsearch's result window.
+ */
+ private int roundReadWindow = 16;
+ /**
+ * How long one conversation view request may take, in seconds, in place of the HTTP server's default of
+ * ten: the floor is the storage read of every landed file plus the fold and the render, which is seconds
+ * for a conversation of a hundred megabytes.
+ */
+ private int viewRequestTimeout = 120;
+ /**
+ * The most rounds one list query reads before folding to one row per conversation, and the ceiling of the
+ * query's own limit argument.
+ */
+ private int maxListLimit = 10000;
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/AIAgentConversationModule.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/AIAgentConversationModule.java
new file mode 100644
index 000000000000..13163784d866
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/AIAgentConversationModule.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation;
+
+import org.apache.skywalking.oap.server.ai.agent.conversation.query.IConversationQueryService;
+import org.apache.skywalking.oap.server.library.module.ModuleDefine;
+
+/**
+ * Conversations of long-lived AI agents, landed by the AI Sessionizer as Session Data (.sd) and
+ * Session Flow (.sf) files over OTLP logs under the AI_AGENT layer.
+ *
+ *
Ingest is a LAL output builder, {@code ConversationFile}, that the bundled lal/ai-agent.yaml
+ * rule names; it verifies each file and stores it verbatim. The read side folds a conversation's rounds and
+ * resolves every reference into the landed records, answering with one asz.view document.
+ */
+public class AIAgentConversationModule extends ModuleDefine {
+ public static final String NAME = "ai-agent-conversation";
+
+ public AIAgentConversationModule() {
+ super(NAME);
+ }
+
+ @Override
+ public Class[] services() {
+ return new Class[] {
+ IConversationQueryService.class
+ };
+ }
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/AIAgentConversationProvider.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/AIAgentConversationProvider.java
new file mode 100644
index 000000000000..09ef66c3227a
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/AIAgentConversationProvider.java
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation;
+
+import com.linecorp.armeria.common.HttpMethod;
+import java.time.Duration;
+import java.util.Collections;
+import org.apache.skywalking.oap.server.ai.agent.conversation.ingest.ConversationFileBuilder;
+import org.apache.skywalking.oap.server.ai.agent.conversation.query.ConversationQueryService;
+import org.apache.skywalking.oap.server.ai.agent.conversation.query.IConversationQueryService;
+import org.apache.skywalking.oap.server.ai.agent.conversation.query.http.ConversationViewHandler;
+import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.server.HTTPHandlerRegister;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.library.module.ModuleConfig;
+import org.apache.skywalking.oap.server.library.module.ModuleDefine;
+import org.apache.skywalking.oap.server.library.module.ModuleProvider;
+import org.apache.skywalking.oap.server.library.module.ModuleStartException;
+import org.apache.skywalking.oap.server.library.module.ServiceNotProvidedException;
+import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
+import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
+import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
+
+public class AIAgentConversationProvider extends ModuleProvider {
+ private AIAgentConversationConfig config = new AIAgentConversationConfig();
+ private ConversationQueryService queryService;
+
+ @Override
+ public String name() {
+ return "default";
+ }
+
+ @Override
+ public Class extends ModuleDefine> module() {
+ return AIAgentConversationModule.class;
+ }
+
+ @Override
+ public ConfigCreator extends ModuleConfig> newConfigCreator() {
+ return new ConfigCreator() {
+ @Override
+ public Class type() {
+ return AIAgentConversationConfig.class;
+ }
+
+ @Override
+ public void onInitialized(final AIAgentConversationConfig initialized) {
+ config = initialized;
+ }
+ };
+ }
+
+ @Override
+ public void prepare() throws ServiceNotProvidedException, ModuleStartException {
+ if (config.getFileReadWindow() <= 0) {
+ throw new ModuleStartException("fileReadWindow should be greater than 0");
+ }
+ if (config.getRoundReadWindow() <= 0) {
+ throw new ModuleStartException("roundReadWindow should be greater than 0");
+ }
+ if (config.getMaxListLimit() <= 0) {
+ throw new ModuleStartException("maxListLimit should be greater than 0");
+ }
+ if (config.getViewRequestTimeout() <= 0) {
+ throw new ModuleStartException("viewRequestTimeout should be greater than 0");
+ }
+ queryService = new ConversationQueryService(getManager(), config);
+ registerServiceImplementation(IConversationQueryService.class, queryService);
+ }
+
+ @Override
+ public void start() throws ServiceNotProvidedException, ModuleStartException {
+ getManager().find(CoreModule.NAME)
+ .provider()
+ .getService(HTTPHandlerRegister.class)
+ .addHandler(
+ new ConversationViewHandler(queryService, Duration.ofSeconds(config.getViewRequestTimeout())),
+ Collections.singletonList(HttpMethod.GET));
+ final MetricsCreator metricsCreator = getManager().find(TelemetryModule.NAME)
+ .provider()
+ .getService(MetricsCreator.class);
+ ConversationFileBuilder.setMetrics(
+ metricsCreator.createCounter(
+ "ai_agent_conversation_files_accepted", "AI agent conversation files verified and stored",
+ new MetricsTag.Keys("format"), new MetricsTag.Values("sd")),
+ metricsCreator.createCounter(
+ "ai_agent_conversation_files_accepted", "AI agent conversation files verified and stored",
+ new MetricsTag.Keys("format"), new MetricsTag.Values("sf")),
+ metricsCreator.createCounter(
+ "ai_agent_conversation_files_rejected", "AI agent conversation files rejected at ingest",
+ new MetricsTag.Keys("reason"), new MetricsTag.Values("digest")),
+ metricsCreator.createCounter(
+ "ai_agent_conversation_files_rejected", "AI agent conversation files rejected at ingest",
+ new MetricsTag.Keys("reason"), new MetricsTag.Values("lines")),
+ metricsCreator.createCounter(
+ "ai_agent_conversation_files_rejected", "AI agent conversation files rejected at ingest",
+ new MetricsTag.Keys("reason"), new MetricsTag.Values("attributes"))
+ );
+ }
+
+ @Override
+ public void notifyAfterCompleted() throws ServiceNotProvidedException, ModuleStartException {
+ }
+
+ @Override
+ public String[] requiredModules() {
+ return new String[] {
+ CoreModule.NAME,
+ StorageModule.NAME,
+ TelemetryModule.NAME
+ };
+ }
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/fold/ConversationFold.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/fold/ConversationFold.java
new file mode 100644
index 000000000000..f76e0712ec6d
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/fold/ConversationFold.java
@@ -0,0 +1,251 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.fold;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
+import lombok.Getter;
+import org.apache.skywalking.oap.server.ai.agent.conversation.format.Ref;
+import org.apache.skywalking.oap.server.ai.agent.conversation.format.SessionFlowRound;
+
+/**
+ * The fold of a conversation's rounds, as the Sessionizer's sessionflow.View folds them: entities
+ * keyed by id, a higher revision replaces a lower one, a tombstone removes, absence means unchanged.
+ *
+ *
Rounds must fold in order and each must name the previous round's commit digest. A round that breaks the
+ * chain is still folded, so a viewer sees what could be folded, and the break is recorded as a problem.
+ */
+@Getter
+public final class ConversationFold {
+ private final Map nodes = new LinkedHashMap<>();
+ private final Map relations = new LinkedHashMap<>();
+ private final Map unresolved = new LinkedHashMap<>();
+ private String conversation;
+ private String session;
+ private String parser;
+ private String policy;
+ private long round;
+ private String digest;
+ private long throughSeq;
+ private String inputDigest;
+ private String sessionFromTime;
+ private String sessionThroughTime;
+ private Map> kids;
+ private Map> from;
+ private Map> to;
+
+ /**
+ * Fold one round on top of the current head, or refuse it the way the Sessionizer's Apply
+ * does: a round out of order, naming another head, or from another conversation, session, parser or policy
+ * is not merged, and the reason is returned in the Sessionizer's words for the document to carry.
+ *
+ * @param r the next round of the chain
+ * @return null when the round was folded, else why it was refused
+ */
+ @Nullable
+ public String apply(final SessionFlowRound r) {
+ final SessionFlowRound.Header h = r.getHeader();
+ if (h.getRound() != round + 1) {
+ return "sessionflow: cannot apply round " + h.getRound() + " to a view at round " + round
+ + ": rounds must fold in order";
+ }
+ if (round > 0 && !nullToEmpty(h.getPrevious()).equals(nullToEmpty(digest))) {
+ return "sessionflow: round " + h.getRound() + " names previous \"" + nullToEmpty(h.getPrevious())
+ + "\", but the view's head is \"" + nullToEmpty(digest) + "\"";
+ }
+ if (round == 0) {
+ conversation = h.getConversation();
+ session = h.getSession();
+ parser = h.getParser();
+ policy = h.getPolicy();
+ } else {
+ if (!nullToEmpty(conversation).equals(nullToEmpty(h.getConversation()))) {
+ return "sessionflow: round " + h.getRound() + " belongs to conversation \"" + nullToEmpty(h.getConversation())
+ + "\", the chain to \"" + nullToEmpty(conversation) + "\"";
+ }
+ if (!nullToEmpty(session).equals(nullToEmpty(h.getSession()))) {
+ return "sessionflow: round " + h.getRound() + " carries session \"" + nullToEmpty(h.getSession())
+ + "\", the chain \"" + nullToEmpty(session) + "\"";
+ }
+ if (!nullToEmpty(parser).equals(nullToEmpty(h.getParser()))) {
+ return "sessionflow: round " + h.getRound() + " was produced by parser \"" + nullToEmpty(h.getParser())
+ + "\", the chain by \"" + nullToEmpty(parser) + "\"; a chain is one interpretation";
+ }
+ if (!nullToEmpty(policy).equals(nullToEmpty(h.getPolicy()))) {
+ return "sessionflow: round " + h.getRound() + " was produced under policy \"" + nullToEmpty(h.getPolicy())
+ + "\", the chain under \"" + nullToEmpty(policy) + "\"";
+ }
+ }
+ for (final SessionFlowRound.Node n : r.getNodes()) {
+ if (n.isTombstone()) {
+ nodes.remove(n.getId());
+ } else {
+ nodes.put(n.getId(), n);
+ }
+ }
+ for (final SessionFlowRound.Relation rel : r.getRelations()) {
+ if (rel.isTombstone()) {
+ relations.remove(rel.getId());
+ } else {
+ relations.put(rel.getId(), rel);
+ }
+ }
+ for (final SessionFlowRound.Unresolved u : r.getUnresolved()) {
+ if (u.isTombstone()) {
+ unresolved.remove(u.getId());
+ } else {
+ unresolved.put(u.getId(), u);
+ }
+ }
+ kids = null;
+ from = null;
+ to = null;
+ round = h.getRound();
+ digest = r.getCommitDigest();
+ throughSeq = h.getThroughSeq();
+ inputDigest = h.getInputDigest();
+ sessionFromTime = h.getSessionFromTime();
+ sessionThroughTime = h.getSessionThroughTime();
+ return null;
+ }
+
+ private static String nullToEmpty(@Nullable final String s) {
+ return s == null ? "" : s;
+ }
+
+ /**
+ * @param id a node id
+ * @return the node's children in record order
+ */
+ public List children(final String id) {
+ index();
+ return kids.getOrDefault(id, Collections.emptyList());
+ }
+
+ public List relationsFrom(final String id) {
+ index();
+ return from.getOrDefault(id, Collections.emptyList());
+ }
+
+ public List relationsTo(final String id) {
+ index();
+ return to.getOrDefault(id, Collections.emptyList());
+ }
+
+ @Nullable
+ public SessionFlowRound.Node node(final String id) {
+ return nodes.get(id);
+ }
+
+ /**
+ * @param kind a node kind
+ * @return the nodes of that kind, in record order
+ */
+ public List nodesOfKind(final String kind) {
+ final List out = new ArrayList<>();
+ for (final SessionFlowRound.Node n : nodes.values()) {
+ if (kind.equals(n.getKind())) {
+ out.add(n);
+ }
+ }
+ return inOrder(out);
+ }
+
+ /**
+ * @return the open unresolved references, by id
+ */
+ public List openUnresolved() {
+ final List out = new ArrayList<>();
+ for (final SessionFlowRound.Unresolved u : unresolved.values()) {
+ if ("open".equals(u.getState())) {
+ out.add(u);
+ }
+ }
+ out.sort(Comparator.comparing(SessionFlowRound.Unresolved::getId));
+ return out;
+ }
+
+ /**
+ * Record order: by the record a node stands on, then by id; a positioned node before one without a
+ * reference.
+ *
+ * @param nodes the nodes
+ * @return a sorted copy
+ */
+ public static List inOrder(final List nodes) {
+ final List out = new ArrayList<>(nodes);
+ out.sort(ConversationFold::compare);
+ return out;
+ }
+
+ public static int compare(final SessionFlowRound.Node a, final SessionFlowRound.Node b) {
+ final Ref ap = a.getRef();
+ final Ref bp = b.getRef();
+ if (ap != null && bp != null) {
+ if (ap.getSeq() != bp.getSeq()) {
+ return Long.compare(ap.getSeq(), bp.getSeq());
+ }
+ if (ap.getRow() != bp.getRow()) {
+ return Long.compare(ap.getRow(), bp.getRow());
+ }
+ if (ap.getBlock() != null && bp.getBlock() != null && !ap.getBlock().equals(bp.getBlock())) {
+ return Integer.compare(ap.getBlock(), bp.getBlock());
+ }
+ } else if (ap != null) {
+ return -1;
+ } else if (bp != null) {
+ return 1;
+ }
+ return a.getId().compareTo(b.getId());
+ }
+
+ private void index() {
+ if (kids != null) {
+ return;
+ }
+ final Map> k = new HashMap<>();
+ for (final SessionFlowRound.Node n : nodes.values()) {
+ if (n.getParent() != null && !n.getParent().isEmpty()) {
+ k.computeIfAbsent(n.getParent(), x -> new ArrayList<>()).add(n);
+ }
+ }
+ for (final List list : k.values()) {
+ list.sort(ConversationFold::compare);
+ }
+ final Map> f = new HashMap<>();
+ final Map> t = new HashMap<>();
+ for (final SessionFlowRound.Relation r : relations.values()) {
+ f.computeIfAbsent(r.getFrom(), x -> new ArrayList<>()).add(r);
+ t.computeIfAbsent(r.getTo(), x -> new ArrayList<>()).add(r);
+ }
+ kids = k;
+ from = f;
+ to = t;
+ }
+
+ private static String first12(final String s) {
+ return s == null ? "" : s.substring(0, Math.min(12, s.length()));
+ }
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/Digests.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/Digests.java
new file mode 100644
index 000000000000..29a70bb1930f
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/Digests.java
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.format;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * The digests the Sessionizer writes, reproduced so a stored file verifies and a chain checks.
+ */
+public final class Digests {
+ private static final char[] HEX = "0123456789abcdef".toCharArray();
+
+ private Digests() {
+ }
+
+ /**
+ * @param bytes the bytes
+ * @return lowercase hex sha256 of the bytes, the file digest on the wire and on every closing line
+ */
+ public static String sha256Hex(final byte[] bytes) {
+ return hex(sha256().digest(bytes));
+ }
+
+ /**
+ * A round's input_digest: the previous round's input digest hashed with the digests of the files
+ * the round newly consumed, sorted, each preceded by a zero byte. Empty previous for round 1.
+ *
+ * @param previous the previous round's input digest, or empty
+ * @param added the digests of the files in the round's window
+ * @return the chained digest
+ */
+ public static String chainInputDigest(final String previous, final List added) {
+ final List sorted = new ArrayList<>(added);
+ Collections.sort(sorted);
+ final MessageDigest md = sha256();
+ md.update(previous.getBytes(StandardCharsets.UTF_8));
+ for (final String digest : sorted) {
+ md.update((byte) 0);
+ md.update(digest.getBytes(StandardCharsets.UTF_8));
+ }
+ return hex(md.digest());
+ }
+
+ /**
+ * @return a fresh sha256 digest
+ */
+ public static MessageDigest sha256() {
+ try {
+ return MessageDigest.getInstance("SHA-256");
+ } catch (final NoSuchAlgorithmException e) {
+ throw new IllegalStateException("SHA-256 is not available", e);
+ }
+ }
+
+ /**
+ * @param digest the digest bytes
+ * @return lowercase hex
+ */
+ public static String hex(final byte[] digest) {
+ final char[] out = new char[digest.length * 2];
+ for (int i = 0; i < digest.length; i++) {
+ final int b = digest[i] & 0xff;
+ out[i * 2] = HEX[b >>> 4];
+ out[i * 2 + 1] = HEX[b & 0x0f];
+ }
+ return new String(out);
+ }
+
+ /**
+ * @param body the file bytes
+ * @return how many lines the body has, counting newline terminators, the way asz.lines counts
+ */
+ public static int countLines(final byte[] body) {
+ int n = 0;
+ for (final byte b : body) {
+ if (b == '\n') {
+ n++;
+ }
+ }
+ return n;
+ }
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/FileNames.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/FileNames.java
new file mode 100644
index 000000000000..d2c795a8f146
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/FileNames.java
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.format;
+
+import java.util.Locale;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import javax.annotation.Nullable;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+
+/**
+ * A landed file's name follows the storage-root layout from its header line, so the name is never stored and is
+ * derived on read, and a name given to a query is parsed back into what it encodes: a session and a seq, or a
+ * round.
+ *
+ *
+ */
+public final class FileNames {
+ private static final Pattern DATA_FILE =
+ Pattern.compile("^(?[^/]+)/(streams|runs)/[^/]+/[a-z]+-[^/-]+-(?\\d{6,})\\.sd$");
+ private static final Pattern ROUND_FILE =
+ Pattern.compile("^_conversations/(?[^/]+)/rounds/r(?\\d{6,})-[0-9a-f]+\\.sf$");
+
+ private FileNames() {
+ }
+
+ /**
+ * @param header the header line of a Session Data file
+ * @return the file's relative path in the storage root
+ */
+ public static String dataFile(final SessionDataFile.Header header) {
+ final String prefix;
+ final String dir;
+ switch (header.getKind() == null ? "" : header.getKind()) {
+ case "transcript":
+ prefix = "transcript";
+ dir = "streams/" + header.getStream();
+ break;
+ case "agent_meta":
+ prefix = "meta";
+ dir = "streams/" + header.getStream();
+ break;
+ case "journal":
+ prefix = "journal";
+ dir = "runs/" + header.getBatch();
+ break;
+ case "workflow_manifest":
+ prefix = "manifest";
+ dir = "runs/" + header.getBatch();
+ break;
+ case "workflow_script":
+ prefix = "script";
+ dir = "runs/" + header.getBatch();
+ break;
+ default:
+ prefix = header.getKind() == null ? "file" : header.getKind();
+ dir = StringUtil.isNotEmpty(header.getStream())
+ ? "streams/" + header.getStream()
+ : "runs/" + header.getBatch();
+ break;
+ }
+ final String stamp = Times.fileStamp(header.getAt());
+ return header.getSession() + "/" + dir + "/" + prefix + "-" + (stamp == null ? "unknown" : stamp)
+ + "-" + String.format(Locale.ROOT, "%06d", header.getSeq()) + ".sd";
+ }
+
+ /**
+ * @param conversation the conversation
+ * @param round the round number
+ * @param commitDigest the round's commit digest
+ * @return the round file's relative path in the storage root
+ */
+ public static String roundFile(final String conversation, final long round, final String commitDigest) {
+ final String digest12 = commitDigest == null ? "" : commitDigest.substring(0, Math.min(12, commitDigest.length()));
+ return "_conversations/" + conversation + "/rounds/r" + String.format(Locale.ROOT, "%06d", round)
+ + "-" + digest12 + ".sf";
+ }
+
+ /**
+ * @param id a file id as returned by the raw-files query
+ * @return what the id names, or null when it is not a landed file or round name
+ */
+ @Nullable
+ public static Parsed parse(final String id) {
+ if (StringUtil.isEmpty(id)) {
+ return null;
+ }
+ final Matcher data = DATA_FILE.matcher(id);
+ if (data.matches()) {
+ return new Parsed(data.group("session"), Long.parseLong(data.group("seq")), null, -1);
+ }
+ final Matcher round = ROUND_FILE.matcher(id);
+ if (round.matches()) {
+ return new Parsed(null, -1, round.group("conversation"), Long.parseLong(round.group("round")));
+ }
+ return null;
+ }
+
+ @Getter
+ @RequiredArgsConstructor
+ public static final class Parsed {
+ private final String session;
+ private final long seq;
+ private final String conversation;
+ private final long round;
+
+ public boolean isDataFile() {
+ return session != null;
+ }
+ }
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/RawJson.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/RawJson.java
new file mode 100644
index 000000000000..2fe9e9d0d61e
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/RawJson.java
@@ -0,0 +1,195 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.format;
+
+import java.util.ArrayList;
+import java.util.List;
+import javax.annotation.Nullable;
+
+/**
+ * Finds the raw text of each part's data value in one record line, so a part's data is rendered
+ * as the Sessionizer wrote it. Go keeps that value as json.RawMessage and prints it verbatim, escapes,
+ * key order and spacing included; a value that was parsed and printed again would differ in all three and break
+ * the document's equality with the Sessionizer's, and clip at another byte.
+ *
+ *
This is a walk over the line's JSON syntax, not a parser: it only needs the start and the end of each
+ * value, and it leaves the decoding to Gson, which has already accepted the line.
+ */
+final class RawJson {
+ private final String text;
+ private int pos;
+
+ private RawJson(final String text) {
+ this.text = text;
+ }
+
+ /**
+ * @param line one record line, a JSON object
+ * @return the raw data text of each element of the line's parts array, in order,
+ * a literal null included, null for a part without one; empty when the line has no parts or is not the JSON
+ * expected
+ */
+ static List partData(final String line) {
+ try {
+ return new RawJson(line).parts();
+ } catch (final IllegalStateException | StringIndexOutOfBoundsException e) {
+ return new ArrayList<>();
+ }
+ }
+
+ private List parts() {
+ final List out = new ArrayList<>();
+ skipSpace();
+ expect('{');
+ while (true) {
+ skipSpace();
+ if (peek() == '}') {
+ return out;
+ }
+ final String key = string();
+ skipSpace();
+ expect(':');
+ skipSpace();
+ if ("parts".equals(key) && peek() == '[') {
+ pos++;
+ while (true) {
+ skipSpace();
+ if (peek() == ']') {
+ pos++;
+ break;
+ }
+ out.add(partDataOf());
+ skipSpace();
+ if (peek() == ',') {
+ pos++;
+ }
+ }
+ } else {
+ skipValue();
+ }
+ skipSpace();
+ if (peek() == ',') {
+ pos++;
+ }
+ }
+ }
+
+ /**
+ * @return the raw data of the part object at the cursor, or null; the cursor ends after the object
+ */
+ @Nullable
+ private String partDataOf() {
+ String data = null;
+ expect('{');
+ while (true) {
+ skipSpace();
+ if (peek() == '}') {
+ pos++;
+ return data;
+ }
+ final String key = string();
+ skipSpace();
+ expect(':');
+ skipSpace();
+ final int start = pos;
+ skipValue();
+ if ("data".equals(key)) {
+ data = text.substring(start, pos);
+ }
+ skipSpace();
+ if (peek() == ',') {
+ pos++;
+ }
+ }
+ }
+
+ private void skipValue() {
+ final char c = peek();
+ if (c == '"') {
+ string();
+ } else if (c == '{' || c == '[') {
+ final char close = c == '{' ? '}' : ']';
+ pos++;
+ while (true) {
+ skipSpace();
+ if (peek() == close) {
+ pos++;
+ return;
+ }
+ if (c == '{') {
+ string();
+ skipSpace();
+ expect(':');
+ skipSpace();
+ }
+ skipValue();
+ skipSpace();
+ if (peek() == ',') {
+ pos++;
+ }
+ }
+ } else {
+ while (pos < text.length() && ",}] \t\r\n".indexOf(text.charAt(pos)) < 0) {
+ pos++;
+ }
+ }
+ }
+
+ /**
+ * @return the decoded text of the string at the cursor, escapes resolved only as far as a key needs them
+ */
+ private String string() {
+ expect('"');
+ final StringBuilder out = new StringBuilder();
+ while (true) {
+ final char c = text.charAt(pos++);
+ if (c == '"') {
+ return out.toString();
+ }
+ if (c == '\\') {
+ final char e = text.charAt(pos++);
+ if (e == 'u') {
+ out.append((char) Integer.parseInt(text.substring(pos, pos + 4), 16));
+ pos += 4;
+ } else {
+ out.append(e);
+ }
+ } else {
+ out.append(c);
+ }
+ }
+ }
+
+ private void skipSpace() {
+ while (pos < text.length() && " \t\r\n".indexOf(text.charAt(pos)) >= 0) {
+ pos++;
+ }
+ }
+
+ private char peek() {
+ return text.charAt(pos);
+ }
+
+ private void expect(final char c) {
+ if (text.charAt(pos) != c) {
+ throw new IllegalStateException("expected " + c + " at " + pos);
+ }
+ pos++;
+ }
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/Ref.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/Ref.java
new file mode 100644
index 000000000000..df8676cf47e2
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/Ref.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.format;
+
+import com.google.gson.JsonObject;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import javax.annotation.Nullable;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+/**
+ * A reference into the landed records: the file by seq, the line by row, and, when the
+ * node stands on one part of a record, the part by block.
+ */
+@Getter
+@RequiredArgsConstructor
+@EqualsAndHashCode
+public final class Ref {
+ private final long seq;
+ private final long row;
+ @Nullable
+ private final Integer block;
+
+ @Nullable
+ public static Ref of(@Nullable final JsonObject json) {
+ if (json == null) {
+ return null;
+ }
+ return new Ref(
+ json.get("seq").getAsLong(),
+ json.get("row").getAsLong(),
+ json.has("block") && !json.get("block").isJsonNull() ? json.get("block").getAsInt() : null
+ );
+ }
+
+ /**
+ * @return the reference as a view map, in the order the format page lists the keys
+ */
+ public Map toMap() {
+ final Map out = new LinkedHashMap<>();
+ out.put("seq", seq);
+ out.put("row", row);
+ if (block != null) {
+ out.put("block", block);
+ }
+ return out;
+ }
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/SessionDataFile.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/SessionDataFile.java
new file mode 100644
index 000000000000..41a4a68f8db2
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/SessionDataFile.java
@@ -0,0 +1,332 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.format;
+
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import javax.annotation.Nullable;
+import lombok.Getter;
+
+/**
+ * One Session Data (.sd) file, decoded from its stored bytes: the header line, the records one per
+ * line in source order, and the closing line with the count and the digest.
+ *
+ *
Row numbers are line numbers: the header is row 0, the first record is row 1, which is how a Session Flow
+ * reference addresses a record.
+ */
+@Getter
+public final class SessionDataFile {
+ private final Header header;
+ private final List records;
+ private final int declaredRecords;
+ private final String declaredDigest;
+ private final int lines;
+ private final int bytes;
+ /** sha256 of the whole file, the digest on the wire and the one a round's input digest chains. */
+ private final String fileDigest;
+ /** The earliest and the latest record time in the file, in milliseconds; 0 when no record carries a time. */
+ private final long fromTime;
+ private final long throughTime;
+
+ private SessionDataFile(final Header header, final List records, final int declaredRecords,
+ final String declaredDigest, final int lines, final int bytes,
+ final String fileDigest, final long fromTime, final long throughTime) {
+ this.header = header;
+ this.records = records;
+ this.declaredRecords = declaredRecords;
+ this.declaredDigest = declaredDigest;
+ this.lines = lines;
+ this.bytes = bytes;
+ this.fileDigest = fileDigest;
+ this.fromTime = fromTime;
+ this.throughTime = throughTime;
+ }
+
+ /**
+ * @param body the file bytes as stored
+ * @return the decoded file
+ * @throws IllegalArgumentException when the first line is not a Session Data header
+ */
+ public static SessionDataFile parse(final byte[] body) {
+ final String text = new String(body, StandardCharsets.UTF_8);
+ final String[] rawLines = text.split("\n", -1);
+ int lineCount = rawLines.length;
+ if (lineCount > 0 && rawLines[lineCount - 1].isEmpty()) {
+ lineCount--;
+ }
+ if (lineCount < 1) {
+ throw new IllegalArgumentException("empty Session Data file");
+ }
+ final JsonObject headerJson = JsonParser.parseString(rawLines[0]).getAsJsonObject();
+ if (!headerJson.has("h")) {
+ throw new IllegalArgumentException("the first line is not a Session Data header");
+ }
+ final Header header = new Header(headerJson);
+ final List records = new ArrayList<>(Math.max(0, lineCount - 2));
+ int declaredRecords = -1;
+ String declaredDigest = null;
+ long from = 0;
+ long through = 0;
+ // as the Sessionizer's reader: a header it would refuse yields no records, and an empty or undecodable
+ // line ends the records there, so a later row is never read and the rows stay contiguous
+ for (int i = 1; header.isValid() && i < lineCount; i++) {
+ final String line = rawLines[i];
+ if (line.isEmpty()) {
+ break;
+ }
+ final JsonObject json;
+ try {
+ json = JsonParser.parseString(line).getAsJsonObject();
+ } catch (final RuntimeException e) {
+ break;
+ }
+ if (i == lineCount - 1 && "end".equals(string(json, "t"))) {
+ declaredRecords = json.has("records") ? json.get("records").getAsInt() : -1;
+ declaredDigest = string(json, "digest");
+ break;
+ }
+ final Record record = new Record(i, line, json);
+ records.add(record);
+ final long time = record.getTime();
+ if (time != 0) {
+ if (from == 0 || time < from) {
+ from = time;
+ }
+ if (time > through) {
+ through = time;
+ }
+ }
+ }
+ return new SessionDataFile(
+ header, Collections.unmodifiableList(records), declaredRecords, declaredDigest,
+ Digests.countLines(body), body.length, Digests.sha256Hex(body), from, through);
+ }
+
+ /**
+ * @param row the line number, the header being row 0
+ * @return the record on that line, or null when there is none
+ */
+ @Nullable
+ public Record record(final long row) {
+ final int index = (int) row - 1;
+ if (index < 0 || index >= records.size()) {
+ return null;
+ }
+ return records.get(index);
+ }
+
+ @Nullable
+ static String string(final JsonObject json, final String key) {
+ final JsonElement element = json.get(key);
+ return element == null || element.isJsonNull() ? null : element.getAsString();
+ }
+
+ static long longOf(final JsonObject json, final String key) {
+ final JsonElement element = json.get(key);
+ return element == null || element.isJsonNull() ? 0L : element.getAsLong();
+ }
+
+ /**
+ * The header line: what the file is and where its records came from.
+ */
+ @Getter
+ public static final class Header {
+ private final JsonObject json;
+ private final String schema;
+ private final long seq;
+ private final String at;
+ private final String kind;
+ private final String adapter;
+ private final String dialect;
+ private final String src;
+ private final String session;
+ private final String stream;
+ private final String batch;
+
+ Header(final JsonObject json) {
+ this.json = json;
+ this.schema = string(json, "schema");
+ this.seq = longOf(json, "seq");
+ this.at = string(json, "at");
+ this.kind = string(json, "kind");
+ this.adapter = string(json, "adapter");
+ this.dialect = string(json, "dialect");
+ this.src = string(json, "src");
+ this.session = string(json, "session");
+ this.stream = string(json, "stream");
+ this.batch = string(json, "batch");
+ }
+
+ /**
+ * @return whether the Sessionizer's reader would open the file: the envelope version, the schema, the
+ * kind, the session, the source and the dialect are all there
+ */
+ public boolean isValid() {
+ return longOf(json, "h") == 1 && "sd/1".equals(schema) && kind != null && !kind.isEmpty()
+ && session != null && !session.isEmpty() && src != null && !src.isEmpty()
+ && dialect != null && !dialect.isEmpty();
+ }
+ }
+
+ /**
+ * One record, kept as its JSON so any field the viewer wants is one lookup away.
+ */
+ @Getter
+ public static final class Record {
+ private final int row;
+ private final JsonObject json;
+ private final String id;
+ /** When the runtime wrote it, in milliseconds; 0 when the record carries no time. */
+ private final long time;
+ /** The same moment in nanoseconds, the precision the Sessionizer computes intervals with. */
+ private final long timeNanos;
+ private final List parts;
+
+ Record(final int row, final String line, final JsonObject json) {
+ this.row = row;
+ this.json = json;
+ this.id = string(json, "id");
+ this.timeNanos = Times.nanos(string(json, "time"));
+ this.time = Math.floorDiv(timeNanos, 1_000_000L);
+ final List list = new ArrayList<>();
+ if (json.has("parts") && json.get("parts").isJsonArray()) {
+ // the raw data text of each part, as the Sessionizer wrote it; see RawJson
+ final List raw = RawJson.partData(line);
+ int i = 0;
+ for (final JsonElement e : json.getAsJsonArray("parts")) {
+ list.add(new Part(e.getAsJsonObject(), i < raw.size() ? raw.get(i) : null));
+ i++;
+ }
+ }
+ this.parts = Collections.unmodifiableList(list);
+ }
+
+ /**
+ * @return the record's readable text: every text part joined by a newline, as the
+ * Sessionizer's Record.Text() returns it
+ */
+ public String text() {
+ final StringBuilder out = new StringBuilder();
+ for (final Part p : parts) {
+ if ("text".equals(p.getKind()) && p.getText() != null && !p.getText().isEmpty()) {
+ if (out.length() > 0) {
+ out.append('\n');
+ }
+ out.append(p.getText());
+ }
+ }
+ return out.toString();
+ }
+
+ /**
+ * @return the record's flags, empty when none
+ */
+ public List flags() {
+ return strings("flags");
+ }
+
+ /**
+ * @return the record's usage object, or null
+ */
+ @Nullable
+ public JsonObject usage() {
+ final JsonElement u = json.get("usage");
+ return u != null && u.isJsonObject() ? u.getAsJsonObject() : null;
+ }
+
+ /**
+ * @return the record's dropped list, or null
+ */
+ @Nullable
+ public JsonArray dropped() {
+ final JsonElement d = json.get("dropped");
+ return d != null && d.isJsonArray() ? d.getAsJsonArray() : null;
+ }
+
+ /**
+ * @return the record's child, or null
+ */
+ @Nullable
+ public String child() {
+ return string(json, "child");
+ }
+
+ private List strings(final String key) {
+ final JsonElement e = json.get(key);
+ if (e == null || !e.isJsonArray()) {
+ return Collections.emptyList();
+ }
+ final List out = new ArrayList<>();
+ for (final JsonElement x : e.getAsJsonArray()) {
+ out.add(x.getAsString());
+ }
+ return out;
+ }
+ }
+
+ /**
+ * One piece of a record: readable text, a call, a result, or data kept whole.
+ */
+ @Getter
+ public static final class Part {
+ private final JsonObject json;
+ private final String kind;
+ private final String text;
+ private final String name;
+ private final String id;
+ private final String of;
+ private final String state;
+ private final int bytes;
+ private final Boolean failed;
+ private final String rawData;
+
+ Part(final JsonObject json, @Nullable final String rawData) {
+ this.json = json;
+ this.rawData = rawData;
+ this.kind = string(json, "k");
+ this.text = string(json, "text");
+ this.name = string(json, "name");
+ this.id = string(json, "id");
+ this.of = string(json, "of");
+ this.state = string(json, "state");
+ this.bytes = (int) longOf(json, "bytes");
+ final JsonElement f = json.get("failed");
+ this.failed = f == null || f.isJsonNull() ? null : f.getAsBoolean();
+ }
+
+ /**
+ * @return the part's data as the Sessionizer wrote it, or null when it has none
+ */
+ @Nullable
+ public String data() {
+ if (rawData != null) {
+ return rawData;
+ }
+ final JsonElement d = json.get("data");
+ // a literal null is kept as the text "null": Go reads it into a raw message and prints it
+ return d == null ? null : d.toString();
+ }
+ }
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/SessionFlowRound.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/SessionFlowRound.java
new file mode 100644
index 000000000000..be245619c066
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/SessionFlowRound.java
@@ -0,0 +1,419 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.format;
+
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
+import lombok.Getter;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+
+/**
+ * One Session Flow (.sf) round, decoded from its stored bytes: the header frame, the node, relation
+ * and unresolved frames, and the commit frame whose digest covers every line before it.
+ */
+@Getter
+public final class SessionFlowRound {
+ private final Header header;
+ private final List nodes;
+ private final List relations;
+ private final List unresolved;
+ /** The digest the commit frame claims. */
+ private final String commitDigest;
+ /** The digest computed over every line before the commit frame; equal to {@link #commitDigest} when intact. */
+ private final String computedDigest;
+ private final int lines;
+ private final int bytes;
+
+ private SessionFlowRound(final Header header, final List nodes, final List relations,
+ final List unresolved, final String commitDigest,
+ final String computedDigest, final int lines, final int bytes) {
+ this.header = header;
+ this.nodes = nodes;
+ this.relations = relations;
+ this.unresolved = unresolved;
+ this.commitDigest = commitDigest;
+ this.computedDigest = computedDigest;
+ this.lines = lines;
+ this.bytes = bytes;
+ }
+
+ /**
+ * @param body the round bytes as stored
+ * @return the decoded round
+ * @throws IllegalArgumentException when the first frame is not a header or the round has no commit frame
+ */
+ public static SessionFlowRound parse(final byte[] body) {
+ final String text = new String(body, StandardCharsets.UTF_8);
+ final String[] rawLines = text.split("\n", -1);
+ final MessageDigest hashed = Digests.sha256();
+ Header header = null;
+ final List nodes = new ArrayList<>();
+ final List relations = new ArrayList<>();
+ final List unresolved = new ArrayList<>();
+ String commitDigest = null;
+ JsonObject commitCounts = null;
+ boolean sawCommit = false;
+ // the checks the Sessionizer's own reader makes; a round that fails one is not a round
+ final Map ids = new HashMap<>();
+ int lineNo = 0;
+ for (final String line : rawLines) {
+ lineNo++;
+ if (line.isEmpty()) {
+ continue;
+ }
+ final JsonObject json = JsonParser.parseString(line).getAsJsonObject();
+ final String t = SessionDataFile.string(json, "t");
+ if (t == null) {
+ throw new IllegalArgumentException("a Session Flow frame without a type");
+ }
+ if (sawCommit) {
+ throw new IllegalArgumentException("content after the commit frame");
+ }
+ switch (t) {
+ case "header":
+ if (header != null) {
+ throw new IllegalArgumentException("a second header");
+ }
+ header = new Header(json);
+ header.validate();
+ break;
+ case "node": {
+ final Node n = new Node(json);
+ claim(ids, lineNo, "node", n.getId());
+ checkRevision(lineNo, n.getRevision(), header);
+ checkRefs(lineNo, n.getRef(), n.getRefs(), header);
+ nodes.add(n);
+ break;
+ }
+ case "relation": {
+ final Relation rel = new Relation(json);
+ claim(ids, lineNo, "relation", rel.getId());
+ checkRevision(lineNo, rel.getRevision(), header);
+ if (!rel.isTombstone() && (StringUtil.isEmpty(rel.getFrom()) || StringUtil.isEmpty(rel.getTo())
+ || StringUtil.isEmpty(rel.getType()))) {
+ throw new IllegalArgumentException(
+ "line " + lineNo + ": relation " + rel.getId() + " is missing an endpoint or a type");
+ }
+ checkRefs(lineNo, null, rel.getEvidence(), header);
+ relations.add(rel);
+ break;
+ }
+ case "unresolved": {
+ final Unresolved u = new Unresolved(json);
+ claim(ids, lineNo, "unresolved", u.getId());
+ checkRevision(lineNo, u.getRevision(), header);
+ if (!u.isTombstone() && !"open".equals(u.getState()) && !"resolved".equals(u.getState())
+ && !"terminal".equals(u.getState())) {
+ throw new IllegalArgumentException(
+ "line " + lineNo + ": unresolved entry " + u.getId() + " has state " + u.getState());
+ }
+ unresolved.add(u);
+ break;
+ }
+ case "commit":
+ commitDigest = SessionDataFile.string(json, "digest");
+ commitCounts = json.has("counts") && json.get("counts").isJsonObject()
+ ? json.getAsJsonObject("counts") : null;
+ sawCommit = true;
+ continue;
+ default:
+ throw new IllegalArgumentException("unknown frame type " + t);
+ }
+ if (header == null) {
+ throw new IllegalArgumentException("the first frame is not a header");
+ }
+ hashed.update(line.getBytes(StandardCharsets.UTF_8));
+ hashed.update((byte) '\n');
+ }
+ if (header == null) {
+ throw new IllegalArgumentException("the round has no header");
+ }
+ if (!sawCommit) {
+ throw new IllegalArgumentException("the round has no commit frame; it is truncated");
+ }
+ // the commit frame is outside the digest, so its counts are what catches a tampered commit
+ final long claimedNodes = commitCounts == null ? 0 : SessionDataFile.longOf(commitCounts, "nodes");
+ final long claimedRelations = commitCounts == null ? 0 : SessionDataFile.longOf(commitCounts, "relations");
+ final long claimedUnresolved = commitCounts == null ? 0 : SessionDataFile.longOf(commitCounts, "unresolved");
+ if (claimedNodes != nodes.size() || claimedRelations != relations.size()
+ || claimedUnresolved != unresolved.size()) {
+ throw new IllegalArgumentException("counts mismatch: read nodes " + nodes.size() + " relations "
+ + relations.size() + " unresolved " + unresolved.size()
+ + ", round claims nodes " + claimedNodes + " relations "
+ + claimedRelations + " unresolved " + claimedUnresolved);
+ }
+ return new SessionFlowRound(
+ header, Collections.unmodifiableList(nodes), Collections.unmodifiableList(relations),
+ Collections.unmodifiableList(unresolved), commitDigest, Digests.hex(hashed.digest()),
+ Digests.countLines(body), body.length);
+ }
+
+ private static void claim(final Map ids, final int line, final String kind, final String id) {
+ if (StringUtil.isEmpty(id)) {
+ throw new IllegalArgumentException("line " + line + ": " + kind + " frame has no id");
+ }
+ final String prev = ids.put(id, kind);
+ if (prev != null) {
+ throw new IllegalArgumentException(
+ "line " + line + ": id " + id + " appears twice in one round (as " + prev + " and " + kind + ")");
+ }
+ }
+
+ /**
+ * An entity names the round that produced it; one that names another was not produced by this round.
+ */
+ private static void checkRevision(final int line, final long revision, final Header header) {
+ if (revision != header.getRound()) {
+ throw new IllegalArgumentException("line " + line + ": revision " + revision + " in round " + header.getRound());
+ }
+ }
+
+ /**
+ * A reference past the range the header declares it read describes evidence the round did not claim to
+ * have seen, and its input digest does not cover it.
+ */
+ private static void checkRefs(final int line, @Nullable final Ref one, final List many, final Header header) {
+ final List all = new ArrayList<>();
+ if (one != null) {
+ all.add(one);
+ }
+ all.addAll(many);
+ for (final Ref r : all) {
+ if (r.getSeq() == 0 && r.getRow() == 0) {
+ throw new IllegalArgumentException("line " + line + ": a reference to seq 0 row 0 is not a position");
+ }
+ if (r.getSeq() > header.getThroughSeq()) {
+ throw new IllegalArgumentException("line " + line + ": reference to landed sequence " + r.getSeq()
+ + ", past the round's declared " + header.getThroughSeq());
+ }
+ }
+ }
+
+ public boolean isIntact() {
+ return commitDigest != null && commitDigest.equals(computedDigest);
+ }
+
+ /**
+ * The header frame: which chain the round belongs to, where it sits in it, and what it consumed.
+ */
+ @Getter
+ public static final class Header {
+ private final String schema;
+ private final String conversation;
+ private final String session;
+ private final long round;
+ private final String previous;
+ private final long fromSeq;
+ private final long throughSeq;
+ private final String inputDigest;
+ private final String parser;
+ private final String policy;
+ private final String fromTime;
+ private final String throughTime;
+ private final String sessionFromTime;
+ private final String sessionThroughTime;
+
+ Header(final JsonObject json) {
+ this.schema = SessionDataFile.string(json, "schema");
+ this.conversation = SessionDataFile.string(json, "conversation");
+ this.session = SessionDataFile.string(json, "session");
+ this.round = SessionDataFile.longOf(json, "round");
+ this.previous = SessionDataFile.string(json, "previous");
+ this.fromSeq = SessionDataFile.longOf(json, "from_seq");
+ this.throughSeq = SessionDataFile.longOf(json, "through_seq");
+ this.inputDigest = SessionDataFile.string(json, "input_digest");
+ this.parser = SessionDataFile.string(json, "parser");
+ this.policy = SessionDataFile.string(json, "policy");
+ this.fromTime = SessionDataFile.string(json, "from_time");
+ this.throughTime = SessionDataFile.string(json, "through_time");
+ this.sessionFromTime = SessionDataFile.string(json, "session_from_time");
+ this.sessionThroughTime = SessionDataFile.string(json, "session_through_time");
+ }
+
+ /**
+ * The header the Sessionizer's own reader would refuse: it cannot be acted on.
+ */
+ void validate() {
+ if (!"sf/1".equals(schema)) {
+ throw new IllegalArgumentException("unsupported schema " + schema + ", want sf/1");
+ }
+ if (StringUtil.isEmpty(conversation)) {
+ throw new IllegalArgumentException("header missing conversation");
+ }
+ if (StringUtil.isEmpty(session)) {
+ throw new IllegalArgumentException("header missing session");
+ }
+ if (round == 0) {
+ throw new IllegalArgumentException("round must count from 1");
+ }
+ if (round > 1 && StringUtil.isEmpty(previous)) {
+ throw new IllegalArgumentException("round " + round + " has no previous digest; the chain would be unverifiable");
+ }
+ if (round == 1 && StringUtil.isNotEmpty(previous)) {
+ throw new IllegalArgumentException("round 1 must not name a previous digest");
+ }
+ if (StringUtil.isEmpty(parser)) {
+ throw new IllegalArgumentException("header missing parser version");
+ }
+ if (StringUtil.isEmpty(policy)) {
+ throw new IllegalArgumentException("header missing policy version");
+ }
+ if (StringUtil.isEmpty(inputDigest)) {
+ throw new IllegalArgumentException("header missing input digest");
+ }
+ if (fromSeq == 0) {
+ throw new IllegalArgumentException("landed sequences count from 1, so from_seq must not be 0");
+ }
+ if (throughSeq < fromSeq - 1) {
+ throw new IllegalArgumentException("round " + round + " consumes sequences " + fromSeq + ".." + throughSeq + ", which is not a range");
+ }
+ }
+ }
+
+ /**
+ * What every entity frame has: an id, the revision that produced it, and whether it is a tombstone.
+ */
+ @Getter
+ public abstract static class Entity {
+ private final String id;
+ private final long revision;
+ private final boolean tombstone;
+
+ Entity(final JsonObject json) {
+ this.id = SessionDataFile.string(json, "id");
+ this.revision = SessionDataFile.longOf(json, "revision");
+ final JsonElement t = json.get("tombstone");
+ this.tombstone = t != null && !t.isJsonNull() && t.getAsBoolean();
+ }
+ }
+
+ @Getter
+ public static final class Node extends Entity {
+ private final String kind;
+ private final String parent;
+ private final String stream;
+ @Nullable
+ private final Ref ref;
+ private final List refs;
+ @Nullable
+ /** The attrs as written, any JSON value, for rendering; null when the frame has none. */
+ private final JsonElement rawAttrs;
+ /** The attrs when they are an object, for lookups. */
+ private final JsonObject attrs;
+
+ Node(final JsonObject json) {
+ super(json);
+ this.kind = SessionDataFile.string(json, "kind");
+ this.parent = SessionDataFile.string(json, "parent");
+ this.stream = SessionDataFile.string(json, "stream");
+ this.ref = json.has("ref") && json.get("ref").isJsonObject() ? Ref.of(json.getAsJsonObject("ref")) : null;
+ final List list = new ArrayList<>();
+ if (json.has("refs") && json.get("refs").isJsonArray()) {
+ for (final JsonElement e : json.getAsJsonArray("refs")) {
+ list.add(Ref.of(e.getAsJsonObject()));
+ }
+ }
+ this.refs = Collections.unmodifiableList(list);
+ this.rawAttrs = json.get("attrs");
+ this.attrs = rawAttrs != null && rawAttrs.isJsonObject() ? rawAttrs.getAsJsonObject() : null;
+ }
+
+ @Nullable
+ /**
+ * @return the attr when it is a string, as the Sessionizer's attrString; null for a number,
+ * an object or nothing
+ */
+ public String attr(final String key) {
+ if (attrs == null) {
+ return null;
+ }
+ final JsonElement e = attrs.get(key);
+ return e != null && e.isJsonPrimitive() && e.getAsJsonPrimitive().isString() ? e.getAsString() : null;
+ }
+
+ public double attrNumber(final String key) {
+ if (attrs == null) {
+ return 0;
+ }
+ final JsonElement e = attrs.get(key);
+ return e == null || e.isJsonNull() || !e.isJsonPrimitive() || !e.getAsJsonPrimitive().isNumber()
+ ? 0 : e.getAsDouble();
+ }
+
+ public boolean attrBool(final String key) {
+ if (attrs == null) {
+ return false;
+ }
+ final JsonElement e = attrs.get(key);
+ return e != null && !e.isJsonNull() && e.isJsonPrimitive() && e.getAsJsonPrimitive().isBoolean()
+ && e.getAsBoolean();
+ }
+ }
+
+ @Getter
+ public static final class Relation extends Entity {
+ private final String type;
+ private final String from;
+ private final String to;
+ private final String quality;
+ private final String via;
+ private final List evidence;
+
+ Relation(final JsonObject json) {
+ super(json);
+ this.type = SessionDataFile.string(json, "type");
+ this.from = SessionDataFile.string(json, "from");
+ this.to = SessionDataFile.string(json, "to");
+ this.quality = SessionDataFile.string(json, "quality");
+ this.via = SessionDataFile.string(json, "via");
+ final List list = new ArrayList<>();
+ if (json.has("evidence") && json.get("evidence").isJsonArray()) {
+ for (final JsonElement e : json.getAsJsonArray("evidence")) {
+ list.add(Ref.of(e.getAsJsonObject()));
+ }
+ }
+ this.evidence = Collections.unmodifiableList(list);
+ }
+ }
+
+ @Getter
+ public static final class Unresolved extends Entity {
+ private final String kind;
+ private final String ref;
+ private final String reason;
+ private final String state;
+
+ Unresolved(final JsonObject json) {
+ super(json);
+ this.kind = SessionDataFile.string(json, "kind");
+ this.ref = SessionDataFile.string(json, "ref");
+ this.reason = SessionDataFile.string(json, "reason");
+ this.state = SessionDataFile.string(json, "state");
+ }
+ }
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/Times.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/Times.java
new file mode 100644
index 000000000000..b057f68c669d
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/format/Times.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.format;
+
+import java.time.Instant;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
+import javax.annotation.Nullable;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+
+/**
+ * Session Data and Session Flow carry RFC 3339 strings, because their bytes are digested. A view carries unix
+ * milliseconds, because it is read and never digested.
+ */
+public final class Times {
+ /**
+ * The stamp in a landed file's name: 20260904T033423.840913000Z, from the header's at.
+ */
+ private static final DateTimeFormatter FILE_STAMP =
+ DateTimeFormatter.ofPattern("yyyyMMdd'T'HHmmss.nnnnnnnnn'Z'").withZone(ZoneOffset.UTC);
+
+ private Times() {
+ }
+
+ /**
+ * @param rfc3339 a time as the runtime wrote it, or null or empty
+ * @return unix milliseconds, or 0 when absent or unparseable
+ */
+ public static long millis(final String rfc3339) {
+ final Instant t = instant(rfc3339);
+ return t == null ? 0L : t.toEpochMilli();
+ }
+
+ /**
+ * @param rfc3339 a time as the Sessionizer writes it, with a zone offset or Z
+ * @return the time in nanoseconds since the epoch, the precision the Sessionizer computes with, or 0
+ */
+ public static long nanos(final String rfc3339) {
+ final Instant t = instant(rfc3339);
+ return t == null ? 0L : t.getEpochSecond() * 1_000_000_000L + t.getNano();
+ }
+
+ /**
+ * RFC 3339 as Go's time.RFC3339Nano reads it: an offset is accepted, not only Z.
+ */
+ @Nullable
+ private static Instant instant(@Nullable final String rfc3339) {
+ if (StringUtil.isEmpty(rfc3339)) {
+ return null;
+ }
+ try {
+ return OffsetDateTime.parse(rfc3339, DateTimeFormatter.ISO_OFFSET_DATE_TIME).toInstant();
+ } catch (final DateTimeParseException e) {
+ return null;
+ }
+ }
+
+ /**
+ * @param rfc3339 a header's at
+ * @return the stamp used in the landed file's name, or null when absent or unparseable
+ */
+ public static String fileStamp(final String rfc3339) {
+ if (StringUtil.isEmpty(rfc3339)) {
+ return null;
+ }
+ final Instant t = instant(rfc3339);
+ return t == null ? null : FILE_STAMP.format(t);
+ }
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/ingest/ConversationFileBuilder.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/ingest/ConversationFileBuilder.java
new file mode 100644
index 000000000000..2d6145ffe3a4
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/ingest/ConversationFileBuilder.java
@@ -0,0 +1,315 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.ingest;
+
+import com.google.gson.JsonObject;
+import java.nio.charset.StandardCharsets;
+import javax.annotation.Nullable;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.apm.network.common.v3.KeyStringValuePair;
+import org.apache.skywalking.apm.network.logging.v3.LogData;
+import org.apache.skywalking.oap.server.ai.agent.conversation.format.Digests;
+import org.apache.skywalking.oap.server.ai.agent.conversation.format.Times;
+import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.analysis.IDManager;
+import org.apache.skywalking.oap.server.core.analysis.Layer;
+import org.apache.skywalking.oap.server.core.analysis.TimeBucket;
+import org.apache.skywalking.oap.server.core.analysis.manual.aiagent.AIAgentSessionDataRecord;
+import org.apache.skywalking.oap.server.core.analysis.manual.aiagent.AIAgentSessionFlowRecord;
+import org.apache.skywalking.oap.server.core.analysis.record.Record;
+import org.apache.skywalking.oap.server.core.analysis.worker.RecordStreamProcessor;
+import org.apache.skywalking.oap.server.core.config.NamingControl;
+import org.apache.skywalking.oap.server.core.source.LALOutputBuilder;
+import org.apache.skywalking.oap.server.core.source.LogMetadata;
+import org.apache.skywalking.oap.server.core.source.SourceReceiver;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+import org.apache.skywalking.oap.server.telemetry.api.CounterMetrics;
+
+/**
+ * The LAL output builder named by outputType: ConversationFile in lal/ai-agent.yaml.
+ *
+ *
The rule's extractor sets the fields below from the record's asz.* attributes, then
+ * {@link #init} takes the body and the service, instance and timestamp the OTLP handler already resolved, and
+ * {@link #complete} verifies the file and dispatches one row to the Session Data or the Session Flow table. One
+ * rule, one builder: the branch on format lives here.
+ *
+ *
The service and instance traffic is registered by the log analyzer's own traffic listener for every record
+ * the rule keeps, so this builder registers none.
+ */
+@Slf4j
+public class ConversationFileBuilder implements LALOutputBuilder {
+ public static final String NAME = "ConversationFile";
+ private static final String FORMAT_SD = "sd";
+ private static final String FORMAT_SF = "sf";
+
+ private static volatile CounterMetrics ACCEPTED_DATA;
+ private static volatile CounterMetrics ACCEPTED_FLOW;
+ private static volatile CounterMetrics REJECTED_DIGEST;
+ private static volatile CounterMetrics REJECTED_LINES;
+ private static volatile CounterMetrics REJECTED_ATTRIBUTES;
+ private static volatile NamingControl NAMING_CONTROL;
+
+ // every record
+ @Getter
+ @Setter
+ private String format;
+ @Getter
+ @Setter
+ private String digest;
+ /** -1 until the extractor sets it; the LAL compiler generates primitive setter calls. */
+ @Getter
+ @Setter
+ private long lines = -1;
+ @Getter
+ @Setter
+ private String throughTime;
+ // sd
+ @Getter
+ @Setter
+ private String session;
+ @Getter
+ @Setter
+ private long seq = -1;
+ // sf
+ @Getter
+ @Setter
+ private String conversation;
+ @Getter
+ @Setter
+ private long round = -1;
+ @Getter
+ @Setter
+ private String sessionFromTime;
+ @Getter
+ @Setter
+ private String sessionThroughTime;
+ @Getter
+ @Setter
+ private String title;
+ @Getter
+ @Setter
+ private long talks;
+ @Getter
+ @Setter
+ private long steps;
+ @Getter
+ @Setter
+ private long streams;
+ @Getter
+ @Setter
+ private long segments;
+ @Getter
+ @Setter
+ private long unresolved;
+
+ // from the handler, through init
+ private String serviceName;
+ private String instanceName;
+ private Layer layer = Layer.AI_AGENT;
+ private long timestamp;
+ private byte[] body;
+ private String fileName;
+
+ /**
+ * Set once by the module provider; the builder itself is created per record by the LAL runtime.
+ *
+ * @param dataAccepted files stored to the Session Data table
+ * @param flowAccepted rounds stored to the Session Flow table
+ * @param digest files rejected because the body's digest is not the declared one
+ * @param lines files rejected because the body's line count is not the declared one
+ * @param attributes files rejected because a required attribute is missing
+ */
+ public static void setMetrics(final CounterMetrics dataAccepted,
+ final CounterMetrics flowAccepted,
+ final CounterMetrics digest,
+ final CounterMetrics lines,
+ final CounterMetrics attributes) {
+ ACCEPTED_DATA = dataAccepted;
+ ACCEPTED_FLOW = flowAccepted;
+ REJECTED_DIGEST = digest;
+ REJECTED_LINES = lines;
+ REJECTED_ATTRIBUTES = attributes;
+ }
+
+ /**
+ * Normally resolved from the core module on the first record; a test sets it directly.
+ *
+ * @param control the naming control
+ */
+ public static void setNamingControl(final NamingControl control) {
+ NAMING_CONTROL = control;
+ }
+
+ @Override
+ public String name() {
+ return NAME;
+ }
+
+ @Override
+ public void init(final LogMetadata metadata, final Object input, final ModuleManager moduleManager) {
+ if (NAMING_CONTROL == null && moduleManager != null) {
+ NAMING_CONTROL = moduleManager.find(CoreModule.NAME).provider().getService(NamingControl.class);
+ }
+ serviceName = metadata.getService();
+ instanceName = metadata.getServiceInstance();
+ if (StringUtil.isNotEmpty(metadata.getLayer())) {
+ layer = Layer.valueOf(metadata.getLayer());
+ }
+ timestamp = metadata.getTimestamp();
+ if (input instanceof LogData.Builder) {
+ final LogData.Builder logData = (LogData.Builder) input;
+ body = logData.getBody().getText().getText().getBytes(StandardCharsets.UTF_8);
+ for (final KeyStringValuePair tag : logData.getTags().getDataList()) {
+ if ("asz.file".equals(tag.getKey())) {
+ fileName = tag.getValue();
+ }
+ }
+ }
+ }
+
+ @Override
+ public void complete(final SourceReceiver sourceReceiver) {
+ if (body == null || StringUtil.isEmpty(format) || StringUtil.isEmpty(digest) || lines < 0) {
+ reject(REJECTED_ATTRIBUTES, "a required attribute is missing");
+ return;
+ }
+ if (!digest.equals(Digests.sha256Hex(body))) {
+ reject(REJECTED_DIGEST, "the body's digest is not the declared " + digest);
+ return;
+ }
+ if (Digests.countLines(body) != lines) {
+ reject(REJECTED_LINES, "the body has " + Digests.countLines(body) + " lines, the record declares " + lines);
+ return;
+ }
+ final String formattedService = NAMING_CONTROL.formatServiceName(serviceName);
+ final String serviceId = IDManager.ServiceID.buildId(formattedService, layer.isNormal());
+ final String instanceId = IDManager.ServiceInstanceID.buildId(
+ serviceId, NAMING_CONTROL.formatInstanceName(StringUtil.isEmpty(instanceName) ? "unknown" : instanceName));
+ switch (format) {
+ case FORMAT_SD:
+ if (StringUtil.isEmpty(session) || seq < 0) {
+ reject(REJECTED_ATTRIBUTES, "a Session Data record without asz.session or asz.seq");
+ return;
+ }
+ dispatch(dataRecord(serviceId, instanceId));
+ count(ACCEPTED_DATA);
+ break;
+ case FORMAT_SF:
+ if (StringUtil.isEmpty(conversation) || round < 0) {
+ reject(REJECTED_ATTRIBUTES, "a Session Flow record without asz.conversation or asz.round");
+ return;
+ }
+ dispatch(flowRecord(serviceId, instanceId));
+ count(ACCEPTED_FLOW);
+ break;
+ default:
+ reject(REJECTED_ATTRIBUTES, "unknown asz.format " + format);
+ break;
+ }
+ }
+
+ /**
+ * Hands a verified row to the record stream. A test overrides it to capture the row.
+ *
+ * @param record the verified row
+ */
+ protected void dispatch(final Record record) {
+ RecordStreamProcessor.getInstance().in(record);
+ }
+
+ private AIAgentSessionDataRecord dataRecord(final String serviceId, final String instanceId) {
+ final AIAgentSessionDataRecord record = new AIAgentSessionDataRecord();
+ record.setServiceId(serviceId);
+ record.setServiceInstanceId(instanceId);
+ record.setSession(session);
+ record.setSeq(seq);
+ record.setDigest(digest);
+ // The file's own through time places it inside the conversation's range; the record's timestamp is the
+ // sender's fallback for a file whose records carry no time.
+ final long fileTime = Times.millis(throughTime);
+ final long ts = fileTime != 0 ? fileTime : timestamp;
+ record.setTimestamp(ts);
+ record.setTimeBucket(TimeBucket.getRecordTimeBucket(ts));
+ record.setBody(body);
+ return record;
+ }
+
+ private AIAgentSessionFlowRecord flowRecord(final String serviceId, final String instanceId) {
+ final AIAgentSessionFlowRecord record = new AIAgentSessionFlowRecord();
+ record.setServiceId(serviceId);
+ record.setServiceInstanceId(instanceId);
+ record.setConversation(conversation);
+ record.setRound(round);
+ record.setSessionFromTime(Times.millis(sessionFromTime));
+ record.setTitle(clip(title, AIAgentSessionFlowRecord.TITLE_MAX_LENGTH));
+ record.setTalks(talks);
+ record.setSteps(steps);
+ record.setStreams(streams);
+ record.setSegments(segments);
+ record.setUnresolved(unresolved);
+ record.setDigest(digest);
+ // The conversation's last activity as of this round is the row's time, so the newest row is the head.
+ final long through = Times.millis(sessionThroughTime);
+ final long ts = through != 0 ? through : timestamp;
+ record.setTimestamp(ts);
+ record.setTimeBucket(TimeBucket.getRecordTimeBucket(ts));
+ record.setBody(body);
+ return record;
+ }
+
+ private void reject(@Nullable final CounterMetrics counter, final String why) {
+ count(counter);
+ log.warn("AI agent conversation file {} from {}/{} rejected: {}", fileName, serviceName, instanceName, why);
+ }
+
+ private static void count(@Nullable final CounterMetrics counter) {
+ if (counter != null) {
+ counter.inc();
+ }
+ }
+
+ @Nullable
+ private static String clip(@Nullable final String value, final int max) {
+ return value == null || value.length() <= max ? value : value.substring(0, max);
+ }
+
+ @Override
+ public String outputToJson() {
+ final JsonObject obj = new JsonObject();
+ obj.addProperty("type", getClass().getSimpleName());
+ obj.addProperty("name", name());
+ obj.addProperty("format", format);
+ obj.addProperty("file", fileName);
+ obj.addProperty("digest", digest);
+ obj.addProperty("lines", lines);
+ obj.addProperty("session", session);
+ obj.addProperty("seq", seq);
+ obj.addProperty("conversation", conversation);
+ obj.addProperty("round", round);
+ obj.addProperty("title", title);
+ obj.addProperty("serviceName", serviceName);
+ obj.addProperty("instanceName", instanceName);
+ obj.addProperty("timestamp", timestamp);
+ obj.addProperty("bytes", body == null ? 0 : body.length);
+ return obj.toString();
+ }
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/ConversationQueryService.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/ConversationQueryService.java
new file mode 100644
index 000000000000..19cf3c96db10
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/ConversationQueryService.java
@@ -0,0 +1,395 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.query;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import javax.annotation.Nullable;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.ai.agent.conversation.AIAgentConversationConfig;
+import org.apache.skywalking.oap.server.ai.agent.conversation.fold.ConversationFold;
+import org.apache.skywalking.oap.server.ai.agent.conversation.format.FileNames;
+import org.apache.skywalking.oap.server.ai.agent.conversation.format.SessionDataFile;
+import org.apache.skywalking.oap.server.ai.agent.conversation.format.SessionFlowRound;
+import org.apache.skywalking.oap.server.ai.agent.conversation.format.Times;
+import org.apache.skywalking.oap.server.ai.agent.conversation.query.type.ConversationFileFormat;
+import org.apache.skywalking.oap.server.ai.agent.conversation.query.type.ConversationList;
+import org.apache.skywalking.oap.server.ai.agent.conversation.query.type.ConversationRawFile;
+import org.apache.skywalking.oap.server.ai.agent.conversation.query.type.ConversationRawFiles;
+import org.apache.skywalking.oap.server.ai.agent.conversation.query.type.ConversationRow;
+import org.apache.skywalking.oap.server.ai.agent.conversation.view.ConversationViewBuilder;
+import org.apache.skywalking.oap.server.core.analysis.IDManager;
+import org.apache.skywalking.oap.server.core.analysis.manual.aiagent.AIAgentSessionDataRecord;
+import org.apache.skywalking.oap.server.core.analysis.manual.aiagent.AIAgentSessionFlowRecord;
+import org.apache.skywalking.oap.server.core.query.input.Duration;
+import org.apache.skywalking.oap.server.core.storage.StorageModule;
+import org.apache.skywalking.oap.server.core.storage.query.IAIAgentConversationQueryDAO;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+
+/**
+ * The read side. A conversation is read in two storage passes and never a read per file: its rounds by
+ * conversation over the whole retention window, then, for each session the head round names, the
+ * files by series id over the range the head round carries, in seq windows that keep one storage response under
+ * the BanyanDB client's inbound cap. The fold, the chain check and the view are built once per head digest and
+ * cached.
+ */
+@Slf4j
+public class ConversationQueryService implements IConversationQueryService {
+ private static final int DEFAULT_LIST_LIMIT = 1000;
+
+ private final ModuleManager moduleManager;
+ private final AIAgentConversationConfig config;
+ private IAIAgentConversationQueryDAO dao;
+
+ public ConversationQueryService(final ModuleManager moduleManager, final AIAgentConversationConfig config) {
+ this.moduleManager = moduleManager;
+ this.config = config;
+ }
+
+ private IAIAgentConversationQueryDAO dao() {
+ if (dao == null) {
+ dao = moduleManager.find(StorageModule.NAME).provider().getService(IAIAgentConversationQueryDAO.class);
+ }
+ return dao;
+ }
+
+ @Override
+ public ConversationList listConversations(final String serviceId,
+ @Nullable final String serviceInstanceId,
+ final Duration duration,
+ @Nullable final Integer limit) throws IOException {
+ final int rounds = Math.min(
+ limit == null || limit <= 0 ? DEFAULT_LIST_LIMIT : limit, config.getMaxListLimit());
+ final List newestFirst =
+ dao().queryRoundsDebuggable(serviceId, serviceInstanceId, null, duration, rounds, false);
+ final Map rows = new LinkedHashMap<>();
+ for (final AIAgentSessionFlowRecord r : newestFirst) {
+ final ConversationRow existing = rows.get(r.getConversation());
+ if (existing != null && existing.getRound() >= r.getRound()) {
+ continue;
+ }
+ rows.put(r.getConversation(), row(r));
+ }
+ final ConversationList list = new ConversationList();
+ list.setConversations(new ArrayList<>(rows.values()));
+ return list;
+ }
+
+ private static ConversationRow row(final AIAgentSessionFlowRecord r) {
+ final ConversationRow row = new ConversationRow();
+ row.setConversation(r.getConversation());
+ row.setServiceInstanceId(r.getServiceInstanceId());
+ row.setServiceInstanceName(instanceName(r.getServiceInstanceId()));
+ row.setTitle(r.getTitle());
+ row.setRound((int) r.getRound());
+ row.setTalks((int) r.getTalks());
+ row.setSteps((int) r.getSteps());
+ row.setStreams((int) r.getStreams());
+ row.setSegments((int) r.getSegments());
+ row.setUnresolved((int) r.getUnresolved());
+ row.setFrom(r.getSessionFromTime());
+ row.setTo(r.getTimestamp());
+ return row;
+ }
+
+ private static String instanceName(final String instanceId) {
+ if (StringUtil.isEmpty(instanceId)) {
+ return "";
+ }
+ try {
+ return IDManager.ServiceInstanceID.analysisId(instanceId).getName();
+ } catch (final RuntimeException e) {
+ return instanceId;
+ }
+ }
+
+ @Override
+ @Nullable
+ public Map buildConversationView(final String serviceId,
+ @Nullable final String serviceInstanceId,
+ final String conversation) throws IOException {
+ final List heads =
+ dao().queryRoundsDebuggable(serviceId, serviceInstanceId, conversation, null, 1, false);
+ if (heads.isEmpty()) {
+ return null;
+ }
+ final Chain chain = readChain(serviceId, serviceInstanceId, conversation);
+ return new ConversationViewBuilder(chain.fold, chain.roundInputs, chain.files, chain.problems).build();
+ }
+
+ @Override
+ public ConversationRawFiles getConversationRawFiles(final String serviceId,
+ @Nullable final String serviceInstanceId,
+ final String conversation,
+ @Nullable final List files,
+ final boolean includeBody) throws IOException {
+ final Set wanted = new LinkedHashSet<>();
+ if (files != null) {
+ for (final String id : files) {
+ final FileNames.Parsed p = FileNames.parse(id);
+ if (p != null) {
+ wanted.add(p);
+ }
+ }
+ }
+ final List rounds = readRounds(serviceId, serviceInstanceId, conversation);
+ final ConversationRawFiles out = new ConversationRawFiles();
+ if (rounds.isEmpty()) {
+ out.setErrorReason("no round of conversation " + conversation + " is stored for this service");
+ return out;
+ }
+ final AIAgentSessionFlowRecord headRow = rounds.get(rounds.size() - 1);
+ final SessionFlowRound head = SessionFlowRound.parse(headRow.getBody());
+ // the caller's sender, or every sender of the service: a Sessionizer renamed between pushes leaves a
+ // conversation's files under two instances, and a read must see both
+ final String instance = StringUtil.isNotEmpty(serviceInstanceId) ? serviceInstanceId : null;
+ final long from = Times.millis(head.getHeader().getSessionFromTime());
+ final long to = rangeEnd(head.getHeader().getSessionThroughTime(), headRow.getTimestamp());
+
+ // Session Data files first, then the rounds, each list in its own order.
+ final Set sessions = new LinkedHashSet<>();
+ if (StringUtil.isNotEmpty(head.getHeader().getSession())) {
+ sessions.add(head.getHeader().getSession());
+ }
+ for (final FileNames.Parsed p : wanted) {
+ if (p.isDataFile()) {
+ sessions.add(p.getSession());
+ }
+ }
+ for (final String session : sessions) {
+ final Set seqs = new HashSet<>();
+ boolean all = files == null;
+ for (final FileNames.Parsed p : wanted) {
+ if (p.isDataFile() && session.equals(p.getSession())) {
+ seqs.add(p.getSeq());
+ }
+ }
+ if (files != null && seqs.isEmpty()) {
+ continue;
+ }
+ final long throughSeq = all ? head.getHeader().getThroughSeq() : seqs.stream().mapToLong(Long::longValue).max().orElse(0);
+ final long fromSeq = all ? 1 : seqs.stream().mapToLong(Long::longValue).min().orElse(1);
+ final Set seen = new HashSet<>();
+ for (final AIAgentSessionDataRecord f : readFiles(serviceId, instance, session, from, to, fromSeq, throughSeq)) {
+ if (!all && !seqs.contains(f.getSeq()) || !seen.add(f.getSeq())) {
+ continue;
+ }
+ final SessionDataFile parsed = SessionDataFile.parse(f.getBody());
+ final ConversationRawFile raw = new ConversationRawFile();
+ raw.setId(FileNames.dataFile(parsed.getHeader()));
+ raw.setFormat(ConversationFileFormat.SD);
+ raw.setSession(session);
+ raw.setSeq((int) f.getSeq());
+ raw.setDigest(f.getDigest());
+ raw.setBytes(f.getBody().length);
+ raw.setTimestamp(f.getTimestamp());
+ if (includeBody) {
+ raw.setBody(new String(f.getBody(), StandardCharsets.UTF_8));
+ }
+ out.getFiles().add(raw);
+ }
+ }
+ for (final AIAgentSessionFlowRecord r : rounds) {
+ if (files != null) {
+ boolean named = false;
+ for (final FileNames.Parsed p : wanted) {
+ if (!p.isDataFile() && p.getRound() == r.getRound()) {
+ named = true;
+ break;
+ }
+ }
+ if (!named) {
+ continue;
+ }
+ }
+ final SessionFlowRound parsed = SessionFlowRound.parse(r.getBody());
+ final ConversationRawFile raw = new ConversationRawFile();
+ raw.setId(FileNames.roundFile(conversation, r.getRound(), parsed.getCommitDigest()));
+ raw.setFormat(ConversationFileFormat.SF);
+ raw.setRound((int) r.getRound());
+ raw.setDigest(r.getDigest());
+ raw.setBytes(r.getBody().length);
+ raw.setTimestamp(r.getTimestamp());
+ if (includeBody) {
+ raw.setBody(new String(r.getBody(), StandardCharsets.UTF_8));
+ }
+ out.getFiles().add(raw);
+ }
+ return out;
+ }
+
+ // ---------------------------------------------------------------- the two-pass read
+
+ private static final class Chain {
+ final ConversationFold fold = new ConversationFold();
+ /** Every stored round in number order, readable or not, for the document's listing. */
+ final List roundInputs = new ArrayList<>();
+ final Map files = new TreeMap<>();
+ /** What stopped the fold short of the chain's last round, in words, as the Sessionizer's FoldPartial. */
+ final List problems = new ArrayList<>();
+ }
+
+ /**
+ * Every round of a conversation, in round order, read window by window up to the head: a round is up to
+ * 2 MiB and a long conversation has hundreds, so one read of them all would not fit a storage response.
+ * The head is the highest round number stored, fixed first, so a round landing during the read is left for
+ * the next call; it is not the newest row by time, because the Sessionizer can write a later round that
+ * carries no later activity, over metadata or older records. A round stored more than once, by two
+ * senders or by a redelivery, is kept once, the first copy.
+ */
+ private List readRounds(final String serviceId, @Nullable final String instance,
+ final String conversation) throws IOException {
+ long headRound = 0;
+ for (final AIAgentSessionFlowRecord r : dao().queryRoundsDebuggable(
+ serviceId, instance, conversation, null, config.getMaxListLimit(), false)) {
+ headRound = Math.max(headRound, r.getRound());
+ }
+ if (headRound == 0) {
+ return new ArrayList<>();
+ }
+ final Map byRound = new TreeMap<>();
+ final int window = config.getRoundReadWindow();
+ for (long start = 1; start <= headRound; start += window) {
+ final long end = Math.min(headRound, start + window - 1);
+ for (final AIAgentSessionFlowRecord r : dao().queryRoundsByNumberDebuggable(
+ serviceId, instance, conversation, start, end)) {
+ byRound.putIfAbsent(r.getRound(), r);
+ }
+ }
+ return new ArrayList<>(byRound.values());
+ }
+
+ /**
+ * The fold goes as far as the chain holds, as the Sessionizer's FoldPartial: a round that is
+ * missing, that does not read, or that the fold refuses stops it, and what was wrong is carried in words.
+ * Every stored round is still listed, so the document shows the rounds after the gap, unverified.
+ */
+ private Chain readChain(final String serviceId, @Nullable final String serviceInstanceId,
+ final String conversation) throws IOException {
+ final Chain chain = new Chain();
+ final List rounds = readRounds(serviceId, serviceInstanceId, conversation);
+ final String instance = StringUtil.isNotEmpty(serviceInstanceId) ? serviceInstanceId : null;
+ boolean folding = true;
+ long throughSeq = 0;
+ AIAgentSessionFlowRecord headRow = null;
+ for (final AIAgentSessionFlowRecord r : rounds) {
+ if (folding && r.getRound() != chain.fold.getRound() + 1) {
+ chain.problems.add("round " + (chain.fold.getRound() + 1) + " is missing; the chain stops at round "
+ + chain.fold.getRound());
+ folding = false;
+ }
+ SessionFlowRound parsed = null;
+ String unreadable = null;
+ try {
+ parsed = SessionFlowRound.parse(r.getBody());
+ if (!parsed.isIntact()) {
+ unreadable = "sessionflow: digest mismatch: computed " + first12(parsed.getComputedDigest())
+ + ", round claims " + first12(parsed.getCommitDigest());
+ }
+ } catch (final RuntimeException e) {
+ unreadable = "sessionflow: " + e.getMessage();
+ }
+ if (unreadable != null) {
+ chain.roundInputs.add(ConversationViewBuilder.RoundInput.unreadable(r.getRound(), unreadable));
+ if (folding) {
+ chain.problems.add("round " + r.getRound() + " does not read: " + unreadable);
+ folding = false;
+ }
+ continue;
+ }
+ chain.roundInputs.add(new ConversationViewBuilder.RoundInput(parsed, r.getDigest()));
+ throughSeq = Math.max(throughSeq, parsed.getHeader().getThroughSeq());
+ if (!folding) {
+ continue;
+ }
+ final String refused = chain.fold.apply(parsed);
+ if (refused != null) {
+ chain.problems.add("round " + r.getRound() + " does not fold: " + refused);
+ folding = false;
+ continue;
+ }
+ headRow = r;
+ }
+ if (headRow == null) {
+ return chain;
+ }
+ // every landed file of the conversation's sessions, as far as any listed round reaches
+ final long from = Times.millis(chain.fold.getSessionFromTime());
+ final long to = rangeEnd(chain.fold.getSessionThroughTime(), headRow.getTimestamp());
+ final Set sessions = new LinkedHashSet<>();
+ if (StringUtil.isNotEmpty(chain.fold.getSession())) {
+ sessions.add(chain.fold.getSession());
+ }
+ for (final SessionFlowRound.Node n : chain.fold.nodesOfKind("session")) {
+ final String id = n.getId();
+ sessions.add(id.startsWith("session/") ? id.substring("session/".length()) : id);
+ }
+ for (final String session : sessions) {
+ for (final AIAgentSessionDataRecord f : readFiles(serviceId, instance, session, from, to, 1, throughSeq)) {
+ if (chain.files.containsKey(f.getSeq())) {
+ // the same file under two senders; the chain check judges the copy that was kept
+ continue;
+ }
+ try {
+ chain.files.put(f.getSeq(), SessionDataFile.parse(f.getBody()));
+ } catch (final RuntimeException e) {
+ // as the Sessionizer's reader, a file that does not decode contributes no records and is
+ // named by the chain check as missing
+ chain.problems.add("file seq " + f.getSeq() + " of session " + session + " cannot be decoded: "
+ + e.getMessage());
+ }
+ }
+ }
+ return chain;
+ }
+
+ private static String first12(@Nullable final String s) {
+ return s == null ? "" : s.substring(0, Math.min(12, s.length()));
+ }
+
+ private List readFiles(final String serviceId, @Nullable final String instance,
+ final String session, final long from, final long to,
+ final long fromSeq, final long throughSeq) throws IOException {
+ final List out = new ArrayList<>();
+ final int window = config.getFileReadWindow();
+ for (long start = fromSeq; start <= throughSeq; start += window) {
+ final long end = Math.min(throughSeq, start + window - 1);
+ out.addAll(dao().queryFilesDebuggable(serviceId, instance, session, from, to, start, end));
+ }
+ return out;
+ }
+
+ /**
+ * The end of the files read: the conversation's last activity as of the head round, or the head row's own
+ * timestamp when the header carries none. A file is stamped at or before that moment.
+ */
+ private static long rangeEnd(@Nullable final String sessionThroughTime, final long headRowTimestamp) {
+ final long through = Times.millis(sessionThroughTime);
+ return Math.max(through, headRowTimestamp);
+ }
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/IConversationQueryService.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/IConversationQueryService.java
new file mode 100644
index 000000000000..007867c0952a
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/IConversationQueryService.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.query;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.skywalking.oap.server.ai.agent.conversation.query.type.ConversationList;
+import org.apache.skywalking.oap.server.ai.agent.conversation.query.type.ConversationRawFiles;
+import org.apache.skywalking.oap.server.core.query.input.Duration;
+import org.apache.skywalking.oap.server.library.module.Service;
+
+/**
+ * The two GraphQL operations of ai-agent-conversation.graphqls and the document behind the
+ * conversation view route.
+ */
+public interface IConversationQueryService extends Service {
+ /**
+ * One row per conversation, from the newest round's attributes; nothing decoded.
+ *
+ * @param serviceId the service
+ * @param serviceInstanceId the sender, or null
+ * @param duration the window
+ * @param limit at most this many rounds read, newest first, before folding; null for the default
+ * @return the rows, newest first
+ * @throws IOException on a storage failure
+ */
+ ConversationList listConversations(String serviceId, @Nullable String serviceInstanceId, Duration duration,
+ @Nullable Integer limit) throws IOException;
+
+ /**
+ * The whole conversation, once, as one asz.view document, built on every call.
+ *
+ * @param serviceId the service
+ * @param serviceInstanceId the sender, or null
+ * @param conversation the conversation
+ * @return the document as ordered maps, or null when the service stores no round of the conversation
+ * @throws IOException on a storage failure
+ */
+ @Nullable
+ Map buildConversationView(String serviceId, @Nullable String serviceInstanceId,
+ String conversation) throws IOException;
+
+ /**
+ * Every landed file and round of a conversation as stored, or only the named ones.
+ *
+ * @param serviceId the service
+ * @param serviceInstanceId the sender, or null
+ * @param conversation the conversation
+ * @param files only these file ids, or null for every file
+ * @param includeBody whether the caller selected the body field
+ * @return the files
+ * @throws IOException on a storage failure
+ */
+ ConversationRawFiles getConversationRawFiles(String serviceId, @Nullable String serviceInstanceId,
+ String conversation, @Nullable List files,
+ boolean includeBody) throws IOException;
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/http/CompressResponse.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/http/CompressResponse.java
new file mode 100644
index 000000000000..a437745edfbd
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/http/CompressResponse.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.query.http;
+
+import com.linecorp.armeria.common.HttpRequest;
+import com.linecorp.armeria.common.HttpResponse;
+import com.linecorp.armeria.server.HttpService;
+import com.linecorp.armeria.server.ServiceRequestContext;
+import com.linecorp.armeria.server.DecoratingHttpServiceFunction;
+import com.linecorp.armeria.server.encoding.EncodingService;
+import java.util.function.Function;
+
+/**
+ * Compresses a JSON or YAML response when the client's Accept-Encoding allows it, chunk by chunk,
+ * so a streamed document stays streamed. A document is repetitive text and shrinks several times over.
+ */
+public final class CompressResponse implements DecoratingHttpServiceFunction {
+ private static final Function super HttpService, EncodingService> ENCODING = EncodingService.builder()
+ .encodableContentTypes(ConversationViewHandler.JSON, ConversationViewHandler.YAML)
+ .newDecorator();
+
+ @Override
+ public HttpResponse serve(final HttpService delegate, final ServiceRequestContext ctx, final HttpRequest req)
+ throws Exception {
+ return delegate.decorate(ENCODING).serve(ctx, req);
+ }
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/http/ConversationViewHandler.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/http/ConversationViewHandler.java
new file mode 100644
index 000000000000..d695ea4e08dc
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/http/ConversationViewHandler.java
@@ -0,0 +1,224 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.query.http;
+
+import com.google.gson.JsonObject;
+import com.linecorp.armeria.common.HttpData;
+import com.linecorp.armeria.common.HttpResponse;
+import com.linecorp.armeria.common.HttpResponseWriter;
+import com.linecorp.armeria.common.HttpStatus;
+import com.linecorp.armeria.common.MediaType;
+import com.linecorp.armeria.common.ResponseHeaders;
+import com.linecorp.armeria.common.util.TimeoutMode;
+import com.linecorp.armeria.server.ServiceRequestContext;
+import com.linecorp.armeria.server.annotation.Decorator;
+import com.linecorp.armeria.server.annotation.Get;
+import com.linecorp.armeria.server.annotation.Header;
+import com.linecorp.armeria.server.annotation.Param;
+import java.io.IOException;
+import java.io.Writer;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import javax.annotation.Nullable;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.ai.agent.conversation.query.IConversationQueryService;
+import org.apache.skywalking.oap.server.ai.agent.conversation.view.ViewJson;
+import org.apache.skywalking.oap.server.ai.agent.conversation.view.ViewYaml;
+import org.apache.skywalking.oap.server.core.analysis.IDManager;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+
+/**
+ * GET /ai-agent/conversations/{conversation}/v1/view: the whole conversation as one
+ * asz.view 1.0 document, streamed. It lives on the core HTTP server beside /graphql,
+ * and is not a GraphQL query because the document is as large as the conversation, tens of megabytes for a
+ * long one: it is written to the response as it is rendered, never held whole as one string, compressed when
+ * the client allows, and given its own request timeout in place of the server's default.
+ *
+ *
Query parameters: service, the service name, or serviceId; optionally
+ * instance, the sender's instance name. What the body is, the HTTP layer says: the media type
+ * names the document format and its version, application/vnd.skywalking.asz.view+json; version=1.0, or the
+ * +yaml twin when Accept asks for YAML. The document's own first two keys repeat it.
+ *
+ *
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):
+ * {"type": "about:blank", "title": "...", "status": 404, "detail": "..."}.
+ */
+@Slf4j
+public class ConversationViewHandler {
+ public static final String PATH = "/ai-agent/conversations/{conversation}/v1/view";
+ /** The document format and version as a media type; the format is the type, the version a parameter. */
+ static final MediaType JSON = MediaType.parse("application/vnd.skywalking.asz.view+json");
+ static final MediaType YAML = MediaType.parse("application/vnd.skywalking.asz.view+yaml");
+ private static final String VERSION_PARAMETER = "version";
+ private static final MediaType PROBLEM = MediaType.parse("application/problem+json").withCharset(StandardCharsets.UTF_8);
+ private static final MediaType JSON_UTF_8 = JSON.withParameter(VERSION_PARAMETER, ViewYaml.VERSION).withCharset(StandardCharsets.UTF_8);
+ private static final MediaType YAML_UTF_8 = YAML.withParameter(VERSION_PARAMETER, ViewYaml.VERSION).withCharset(StandardCharsets.UTF_8);
+ /** Bytes rendered between two writes to the response. */
+ private static final int CHUNK_CHARS = 64 * 1024;
+
+ private final IConversationQueryService service;
+ private final Duration timeout;
+
+ public ConversationViewHandler(final IConversationQueryService service, final Duration timeout) {
+ this.service = service;
+ this.timeout = timeout;
+ }
+
+ @Get(PATH)
+ @Decorator(CompressResponse.class)
+ public HttpResponse view(final ServiceRequestContext ctx,
+ @Param("conversation") final String conversation,
+ @Param("service") @Nullable final String serviceName,
+ @Param("serviceId") @Nullable final String serviceIdParam,
+ @Param("instance") @Nullable final String instanceName,
+ @Header("Accept") @Nullable final String accept) {
+ final String serviceId;
+ if (StringUtil.isNotEmpty(serviceIdParam)) {
+ serviceId = serviceIdParam;
+ } else if (StringUtil.isNotEmpty(serviceName)) {
+ serviceId = IDManager.ServiceID.buildId(serviceName, true);
+ } else {
+ return HttpResponse.of(HttpStatus.BAD_REQUEST, PROBLEM, problem(HttpStatus.BAD_REQUEST, "service or serviceId is required"));
+ }
+ final String instanceId = StringUtil.isEmpty(instanceName)
+ ? null : IDManager.ServiceInstanceID.buildId(serviceId, instanceName);
+ final boolean yaml = accept != null && accept.contains("yaml");
+
+ ctx.setRequestTimeout(TimeoutMode.SET_FROM_NOW, timeout);
+ final HttpResponseWriter res = HttpResponse.streaming();
+ ctx.blockingTaskExecutor().execute(() -> stream(res, serviceId, instanceId, conversation, yaml));
+ return res;
+ }
+
+ private void stream(final HttpResponseWriter res, final String serviceId, @Nullable final String instanceId,
+ final String conversation, final boolean yaml) {
+ final Map doc;
+ try {
+ doc = service.buildConversationView(serviceId, instanceId, conversation);
+ } catch (final IOException | RuntimeException e) {
+ log.error("AI agent conversation {} of service {} could not be read", conversation, serviceId, e);
+ problem(res, HttpStatus.INTERNAL_SERVER_ERROR, e.getMessage());
+ return;
+ }
+ if (doc == null) {
+ problem(res, HttpStatus.NOT_FOUND, "no round of conversation " + conversation + " is stored for this service");
+ return;
+ }
+ res.write(ResponseHeaders.builder(HttpStatus.OK).contentType(yaml ? YAML_UTF_8 : JSON_UTF_8).build());
+ try (Writer out = new ChunkWriter(res, timeout)) {
+ if (yaml) {
+ ViewYaml.write(doc, out);
+ } else {
+ ViewJson.write(doc, out);
+ }
+ } catch (final IOException e) {
+ log.debug("AI agent conversation {} response ended early: {}", conversation, e.getMessage());
+ res.close(e);
+ return;
+ }
+ res.close();
+ }
+
+ private static void problem(final HttpResponseWriter res, final HttpStatus status, @Nullable final String detail) {
+ res.write(ResponseHeaders.builder(status).contentType(PROBLEM).build());
+ res.write(HttpData.ofUtf8(problem(status, detail)));
+ res.close();
+ }
+
+ /**
+ * @return an RFC 9457 problem: the status and its reason phrase, and what went wrong in words
+ */
+ private static String problem(final HttpStatus status, @Nullable final String detail) {
+ final JsonObject body = new JsonObject();
+ body.addProperty("type", "about:blank");
+ body.addProperty("title", status.reasonPhrase());
+ body.addProperty("status", status.code());
+ body.addProperty("detail", detail == null ? "" : detail);
+ return body.toString();
+ }
+
+ /**
+ * Hands the rendered text to the response in UTF-8 chunks and waits for each to be consumed before
+ * rendering more, so a slow client holds back the render instead of growing a buffer. A chunk never ends
+ * between the two halves of a surrogate pair: a trailing high surrogate waits for the next chunk, so every
+ * code point is encoded whole. A client that went away ends the render with an IOException at the next
+ * chunk.
+ */
+ static final class ChunkWriter extends Writer {
+ private final HttpResponseWriter res;
+ private final Duration timeout;
+ private final StringBuilder buffer = new StringBuilder(CHUNK_CHARS + 1024);
+
+ ChunkWriter(final HttpResponseWriter res, final Duration timeout) {
+ this.res = res;
+ this.timeout = timeout;
+ }
+
+ @Override
+ public void write(final char[] cbuf, final int off, final int len) throws IOException {
+ int from = off;
+ int left = len;
+ while (left > 0) {
+ final int take = Math.min(left, CHUNK_CHARS - buffer.length());
+ buffer.append(cbuf, from, take);
+ from += take;
+ left -= take;
+ if (buffer.length() >= CHUNK_CHARS) {
+ flush();
+ }
+ }
+ }
+
+ @Override
+ public void flush() throws IOException {
+ write(false);
+ }
+
+ private void write(final boolean last) throws IOException {
+ int end = buffer.length();
+ if (!last && end > 0 && Character.isHighSurrogate(buffer.charAt(end - 1))) {
+ end--;
+ }
+ if (end == 0) {
+ return;
+ }
+ if (!res.tryWrite(HttpData.ofUtf8(buffer.substring(0, end)))) {
+ throw new IOException("the response is closed");
+ }
+ buffer.delete(0, end);
+ try {
+ res.whenConsumed().get(timeout.toMillis(), TimeUnit.MILLISECONDS);
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException("interrupted while the client read the response", e);
+ } catch (final ExecutionException | TimeoutException e) {
+ throw new IOException("the client stopped reading the response", e);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ write(true);
+ }
+ }
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/input/ConversationCondition.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/input/ConversationCondition.java
new file mode 100644
index 000000000000..5804d5caf660
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/input/ConversationCondition.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.query.input;
+
+import lombok.Data;
+import org.apache.skywalking.oap.server.core.query.input.InstanceCondition;
+import org.apache.skywalking.oap.server.core.query.input.ServiceCondition;
+
+@Data
+public class ConversationCondition {
+ private ServiceCondition service;
+ private String conversation;
+ private InstanceCondition instance;
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/input/ConversationListCondition.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/input/ConversationListCondition.java
new file mode 100644
index 000000000000..7f43827d14ba
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/input/ConversationListCondition.java
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.query.input;
+
+import lombok.Data;
+import org.apache.skywalking.oap.server.core.query.input.InstanceCondition;
+import org.apache.skywalking.oap.server.core.query.input.ServiceCondition;
+
+@Data
+public class ConversationListCondition {
+ private ServiceCondition service;
+ private InstanceCondition instance;
+ private Integer limit;
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/type/ConversationFileFormat.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/type/ConversationFileFormat.java
new file mode 100644
index 000000000000..57973ead739e
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/type/ConversationFileFormat.java
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.query.type;
+
+/**
+ * Which of the two landed formats a raw file is.
+ */
+public enum ConversationFileFormat {
+ /** Session Data: the records of one stream, an agent's meta file, a run journal, a workflow manifest or script. */
+ SD,
+ /** Session Flow: one round of the conversation's chain. */
+ SF
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/type/ConversationList.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/type/ConversationList.java
new file mode 100644
index 000000000000..9af0abb3d2f2
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/type/ConversationList.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.query.type;
+
+import java.util.ArrayList;
+import java.util.List;
+import lombok.Data;
+import org.apache.skywalking.oap.server.core.query.type.debugging.DebuggingTrace;
+
+@Data
+public class ConversationList {
+ private String errorReason;
+ private List conversations = new ArrayList<>();
+ private DebuggingTrace debuggingTrace;
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/type/ConversationRawFile.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/type/ConversationRawFile.java
new file mode 100644
index 000000000000..a0fa97efe11e
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/type/ConversationRawFile.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.query.type;
+
+import lombok.Data;
+
+/**
+ * One landed file or round, as stored.
+ */
+@Data
+public class ConversationRawFile {
+ private String id;
+ private ConversationFileFormat format;
+ private String session;
+ private Integer seq;
+ private Integer round;
+ private String digest;
+ private int bytes;
+ private long timestamp;
+ private String body;
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/type/ConversationRawFiles.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/type/ConversationRawFiles.java
new file mode 100644
index 000000000000..341179e50d1d
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/type/ConversationRawFiles.java
@@ -0,0 +1,31 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.query.type;
+
+import java.util.ArrayList;
+import java.util.List;
+import lombok.Data;
+import org.apache.skywalking.oap.server.core.query.type.debugging.DebuggingTrace;
+
+@Data
+public class ConversationRawFiles {
+ private String errorReason;
+ private List files = new ArrayList<>();
+ private DebuggingTrace debuggingTrace;
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/type/ConversationRow.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/type/ConversationRow.java
new file mode 100644
index 000000000000..8f166a76a3b1
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/query/type/ConversationRow.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.query.type;
+
+import lombok.Data;
+
+/**
+ * One row per conversation on the list page, from the newest round's attributes.
+ */
+@Data
+public class ConversationRow {
+ private String conversation;
+ private String serviceInstanceId;
+ private String serviceInstanceName;
+ private String title;
+ private int round;
+ private int talks;
+ private int steps;
+ private int streams;
+ private int segments;
+ private int unresolved;
+ private long from;
+ private long to;
+}
diff --git a/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/view/ConversationViewBuilder.java b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/view/ConversationViewBuilder.java
new file mode 100644
index 000000000000..5437dfaa24ac
--- /dev/null
+++ b/oap-server/analyzer/ai-agent-conversation/src/main/java/org/apache/skywalking/oap/server/ai/agent/conversation/view/ConversationViewBuilder.java
@@ -0,0 +1,1389 @@
+/*
+ * Licensed to the 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.
+ * The 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 org.apache.skywalking.oap.server.ai.agent.conversation.view;
+
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.regex.Pattern;
+import javax.annotation.Nullable;
+import lombok.Getter;
+import org.apache.skywalking.oap.server.ai.agent.conversation.fold.ConversationFold;
+import org.apache.skywalking.oap.server.ai.agent.conversation.format.Digests;
+import org.apache.skywalking.oap.server.ai.agent.conversation.format.FileNames;
+import org.apache.skywalking.oap.server.ai.agent.conversation.format.Ref;
+import org.apache.skywalking.oap.server.ai.agent.conversation.format.SessionDataFile;
+import org.apache.skywalking.oap.server.ai.agent.conversation.format.SessionFlowRound;
+import org.apache.skywalking.oap.server.ai.agent.conversation.format.Times;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
+
+/**
+ * Builds the asz.view 1.0 document of one conversation, key for key as the Sessionizer's
+ * pkg/sessionview defines it and its internal/view builds it, so the document the OAP
+ * answers with equals the one asz conversation -json prints for the same files.
+ *
+ *
The document is deterministic: no wall-clock time, keys in the order the format page lists them, lists in
+ * record order. Keys a node has no value for are absent; the fixed keys of rounds and files are null when absent.
+ */
+public final class ConversationViewBuilder {
+ /** The readable text a node carries is clipped to this many bytes; the full size is in bytes. */
+ static final int PREVIEW_BYTES = 2000;
+ private static final int MAX_DEPTH = 12;
+ private static final Pattern INTEGER_LITERAL = Pattern.compile("-?\\d+");
+ /** A parent walk stops here; a well-formed fold is far shallower, a malformed one must not loop. */
+ private static final int MAX_ANCESTORS = 64;
+ static final String STATE_VERIFIED = "verified";
+ static final String STATE_INCOMPLETE = "incomplete";
+ static final String STATE_MISMATCH = "mismatch";
+
+ private final ConversationFold fold;
+ private final List rounds;
+ private final Map files;
+ private final List problems;
+ /** Every timed record's moment in nanoseconds, the precision the Sessionizer computes intervals with. */
+ private final Map at = new HashMap<>();
+
+ /**
+ * @param fold the fold of the rounds, in order
+ * @param rounds every stored round in number order, decoded or with the reason it does not read
+ * @param files the session's landed files by seq
+ * @param problems what stopped the fold short of the chain's last round, in words; they lead the document's
+ * problems, and a fold that stopped short leaves the document incomplete at best
+ */
+ public ConversationViewBuilder(final ConversationFold fold,
+ final List rounds,
+ final Map files,
+ final List problems) {
+ this.fold = fold;
+ this.rounds = rounds;
+ this.files = files;
+ this.problems = problems;
+ for (final SessionDataFile f : files.values()) {
+ for (final SessionDataFile.Record r : f.getRecords()) {
+ if (r.getTimeNanos() != 0) {
+ at.put(new Ref(f.getHeader().getSeq(), r.getRow(), null), r.getTimeNanos());
+ }
+ }
+ }
+ }
+
+ /**
+ * @return the document as ordered maps
+ */
+ public Map build() {
+ final Overview o = overview();
+ final Chain chain = chain();
+
+ final Map doc = new LinkedHashMap<>();
+ doc.put("format", ViewYaml.FORMAT);
+ doc.put("version", ViewYaml.VERSION);
+ doc.put("conversation", nullToEmpty(fold.getConversation()));
+ doc.put("sessions", sessions());
+ final Map head = new LinkedHashMap<>();
+ head.put("round", fold.getRound());
+ head.put("digest", nullToEmpty(fold.getDigest()));
+ doc.put("head", head);
+ doc.put("parser", nullToEmpty(fold.getParser()));
+ doc.put("policy", nullToEmpty(fold.getPolicy()));
+
+ final Map summary = new LinkedHashMap<>();
+ summary.put("title", o.title);
+ summary.put("state", chain.state);
+ summary.put("problems", chain.problems);
+ summary.put("talks", o.talks.size());
+ summary.put("steps", stepNodes());
+ summary.put("streams", o.streams.size());
+ summary.put("segments", o.segments.size());
+ summary.put("rounds", chain.rounds.size());
+ summary.put("unresolved", fold.openUnresolved().size());
+ final SessionFlowRound.Node sessionNode = fold.node(sessionNodeId());
+ summary.put("from", sessionNode == null ? 0L : Times.millis(sessionNode.attr("from_time")));
+ summary.put("to", sessionNode == null ? 0L : Times.millis(sessionNode.attr("through_time")));
+ summary.put("kinds", o.kinds);
+ summary.put("relation_types", o.relationTypes);
+ summary.put("quality", o.quality);
+ doc.put("summary", summary);
+ doc.put("rounds", chain.rounds);
+ final List