This is an automated email from the ASF dual-hosted git repository.
hsluoyz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/casbin-gateway.git
The following commit(s) were added to refs/heads/master by this push:
new 441127c feat(sessions): read dsh's zstd-framed transcripts
441127c is described below
commit 441127c1034cb98ce07501d0733f279a1a4322b7
Author: Yang Luo <[email protected]>
AuthorDate: Fri Sep 4 11:57:54 2026 +0800
feat(sessions): read dsh's zstd-framed transcripts
---
README.md | 2 +-
README_zh.md | 2 +-
agenthistory/dsh.go | 231 ++++++++++++++++++++++++++++++++++++++++++++++++
agenthistory/history.go | 36 ++++++--
go.mod | 1 +
go.sum | 2 +
6 files changed, 263 insertions(+), 11 deletions(-)
diff --git a/README.md b/README.md
index 8e820bf..de12238 100644
--- a/README.md
+++ b/README.md
@@ -100,7 +100,7 @@
| <img src="https://www.google.com/s2/favicons?domain=openagentai.org&sz=64";
width="16" height="16" alt=""> **OpenAgent** | ✅ | — | — | — | — | — | — |
| <img src="https://www.google.com/s2/favicons?domain=openclaw.ai&sz=64";
width="16" height="16" alt=""> **OpenClaw** | ✅ | ✅ OpenAI | ✅ | ✅ | ✅ | ✅ |
npm |
| <img src="https://www.google.com/s2/favicons?domain=nousresearch.com&sz=64";
width="16" height="16" alt=""> **Hermes Agent** | ✅ | ✅ OpenAI | — | — | — | —
| — |
-| <img src="https://www.google.com/s2/favicons?domain=deepseek.com&sz=64";
width="16" height="16" alt=""> **DeepSeek Harness** | ✅ | ✅ OpenAI | ✅ | ✅ | —
| — | npm |
+| <img src="https://www.google.com/s2/favicons?domain=deepseek.com&sz=64";
width="16" height="16" alt=""> **DeepSeek Harness** | ✅ | ✅ OpenAI | ✅ | ✅ | —
| ✅ | npm |
- **Monitoring** — audit-only records of what the agent did: prompts, tool
calls, permission prompts. Nothing an agent does waits on Gateway, and no
answer changes because monitoring is on.
- **Provider** — Gateway writes the agent's own configuration to point it at a
bound provider, in the wire format that agent's client speaks. An agent without
it still reaches Gateway through the environment variables the UI shows.
diff --git a/README_zh.md b/README_zh.md
index 3360c58..b8e845a 100644
--- a/README_zh.md
+++ b/README_zh.md
@@ -100,7 +100,7 @@
| <img src="https://www.google.com/s2/favicons?domain=openagentai.org&sz=64";
width="16" height="16" alt=""> **OpenAgent** | ✅ | — | — | — | — | — | — |
| <img src="https://www.google.com/s2/favicons?domain=openclaw.ai&sz=64";
width="16" height="16" alt=""> **OpenClaw** | ✅ | ✅ OpenAI | ✅ | ✅ | ✅ | ✅ |
npm |
| <img src="https://www.google.com/s2/favicons?domain=nousresearch.com&sz=64";
width="16" height="16" alt=""> **Hermes Agent** | ✅ | ✅ OpenAI | — | — | — | —
| — |
-| <img src="https://www.google.com/s2/favicons?domain=deepseek.com&sz=64";
width="16" height="16" alt=""> **DeepSeek Harness** | ✅ | ✅ OpenAI | ✅ | ✅ | —
| — | npm |
+| <img src="https://www.google.com/s2/favicons?domain=deepseek.com&sz=64";
width="16" height="16" alt=""> **DeepSeek Harness** | ✅ | ✅ OpenAI | ✅ | ✅ | —
| ✅ | npm |
- **行为监控** —— 只读地记录 Agent 做了什么：提示词、工具调用、权限询问。Agent 的任何动作都不会等待
Gateway，开着监控也不会改变任何一次回答。
- **换供应商** —— Gateway 按该 Agent 客户端自己说的协议格式，改写它的配置文件，把它指向绑定的供应商。不支持的 Agent
仍可以用界面上给出的环境变量接进来。
diff --git a/agenthistory/dsh.go b/agenthistory/dsh.go
new file mode 100644
index 0000000..54f3ce6
--- /dev/null
+++ b/agenthistory/dsh.go
@@ -0,0 +1,231 @@
+// Copyright 2026 The casbin Authors. All Rights Reserved.
+//
+// Licensed 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 agenthistory
+
+import (
+ "encoding/json"
+ "io"
+ "os"
+ "path/filepath"
+ "strings"
+ "time"
+
+ "github.com/klauspost/compress/zstd"
+)
+
+const dshAgent = "dsh"
+
+// dshTranscript is the fixed name every dsh session log is written under, in a
+// directory of its own. The default encoding is a concatenation of Zstandard
+// frames; a harness configured with compression off writes the same lines raw.
+const (
+ dshTranscript = "session.jsonl"
+ dshCompressedSuffix = ".zstd"
+)
+
+// scanDsh reads the sessions dsh keeps under its own home. The profile the
+// harness ships points its persistence at $DSH_HOME/sessions, and every
session
+// is one directory under the project it ran in.
+func scanDsh(home string) []Session {
+ root := filepath.Join(home, ".dsh", "sessions")
+ sessions := []Session{}
+ for _, file := range newestTranscripts(root, isDshTranscript) {
+ if session, ok := read(file, parseDsh); ok {
+ sessions = append(sessions, session)
+ }
+ }
+ return sessions
+}
+
+func isDshTranscript(name string) bool {
+ return name == dshTranscript || name ==
dshTranscript+dshCompressedSuffix
+}
+
+// dshLine is the union of the two kinds of line a session log holds: the
header
+// that opens it, and the events appended after it. A packed chunk row carries
+// neither an id nor an event type this reads, and is passed over.
+type dshLine struct {
+ Type string `json:"type"`
+ Id string `json:"id"`
+ Cwd string `json:"cwd"`
+ CreatedAt json.RawMessage `json:"createdAt"`
+ Time int64 `json:"time"`
+ Data json.RawMessage `json:"data"`
+}
+
+func parseDsh(file transcript) (Session, bool) {
+ session := Session{
+ Agent: dshAgent,
+ Path: file.path,
+ LastTime: file.info.ModTime().Local().Format(time.RFC3339),
+ Historical: true,
+ }
+ usage := newUsageReader()
+
+ source, closer, err := openDshTranscript(file.path)
+ if err != nil {
+ return Session{}, false
+ }
+ defer closer()
+
+ // A session being written right now ends in a frame the encoder has not
+ // closed, which is what the harness's own reader expects to find. The
+ // records read before it are kept rather than thrown away with the
error.
+ _, _ = eachLineIn(source, func(data []byte) {
+ var entry dshLine
+ if err := json.Unmarshal(data, &entry); err != nil {
+ return
+ }
+
+ switch entry.Type {
+ case "session":
+ session.SessionKey = entry.Id
+ session.Cwd = entry.Cwd
+ if when := dshHeaderTime(entry.CreatedAt); when != "" {
+ session.FirstTime = when
+ }
+ return
+ case "user/message", "assistant/message":
+ session.RecordCount++
+ default:
+ return
+ }
+
+ if when := dshEventTime(entry.Time); when != "" {
+ if session.FirstTime == "" {
+ session.FirstTime = when
+ }
+ session.LastTime = when
+ }
+ if entry.Type == "user/message" {
+ if session.Title == "" {
+ session.Title =
usableTitle(dshContent(entry.Data))
+ }
+ return
+ }
+ addDshUsage(usage, entry)
+ })
+
+ if session.SessionKey == "" {
+ return Session{}, false
+ }
+ if session.FirstTime == "" {
+ session.FirstTime = session.LastTime
+ }
+ session.Usage = usage.buckets(dayOf(session.LastTime))
+ return session, true
+}
+
+// openDshTranscript reads the log as it was written. Zstandard frames are
+// decoded on the way through, so a truncated tail ends the read with what came
+// before it rather than failing the session.
+func openDshTranscript(path string) (io.Reader, func(), error) {
+ handle, err := os.Open(path)
+ if err != nil {
+ return nil, nil, err
+ }
+ if !strings.HasSuffix(path, dshCompressedSuffix) {
+ return handle, func() { handle.Close() }, nil
+ }
+
+ decoder, err := zstd.NewReader(handle)
+ if err != nil {
+ handle.Close()
+ return nil, nil, err
+ }
+ return decoder.IOReadCloser(), func() {
+ decoder.Close()
+ handle.Close()
+ }, nil
+}
+
+// dshContent is the model-facing blocks of one message, which is what the
event
+// data is for a user turn.
+func dshContent(data json.RawMessage) json.RawMessage {
+ var message struct {
+ Content json.RawMessage `json:"content"`
+ }
+ if err := json.Unmarshal(data, &message); err != nil {
+ return nil
+ }
+ return message.Content
+}
+
+// addDshUsage reads what one model call spent. The counts ride the assistant
+// message the call produced, and the model that produced it is named there
too.
+func addDshUsage(reader *usageReader, entry dshLine) {
+ var event struct {
+ Message struct {
+ Id string `json:"id"`
+ Source struct {
+ Provider string `json:"provider"`
+ Model string `json:"model"`
+ } `json:"source"`
+ } `json:"message"`
+ Usage *struct {
+ InputTokens int `json:"inputTokens"`
+ OutputTokens int `json:"outputTokens"`
+ CacheReadTokens int `json:"cacheReadTokens"`
+ CacheWriteTokens int `json:"cacheWriteTokens"`
+ ReasoningTokens int `json:"reasoningTokens"`
+ } `json:"usage"`
+ }
+ if err := json.Unmarshal(entry.Data, &event); err != nil || event.Usage
== nil {
+ return
+ }
+
+ spent := event.Usage
+ if spent.InputTokens == 0 && spent.OutputTokens == 0 &&
spent.CacheReadTokens == 0 && spent.CacheWriteTokens == 0 {
+ return
+ }
+ model := event.Message.Source.Model
+ if provider := event.Message.Source.Provider; provider != "" && model
!= "" {
+ model = provider + "/" + model
+ }
+ reader.put(event.Message.Id, turn{
+ model: model,
+ day: dayOf(dshEventTime(entry.Time)),
+ prompt: spent.InputTokens,
+ completion: spent.OutputTokens,
+ cacheRead: spent.CacheReadTokens,
+ cacheWrite: spent.CacheWriteTokens,
+ reasoning: spent.ReasoningTokens,
+ })
+}
+
+// dshEventTime reads the epoch milliseconds an event is stamped with.
+func dshEventTime(value int64) string {
+ if value <= 0 {
+ return ""
+ }
+ return time.UnixMilli(value).Local().Format(time.RFC3339)
+}
+
+// dshHeaderTime reads when the session opened, which the header spells either
+// as epoch milliseconds or as a timestamp.
+func dshHeaderTime(value json.RawMessage) string {
+ if len(value) == 0 {
+ return ""
+ }
+ var milliseconds int64
+ if err := json.Unmarshal(value, &milliseconds); err == nil {
+ return dshEventTime(milliseconds)
+ }
+ var text string
+ if err := json.Unmarshal(value, &text); err == nil {
+ return strings.TrimSpace(text)
+ }
+ return ""
+}
diff --git a/agenthistory/history.go b/agenthistory/history.go
index 1d018fb..e0c30e9 100644
--- a/agenthistory/history.go
+++ b/agenthistory/history.go
@@ -132,15 +132,20 @@ func scan(home string) []Session {
sessions := []Session{}
for _, source := range transcriptDirs {
root := filepath.Join(append([]string{home},
source.parts...)...)
- for _, file := range newestTranscripts(root) {
- if session, ok := read(source.agent, file,
source.keyOf); ok {
+ agent, keyOf := source.agent, source.keyOf
+ for _, file := range newestTranscripts(root, isJSONL) {
+ if session, ok := read(file, func(file transcript)
(Session, bool) {
+ return parse(agent, file, keyOf)
+ }); ok {
sessions = append(sessions, session)
}
}
}
// opencode keeps its sessions in a database rather than a transcript
file,
- // and already totals what each one spent.
+ // and already totals what each one spent. dsh writes a transcript, but
a
+ // compressed one its own reader has to open.
sessions = append(sessions, scanOpencode(home)...)
+ sessions = append(sessions, scanDsh(home)...)
sort.SliceStable(sessions, func(left, right int) bool {
return sessions[left].LastTime > sessions[right].LastTime
@@ -153,13 +158,18 @@ type transcript struct {
info os.FileInfo
}
-// newestTranscripts finds the .jsonl files under root, newest first and
capped.
-func newestTranscripts(root string) []transcript {
+// isJSONL is the transcript name rule of an agent that writes plain JSONL.
+func isJSONL(name string) bool {
+ return strings.HasSuffix(name, ".jsonl")
+}
+
+// newestTranscripts finds the transcripts under root, newest first and capped.
+func newestTranscripts(root string, match func(name string) bool) []transcript
{
found := []transcript{}
// A missing directory means the agent was never installed here, which
is not
// an error worth reporting from a page that lists whatever it finds.
filepath.WalkDir(root, func(path string, entry os.DirEntry, err error)
error {
- if err != nil || entry.IsDir() ||
!strings.HasSuffix(entry.Name(), ".jsonl") {
+ if err != nil || entry.IsDir() || !match(entry.Name()) {
return nil
}
info, err := entry.Info()
@@ -179,7 +189,9 @@ func newestTranscripts(root string) []transcript {
return found
}
-func read(agent string, file transcript, keyOf func(string) string) (Session,
bool) {
+// read is one transcript, from the cache when the file has not changed since
it
+// was last parsed.
+func read(file transcript, parse func(transcript) (Session, bool)) (Session,
bool) {
key := cacheKey{path: file.path, size: file.info.Size(), unix:
file.info.ModTime().UnixNano()}
cacheMutex.Lock()
cached, found := cache[key]
@@ -188,7 +200,7 @@ func read(agent string, file transcript, keyOf func(string)
string) (Session, bo
return cached, true
}
- session, ok := parse(agent, file, keyOf)
+ session, ok := parse(file)
if !ok {
return Session{}, false
}
@@ -253,7 +265,13 @@ func eachLine(path string, visit func(data []byte)) (int,
error) {
}
defer handle.Close()
- reader := bufio.NewReaderSize(handle, readBufferBytes)
+ return eachLineIn(handle, visit)
+}
+
+// eachLineIn is eachLine over an open stream, for a transcript that is decoded
+// on the way in rather than read from the file as it lies.
+func eachLineIn(source io.Reader, visit func(data []byte)) (int, error) {
+ reader := bufio.NewReaderSize(source, readBufferBytes)
buffer := []byte{}
skipping := false
skipped := 0
diff --git a/go.mod b/go.mod
index 0e995c1..a3f7392 100644
--- a/go.mod
+++ b/go.mod
@@ -12,6 +12,7 @@ require (
github.com/go-mysql-org/go-mysql v1.7.0
github.com/go-sql-driver/mysql v1.6.0
github.com/google/uuid v1.6.0
+ github.com/klauspost/compress v1.17.9
github.com/lib/pq v1.10.2
github.com/siddontang/go-log v0.0.0-20190221022429-1e957dd83bed
github.com/xorm-io/core v0.7.4
diff --git a/go.sum b/go.sum
index 8a6d250..56a41dd 100644
--- a/go.sum
+++ b/go.sum
@@ -113,6 +113,8 @@ github.com/json-iterator/go v1.1.11/go.mod
h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/
github.com/julienschmidt/httprouter v1.2.0/go.mod
h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51
h1:Z9n2FFNUXsshfwJMBgNA0RU6/i7WVaAegv3PtuIHPMs=
github.com/kballard/go-shellquote v0.0.0-20180428030007-95032a82bc51/go.mod
h1:CzGEWj7cYgsdH8dAjBGEr58BoE7ScuLd+fwFZ44+/x8=
+github.com/klauspost/compress v1.17.9
h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
+github.com/klauspost/compress v1.17.9/go.mod
h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod
h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod
h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
github.com/kr/pretty v0.1.0/go.mod
h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
